derivepassphrase.git

@@ -1322,6 +1322,44 @@ def _check_for_misleading_passphrase(

             )

+def _key_to_phrase(

+    key_: str | bytes | bytearray,

+    /,

+    *,

+    error_callback: Callable[..., NoReturn] = sys.exit,

+) -> bytes | bytearray:

+    key = base64.standard_b64decode(key_)

+    try:

+        with ssh_agent.SSHAgentClient.ensure_agent_subcontext() as client:

+            try:

+                return vault.Vault.phrase_from_key(key, conn=client)

+            except ssh_agent.SSHAgentFailedError as e:

+                try:

+                    keylist = client.list_keys()

+                except ssh_agent.SSHAgentFailedError:

+                    pass

+                except Exception as e2:  # noqa: BLE001

+                    e.__context__ = e2

+                else:

+                    if not any(  # pragma: no branch

+                        k == key for k, _ in keylist

+                    ):

+                        error_callback(

+                            'The requested SSH key is not loaded '

+                            'into the agent.'

+                        )

+                error_callback(e)

+    except KeyError:

+        error_callback('Cannot find running SSH agent; check SSH_AUTH_SOCK')

+    except NotImplementedError:

+        error_callback(

+            'Cannot connect to SSH agent because '

+            'this Python version does not support UNIX domain sockets'

+        )

+    except OSError as e:

+        error_callback('Cannot connect to SSH agent: %s', e.strerror)

+

+

 # Concrete option groups used by this command-line interface.

 class PasswordGenerationOption(OptionGroupOption):

     """Password generation options for the CLI."""

@@ -1779,36 +1817,6 @@ def derivepassphrase_vault(  # noqa: C901,PLR0912,PLR0913,PLR0914,PLR0915

         logger.error(msg, *args, stacklevel=stacklevel, **kwargs)

         ctx.exit(1)

-    def key_to_phrase(key_: str | bytes | bytearray, /) -> bytes | bytearray:

-        key = base64.standard_b64decode(key_)

-        try:

-            with ssh_agent.SSHAgentClient.ensure_agent_subcontext() as client:

-                try:

-                    return vault.Vault.phrase_from_key(key, conn=client)

-                except ssh_agent.SSHAgentFailedError as e:

-                    try:

-                        keylist = client.list_keys()

-                    except ssh_agent.SSHAgentFailedError:

-                        pass

-                    except Exception as e2:  # noqa: BLE001

-                        e.__context__ = e2

-                    else:

-                        if not any(k == key for k, _ in keylist):

-                            err(

-                                'The requested SSH key is not loaded '

-                                'into the agent.'

-                            )

-                    err(e)

-        except KeyError:

-            err('Cannot find running SSH agent; check SSH_AUTH_SOCK')

-        except NotImplementedError:

-            err(

-                'Cannot connect to SSH agent because '

-                'this Python version does not support UNIX domain sockets'

-            )

-        except OSError as e:

-            err('Cannot connect to SSH agent: %s', e.strerror)

-

     def get_config() -> _types.VaultConfig:

         try:

             return _load_config()

@@ -2184,9 +2192,13 @@ def derivepassphrase_vault(  # noqa: C901,PLR0912,PLR0913,PLR0914,PLR0915

             # cases, set the phrase via vault.Vault.phrase_from_key if

             # a key is given.  Finally, if nothing is set, error out.

             if use_key or use_phrase:

-                kwargs['phrase'] = key_to_phrase(key) if use_key else phrase

+                kwargs['phrase'] = _key_to_phrase(

+                    key, error_callback=err

+                ) if use_key else phrase

             elif kwargs.get('key'):

-                kwargs['phrase'] = key_to_phrase(kwargs['key'])

+                kwargs['phrase'] = _key_to_phrase(

+                    kwargs['key'], error_callback=err

+                )

             elif kwargs.get('phrase'):

                 pass

             else:

@@ -4,6 +4,7 @@

 from __future__ import annotations

+import base64

 import contextlib

 import copy

 import errno

@@ -14,7 +15,7 @@ import shutil

 import socket

 import textwrap

 import warnings

-from typing import TYPE_CHECKING

+from typing import TYPE_CHECKING, NoReturn

 import click.testing

 import hypothesis

@@ -2116,6 +2117,77 @@ Boo.

                     exception is None

                 ), 'exception querying suitable SSH keys'

+    def test_400_key_to_phrase(

+        self,

+        monkeypatch: pytest.MonkeyPatch,

+        skip_if_no_af_unix_support: None,

+        ssh_agent_client_with_test_keys_loaded: ssh_agent.SSHAgentClient,

+    ) -> None:

+

+        class CustomError(RuntimeError):

+            pass

+

+        def err(*args: Any, **_kwargs: Any) -> NoReturn:

+            args = args or ('custom error message',)

+            raise CustomError(*args)

+

+        def fail(*_args: Any, **_kwargs: Any) -> Any:

+            raise ssh_agent.SSHAgentFailedError(

+                _types.SSH_AGENT.FAILURE.value,

+                b'',

+            )

+

+        del skip_if_no_af_unix_support

+        monkeypatch.setattr(ssh_agent.SSHAgentClient, 'sign', fail)

+        loaded_keys = list(ssh_agent_client_with_test_keys_loaded.list_keys())

+        loaded_key = base64.standard_b64encode(loaded_keys[0][0])

+        with monkeypatch.context() as mp:

+            mp.setattr(

+                ssh_agent.SSHAgentClient,

+                'list_keys',

+                lambda *_a, **_kw: [],

+            )

+            with pytest.raises(CustomError, match='not loaded into the agent'):

+                cli._key_to_phrase(loaded_key, error_callback=err)

+        with monkeypatch.context() as mp:

+            mp.setattr(ssh_agent.SSHAgentClient, 'list_keys', fail)

+            with pytest.raises(

+                CustomError, match='SSH agent failed to complete'

+            ):

+                cli._key_to_phrase(loaded_key, error_callback=err)

+        with monkeypatch.context() as mp:

+            mp.setattr(ssh_agent.SSHAgentClient, 'list_keys', err)

+            with pytest.raises(

+                CustomError, match='SSH agent failed to complete'

+            ) as excinfo:

+                cli._key_to_phrase(loaded_key, error_callback=err)

+            assert excinfo.value.args

+            assert isinstance(

+                excinfo.value.args[0], ssh_agent.SSHAgentFailedError

+            )

+            assert excinfo.value.args[0].__context__ is not None

+            assert isinstance(excinfo.value.args[0].__context__, CustomError)

+        with monkeypatch.context() as mp:

+            mp.delenv('SSH_AUTH_SOCK', raising=True)

+            with pytest.raises(

+                CustomError, match='Cannot find running SSH agent'

+            ):

+                cli._key_to_phrase(loaded_key, error_callback=err)

+        with monkeypatch.context() as mp:

+            mp.setenv(

+                'SSH_AUTH_SOCK', os.environ['SSH_AUTH_SOCK'] + '~'

+            )

+            with pytest.raises(

+                CustomError, match='Cannot connect to SSH agent'

+            ):

+                cli._key_to_phrase(loaded_key, error_callback=err)

+        with monkeypatch.context() as mp:

+            mp.delattr(socket, 'AF_UNIX', raising=True)

+            with pytest.raises(

+                CustomError, match='does not support UNIX domain sockets'

+            ):

+                cli._key_to_phrase(loaded_key, error_callback=err)

+

 class TestCLITransition:

     def test_100_help_output(self, monkeypatch: pytest.MonkeyPatch) -> None: