derivepassphrase.git

@@ -511,7 +511,6 @@ def _get_suitable_ssh_keys(

     """

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

-        has_deterministic_signatures = client.has_deterministic_signatures()

         try:

             all_key_comment_pairs = list(client.list_keys())

         except EOFError as e:  # pragma: no cover

@@ -519,10 +518,7 @@ def _get_suitable_ssh_keys(

         suitable_keys = copy.copy(all_key_comment_pairs)

         for pair in all_key_comment_pairs:

             key, _comment = pair

-        if (

-            has_deterministic_signatures

-            or vault.Vault._is_suitable_ssh_key(key)  # noqa: SLF001

-        ):

+            if vault.Vault.is_suitable_ssh_key(key, client=client):

                 yield pair

     if not suitable_keys:  # pragma: no cover

         raise LookupError(_NO_USABLE_KEYS)

@@ -351,12 +351,19 @@ class SSHAgentClient:

             in self.query_extensions()

         )

-    def has_deterministic_signatures(self) -> bool:

-        """Check whether the agent returns deterministic signatures.

+    def has_deterministic_dsa_signatures(self) -> bool:

+        """Check whether the agent returns deterministic DSA signatures.

+

+        This includes ECDSA signatures.

+

+        Generally, this means that the SSH agent implements [RFC 6979][]

+        or a similar system.

+

+        [RFC 6979]: https://www.rfc-editor.org/rfc/rfc6979

         Returns:

             True if a known agent was detected where signatures are

-            deterministic for all SSH key types, false otherwise.

+            deterministic for all DSA key types, false otherwise.

         Note: Known agents with deterministic signatures

             | agent           | detected via                                                  |

@@ -594,24 +601,27 @@ class SSHAgentClient:

         )

     def query_extensions(self) -> AbstractSet[bytes]:

-        """Request a list of extensions supported by the SSH agent.

-

-        Args:

-            raise_if_no_extension_support:

-                If true, and if the agent does not support querying

-                extensions, then raise an error.  If false, silently

-                return an empty result.

+        """Request a listing of extensions supported by the SSH agent.

         Returns:

-            A read-only sequence of extension names.

+            A read-only set of extension names the SSH agent says it

+            supports.

         Raises:

             EOFError:

                 The response from the SSH agent is truncated or missing.

             OSError:

                 There was a communication error with the SSH agent.

-            SSHAgentFailedError:

-                The agent failed to complete the request.

+            RuntimeError:

+                The response from the SSH agent is malformed.

+

+        Note:

+            The set of supported extensions is queried via the `query`

+            extension request.  If the agent does not support the query

+            extension request, or extension requests in general, then an

+            empty set is returned.  This does not however imply that the

+            agent doesn't support *any* extension request... merely that

+            it doesn't support extension autodiscovery.

         """

         try:

@@ -628,9 +638,19 @@ class SSHAgentClient:

             # This isn't necessarily true, e.g. for OpenSSH's ssh-agent.

             return frozenset()

         extensions: set[bytes] = set()

+        msg = 'Malformed response from SSH agent'

+        msg2 = 'Extension response message does not match request'

+        try:

             _query, response_data = self.unstring_prefix(response_data)

-        assert bytes(_query) == b'query'

+        except ValueError as e:

+            raise RuntimeError(msg) from e

+        if bytes(_query) != b'query':

+            raise RuntimeError(msg2)

         while response_data:

+            try:

                 extension, response_data = self.unstring_prefix(response_data)

+            except ValueError as e:

+                raise RuntimeError(msg) from e

+            else:

                 extensions.add(bytes(extension))

         return frozenset(extensions)

@@ -447,23 +447,30 @@ class Vault:

                 return bytes(result)

     @staticmethod

-    def _is_suitable_ssh_key(key: bytes | bytearray, /) -> bool:

+    def is_suitable_ssh_key(

+        key: bytes | bytearray,

+        /,

+        *,

+        client: ssh_agent.SSHAgentClient | None = None,

+    ) -> bool:

         """Check whether the key is suitable for passphrase derivation.

-        Currently, this only statically checks whether signatures with

-        this key type are guaranteed to be deterministic.

+        Some key types are guaranteed to be deterministic.  Other keys

+        types are only deterministic if the SSH agent supports this

+        feature.

         Args:

-            key: SSH public key to check.

+            key:

+                SSH public key to check.

+            client:

+                An optional SSH agent client to check for additional

+                deterministic key types.  If not given, assume no such

+                types.

         Returns:

-            True if and only if the key is guaranteed suitable for use in

-            deriving a passphrase deterministically.

-

-        Note:

-            Some SSH agents additionally guarantee that all signatures

-            are deterministic and thus *all* keys are suitable; see

-            [`ssh_agent.SSHAgentClient.has_deterministic_signatures`][].

+            True if and only if the key is guaranteed suitable for use

+            in deriving a passphrase deterministically (perhaps

+            restricted to the indicated SSH agent).

         """

         TestFunc: TypeAlias = 'Callable[[bytes | bytearray], bool]'

@@ -475,7 +482,31 @@ class Vault:

             'ssh-ed448': lambda k: k.startswith(b'\x00\x00\x00\x09ssh-ed448'),

             'ssh-rsa': lambda k: k.startswith(b'\x00\x00\x00\x07ssh-rsa'),

         }

