Refactor the (SSH agent) testing machinery tests for modularity (3a69028) - derivepassphrase.git

@@ -160,111 +160,71 @@ def minimize_openssh_keyfile_padding(
 class Parametrize:
     """Common test parametrizations."""
 
-    STUBBED_AGENT_ADDRESSES = pytest.mark.parametrize(
-        ["address", "exception", "match"],
+    LIST_IDENTITIES = pytest.mark.parametrize(
+        ["extended_agent", "query_request"],
         [
-            pytest.param(None, KeyError, "SSH_AUTH_SOCK", id="unset"),
-            pytest.param("stub-ssh-agent:", None, "", id="standard"),
             pytest.param(
-                str(pathlib.Path("~").expanduser()),
-                FileNotFoundError,
-                os.strerror(errno.ENOENT),
-                id="invalid-url",
-            ),
-            pytest.param(
-                "stub-ssh-agent:EPROTONOSUPPORT",
-                OSError,
-                os.strerror(errno.EPROTONOSUPPORT),
-                id="protocol-not-supported",
+                None,
+                (
+                    # SSH string header
+                    b"\x00\x00\x00\x01"
+                    # request code: SSH_AGENTC_REQUEST_IDENTITIES
+                    b"\x0b"
                 ),
-            pytest.param(
-                "stub-ssh-agent:ABCDEFGHIJKLMNOPQRSTUVWXYZ",
-                OSError,
-                os.strerror(errno.EINVAL),
-                id="invalid-error-code",
+                id="base",
             ),
-        ],
-    )
-    TEST_KEYS = pytest.mark.parametrize(
-        ["keyname", "key"],
-        data.ALL_KEYS.items(),
-        ids=data.ALL_KEYS.keys(),
-    )
-    INVALID_SSH_AGENT_MESSAGES = pytest.mark.parametrize(
-        "message",
-        [
-            pytest.param(b"\x00\x00\x00\x00", id="empty-message"),
             pytest.param(
-                b"\x00\x00\x00\x06\x1b\x00\x00\x00\x01\xff",
-                id="invalid-extension-name",
-            ),
-            pytest.param(
-                b"\x00\x00\x00\x11\x0d\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00",
-                id="sign-with-trailing-data",
+                True,
+                (
+                    # SSH string header
+                    b"\x00\x00\x00\x2e"
+                    # request code: SSH_AGENTC_REQUEST_IDENTITIES
+                    b"\x1b"
+                    # extension type: list-extended@putty.projects.tartarus.org
+                    b"\x00\x00\x00\x29list-extended@putty.projects.tartarus.org"
+                    # (no payload)
                 ),
-            pytest.param(
-                b"\x00\x00\x00\x05\x0d\x00\x00\x00\x00",
-                id="sign-without-payload",
+                id="extended",
             ),
         ],
     )
-    UNSUPPORTED_SSH_AGENT_MESSAGES = pytest.mark.parametrize(
-        "message",
+    QUERY_EXTENSION = pytest.mark.parametrize(
+        ["extended_agent", "query_response"],
         [
             pytest.param(
-                ssh_agent.SSHAgentClient.string(
-                    b"".join([
-                        b"\x0d",
-                        ssh_agent.SSHAgentClient.string(
-                            data.ALL_KEYS["rsa"].public_key_data
-                        ),
-                        ssh_agent.SSHAgentClient.string(vault.Vault.UUID),
-                        b"\x00\x00\x00\x02",
-                    ])
-                ),
-                id="sign-with-flags",
-            ),
-            pytest.param(
-                ssh_agent.SSHAgentClient.string(
-                    b"".join([
-                        b"\x0d",
-                        ssh_agent.SSHAgentClient.string(
-                            data.ALL_KEYS["ed25519"].public_key_data
-                        ),
-                        b"\x00\x00\x00\x08\x00\x01\x02\x03\x04\x05\x06\x07",
-                        b"\x00\x00\x00\x00",
-                    ])
-                ),
-                id="sign-with-nonstandard-passphrase",
-            ),
-            pytest.param(
-                ssh_agent.SSHAgentClient.string(
-                    b"".join([
-                        b"\x0d",
-                        ssh_agent.SSHAgentClient.string(
-                            data.ALL_KEYS["dsa1024"].public_key_data
-                        ),
-                        ssh_agent.SSHAgentClient.string(vault.Vault.UUID),
-                        b"\x00\x00\x00\x00",
-                    ])
+                None,
+                (
+                    # SSH string header
+                    b"\x00\x00\x00\x01"
+                    # response code: SSH_AGENT_FAILURE
+                    b"\x05"
                 ),
-                id="sign-key-no-expected-signature",
-                # NOTE: only unsupported when stubbed agent has RFC 6979
-                # support disabled
+                id="base",
             ),
             pytest.param(
-                ssh_agent.SSHAgentClient.string(
-                    b"".join([
-                        b"\x0d",
-                        b"\x00\x00\x00\x00",
-                        ssh_agent.SSHAgentClient.string(vault.Vault.UUID),
-                        b"\x00\x00\x00\x00",
-                    ])
+                True,
+                (
+                    # SSH string header
+                    b"\x00\x00\x00\x40"
+                    # response code: SSH_AGENT_EXTENSION_RESPONSE
+                    b"\x1d"
+                    # extension response: extension type ("query")
+                    b"\x00\x00\x00\x05query"
+                    # supported extension #1: query
+                    b"\x00\x00\x00\x05query"
+                    # supported extension #2:
+                    # list-extended@putty.projects.tartarus.org
+                    b"\x00\x00\x00\x29list-extended@putty.projects.tartarus.org"
                 ),
-                id="sign-key-unregistered-test-key",
+                id="extended",
             ),
         ],
     )
+    TEST_KEYS = pytest.mark.parametrize(
+        ["keyname", "key"],
+        data.ALL_KEYS.items(),
+        ids=data.ALL_KEYS.keys(),
+    )
     SUPPORTED_SSH_TEST_KEYS = pytest.mark.parametrize(
         ["ssh_test_key_type", "ssh_test_key"],
         list(data.SUPPORTED_KEYS.items()),
@@ -344,6 +304,94 @@ class Strategies:
         payload.extend(truncated_message)
         return bytes(payload)
 
+    @staticmethod
+    def invalid_ssh_agent_messages() -> strategies.SearchStrategy[bytes]:
+        """Generate invalid agent request messages."""
+        string = ssh_agent.SSHAgentClient.string
+        empty_message = b"\x00\x00\x00\x00"
+        invalid_extension_name = b"\x00\x00\x00\x06\x1b\x00\x00\x00\x01\xff"
+        sign_with_trailing_data = b"\x00\x00\x00\x11\x0d\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
+        sign_without_payload = b"\x00\x00\x00\x05\x0d\x00\x00\x00\x00"
+        return strategies.sampled_from([
+            empty_message,
+            invalid_extension_name,
+            sign_with_trailing_data,
+            sign_without_payload,
+        ]).map(string)
+
+    @staticmethod
+    def unsupported_ssh_agent_messages() -> strategies.SearchStrategy[bytes]:
+        """Generate agent request messages unsupported by the stubbed agent."""
+        string = ssh_agent.SSHAgentClient.string
+        sign_with_flags = [
+            b"\x0d",
+            string(data.ALL_KEYS["rsa"].public_key_data),
+            string(vault.Vault.UUID),
+            b"\x00\x00\x00\x02",
+        ]
+        sign_with_nonstandard_passphrase = [
+            b"\x0d",
+            string(data.ALL_KEYS["ed25519"].public_key_data),
+            b"\x00\x00\x00\x08\x00\x01\x02\x03\x04\x05\x06\x07",
+            b"\x00\x00\x00\x00",
+        ]
+        # NOTE: only unsupported when stubbed agent has RFC 6979 support
+        # disabled
+        sign_key_no_expected_signature = [
+            b"\x0d",
+            string(data.ALL_KEYS["dsa1024"].public_key_data),
+            string(vault.Vault.UUID),
+            b"\x00\x00\x00\x00",
+        ]
+        sign_key_unregistered_test_key = [
+            b"\x0d",
+            b"\x00\x00\x00\x00",
+            string(vault.Vault.UUID),
+            b"\x00\x00\x00\x00",
+        ]
+        return (
+            strategies
+            .sampled_from([
+                sign_with_flags,
+                sign_with_nonstandard_passphrase,
+                sign_key_no_expected_signature,
+                sign_key_unregistered_test_key,
+            ])
+            .map(b"".join)
+            .map(string)
+        )
+
+    @staticmethod
+    def stubbed_agent_addresses() -> strategies.SearchStrategy[
+        tuple[str, type[OSError], str]
+    ]:
+        """Generate agent addresses for the stubbed agents.
+
+        These are all error calls.  Non-error calls are supplied by the
+        explicit examples.
+
+        """
+        invalid_url = (
+            str(pathlib.Path("~").expanduser()),
+            FileNotFoundError,
+            os.strerror(errno.ENOENT),
+        )
+        protocol_not_supported = (
+            "stub-ssh-agent:EPROTONOSUPPORT",
+            OSError,
+            os.strerror(errno.EPROTONOSUPPORT),
+        )
+        invalid_error_code = (
+            "stub-ssh-agent:ABCDEFGHIJKLMNOPQRSTUVWXYZ",
+            OSError,
+            os.strerror(errno.EINVAL),
+        )
+        return strategies.sampled_from([
+            invalid_url,
+            protocol_not_supported,
+            invalid_error_code,
+        ])
+
 
 class TestTestKeys:
     """Tests testing the test keys."""
@@ -444,48 +492,44 @@ class TestTestKeys:
         )
 
 
-class TestStubbedSSHAgentSocketRequests:
-    """Test the stubbed SSH agent socket: normal requests."""
+class TestStubbedSSHAgentSocket:
+    """Test the stubbed SSH agent socket: common machinery."""
 
     @contextlib.contextmanager
-    def _get_addressed_agent(
-        self, *, extended_agent: bool = False
-    ) -> Iterator[machinery.StubbedSSHAgentSocketWithAddress]:
-        agent_class: type[machinery.StubbedSSHAgentSocketWithAddress] = (
+    def _get_agent(
+        self, *, extended_agent: bool | None = False
+    ) -> Iterator[machinery.StubbedSSHAgentSocket]:
+        agent_class: type[machinery.StubbedSSHAgentSocket] = (
             machinery.StubbedSSHAgentSocketWithAddressAndDeterministicDSA
             if extended_agent
             else machinery.StubbedSSHAgentSocketWithAddress
+            if extended_agent is not None
+            else machinery.StubbedSSHAgentSocket
         )
         with contextlib.ExitStack() as stack:
             monkeypatch = stack.enter_context(pytest.MonkeyPatch.context())
+            if issubclass(
+                agent_class, machinery.StubbedSSHAgentSocketWithAddress
+            ):
                 monkeypatch.setenv("SSH_AUTH_SOCK", agent_class.ADDRESS)
+            else:
+                monkeypatch.delenv("SSH_AUTH_SOCK", raising=False)
             yield stack.enter_context(agent_class())
 
-    def test_query_extensions_base(self) -> None:
-        """The base agent implements no extensions."""
-        with self._get_addressed_agent(extended_agent=False) as agent:
-            assert "query" not in agent.enabled_extensions
-            query_request = (
-                # SSH string header
-                b"\x00\x00\x00\x0a"
-                # request code: SSH_AGENTC_EXTENSION
-                b"\x1b"
-                # payload: SSH string "query"
-                b"\x00\x00\x00\x05query"
-            )
-            query_response = (
-                # SSH string header
-                b"\x00\x00\x00\x01"
-                # response code: SSH_AGENT_FAILURE
-                b"\x05"
-            )
-            agent.sendall(query_request)
-            assert agent.recv(1000) == query_response
 
-    def test_query_extensions_extended(self) -> None:
-        """The extended agent implements a known list of extensions."""
-        with self._get_addressed_agent(extended_agent=True) as agent:
-            assert "query" in agent.enabled_extensions
+class TestStubbedSSHAgentSocketRequests(TestStubbedSSHAgentSocket):
+    """Test the stubbed SSH agent socket: normal requests."""
+
+    @Parametrize.QUERY_EXTENSION
+    def test_query_extensions(
+        self, extended_agent: bool, query_response: bytes
+    ) -> None:
+        """The agent implements a known list of extensions.
+
+        The list is empty for the base agent, and non-empty for the
+        extended agent.
+
+        """
         query_request = (
             # SSH string header
             b"\x00\x00\x00\x0a"
@@ -494,75 +538,45 @@ class TestStubbedSSHAgentSocketRequests:
             # payload: SSH string "query"
             b"\x00\x00\x00\x05query"
         )
-            query_response = (
-                # SSH string header
-                b"\x00\x00\x00\x40"
-                # response code: SSH_AGENT_EXTENSION_RESPONSE
-                b"\x1d"
-                # extension response: extension type ("query")
-                b"\x00\x00\x00\x05query"
-                # supported extension #1: query
-                b"\x00\x00\x00\x05query"
-                # supported extension #2:
-                # list-extended@putty.projects.tartarus.org
-                b"\x00\x00\x00\x29list-extended@putty.projects.tartarus.org"
+        with self._get_agent(extended_agent=extended_agent) as agent:
+            assert ("query" in agent.enabled_extensions) == bool(
+                extended_agent
             )
             agent.sendall(query_request)
             assert agent.recv(1000) == query_response
 
-    def test_request_identities(self) -> None:
-        """The agent implements a known list of identities."""
+    @Parametrize.LIST_IDENTITIES
+    def test_request_identities(
+        self, extended_agent: bool | None, query_request: bytes
+    ) -> None:
+        """The agent implements a known list of identities.
+
+        The extended agent implements PuTTY's `list-extended` extension.
+
+        """
         unstring_prefix = ssh_agent.SSHAgentClient.unstring_prefix
-        with machinery.StubbedSSHAgentSocket() as agent:
-            query_request = (
-                # SSH string header
-                b"\x00\x00\x00\x01"
-                # request code: SSH_AGENTC_REQUEST_IDENTITIES
-                b"\x0b"
-            )
+        with self._get_agent(extended_agent=extended_agent) as agent:
             agent.sendall(query_request)
             message_length = int.from_bytes(agent.recv(4), "big")
             orig_message: bytes | bytearray = bytearray(
                 agent.recv(message_length)
             )
-            assert orig_message[0] == _types.SSH_AGENT.IDENTITIES_ANSWER
-            identity_count = int.from_bytes(orig_message[1:5], "big")
-            message = bytes(orig_message[5:])
-            for _ in range(identity_count):
-                key, message = unstring_prefix(message)
-                _comment, message = unstring_prefix(message)
-                assert key
-                assert key in {
-                    k.public_key_data for k in data.ALL_KEYS.values()
-                }
-            assert not message
-
-    def test_request_identities_extended(self) -> None:
-        """The extended agent implements PuTTY's `list-extended` extension."""
-        unstring_prefix = ssh_agent.SSHAgentClient.unstring_prefix
-        with self._get_addressed_agent(extended_agent=True) as agent:
-            extension_request = (
-                # SSH string header
-                b"\x00\x00\x00\x2e"
-                # request code: SSH_AGENTC_REQUEST_IDENTITIES
-                b"\x1b"
-                # extension type: list-extended@putty.projects.tartarus.org
-                b"\x00\x00\x00\x29list-extended@putty.projects.tartarus.org"
-                # (no payload)
-            )
-            agent.sendall(extension_request)
-            message_length = int.from_bytes(agent.recv(4), "big")
-            orig_message: bytes | bytearray = bytearray(
-                agent.recv(message_length)
+            assert (
+                orig_message[0] == _types.SSH_AGENT.SUCCESS
+                if extended_agent
+                else _types.SSH_AGENT.IDENTITIES_ANSWER
             )
-            assert orig_message[0] == _types.SSH_AGENT.SUCCESS
             identity_count = int.from_bytes(orig_message[1:5], "big")
             message = bytes(orig_message[5:])
             for _ in range(identity_count):
                 key, message = unstring_prefix(message)
                 _comment, message = unstring_prefix(message)
-                flags, message = unstring_prefix(message)
-                assert flags == b"\x00\x00\x00\x00"
+                flags, message = (
+                    unstring_prefix(message)
+                    if extended_agent
+                    else (b"", message)
+                )
+                assert not extended_agent or flags == b"\x00\x00\x00\x00"
                 assert key
                 assert key in {
                     k.public_key_data for k in data.ALL_KEYS.values()
@@ -601,20 +615,20 @@ class TestStubbedSSHAgentSocketRequests:
             assert agent.recv(1000) == query_response
 
 
-class TestStubbedSSHAgentSocketProperOperations:
+class TestStubbedSSHAgentSocketProperOperations(TestStubbedSSHAgentSocket):
     """Test the stubbed SSH agent socket: proper use and misuse."""
 
     def test_close_multiple(self) -> None:
         """The agent can be closed repeatedly."""
-        with machinery.StubbedSSHAgentSocket() as agent:
+        with self._get_agent(extended_agent=None) as agent:
             pass
-        with machinery.StubbedSSHAgentSocket() as agent:
+        with agent:
             pass
         del agent
 
     def test_closed_agents_cannot_be_interacted_with(self) -> None:
         """The agent cannot be usefully used after close."""
-        with machinery.StubbedSSHAgentSocket() as agent:
+        with self._get_agent(extended_agent=None) as agent:
             pass
         query_request = (
             # SSH string header
@@ -631,7 +645,7 @@ class TestStubbedSSHAgentSocketProperOperations:
 
     def test_no_recv_without_sendall(self) -> None:
         """The agent requires a message before sending a response."""
-        with machinery.StubbedSSHAgentSocket() as agent:  # noqa: SIM117
+        with self._get_agent(extended_agent=None) as agent:  # noqa: SIM117
             with pytest.raises(
                 AssertionError,
                 match=re.escape(
@@ -657,7 +671,7 @@ class TestStubbedSSHAgentSocketProperOperations:
     )
     def test_piecemeal_sendall(self, query_request_parts: list[bytes]) -> None:
         """The agent supports receiving messages incrementally."""
-        with machinery.StubbedSSHAgentSocket() as agent:
+        with self._get_agent(extended_agent=None) as agent:
             agent.enabled_extensions = frozenset({"query"})
             for part in query_request_parts:
                 agent.sendall(part)
@@ -677,7 +691,7 @@ class TestStubbedSSHAgentSocketProperOperations:
                 _types.SSH_AGENT.EXTENSION_RESPONSE
             )
 
-    @Parametrize.INVALID_SSH_AGENT_MESSAGES
+    @hypothesis.given(message=Strategies.invalid_ssh_agent_messages())
     def test_invalid_ssh_agent_messages(
         self,
         message: Buffer,
@@ -703,14 +717,16 @@ class TestStubbedSSHAgentSocketProperOperations:
                 machinery.StubbedSSHAgentSocket._INCOMPLETE_REQUEST
             ),
         ):
-            with machinery.StubbedSSHAgentSocket() as agent:
+            with self._get_agent(extended_agent=None) as agent:
                 agent.sendall(message)
 
 
-class TestStubbedSSHAgentSocketSupportedAndUnsupportedFeatures:
+class TestStubbedSSHAgentSocketSupportedAndUnsupportedFeatures(
+    TestStubbedSSHAgentSocket
+):
     """Test the stubbed SSH agent socket: supported/unsupported features."""
 
-    @Parametrize.UNSUPPORTED_SSH_AGENT_MESSAGES
+    @hypothesis.given(message=Strategies.unsupported_ssh_agent_messages())
     def test_unsupported_ssh_agent_messages(
         self,
         message: Buffer,
@@ -722,18 +738,19 @@ class TestStubbedSSHAgentSocketSupportedAndUnsupportedFeatures:
             # response code: SSH_AGENT_FAILURE
             b"\x05"
         )
-        with machinery.StubbedSSHAgentSocket() as agent:
+        with self._get_agent(extended_agent=None) as agent:
             agent.sendall(message)
             assert agent.recv(100) == query_response
 
-    @Parametrize.STUBBED_AGENT_ADDRESSES
+    @hypothesis.given(args_tuple=Strategies.stubbed_agent_addresses())
+    @hypothesis.example(args_tuple=(None, KeyError, "SSH_AUTH_SOCK"))
+    @hypothesis.example(args_tuple=("stub-ssh-agent:", None, ""))
     def test_addresses(
         self,
-        address: str | None,
-        exception: type[Exception] | None,
-        match: str,
+        args_tuple: tuple[str | None, type[Exception] | None, str],
     ) -> None:
         """The agent accepts addresses."""
+        address, exception, match = args_tuple
         with contextlib.ExitStack() as stack:
             monkeypatch = stack.enter_context(pytest.MonkeyPatch.context())
             if address:


...	...	@@ -160,111 +160,71 @@ def minimize_openssh_keyfile_padding(
160	160	class Parametrize:
161	161	"""Common test parametrizations."""
162	162
163		- STUBBED_AGENT_ADDRESSES = pytest.mark.parametrize(
164		- ["address", "exception", "match"],
	163	+ LIST_IDENTITIES = pytest.mark.parametrize(
	164	+ ["extended_agent", "query_request"],
165	165	[
166		- pytest.param(None, KeyError, "SSH_AUTH_SOCK", id="unset"),
167		- pytest.param("stub-ssh-agent:", None, "", id="standard"),
168	166	pytest.param(
169		- str(pathlib.Path("~").expanduser()),
170		- FileNotFoundError,
171		- os.strerror(errno.ENOENT),
172		- id="invalid-url",
173		- ),
174		- pytest.param(
175		- "stub-ssh-agent:EPROTONOSUPPORT",
176		- OSError,
177		- os.strerror(errno.EPROTONOSUPPORT),
178		- id="protocol-not-supported",
	167	+ None,
	168	+ (
	169	+ # SSH string header
	170	+ b"\x00\x00\x00\x01"
	171	+ # request code: SSH_AGENTC_REQUEST_IDENTITIES
	172	+ b"\x0b"
179	173	),
180		- pytest.param(
181		- "stub-ssh-agent:ABCDEFGHIJKLMNOPQRSTUVWXYZ",
182		- OSError,
183		- os.strerror(errno.EINVAL),
184		- id="invalid-error-code",
	174	+ id="base",
185	175	),
186		- ],
187		- )
188		- TEST_KEYS = pytest.mark.parametrize(
189		- ["keyname", "key"],
190		- data.ALL_KEYS.items(),
191		- ids=data.ALL_KEYS.keys(),
192		- )
193		- INVALID_SSH_AGENT_MESSAGES = pytest.mark.parametrize(
194		- "message",
195		- [
196		- pytest.param(b"\x00\x00\x00\x00", id="empty-message"),
197	176	pytest.param(
198		- b"\x00\x00\x00\x06\x1b\x00\x00\x00\x01\xff",
199		- id="invalid-extension-name",
200		- ),
201		- pytest.param(
202		- b"\x00\x00\x00\x11\x0d\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00",
203		- id="sign-with-trailing-data",
	177	+ True,
	178	+ (
	179	+ # SSH string header
	180	+ b"\x00\x00\x00\x2e"
	181	+ # request code: SSH_AGENTC_REQUEST_IDENTITIES
	182	+ b"\x1b"
	183	+ # extension type: list-extended@putty.projects.tartarus.org
	184	+ b"\x00\x00\x00\x29list-extended@putty.projects.tartarus.org"
	185	+ # (no payload)
204	186	),
205		- pytest.param(
206		- b"\x00\x00\x00\x05\x0d\x00\x00\x00\x00",
207		- id="sign-without-payload",
	187	+ id="extended",
208	188	),
209	189	],
210	190	)
211		- UNSUPPORTED_SSH_AGENT_MESSAGES = pytest.mark.parametrize(
212		- "message",
	191	+ QUERY_EXTENSION = pytest.mark.parametrize(
	192	+ ["extended_agent", "query_response"],
213	193	[
214	194	pytest.param(
215		- ssh_agent.SSHAgentClient.string(
216		- b"".join([
217		- b"\x0d",
218		- ssh_agent.SSHAgentClient.string(
219		- data.ALL_KEYS["rsa"].public_key_data
220		- ),
221		- ssh_agent.SSHAgentClient.string(vault.Vault.UUID),
222		- b"\x00\x00\x00\x02",
223		- ])
224		- ),
225		- id="sign-with-flags",
226		- ),
227		- pytest.param(
228		- ssh_agent.SSHAgentClient.string(
229		- b"".join([
230		- b"\x0d",
231		- ssh_agent.SSHAgentClient.string(
232		- data.ALL_KEYS["ed25519"].public_key_data
233		- ),
234		- b"\x00\x00\x00\x08\x00\x01\x02\x03\x04\x05\x06\x07",
235		- b"\x00\x00\x00\x00",
236		- ])
237		- ),
238		- id="sign-with-nonstandard-passphrase",
239		- ),
240		- pytest.param(
241		- ssh_agent.SSHAgentClient.string(
242		- b"".join([
243		- b"\x0d",
244		- ssh_agent.SSHAgentClient.string(
245		- data.ALL_KEYS["dsa1024"].public_key_data
246		- ),
247		- ssh_agent.SSHAgentClient.string(vault.Vault.UUID),
248		- b"\x00\x00\x00\x00",
249		- ])
	195	+ None,
	196	+ (
	197	+ # SSH string header
	198	+ b"\x00\x00\x00\x01"
	199	+ # response code: SSH_AGENT_FAILURE
	200	+ b"\x05"
250	201	),
251		- id="sign-key-no-expected-signature",
252		- # NOTE: only unsupported when stubbed agent has RFC 6979
253		- # support disabled
	202	+ id="base",
254	203	),
255	204	pytest.param(
256		- ssh_agent.SSHAgentClient.string(
257		- b"".join([
258		- b"\x0d",
259		- b"\x00\x00\x00\x00",
260		- ssh_agent.SSHAgentClient.string(vault.Vault.UUID),
261		- b"\x00\x00\x00\x00",
262		- ])
	205	+ True,
	206	+ (
	207	+ # SSH string header
	208	+ b"\x00\x00\x00\x40"
	209	+ # response code: SSH_AGENT_EXTENSION_RESPONSE
	210	+ b"\x1d"
	211	+ # extension response: extension type ("query")
	212	+ b"\x00\x00\x00\x05query"
	213	+ # supported extension #1: query
	214	+ b"\x00\x00\x00\x05query"
	215	+ # supported extension #2:
	216	+ # list-extended@putty.projects.tartarus.org
	217	+ b"\x00\x00\x00\x29list-extended@putty.projects.tartarus.org"
263	218	),
264		- id="sign-key-unregistered-test-key",
	219	+ id="extended",
265	220	),
266	221	],
267	222	)
	223	+ TEST_KEYS = pytest.mark.parametrize(
	224	+ ["keyname", "key"],
	225	+ data.ALL_KEYS.items(),
	226	+ ids=data.ALL_KEYS.keys(),
	227	+ )
268	228	SUPPORTED_SSH_TEST_KEYS = pytest.mark.parametrize(
269	229	["ssh_test_key_type", "ssh_test_key"],
270	230	list(data.SUPPORTED_KEYS.items()),
...	...	@@ -344,6 +304,94 @@ class Strategies:
344	304	payload.extend(truncated_message)
345	305	return bytes(payload)
346	306
	307	+ @staticmethod
	308	+ def invalid_ssh_agent_messages() -> strategies.SearchStrategy[bytes]:
	309	+ """Generate invalid agent request messages."""
	310	+ string = ssh_agent.SSHAgentClient.string
	311	+ empty_message = b"\x00\x00\x00\x00"
	312	+ invalid_extension_name = b"\x00\x00\x00\x06\x1b\x00\x00\x00\x01\xff"
	313	+ sign_with_trailing_data = b"\x00\x00\x00\x11\x0d\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00"
	314	+ sign_without_payload = b"\x00\x00\x00\x05\x0d\x00\x00\x00\x00"
	315	+ return strategies.sampled_from([
	316	+ empty_message,
	317	+ invalid_extension_name,
	318	+ sign_with_trailing_data,
	319	+ sign_without_payload,
	320	+ ]).map(string)
	321	+
	322	+ @staticmethod
	323	+ def unsupported_ssh_agent_messages() -> strategies.SearchStrategy[bytes]:
	324	+ """Generate agent request messages unsupported by the stubbed agent."""
	325	+ string = ssh_agent.SSHAgentClient.string
	326	+ sign_with_flags = [
	327	+ b"\x0d",
	328	+ string(data.ALL_KEYS["rsa"].public_key_data),
	329	+ string(vault.Vault.UUID),
	330	+ b"\x00\x00\x00\x02",
	331	+ ]
	332	+ sign_with_nonstandard_passphrase = [
	333	+ b"\x0d",
	334	+ string(data.ALL_KEYS["ed25519"].public_key_data),
	335	+ b"\x00\x00\x00\x08\x00\x01\x02\x03\x04\x05\x06\x07",
	336	+ b"\x00\x00\x00\x00",
	337	+ ]
	338	+ # NOTE: only unsupported when stubbed agent has RFC 6979 support
	339	+ # disabled
	340	+ sign_key_no_expected_signature = [
	341	+ b"\x0d",
	342	+ string(data.ALL_KEYS["dsa1024"].public_key_data),
	343	+ string(vault.Vault.UUID),
	344	+ b"\x00\x00\x00\x00",
	345	+ ]
	346	+ sign_key_unregistered_test_key = [
	347	+ b"\x0d",
	348	+ b"\x00\x00\x00\x00",
	349	+ string(vault.Vault.UUID),
	350	+ b"\x00\x00\x00\x00",
	351	+ ]
	352	+ return (
	353	+ strategies
	354	+ .sampled_from([
	355	+ sign_with_flags,
	356	+ sign_with_nonstandard_passphrase,
	357	+ sign_key_no_expected_signature,
	358	+ sign_key_unregistered_test_key,
	359	+ ])
	360	+ .map(b"".join)
	361	+ .map(string)
	362	+ )
	363	+
	364	+ @staticmethod
	365	+ def stubbed_agent_addresses() -> strategies.SearchStrategy[
	366	+ tuple[str, type[OSError], str]
	367	+ ]:
	368	+ """Generate agent addresses for the stubbed agents.
	369	+
	370	+ These are all error calls. Non-error calls are supplied by the
	371	+ explicit examples.
	372	+
	373	+ """
	374	+ invalid_url = (
	375	+ str(pathlib.Path("~").expanduser()),
	376	+ FileNotFoundError,
	377	+ os.strerror(errno.ENOENT),
	378	+ )
	379	+ protocol_not_supported = (
	380	+ "stub-ssh-agent:EPROTONOSUPPORT",
	381	+ OSError,
	382	+ os.strerror(errno.EPROTONOSUPPORT),
	383	+ )
	384	+ invalid_error_code = (
	385	+ "stub-ssh-agent:ABCDEFGHIJKLMNOPQRSTUVWXYZ",
	386	+ OSError,
	387	+ os.strerror(errno.EINVAL),
	388	+ )
	389	+ return strategies.sampled_from([
	390	+ invalid_url,
	391	+ protocol_not_supported,
	392	+ invalid_error_code,
	393	+ ])
	394	+
347	395
348	396	class TestTestKeys:
349	397	"""Tests testing the test keys."""
...	...	@@ -444,48 +492,44 @@ class TestTestKeys:
444	492	)
445	493
446	494
447		-class TestStubbedSSHAgentSocketRequests:
448		- """Test the stubbed SSH agent socket: normal requests."""
	495	+class TestStubbedSSHAgentSocket:
	496	+ """Test the stubbed SSH agent socket: common machinery."""
449	497
450	498	@contextlib.contextmanager
451		- def _get_addressed_agent(
452		- self, *, extended_agent: bool = False
453		- ) -> Iterator[machinery.StubbedSSHAgentSocketWithAddress]:
454		- agent_class: type[machinery.StubbedSSHAgentSocketWithAddress] = (
	499	+ def _get_agent(
	500	+ self, *, extended_agent: bool \| None = False
	501	+ ) -> Iterator[machinery.StubbedSSHAgentSocket]:
	502	+ agent_class: type[machinery.StubbedSSHAgentSocket] = (
455	503	machinery.StubbedSSHAgentSocketWithAddressAndDeterministicDSA
456	504	if extended_agent
457	505	else machinery.StubbedSSHAgentSocketWithAddress
	506	+ if extended_agent is not None
	507	+ else machinery.StubbedSSHAgentSocket
458	508	)
459	509	with contextlib.ExitStack() as stack:
460	510	monkeypatch = stack.enter_context(pytest.MonkeyPatch.context())
	511	+ if issubclass(
	512	+ agent_class, machinery.StubbedSSHAgentSocketWithAddress
	513	+ ):
461	514	monkeypatch.setenv("SSH_AUTH_SOCK", agent_class.ADDRESS)
	515	+ else:
	516	+ monkeypatch.delenv("SSH_AUTH_SOCK", raising=False)
462	517	yield stack.enter_context(agent_class())
463	518
464		- def test_query_extensions_base(self) -> None:
465		- """The base agent implements no extensions."""
466		- with self._get_addressed_agent(extended_agent=False) as agent:
467		- assert "query" not in agent.enabled_extensions
468		- query_request = (
469		- # SSH string header
470		- b"\x00\x00\x00\x0a"
471		- # request code: SSH_AGENTC_EXTENSION
472		- b"\x1b"
473		- # payload: SSH string "query"
474		- b"\x00\x00\x00\x05query"
475		- )
476		- query_response = (
477		- # SSH string header
478		- b"\x00\x00\x00\x01"
479		- # response code: SSH_AGENT_FAILURE
480		- b"\x05"
481		- )
482		- agent.sendall(query_request)
483		- assert agent.recv(1000) == query_response
484	519
485		- def test_query_extensions_extended(self) -> None:
486		- """The extended agent implements a known list of extensions."""
487		- with self._get_addressed_agent(extended_agent=True) as agent:
488		- assert "query" in agent.enabled_extensions
	520	+class TestStubbedSSHAgentSocketRequests(TestStubbedSSHAgentSocket):
	521	+ """Test the stubbed SSH agent socket: normal requests."""
	522	+
	523	+ @Parametrize.QUERY_EXTENSION
	524	+ def test_query_extensions(
	525	+ self, extended_agent: bool, query_response: bytes
	526	+ ) -> None:
	527	+ """The agent implements a known list of extensions.
	528	+
	529	+ The list is empty for the base agent, and non-empty for the
	530	+ extended agent.
	531	+
	532	+ """
489	533	query_request = (
490	534	# SSH string header
491	535	b"\x00\x00\x00\x0a"
...	...	@@ -494,75 +538,45 @@ class TestStubbedSSHAgentSocketRequests:
494	538	# payload: SSH string "query"
495	539	b"\x00\x00\x00\x05query"
496	540	)
497		- query_response = (
498		- # SSH string header
499		- b"\x00\x00\x00\x40"
500		- # response code: SSH_AGENT_EXTENSION_RESPONSE
501		- b"\x1d"
502		- # extension response: extension type ("query")
503		- b"\x00\x00\x00\x05query"
504		- # supported extension #1: query
505		- b"\x00\x00\x00\x05query"
506		- # supported extension #2:
507		- # list-extended@putty.projects.tartarus.org
508		- b"\x00\x00\x00\x29list-extended@putty.projects.tartarus.org"
	541	+ with self._get_agent(extended_agent=extended_agent) as agent:
	542	+ assert ("query" in agent.enabled_extensions) == bool(
	543	+ extended_agent
509	544	)
510	545	agent.sendall(query_request)
511	546	assert agent.recv(1000) == query_response
512	547
513		- def test_request_identities(self) -> None:
514		- """The agent implements a known list of identities."""
	548	+ @Parametrize.LIST_IDENTITIES
	549	+ def test_request_identities(
	550	+ self, extended_agent: bool \| None, query_request: bytes
	551	+ ) -> None:
	552	+ """The agent implements a known list of identities.
	553	+
	554	+ The extended agent implements PuTTY's `list-extended` extension.
	555	+
	556	+ """
515	557	unstring_prefix = ssh_agent.SSHAgentClient.unstring_prefix
516		- with machinery.StubbedSSHAgentSocket() as agent:
517		- query_request = (
518		- # SSH string header
519		- b"\x00\x00\x00\x01"
520		- # request code: SSH_AGENTC_REQUEST_IDENTITIES
521		- b"\x0b"
522		- )
	558	+ with self._get_agent(extended_agent=extended_agent) as agent:
523	559	agent.sendall(query_request)
524	560	message_length = int.from_bytes(agent.recv(4), "big")
525	561	orig_message: bytes \| bytearray = bytearray(
526	562	agent.recv(message_length)
527	563	)
528		- assert orig_message[0] == _types.SSH_AGENT.IDENTITIES_ANSWER
529		- identity_count = int.from_bytes(orig_message[1:5], "big")
530		- message = bytes(orig_message[5:])
531		- for _ in range(identity_count):
532		- key, message = unstring_prefix(message)
533		- _comment, message = unstring_prefix(message)
534		- assert key
535		- assert key in {
536		- k.public_key_data for k in data.ALL_KEYS.values()
537		- }
538		- assert not message
539		-
540		- def test_request_identities_extended(self) -> None:
541		- """The extended agent implements PuTTY's `list-extended` extension."""
542		- unstring_prefix = ssh_agent.SSHAgentClient.unstring_prefix
543		- with self._get_addressed_agent(extended_agent=True) as agent:
544		- extension_request = (
545		- # SSH string header
546		- b"\x00\x00\x00\x2e"
547		- # request code: SSH_AGENTC_REQUEST_IDENTITIES
548		- b"\x1b"
549		- # extension type: list-extended@putty.projects.tartarus.org
550		- b"\x00\x00\x00\x29list-extended@putty.projects.tartarus.org"
551		- # (no payload)
552		- )
553		- agent.sendall(extension_request)
554		- message_length = int.from_bytes(agent.recv(4), "big")
555		- orig_message: bytes \| bytearray = bytearray(
556		- agent.recv(message_length)
	564	+ assert (
	565	+ orig_message[0] == _types.SSH_AGENT.SUCCESS
	566	+ if extended_agent
	567	+ else _types.SSH_AGENT.IDENTITIES_ANSWER
557	568	)
558		- assert orig_message[0] == _types.SSH_AGENT.SUCCESS
559	569	identity_count = int.from_bytes(orig_message[1:5], "big")
560	570	message = bytes(orig_message[5:])
561	571	for _ in range(identity_count):
562	572	key, message = unstring_prefix(message)
563	573	_comment, message = unstring_prefix(message)
564		- flags, message = unstring_prefix(message)
565		- assert flags == b"\x00\x00\x00\x00"
	574	+ flags, message = (
	575	+ unstring_prefix(message)
	576	+ if extended_agent
	577	+ else (b"", message)
	578	+ )
	579	+ assert not extended_agent or flags == b"\x00\x00\x00\x00"
566	580	assert key
567	581	assert key in {
568	582	k.public_key_data for k in data.ALL_KEYS.values()
...	...	@@ -601,20 +615,20 @@ class TestStubbedSSHAgentSocketRequests:
601	615	assert agent.recv(1000) == query_response
602	616
603	617
604		-class TestStubbedSSHAgentSocketProperOperations:
	618	+class TestStubbedSSHAgentSocketProperOperations(TestStubbedSSHAgentSocket):
605	619	"""Test the stubbed SSH agent socket: proper use and misuse."""
606	620
607	621	def test_close_multiple(self) -> None:
608	622	"""The agent can be closed repeatedly."""
609		- with machinery.StubbedSSHAgentSocket() as agent:
	623	+ with self._get_agent(extended_agent=None) as agent:
610	624	pass
611		- with machinery.StubbedSSHAgentSocket() as agent:
	625	+ with agent:
612	626	pass
613	627	del agent
614	628
615	629	def test_closed_agents_cannot_be_interacted_with(self) -> None:
616	630	"""The agent cannot be usefully used after close."""
617		- with machinery.StubbedSSHAgentSocket() as agent:
	631	+ with self._get_agent(extended_agent=None) as agent:
618	632	pass
619	633	query_request = (
620	634	# SSH string header
...	...	@@ -631,7 +645,7 @@ class TestStubbedSSHAgentSocketProperOperations:
631	645
632	646	def test_no_recv_without_sendall(self) -> None:
633	647	"""The agent requires a message before sending a response."""
634		- with machinery.StubbedSSHAgentSocket() as agent: # noqa: SIM117
	648	+ with self._get_agent(extended_agent=None) as agent: # noqa: SIM117
635	649	with pytest.raises(
636	650	AssertionError,
637	651	match=re.escape(
...	...	@@ -657,7 +671,7 @@ class TestStubbedSSHAgentSocketProperOperations:
657	671	)
658	672	def test_piecemeal_sendall(self, query_request_parts: list[bytes]) -> None:
659	673	"""The agent supports receiving messages incrementally."""
660		- with machinery.StubbedSSHAgentSocket() as agent:
	674	+ with self._get_agent(extended_agent=None) as agent:
661	675	agent.enabled_extensions = frozenset({"query"})
662	676	for part in query_request_parts:
663	677	agent.sendall(part)
...	...	@@ -677,7 +691,7 @@ class TestStubbedSSHAgentSocketProperOperations:
677	691	_types.SSH_AGENT.EXTENSION_RESPONSE
678	692	)
679	693
680		- @Parametrize.INVALID_SSH_AGENT_MESSAGES
	694	+ @hypothesis.given(message=Strategies.invalid_ssh_agent_messages())
681	695	def test_invalid_ssh_agent_messages(
682	696	self,
683	697	message: Buffer,
...	...	@@ -703,14 +717,16 @@ class TestStubbedSSHAgentSocketProperOperations:
703	717	machinery.StubbedSSHAgentSocket._INCOMPLETE_REQUEST
704	718	),
705	719	):
706		- with machinery.StubbedSSHAgentSocket() as agent:
	720	+ with self._get_agent(extended_agent=None) as agent:
707	721	agent.sendall(message)
708	722
709	723
710		-class TestStubbedSSHAgentSocketSupportedAndUnsupportedFeatures:
	724	+class TestStubbedSSHAgentSocketSupportedAndUnsupportedFeatures(
	725	+ TestStubbedSSHAgentSocket
	726	+):
711	727	"""Test the stubbed SSH agent socket: supported/unsupported features."""
712	728
713		- @Parametrize.UNSUPPORTED_SSH_AGENT_MESSAGES
	729	+ @hypothesis.given(message=Strategies.unsupported_ssh_agent_messages())
714	730	def test_unsupported_ssh_agent_messages(
715	731	self,
716	732	message: Buffer,
...	...	@@ -722,18 +738,19 @@ class TestStubbedSSHAgentSocketSupportedAndUnsupportedFeatures:
722	738	# response code: SSH_AGENT_FAILURE
723	739	b"\x05"
724	740	)
725		- with machinery.StubbedSSHAgentSocket() as agent:
	741	+ with self._get_agent(extended_agent=None) as agent:
726	742	agent.sendall(message)
727	743	assert agent.recv(100) == query_response
728	744
729		- @Parametrize.STUBBED_AGENT_ADDRESSES
	745	+ @hypothesis.given(args_tuple=Strategies.stubbed_agent_addresses())
	746	+ @hypothesis.example(args_tuple=(None, KeyError, "SSH_AUTH_SOCK"))
	747	+ @hypothesis.example(args_tuple=("stub-ssh-agent:", None, ""))
730	748	def test_addresses(
731	749	self,
732		- address: str \| None,
733		- exception: type[Exception] \| None,
734		- match: str,
	750	+ args_tuple: tuple[str \| None, type[Exception] \| None, str],
735	751	) -> None:
736	752	"""The agent accepts addresses."""
	753	+ address, exception, match = args_tuple
737	754	with contextlib.ExitStack() as stack:
738	755	monkeypatch = stack.enter_context(pytest.MonkeyPatch.context())
739	756	if address:
740	757