Add more tests for the SSH agent socket provider internals (9093a9b) - derivepassphrase.git

tests/test_derivepassphrase_ssh_agent/test_000_basic.py

@@ -8,6 +8,7 @@ from __future__ import annotations
 
 import base64
 import contextlib
+import ctypes
 import importlib.metadata
 import io
 import re
@@ -30,7 +31,7 @@ from tests.data import callables
 from tests.machinery import pytest as pytest_machinery
 
 if TYPE_CHECKING:
-    from collections.abc import Iterable, Iterator, Mapping
+    from collections.abc import Callable, Iterable, Iterator, Mapping
 
     from typing_extensions import Any, Literal
 
@@ -212,6 +213,14 @@ class Parametrize(types.SimpleNamespace):
             pytest.param(16777216, b"\x01\x00\x00\x00", id="16777216"),
         ],
     )
+    IO_OPERATIONS_ON_HANDLES = pytest.mark.parametrize(
+        "io_operation",
+        [
+            lambda handle: handle.recv(1),
+            lambda handle: handle.sendall(b"Hello, world!"),
+        ],
+        ids=["recv", "sendall"],
+    )
     SIGN_ERROR_RESPONSES = pytest.mark.parametrize(
         [
             "key",
@@ -409,6 +418,31 @@ class Parametrize(types.SimpleNamespace):
     )
 
 
+class Strategies:
+    """Common hypothesis data generation strategies."""
+
+    @staticmethod
+    def select_invalid_pipe_names(candidate: str) -> bool:
+        """Select candidate strings that are invalid named pipe names."""
+        candidate = candidate.lower().replace("/", "\\")
+        return not candidate.startswith(socketprovider.PIPE_PREFIX)
+
+    @classmethod
+    def bad_named_pipe_names(cls) -> strategies.SearchStrategy[str]:
+        r"""Generate invalid names for Windows named pipes.
+
+        Invalid named pipe names do not lie below the `\\.\pipe\` path.
+        They also may or may not contain characters that are invalid
+        in paths on The Annoying OS.
+
+        """
+        # This is an environment string on The Annoying OS, so it must
+        # at least be a non-empty C string, i.e., without embedded NULs.
+        return strategies.text(
+            alphabet=strategies.characters(min_codepoint=1), min_size=1
+        ).filter(cls.select_invalid_pipe_names)
+
+
 class TestStaticFunctionality:
     """Test the static functionality of the `ssh_agent` module."""
 
@@ -476,24 +510,6 @@ class TestShellExportScriptParsing:
                 callables.parse_sh_export_line(line, env_name=env_name)
 
 
-class TestGracefulFailureForSSHAuthSock:
-    """Test the `posix` socket provider if `SSH_AUTH_SOCK` is unset."""
-
-    def test_constructor_posix_no_ssh_auth_sock(
-        self,
-        skip_if_no_af_unix_support: None,
-    ) -> None:
-        """Abort if the running agent cannot be located on POSIX."""
-        del skip_if_no_af_unix_support
-        posix_handler = socketprovider.SocketProvider.resolve("posix")
-        with pytest.MonkeyPatch.context() as monkeypatch:
-            monkeypatch.delenv("SSH_AUTH_SOCK", raising=False)
-            with pytest.raises(
-                KeyError, match="SSH_AUTH_SOCK environment variable"
-            ):
-                posix_handler()
-
-
 class TestSSHProtocolDatatypes(TestStaticFunctionality):
     """Tests for the utility functions for the SSH protocol datatypes."""
 
@@ -1383,3 +1399,111 @@ class TestAgentErrorResponses:
                 match=r"Malformed response|does not match request",
             ):
                 client.query_extensions()
+
+
+class TestSSHAgentSocketProviderFailures:
+    """Test SSH agent socket provider-specific failures."""
+
+    def test_posix_provider_no_ssh_auth_sock(
+        self,
+        skip_if_no_af_unix_support: None,
+    ) -> None:
+        """For `posix`, fail if the running agent cannot be located."""
+        del skip_if_no_af_unix_support
+        provider = socketprovider.SocketProvider.resolve("posix")
+        with pytest.MonkeyPatch.context() as monkeypatch:
+            monkeypatch.delenv("SSH_AUTH_SOCK", raising=False)
+            with pytest.raises(
+                KeyError, match="SSH_AUTH_SOCK environment variable"
+            ):
+                provider()
+
+    @pytest.mark.skipif(
+        not hasattr(ctypes, "WinDLL"),
+        reason="not relevant to systems without Windows named pipe support",
+    )
+    @hypothesis.given(named_pipe_name=Strategies.bad_named_pipe_names())
+    def test_ssh_auth_sock_on_windows_provider_bad_named_pipe_name(
+        self,
+        named_pipe_name: str,
+    ) -> None:
+        """For `ssh_auth_sock_on_windows`, fail for bad named pipe names."""
+        provider = socketprovider.SocketProvider.resolve(
+            "ssh_auth_sock_on_windows"
+        )
+        with pytest.MonkeyPatch.context() as monkeypatch:
+            monkeypatch.setenv("SSH_AUTH_SOCK", named_pipe_name)
+            with pytest.raises(ValueError, match="Invalid named pipe name"):
+                provider()
+
+
+class TestWindowsNamedPipeHandle:
+    """Test the WindowsNamedPipeHandle wrapper class.
+
+    Even works on other operating systems, because the relevant parts
+    are stubbed out anyway.
+
+    """
+
+    DUMMY_ADDRESS = f"{socketprovider.PIPE_PREFIX}dummy-address"
+    IO_ON_CLOSED_HANDLE_PATTERN = re.escape(
+        socketprovider.WindowsNamedPipeHandle._IO_ON_CLOSED_FILE
+    )
+
+    @hypothesis.given(named_pipe_name=Strategies.bad_named_pipe_names())
+    def test_constructor_bad_named_pipe_name(
+        self,
+        named_pipe_name: str,
+    ) -> None:
+        """The constructor fails for bad named pipe names."""
+        with pytest.raises(ValueError, match="Invalid named pipe name"):  # noqa: SIM117
+            with socketprovider.WindowsNamedPipeHandle(named_pipe_name):
+                pass
+
+    @contextlib.contextmanager
+    def stub_createfile_and_closehandle(
+        self,
+    ) -> Iterator[None]:
+        singleton = object()
+
+        def create_file(*_args: Any, **_kwargs: Any) -> object:
+            return singleton
+
+        def close_handle(*_args: Any, **_kwargs: Any) -> bool:
+            return True
+
+        with pytest.MonkeyPatch.context() as monkeypatch:
+            monkeypatch.setattr(socketprovider, "CreateFile", create_file)
+            monkeypatch.setattr(socketprovider, "CloseHandle", close_handle)
+            yield
+
+    @Parametrize.IO_OPERATIONS_ON_HANDLES
+    def test_closed_handle_cannot_be_operated_on(
+        self,
+        io_operation: Callable[[Any], Any],
+    ) -> None:
+        """Operating on a closed handle fails."""
+        with self.stub_createfile_and_closehandle():
+            handle = socketprovider.WindowsNamedPipeHandle(self.DUMMY_ADDRESS)
+            with handle:
+                pass  # close the handle
+            with pytest.raises(
+                ValueError, match=self.IO_ON_CLOSED_HANDLE_PATTERN
+            ):
+                io_operation(handle)
+
+    @Parametrize.IO_OPERATIONS_ON_HANDLES
+    def test_handle_is_non_reentrant(
+        self,
+        io_operation: Callable[[Any], Any],
+    ) -> None:
+        """The handler cannot be sensibly used in nested `with` blocks."""
+        with self.stub_createfile_and_closehandle():
+            handle = socketprovider.WindowsNamedPipeHandle(self.DUMMY_ADDRESS)
+            with handle:
+                with handle:
+                    pass  # close the handle
+                with pytest.raises(
+                    ValueError, match=self.IO_ON_CLOSED_HANDLE_PATTERN
+                ):
+                    io_operation(handle)


...	...	@@ -86,7 +86,7 @@ class Strategies:
86	86	config2.pop(key, None) # type: ignore[misc]
87	87	return config2
88	88
89		- # Prefer this explicit composite strategy oven `strategies.builds`
	89	+ # Prefer this explicit composite strategy over `strategies.builds`
90	90	# because the type checker can better introspect this.
91	91	@strategies.composite
92	92	@staticmethod

...	...	@@ -8,6 +8,7 @@ from __future__ import annotations
8	8
9	9	import base64
10	10	import contextlib
	11	+import ctypes
11	12	import importlib.metadata
12	13	import io
13	14	import re
...	...	@@ -30,7 +31,7 @@ from tests.data import callables
30	31	from tests.machinery import pytest as pytest_machinery
31	32
32	33	if TYPE_CHECKING:
33		- from collections.abc import Iterable, Iterator, Mapping
	34	+ from collections.abc import Callable, Iterable, Iterator, Mapping
