Add ability to gracefully handle an encrypted private key for SSH (can be forced or automated with an env var too)

CHANGELOG.md

@@ -1,6 +1,7 @@
 # 0.4.4 (unreleased)
 
  * Update cryptography dependency
+ * Add capability to handle passphrases on encrypted SSH private keys. Prompting can be forced with `--ask-key-passphrase` or automated (e.g for CI) with `--ssh-key-passphrase env SOMEVAR`
 
 # 0.4.3
 

README.md

@@ -89,8 +89,27 @@ Harvest state about a host and write a harvest bundle.
     - glob (default): supports `*` and `**` (prefix with `glob:` to force)
     - regex: prefix with `re:` or `regex:`
   - Precedence: excludes win over includes.
-* Using remote mode and sudo requires password?
-  - `--ask-become-pass` (or `-K`) will prompt for the password. If you forget, and remote requires password for sudo, it'll still fall back to prompting for a password, but will be a bit slower to do so.
+  * Using remote mode and auth requires secrets?
+    * sudo password:
+      * `--ask-become-pass` (or `-K`) prompts for the sudo password.
+      * If you forget, and remote sudo requires a password, Enroll will still fall back to prompting in interactive mode (slightly slower due to retry).
+    * SSH private-key passphrase:
+      * `--ask-key-passphrase` prompts for the SSH key passphrase.
+      * `--ssh-key-passphrase-env ENV_VAR` reads the SSH key passphrase from an environment variable (useful for CI/non-interactive runs).
+      * If neither is provided, and Enroll detects an encrypted key in an interactive session, it will still fall back to prompting on-demand.
+      * In non-interactive sessions, pass `--ask-key-passphrase` or `--ssh-key-passphrase-env ENV_VAR` when using encrypted private keys.
+    * Note: `--ask-key-passphrase` and `--ssh-key-passphrase-env` are mutually exclusive.
+
+Examples (encrypted SSH key)
+
+```bash
+# Interactive
+enroll harvest --remote-host myhost.example.com --remote-user myuser --ask-key-passphrase --out /tmp/enroll-harvest
+
+# Non-interactive / CI
+export ENROLL_SSH_KEY_PASSPHRASE='correct horse battery staple'
+enroll single-shot --remote-host myhost.example.com --remote-user myuser --ssh-key-passphrase-env ENROLL_SSH_KEY_PASSPHRASE --harvest /tmp/enroll-harvest --out /tmp/enroll-ansible --fqdn myhost.example.com
+```
 
 ---
 

enroll/cli.py

@@ -22,7 +22,11 @@ from .diff import (
 from .explain import explain_state
 from .harvest import harvest
 from .manifest import manifest
-from .remote import remote_harvest, RemoteSudoPasswordRequired
+from .remote import (
+    remote_harvest,
+    RemoteSudoPasswordRequired,
+    RemoteSSHKeyPassphraseRequired,
+)
 from .sopsutil import SopsError, encrypt_file_binary
 from .validate import validate_harvest
 from .version import get_enroll_version
@@ -390,6 +394,24 @@ def _add_remote_args(p: argparse.ArgumentParser) -> None:
         ),
     )
 
+    keyp = p.add_mutually_exclusive_group()
+    keyp.add_argument(
+        "--ask-key-passphrase",
+        action="store_true",
+        help=(
+            "Prompt for the SSH private key passphrase when using --remote-host. "
+            "If not set, enroll will still prompt on-demand if it detects an encrypted key in an interactive session."
+        ),
+    )
+    keyp.add_argument(
+        "--ssh-key-passphrase-env",
+        metavar="ENV_VAR",
+        help=(
+            "Read the SSH private key passphrase from environment variable ENV_VAR "
+            "(useful for non-interactive runs/CI)."
+        ),
+    )
+
 
 def main() -> None:
     ap = argparse.ArgumentParser(prog="enroll")
