add iinit-like behaviour when asking for password

cubi_tk/irods_utils.py

@@ -1,10 +1,13 @@
 import getpass
 import os.path
+from pathlib import Path
 import sys
 import tempfile
 from typing import Tuple
 
 import attr
+from irods.exception import CAT_INVALID_AUTHENTICATION, PAM_AUTH_PASSWORD_FAILED
+from irods.password_obfuscation import encode
 from irods.session import iRODSSession
 from logzero import logger
 from tqdm import tqdm
@@ -36,35 +39,74 @@ def get_irods_error(e: Exception):
     return es if es and es != "None" else e.__class__.__name__
 
 
-def init_irods(irods_env_path: os.PathLike) -> iRODSSession:
+def init_irods(irods_env_path: os.PathLike, ask: bool = False) -> iRODSSession:
     """Connect to iRODS."""
     irods_auth_path = irods_env_path.parent.joinpath(".irodsA")
     if irods_auth_path.exists():
         try:
             session = iRODSSession(irods_env_file=irods_env_path)
             session.server_version  # check for outdated .irodsA file
+            return session
         except Exception as e:  # pragma: no cover
             logger.error(f"iRODS connection failed: {get_irods_error(e)}")
-            logger.error("Are you logged in? try 'iinit'")
-            sys.exit(1)
+            pass
         finally:
             session.cleanup()
-    else:
-        # Query user for password.
-        logger.info("iRODS authentication file not found.")
-        password = getpass.getpass(prompt="Please enter SODAR password:")
+
+    # No valid .irodsA file. Query user for password.
+    logger.info("No valid iRODS authentication file found.")
+    attempts = 0
+    while attempts < 3:
         try:
-            session = iRODSSession(irods_env_file=irods_env_path, password=password)
+            session = iRODSSession(
+                irods_env_file=irods_env_path,
+                password=getpass.getpass(prompt="Please enter SODAR password:"),
+            )
             session.server_version  # check for exceptions
+            break
+        except PAM_AUTH_PASSWORD_FAILED:  # pragma: no cover
+            if attempts < 2:
+                logger.warning("Wrong password. Please try again.")
+                attempts += 1
+                continue
+            else:
+                logger.error("iRODS connection failed.")
+                sys.exit(1)
         except Exception as e:  # pragma: no cover
             logger.error(f"iRODS connection failed: {get_irods_error(e)}")
             sys.exit(1)
         finally:
             session.cleanup()
 
+    if ask and input("Save iRODS session for passwordless operation? [y/N] ").lower().startswith(
+        "y"
+    ):
+        save_irods_token(session)
+    elif not ask:
+        save_irods_token(session)
+
     return session
 
 
+def save_irods_token(session: iRODSSession, irods_env_path=None):
+    """Retrieve PAM temp auth token 'obfuscate' it and save to disk."""
+    if not irods_env_path:
+        irods_auth_path = Path.home().joinpath(".irods", ".irodsA")
+    else:
+        irods_auth_path = Path(irods_env_path).parent.joinpath(".irodsA")
+
+    irods_auth_path.parent.mkdir(parents=True, exist_ok=True)
+
+    try:
+        token = session.pam_pw_negotiated
+    except CAT_INVALID_AUTHENTICATION:
+        raise
+
+    if isinstance(token, list) and token:
+        irods_auth_path.write_text(encode(token[0]))
+        irods_auth_path.chmod(0o600)
+
+
 class iRODSTransfer:
     """Transfers files to and from iRODS."""
 

cubi_tk/sodar/ingest.py

@@ -75,11 +75,17 @@ def setup_argparse(cls, parser: argparse.ArgumentParser) -> None:
             "--yes",
             default=False,
             action="store_true",
-            help="Don't ask for permission prior to transfer.",
+            help="Don't ask for permission.",
         )
         parser.add_argument(
             "--collection", type=str, help="Target iRODS collection. Skips manual selection input."
         )
+        parser.add_argument(
+            "--iinit",
+            default=False,
+            action="store_true",
+            help="Save PAM auth token to disk. Keep login active.",
+        )
         parser.add_argument(
             "sources", help="One or multiple files/directories to ingest.", nargs="+"
         )
@@ -129,7 +135,7 @@ def execute(self):
         source_paths = self.build_file_list()
 
         # Initiate iRODS session
-        irods_session = init_irods(self.irods_env_path)
+        irods_session = init_irods(self.irods_env_path, ask=not self.args.yes)
 
         # Query target collection
         logger.info("Querying landing zone collections…")

tests/test_irods_utils.py

@@ -1,19 +1,54 @@
 from pathlib import Path
 import shutil
-from unittest.mock import patch
+from unittest.mock import MagicMock, PropertyMock, patch
 
 import irods.exception
 from irods.session import iRODSSession
 import pytest
 
-from cubi_tk.irods_utils import TransferJob, get_irods_error, init_irods, iRODSTransfer
+from cubi_tk.irods_utils import (
+    TransferJob,
+    get_irods_error,
+    init_irods,
+    iRODSTransfer,
+    save_irods_token,
+)
 
 
 @pytest.fixture
 def fake_filesystem(fs):
     yield fs
 
 
+@pytest.fixture
+def jobs():
+    return (
+        TransferJob(
+            path_src="myfile.csv",
+            path_dest="dest_dir/myfile.csv",
+            bytes=123,
+            md5="ed3b3cbb18fd148bc925944ff0861ce6",
+        ),
+        TransferJob(
+            path_src="folder/file.csv",
+            path_dest="dest_dir/folder/file.csv",
+            bytes=1024,
+            md5="a6e9e3c859b803adb0f1d5f08a51d0f6",
+        ),
+    )
+
+
+@pytest.fixture
+def itransfer(jobs):
+    session = iRODSSession(
+        irods_host="localhost",
+        irods_port=1247,
+        irods_user_name="pytest",
+        irods_zone_name="pytest",
+    )
+    return iRODSTransfer(session, jobs)
+
+
 def test_get_irods_error():
     e = irods.exception.NetworkException()
     assert get_irods_error(e) == "NetworkException"
@@ -41,34 +76,17 @@ def test_init_irods(mockpass, mocksession, fs):
     mocksession.assert_called_with(irods_env_file=ienv)
 
 
-@pytest.fixture
-def jobs():
-    return (
-        TransferJob(
-            path_src="myfile.csv",
-            path_dest="dest_dir/myfile.csv",
-            bytes=123,
-            md5="ed3b3cbb18fd148bc925944ff0861ce6",
-        ),
-        TransferJob(
-            path_src="folder/file.csv",
-            path_dest="dest_dir/folder/file.csv",
-            bytes=1024,
-            md5="a6e9e3c859b803adb0f1d5f08a51d0f6",
-        ),
-    )
-
+@patch("cubi_tk.irods_utils.encode", return_value="it works")
+def test_write_token(mockencode, fs):
+    ienv = Path(".irods/irods_environment.json")
 
-@pytest.fixture
-def itransfer(jobs):
-    session = iRODSSession(
-        irods_host="localhost",
-        irods_port=1247,
-        irods_user_name="pytest",
-        irods_zone_name="pytest",
-    )
+    mocksession = MagicMock()
+    pam_pw = PropertyMock(return_value=["secure"])
+    type(mocksession).pam_pw_negotiated = pam_pw
 
-    return iRODSTransfer(session, jobs)
+    save_irods_token(mocksession, ienv)
+    assert ienv.parent.joinpath(".irodsA").exists()
+    mockencode.assert_called_with("secure")
 
 
 def test_irods_transfer_init(jobs, itransfer):

tests/test_sodar_ingest.py

@@ -62,4 +62,3 @@ def test_sodar_ingest_build_jobs(mockjob, mockstats, mockmd5, mocksorted, ingest
             bytes=1024,
             md5="5555",
         )
-