34	35
35	36	from typing_extensions import Any, Literal
36	37
...	...	@@ -212,6 +213,14 @@ class Parametrize(types.SimpleNamespace):
212	213	pytest.param(16777216, b"\x01\x00\x00\x00", id="16777216"),
213	214	],
214	215	)
	216	+ IO_OPERATIONS_ON_HANDLES = pytest.mark.parametrize(
	217	+ "io_operation",
	218	+ [
	219	+ lambda handle: handle.recv(1),
	220	+ lambda handle: handle.sendall(b"Hello, world!"),
	221	+ ],
	222	+ ids=["recv", "sendall"],
	223	+ )
215	224	SIGN_ERROR_RESPONSES = pytest.mark.parametrize(
216	225	[
217	226	"key",
...	...	@@ -409,6 +418,31 @@ class Parametrize(types.SimpleNamespace):
409	418	)
410	419
411	420
	421	+class Strategies:
	422	+ """Common hypothesis data generation strategies."""
	423	+
	424	+ @staticmethod
	425	+ def select_invalid_pipe_names(candidate: str) -> bool:
	426	+ """Select candidate strings that are invalid named pipe names."""
	427	+ candidate = candidate.lower().replace("/", "\\")
	428	+ return not candidate.startswith(socketprovider.PIPE_PREFIX)
	429	+
	430	+ @classmethod
	431	+ def bad_named_pipe_names(cls) -> strategies.SearchStrategy[str]:
	432	+ r"""Generate invalid names for Windows named pipes.
	433	+
	434	+ Invalid named pipe names do not lie below the `\\.\pipe\` path.
	435	+ They also may or may not contain characters that are invalid
	436	+ in paths on The Annoying OS.
	437	+
	438	+ """
	439	+ # This is an environment string on The Annoying OS, so it must
	440	+ # at least be a non-empty C string, i.e., without embedded NULs.
	441	+ return strategies.text(
	442	+ alphabet=strategies.characters(min_codepoint=1), min_size=1
	443	+ ).filter(cls.select_invalid_pipe_names)
	444	+
	445	+
412	446	class TestStaticFunctionality:
413	447	"""Test the static functionality of the `ssh_agent` module."""
414	448
...	...	@@ -476,24 +510,6 @@ class TestShellExportScriptParsing:
476	510	callables.parse_sh_export_line(line, env_name=env_name)
477	511
478	512
479		-class TestGracefulFailureForSSHAuthSock:
480		- """Test the `posix` socket provider if `SSH_AUTH_SOCK` is unset."""
481		-
482		- def test_constructor_posix_no_ssh_auth_sock(
483		- self,
484		- skip_if_no_af_unix_support: None,
485		- ) -> None:
486		- """Abort if the running agent cannot be located on POSIX."""
487		- del skip_if_no_af_unix_support
488		- posix_handler = socketprovider.SocketProvider.resolve("posix")
489		- with pytest.MonkeyPatch.context() as monkeypatch:
490		- monkeypatch.delenv("SSH_AUTH_SOCK", raising=False)
491		- with pytest.raises(
492		- KeyError, match="SSH_AUTH_SOCK environment variable"
493		- ):
494		- posix_handler()
495		-
496		-
497	513	class TestSSHProtocolDatatypes(TestStaticFunctionality):
498	514	"""Tests for the utility functions for the SSH protocol datatypes."""
499	515
...	...	@@ -1383,3 +1399,111 @@ class TestAgentErrorResponses:
1383	1399	match=r"Malformed response\|does not match request",
1384	1400	):
1385	1401	client.query_extensions()
	1402	+
	1403	+
	1404	+class TestSSHAgentSocketProviderFailures:
	1405	+ """Test SSH agent socket provider-specific failures."""
	1406	+
	1407	+ def test_posix_provider_no_ssh_auth_sock(
	1408	+ self,
	1409	+ skip_if_no_af_unix_support: None,
	1410	+ ) -> None:
	1411	+ """For `posix`, fail if the running agent cannot be located."""
	1412	+ del skip_if_no_af_unix_support
	1413	+ provider = socketprovider.SocketProvider.resolve("posix")
	1414	+ with pytest.MonkeyPatch.context() as monkeypatch:
	1415	+ monkeypatch.delenv("SSH_AUTH_SOCK", raising=False)
	1416	+ with pytest.raises(
	1417	+ KeyError, match="SSH_AUTH_SOCK environment variable"
	1418	+ ):
	1419	+ provider()
	1420	+
	1421	+ @pytest.mark.skipif(
	1422	+ not hasattr(ctypes, "WinDLL"),
	1423	+ reason="not relevant to systems without Windows named pipe support",
	1424	+ )
	1425	+ @hypothesis.given(named_pipe_name=Strategies.bad_named_pipe_names())
	1426	+ def test_ssh_auth_sock_on_windows_provider_bad_named_pipe_name(
	1427	+ self,
	1428	+ named_pipe_name: str,
	1429	+ ) -> None:
	1430	+ """For `ssh_auth_sock_on_windows`, fail for bad named pipe names."""
	1431	+ provider = socketprovider.SocketProvider.resolve(
	1432	+ "ssh_auth_sock_on_windows"
	1433	+ )
	1434	+ with pytest.MonkeyPatch.context() as monkeypatch:
	1435	+ monkeypatch.setenv("SSH_AUTH_SOCK", named_pipe_name)
	1436	+ with pytest.raises(ValueError, match="Invalid named pipe name"):
	1437	+ provider()
	1438	+
	1439	+
	1440	+class TestWindowsNamedPipeHandle:
	1441	+ """Test the WindowsNamedPipeHandle wrapper class.
	1442	+
	1443	+ Even works on other operating systems, because the relevant parts
	1444	+ are stubbed out anyway.
	1445	+
	1446	+ """
	1447	+
	1448	+ DUMMY_ADDRESS = f"{socketprovider.PIPE_PREFIX}dummy-address"
	1449	+ IO_ON_CLOSED_HANDLE_PATTERN = re.escape(
	1450	+ socketprovider.WindowsNamedPipeHandle._IO_ON_CLOSED_FILE
	1451	+ )
	1452	+
	1453	+ @hypothesis.given(named_pipe_name=Strategies.bad_named_pipe_names())
	1454	+ def test_constructor_bad_named_pipe_name(
	1455	+ self,
	1456	+ named_pipe_name: str,
	1457	+ ) -> None:
	1458	+ """The constructor fails for bad named pipe names."""
	1459	+ with pytest.raises(ValueError, match="Invalid named pipe name"): # noqa: SIM117
	1460	+ with socketprovider.WindowsNamedPipeHandle(named_pipe_name):
	1461	+ pass
	1462	+
	1463	+ @contextlib.contextmanager
	1464	+ def stub_createfile_and_closehandle(
	1465	+ self,
	1466	+ ) -> Iterator[None]:
	1467	+ singleton = object()
	1468	+
	1469	+ def create_file(_args: Any, *_kwargs: Any) -> object:
	1470	+ return singleton
	1471	+
	1472	+ def close_handle(_args: Any, *_kwargs: Any) -> bool:
	1473	+ return True
	1474	+
	1475	+ with pytest.MonkeyPatch.context() as monkeypatch:
	1476	+ monkeypatch.setattr(socketprovider, "CreateFile", create_file)
	1477	+ monkeypatch.setattr(socketprovider, "CloseHandle", close_handle)
	1478	+ yield
	1479	+
	1480	+ @Parametrize.IO_OPERATIONS_ON_HANDLES
	1481	+ def test_closed_handle_cannot_be_operated_on(
	1482	+ self,
	1483	+ io_operation: Callable[[Any], Any],
	1484	+ ) -> None:
	1485	+ """Operating on a closed handle fails."""
	1486	+ with self.stub_createfile_and_closehandle():
	1487	+ handle = socketprovider.WindowsNamedPipeHandle(self.DUMMY_ADDRESS)
	1488	+ with handle:
	1489	+ pass # close the handle
	1490	+ with pytest.raises(
	1491	+ ValueError, match=self.IO_ON_CLOSED_HANDLE_PATTERN
	1492	+ ):
	1493	+ io_operation(handle)
	1494	+
	1495	+ @Parametrize.IO_OPERATIONS_ON_HANDLES
	1496	+ def test_handle_is_non_reentrant(
	1497	+ self,
	1498	+ io_operation: Callable[[Any], Any],
	1499	+ ) -> None:
	1500	+ """The handler cannot be sensibly used in nested `with` blocks."""
	1501	+ with self.stub_createfile_and_closehandle():
	1502	+ handle = socketprovider.WindowsNamedPipeHandle(self.DUMMY_ADDRESS)
	1503	+ with handle:
	1504	+ with handle:
	1505	+ pass # close the handle
	1506	+ with pytest.raises(
	1507	+ ValueError, match=self.IO_ON_CLOSED_HANDLE_PATTERN
	1508	+ ):
	1509	+ io_operation(handle)
1386	1510