-        return any(v(key) for v in deterministic_signature_types.values())

+        dsa_signature_types = {

+            'ssh-dss': lambda k: k.startswith(b'\x00\x00\x00\x07ssh-dss'),

+            'ecdsa-sha2-nistp256': lambda k: k.startswith(

+                b'\x00\x00\x00\x13ecdsa-sha2-nistp256'

+            ),

+            'ecdsa-sha2-nistp384': lambda k: k.startswith(

+                b'\x00\x00\x00\x13ecdsa-sha2-nistp384'

+            ),

+            'ecdsa-sha2-nistp521': lambda k: k.startswith(

+                b'\x00\x00\x00\x13ecdsa-sha2-nistp521'

+            ),

+        }

+        criteria = [

+            lambda: any(

+                v(key) for v in deterministic_signature_types.values()

+            ),

+        ]

+        if client is not None:

+            criteria.append(

+                lambda: (

+                    client.has_deterministic_dsa_signatures()

+                    and any(v(key) for v in dsa_signature_types.values())

+                )

+            )

+        return any(crit() for crit in criteria)

     @classmethod

     def phrase_from_key(

@@ -489,8 +520,8 @@ class Vault:

         vault allows the usage of certain SSH keys to derive a master

         passphrase, by signing the vault UUID with the SSH key.  The key

-        type or the SSH agent must ensure that signatures are

-        deterministic.

+        type must ensure that signatures are deterministic (perhaps only

+        in conjunction with the given SSH agent).

         Args:

             key:

@@ -544,10 +575,7 @@ class Vault:

         """

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

-            if not (

-                client.has_deterministic_signatures()

-                or cls._is_suitable_ssh_key(key)

-            ):

+            if not cls.is_suitable_ssh_key(key, client=client):

                 msg = (

                     'unsuitable SSH key: bad key, or '

                     'signature not deterministic under this agent'

@@ -256,7 +256,8 @@ class TestAgentInteraction:

         )

         assert signature2 == expected_signature, 'SSH signature mismatch'

         assert (

-            vault.Vault.phrase_from_key(public_key_data) == derived_passphrase

+            vault.Vault.phrase_from_key(public_key_data, conn=client)

+            == derived_passphrase

         ), 'SSH signature mismatch'

     @pytest.mark.parametrize(

@@ -275,10 +276,13 @@ class TestAgentInteraction:

         _ = data_dict['expected_signature']

         if public_key_data not in key_comment_pairs:  # pragma: no cover

             pytest.skip('prerequisite SSH key not loaded')

-        if client.has_deterministic_signatures():

-            pytest.skip('agent ensures all keys are suitable')

+        assert not vault.Vault.is_suitable_ssh_key(

+            public_key_data, client=None

+        ), 'Expected key to be unsuitable in general'

+        if vault.Vault.is_suitable_ssh_key(public_key_data, client=client):

+            pytest.skip('agent automatically ensures key is suitable')

         with pytest.raises(ValueError, match='unsuitable SSH key'):

-            vault.Vault.phrase_from_key(public_key_data)

+            vault.Vault.phrase_from_key(public_key_data, conn=client)

     @pytest.mark.parametrize(

         ['key', 'single'],

@@ -298,9 +302,17 @@ class TestAgentInteraction:

         client = ssh_agent_client_with_test_keys_loaded

         def key_is_suitable(key: bytes) -> bool:

-            return client.has_deterministic_signatures() or key in {

+            always = {

                 v['public_key_data'] for v in tests.SUPPORTED_KEYS.values()

             }

+            dsa = {

+                v['public_key_data']

+                for k, v in tests.UNSUITABLE_KEYS.items()

+                if k.startswith(('dsa', 'ecdsa'))

+            }

+            return key in always or (

+                client.has_deterministic_dsa_signatures() and key in dsa

+            )

         if single:

             monkeypatch.setattr(