@@ -771,6 +793,10 @@ def main() -> None:
                             pass
                         remote_harvest(
                             ask_become_pass=args.ask_become_pass,
+                            ask_key_passphrase=bool(args.ask_key_passphrase),
+                            ssh_key_passphrase_env=getattr(
+                                args, "ssh_key_passphrase_env", None
+                            ),
                             local_out_dir=tmp_bundle,
                             remote_host=args.remote_host,
                             remote_port=args.remote_port,
@@ -793,6 +819,10 @@ def main() -> None:
                     )
                     state = remote_harvest(
                         ask_become_pass=args.ask_become_pass,
+                        ask_key_passphrase=bool(args.ask_key_passphrase),
+                        ssh_key_passphrase_env=getattr(
+                            args, "ssh_key_passphrase_env", None
+                        ),
                         local_out_dir=out_dir,
                         remote_host=args.remote_host,
                         remote_port=args.remote_port,
@@ -996,6 +1026,10 @@ def main() -> None:
                             pass
                         remote_harvest(
                             ask_become_pass=args.ask_become_pass,
+                            ask_key_passphrase=bool(args.ask_key_passphrase),
+                            ssh_key_passphrase_env=getattr(
+                                args, "ssh_key_passphrase_env", None
+                            ),
                             local_out_dir=tmp_bundle,
                             remote_host=args.remote_host,
                             remote_port=args.remote_port,
@@ -1027,6 +1061,10 @@ def main() -> None:
                     )
                     remote_harvest(
                         ask_become_pass=args.ask_become_pass,
+                        ask_key_passphrase=bool(args.ask_key_passphrase),
+                        ssh_key_passphrase_env=getattr(
+                            args, "ssh_key_passphrase_env", None
+                        ),
                         local_out_dir=harvest_dir,
                         remote_host=args.remote_host,
                         remote_port=args.remote_port,
@@ -1096,6 +1134,12 @@ def main() -> None:
         raise SystemExit(
             "error: remote sudo requires a password. Re-run with --ask-become-pass."
         ) from None
+    except RemoteSSHKeyPassphraseRequired as e:
+        msg = str(e).strip() or (
+            "SSH private key passphrase is required. "
+            "Re-run with --ask-key-passphrase or --ssh-key-passphrase-env VAR."
+        )
+        raise SystemExit(f"error: {msg}") from None
     except RuntimeError as e:
         raise SystemExit(f"error: {e}") from None
     except SopsError as e:

enroll/remote.py

@@ -18,6 +18,10 @@ class RemoteSudoPasswordRequired(RuntimeError):
     """Raised when sudo requires a password but none was provided."""
 
 
+class RemoteSSHKeyPassphraseRequired(RuntimeError):
+    """Raised when SSH private key decryption needs a passphrase."""
+
+
 def _sudo_password_required(out: str, err: str) -> bool:
     """Return True if sudo output indicates it needs a password/TTY."""
     blob = (out + "\n" + err).lower()
@@ -68,11 +72,42 @@ def _resolve_become_password(
     return None
 
 
+def _resolve_ssh_key_passphrase(
+    ask_key_passphrase: bool,
+    *,
+    env_var: Optional[str] = None,
+    prompt: str = "SSH key passphrase: ",
+    getpass_fn: Callable[[str], str] = getpass.getpass,
+) -> Optional[str]:
+    """Resolve SSH private-key passphrase from env and/or prompt.
+
+    Precedence:
+      1) --ssh-key-passphrase-env style input (env_var)
+      2) --ask-key-passphrase style interactive prompt
+      3) None
+    """
+    if env_var:
+        val = os.environ.get(str(env_var))
+        if val is None:
+            raise RuntimeError(
+                "SSH key passphrase environment variable is not set: " f"{env_var}"
+            )
+        return val
+
+    if ask_key_passphrase:
+        return getpass_fn(prompt)
+
+    return None
+
+
 def remote_harvest(
     *,
     ask_become_pass: bool = False,
+    ask_key_passphrase: bool = False,
+    ssh_key_passphrase_env: Optional[str] = None,
     no_sudo: bool = False,
     prompt: str = "sudo password: ",
+    key_prompt: str = "SSH key passphrase: ",
     getpass_fn: Optional[Callable[[str], str]] = None,
     stdin: Optional[TextIO] = None,
     **kwargs,
@@ -97,17 +132,48 @@ def remote_harvest(
         prompt=prompt,
         getpass_fn=getpass_fn,
     )
+    ssh_key_passphrase = _resolve_ssh_key_passphrase(
+        ask_key_passphrase,
+        env_var=ssh_key_passphrase_env,
+        prompt=key_prompt,
+        getpass_fn=getpass_fn,
+    )
 
+    while True:
         try:
-        return _remote_harvest(sudo_password=sudo_password, no_sudo=no_sudo, **kwargs)
+            return _remote_harvest(
+                sudo_password=sudo_password,
+                no_sudo=no_sudo,
+                ssh_key_passphrase=ssh_key_passphrase,
+                **kwargs,
+            )
+        except RemoteSSHKeyPassphraseRequired:
+            # Already tried a passphrase and still failed.
+            if ssh_key_passphrase is not None:
+                raise RemoteSSHKeyPassphraseRequired(
+                    "SSH private key could not be decrypted with the supplied "
+                    "passphrase."
+                ) from None
+
+            # Fallback prompt if interactive.
+            if stdin is not None and getattr(stdin, "isatty", lambda: False)():
+                ssh_key_passphrase = getpass_fn(key_prompt)
+                continue
+
+            raise RemoteSSHKeyPassphraseRequired(
+                "SSH private key is encrypted and needs a passphrase. "
+                "Re-run with --ask-key-passphrase or "
+                "--ssh-key-passphrase-env VAR."
+            )
+
         except RemoteSudoPasswordRequired:
             if sudo_password is not None:
                 raise
 
-        # Fallback prompt if interactive
+            # Fallback prompt if interactive.
             if stdin is not None and getattr(stdin, "isatty", lambda: False)():
-            pw = getpass_fn(prompt)
-            return _remote_harvest(sudo_password=pw, no_sudo=no_sudo, **kwargs)
+                sudo_password = getpass_fn(prompt)
+                continue
 
             raise RemoteSudoPasswordRequired(
                 "Remote sudo requires a password. Re-run with --ask-become-pass."
@@ -337,6 +403,7 @@ def _remote_harvest(
     dangerous: bool = False,
     no_sudo: bool = False,
     sudo_password: Optional[str] = None,
+    ssh_key_passphrase: Optional[str] = None,
     include_paths: Optional[list[str]] = None,
     exclude_paths: Optional[list[str]] = None,
 ) -> Path:
@@ -467,6 +534,7 @@ def _remote_harvest(
 
         # If we created a socket (sock!=None), pass hostkey_name as hostname so
         # known_hosts lookup uses HostKeyAlias (or whatever hostkey_name resolved to).
+        try:
             ssh.connect(
                 hostname=hostkey_name if sock is not None else connect_host,
                 port=connect_port,
@@ -478,7 +546,12 @@ def _remote_harvest(
                 timeout=connect_timeout,
                 banner_timeout=connect_timeout,
                 auth_timeout=connect_timeout,
+                passphrase=ssh_key_passphrase,
             )
+        except paramiko.PasswordRequiredException as e:  # type: ignore[attr-defined]
+            raise RemoteSSHKeyPassphraseRequired(
+                "SSH private key is encrypted and no passphrase was provided."
+            ) from e
 
         # If no username was explicitly provided, SSH may have selected a default.
         # We need a concrete username for the (sudo) chown step below.