derivepassphrase.git

@@ -828,8 +828,8 @@ def select_ssh_key(

     /,

     *,

     ctx: click.Context | None = None,

-    error_callback: Callable[..., NoReturn] = default_error_callback,

-    warning_callback: Callable[..., None] = lambda *_args: None,

+    error_callback: Callable[..., NoReturn],

+    warning_callback: Callable[..., None],

 ) -> bytes | bytearray:

     """Interactively select an SSH key for passphrase derivation.

@@ -1068,8 +1068,8 @@ def key_to_phrase(

     key: str | Buffer,

     /,

     *,

-    error_callback: Callable[..., NoReturn] = default_error_callback,

-    warning_callback: Callable[..., None] = lambda *_args: None,

+    error_callback: Callable[..., NoReturn],

+    warning_callback: Callable[..., None],

     conn: ssh_agent.SSHAgentClient

     | _types.SSHAgentSocket

     | Sequence[str]

@@ -1082,6 +1082,26 @@ def key_to_phrase(

     obtained from the key, because this is the first point of contact

     with the SSH agent.

+    Args:

+        key:

+            The SSH key for which the equivalent master passphrase shall

+            be computed, encoded in base64.

+        conn:

+            An optional connection hint to the SSH agent.  See

+            [`ssh_agent.SSHAgentClient.ensure_agent_subcontext`][].

+        error_callback:

+            A callback function for an error message, if any.  The

+            callback function is responsible for aborting this function

+            call after acknowledging, formatting and/or forwarding the

+            error message; it would typically call [`sys.exit`][] or

+            raise an exception of its own, based on the provided error

+            message.

+        warning_callback:

+            A callback function for a warning message, if any.  The

+            callback function is responsible for formatting the warning

+            message and dispatching it into the warning system, if so

+            desired.

+

     """

     key = base64.standard_b64decode(key)

     with exceptiongroup.catch({  # noqa: SIM117

@@ -1379,7 +1379,9 @@ class _VaultContext:  # noqa: PLR0904

             phrase = cast("str", overrides["phrase"])

         elif settings.get("key"):

             phrase = cli_helpers.key_to_phrase(

-                cast("str", settings["key"]), error_callback=self.err

+                cast("str", settings["key"]),

+                error_callback=self.err,

+                warning_callback=self.warning,

             )

         elif settings.get("phrase"):

             phrase = cast("str", settings["phrase"])

@@ -445,6 +445,7 @@ class Parametrize(types.SimpleNamespace):

             "system_support_action",

             "sign_action",

             "pattern",

+            "warnings_patterns",

         ],

         [

             pytest.param(

@@ -453,6 +454,7 @@ class Parametrize(types.SimpleNamespace):

                 None,

                 SignAction.FAIL,

                 "not loaded into the agent",

+                [],

                 id="key-not-loaded",

             ),

             pytest.param(

@@ -461,6 +463,7 @@ class Parametrize(types.SimpleNamespace):

                 None,

                 SignAction.FAIL,

                 "SSH agent failed to or refused to",

+                [],

                 id="list-keys-refused",

             ),

             pytest.param(

@@ -469,6 +472,7 @@ class Parametrize(types.SimpleNamespace):

                 None,

                 SignAction.FAIL,

                 "SSH agent failed to or refused to",

+                [],

                 id="list-keys-protocol-error",

             ),

             pytest.param(

@@ -477,6 +481,7 @@ class Parametrize(types.SimpleNamespace):

                 None,

                 SignAction.FAIL,

                 "Cannot find any running SSH agent",

+                [],

                 id="agent-address-missing",

             ),

             pytest.param(

@@ -485,6 +490,7 @@ class Parametrize(types.SimpleNamespace):

                 None,

                 SignAction.FAIL,

                 "Cannot connect to the SSH agent",

+                [],

                 id="agent-address-mangled",

             ),

             pytest.param(

@@ -493,6 +499,7 @@ class Parametrize(types.SimpleNamespace):

                 SystemSupportAction.UNSET_NATIVE,

                 SignAction.FAIL,

                 "does not support communicating with it",

+                [],

                 id="no-agent-support",

             ),

             pytest.param(

@@ -501,6 +508,7 @@ class Parametrize(types.SimpleNamespace):

                 SystemSupportAction.UNSET_PROVIDER_LIST,

                 SignAction.FAIL,

                 "does not support communicating with it",

+                [],

                 id="no-agent-support",

             ),

             pytest.param(

@@ -509,6 +517,7 @@ class Parametrize(types.SimpleNamespace):

                 SystemSupportAction.UNSET_AF_UNIX_AND_ENSURE_USE,

                 SignAction.FAIL,

                 "does not support communicating with it",

+                ["Cannot connect to an SSH agent via UNIX domain sockets"],

                 id="no-agent-support",

             ),

             pytest.param(

@@ -517,6 +526,7 @@ class Parametrize(types.SimpleNamespace):

                 SystemSupportAction.UNSET_WINDLL_AND_ENSURE_USE,

                 SignAction.FAIL,

                 "does not support communicating with it",

+                ["Cannot connect to an SSH agent via Windows named pipes"],

                 id="no-agent-support",

             ),

             pytest.param(

@@ -525,6 +535,7 @@ class Parametrize(types.SimpleNamespace):

                 None,

                 SignAction.FAIL_RUNTIME,

                 "violates the communication protocol",

+                [],

                 id="sign-violates-protocol",

             ),

         ],

@@ -1427,17 +1438,24 @@ class TestMisc:

         address_action: SocketAddressAction | None,

         sign_action: SignAction,

         pattern: str,

+        warnings_patterns: list[str],

     ) -> None:

         """All errors in [`cli_helpers.key_to_phrase`][] are handled."""

+        captured_warnings: list[str] = []

+

         class ErrCallback(BaseException):

             def __init__(self, *args: Any, **kwargs: Any) -> None:

                 super().__init__(*args[:1])

                 self.args = args

                 self.kwargs = kwargs

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

-            raise ErrCallback(*args, **_kwargs)

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

+            raise ErrCallback(*args, **kwargs)

+

+        def warn(*args: Any) -> None:

+            if args:  # pragma: no branch

+                captured_warnings.append(str(args[0]))

         with pytest.MonkeyPatch.context() as monkeypatch:

             loaded_keys = list(

@@ -1454,7 +1472,14 @@ class TestMisc:

             if system_support_action:

                 system_support_action(monkeypatch)

             with pytest.raises(ErrCallback, match=pattern) as excinfo:

-                cli_helpers.key_to_phrase(loaded_key, error_callback=err)

+                cli_helpers.key_to_phrase(

+                    loaded_key, error_callback=err, warning_callback=warn

+                )

+

+            for pat in warnings_patterns:

+                assert any([pat in string for string in captured_warnings]), (

+                    f"expected some warning message to match {pat}"

+                )

             if list_keys_action == ListKeysAction.FAIL_RUNTIME:

                 assert excinfo.value.kwargs

                 assert isinstance(

@@ -1029,7 +1029,11 @@ class TestSuitableKeys:

         @click.command()

         def driver() -> None:

             """Call [`cli_helpers.select_ssh_key`][] directly, as a command."""

-            key = cli_helpers.select_ssh_key(client)

+            key = cli_helpers.select_ssh_key(

+                client,

+                error_callback=lambda s, *_args: pytest.fail(str(s)),

+                warning_callback=lambda s, *_args: pytest.fail(str(s)),

+            )

             click.echo(base64.standard_b64encode(key).decode("ASCII"))

         runner = machinery.CliRunner(mix_stderr=True)