derivepassphrase.git

@@ -183,6 +183,10 @@ class SSH_AGENT(enum.Enum):  # noqa: N801

     """SSH agent protocol numbers: server replies.

     Attributes:

+        FAILURE:

+            Generic failure code.

+        SUCCESS:

+            Generic success code.

         IDENTITIES_ANSWER:

             Successful answer to `SSH_AGENTC.REQUEST_IDENTITIES`.

         SIGN_RESPONSE:

@@ -190,5 +194,7 @@ class SSH_AGENT(enum.Enum):  # noqa: N801

     """

+    FAILURE: int = 5

+    SUCCESS: int = 6

     IDENTITIES_ANSWER: int = 12

     SIGN_RESPONSE: int = 14

@@ -168,6 +168,8 @@ def _get_suitable_ssh_keys(

             SSH agent.

         RuntimeError:

             There was an error communicating with the SSH agent.

+        SSHAgentFailedError:

+            The agent failed to supply a list of loaded keys.

     """

     client: ssh_agent.SSHAgentClient

@@ -310,6 +312,8 @@ def _select_ssh_key(

             SSH agent.

         RuntimeError:

             There was an error communicating with the SSH agent.

+        SSHAgentFailedError:

+            The agent failed to supply a list of loaded keys.

     """

     suitable_keys = list(_get_suitable_ssh_keys(conn))

     key_listing: list[str] = []

@@ -1036,7 +1040,11 @@ def derivepassphrase(

                     f'Cannot connect to SSH agent: {e.strerror}: '

                     f'{e.filename!r}'

                 )

-            except (LookupError, RuntimeError) as e:

+            except (

+                LookupError,

+                RuntimeError,

+                ssh_agent.SSHAgentFailedError,

+            ) as e:

                 err(str(e))

         elif use_phrase:

             maybe_phrase = _prompt_for_passphrase()

@@ -37,6 +37,24 @@ class TrailingDataError(RuntimeError):

         super().__init__('Overlong response from SSH agent')

+class SSHAgentFailedError(RuntimeError):

+    """The SSH agent failed to complete the requested operation."""

+

+    def __str__(self) -> str:

+        match self.args:

+            case (_types.SSH_AGENT.FAILURE.value, b''):  # pragma: no branch

+                return 'The SSH agent failed to complete the request'

+            case (_, _msg) if _msg:  # pragma: no cover

+                code = self.args[0]

+                msg = self.args[1].decode('utf-8', 'surrogateescape')

+                return f'[Code {code:d}] {msg:s}'

+            case _:  # pragma: no cover

+                return repr(self)

+

+    def __repr__(self) -> str:  # pragma: no cover

+        return f'{self.__class__.__name__}{self.args!r}'

+

+

 class SSHAgentClient:

     """A bare-bones SSH agent client supporting signing and key listing.

@@ -284,7 +302,7 @@ class SSHAgentClient:

                 The response from the SSH agent is truncated or missing.

             TrailingDataError:

                 The response from the SSH agent is too long.

-            RuntimeError:

+            SSHAgentFailedError:

                 The agent failed to complete the request.

         """

@@ -292,11 +310,7 @@ class SSHAgentClient:

             _types.SSH_AGENTC.REQUEST_IDENTITIES.value, b''

         )

         if response_code != _types.SSH_AGENT.IDENTITIES_ANSWER.value:

-            msg = (

-                f'error return from SSH agent: '

-                f'{response_code = }, {response = }'

-            )

-            raise RuntimeError(msg)

+            raise SSHAgentFailedError(response_code, response)

         response_stream = collections.deque(response)

         def shift(num: int) -> bytes:

@@ -362,7 +376,7 @@ class SSHAgentClient:

                 The response from the SSH agent is truncated or missing.

             TrailingDataError:

                 The response from the SSH agent is too long.

-            RuntimeError:

+            SSHAgentFailedError:

                 The agent failed to complete the request.

             KeyError:

                 `check_if_key_loaded` is true, and the `key` was not

@@ -381,6 +395,5 @@ class SSHAgentClient:

             _types.SSH_AGENTC.SIGN_REQUEST.value, request_data

         )

         if response_code != _types.SSH_AGENT.SIGN_RESPONSE.value:

-            msg = f'signing data failed: {response_code = }, {response = }'

-            raise RuntimeError(msg)

+            raise SSHAgentFailedError(response_code, response)

         return self.unstring(response)

@@ -360,10 +360,20 @@ class TestAgentInteraction:

     @pytest.mark.parametrize(

         ['response_code', 'response', 'exc_type', 'exc_pattern'],

         [

-            (255, b'', RuntimeError, 'error return from SSH agent:'),

-            (12, b'\x00\x00\x00\x01', EOFError, 'truncated response'),

             (

-                12,

+                _types.SSH_AGENT.FAILURE,

+                b'',

+                ssh_agent.SSHAgentFailedError,

+                'failed to complete the request',

+            ),

+            (

+                _types.SSH_AGENT.IDENTITIES_ANSWER,

+                b'\x00\x00\x00\x01',

+                EOFError,

+                'truncated response',

+            ),

+            (

+                _types.SSH_AGENT.IDENTITIES_ANSWER,

                 b'\x00\x00\x00\x00abc',

                 ssh_agent.TrailingDataError,

                 'Overlong response',

@@ -373,7 +383,7 @@ class TestAgentInteraction:

     def test_320_list_keys_error_responses(

         self,

         monkeypatch: Any,

-        response_code: int,

+        response_code: _types.SSH_AGENT,

         response: bytes | bytearray,

         exc_type: type[Exception],

         exc_pattern: str,

@@ -382,7 +392,7 @@ class TestAgentInteraction:

         monkeypatch.setattr(

             client,

             'request',

-            lambda *a, **kw: (response_code, response),  # noqa: ARG005

+            lambda *a, **kw: (response_code.value, response),  # noqa: ARG005

         )

         with pytest.raises(exc_type, match=exc_pattern):

             client.list_keys()

@@ -394,16 +404,16 @@ class TestAgentInteraction:

             (

                 b'invalid-key',

                 True,

-                (255, b''),

+                (_types.SSH_AGENT.FAILURE, b''),

                 KeyError,

                 'target SSH key not loaded into agent',

             ),

             (

                 tests.SUPPORTED_KEYS['ed25519']['public_key_data'],

                 True,

-                (255, b''),

-                RuntimeError,

-                'signing data failed:',

+                (_types.SSH_AGENT.FAILURE, b''),

+                ssh_agent.SSHAgentFailedError,

+                'failed to complete the request',

             ),

         ],

     )

@@ -412,12 +422,16 @@ class TestAgentInteraction:

         monkeypatch: Any,

         key: bytes | bytearray,

         check: bool,

-        response: tuple[int, bytes | bytearray],

+        response: tuple[_types.SSH_AGENT, bytes | bytearray],

         exc_type: type[Exception],

         exc_pattern: str,

     ) -> None:

         client = ssh_agent.SSHAgentClient()

-        monkeypatch.setattr(client, 'request', lambda a, b: response)  # noqa: ARG005

+        monkeypatch.setattr(

+            client,

+            'request',

+            lambda a, b: (response[0].value, response[1]),  # noqa: ARG005

+        )

         KeyCommentPair = _types.KeyCommentPair  # noqa: N806

         loaded_keys = [

             KeyCommentPair(v['public_key_data'], b'no comment')