Add tests for the SSH agent socket provider machinery (da49d3b) - derivepassphrase.git

Add tests for the SSH agent socket provider machinery

Marco Ricci commited on 2025-08-02 13:50:27
Zeige 4 geänderte Dateien mit 1416 Einfügungen und 111 Löschungen.

We add the stub agent to the list of spawnable SSH agents, since it is
always available.  This necessitates a lot of cleanup to old test suite
functionality that assumes SSH agents are available if and only if UNIX
domain sockets are supported.  The `PERMITTED_SSH_AGENTS` environment
variable now controls which (external) SSH agents will be attempted to
be spawned, if supported by the system.  Additionally, the existing
agents that can be spawned are all UNIX-only, and correctly marked as
such; this avoids attempting to spawn Pageant on The Annoying Operating
System using a UNIX Pageant command-line.

We include a stub for the entry points list/table from
`importlib.metadata`, which sadly is rather ugly due to having to
reflect both an explicit API change in Python 3.12, and an undocumented
and hairy compatibility hack in Python 3.10.

We add a `SSHAgentSocketProviderRegistryStateMachine` class for testing
socket provider registrations via a `hypothesis` state machine, and
quite a few new entries in the `TestStaticFunctionality` class relating
to locating and registering socket providers on handcrafted explicit
examples.  I have resisted the temptation to implement all registry
tests as `hypothesis` tests, in favor of having at least *some* unit
tests left over to test the registry, even if skipping
`hypothesis`-based tests for speed reasons.

Finally, we split the monolithic test
`TestCLIUtils.test_400_key_to_phrase` from the CLI tests into
a parametrized series of smaller tests.

tests/__init__.py e985b5e..43d097e
tests/conftest.py f09add8..03b9381
tests/test_derivepassphrase_cli.py 5609fb3..e6c090e
tests/test_derivepassphrase_ssh_agent.py 37a2165..a6e346d

tests/__init__.py

Zeige Datei @ da49d3b

@@ -27,7 +27,7 @@ import click.testing
 import hypothesis
 import pytest
 from hypothesis import strategies
-from typing_extensions import NamedTuple, assert_never
+from typing_extensions import NamedTuple, assert_never, overload
 
 from derivepassphrase import _types, cli, ssh_agent, vault
 from derivepassphrase._internals import cli_helpers, cli_machinery
@@ -670,11 +670,19 @@ class RunningSSHAgentInfo(NamedTuple):
 
     """
 
-    socket: str
+    socket: str | type[_types.SSHAgentSocket]
     """"""
     agent_type: KnownSSHAgent
     """"""
 
+    def require_external_address(self) -> str:  # pragma: no cover
+        if not isinstance(self.socket, str):
+            pytest.skip(
+                reason='This test requires a real, externally resolvable '
+                'address for the SSH agent socket.'
+            )
+        return self.socket
+
 
 ALL_KEYS: Mapping[str, SSHTestKey] = {
     'ed25519': SSHTestKey(
@@ -2306,6 +2314,214 @@ def phrase_from_key(
     raise KeyError(key)  # pragma: no cover
 
 
+def provider_entry_provider() -> _types.SSHAgentSocket:  # pragma: no cover
+    """A pseudo provider for a [`_types.SSHAgentSocketProviderEntry`][]."""
+    msg = 'We are not supposed to be called!'
+    raise AssertionError(msg)
+
+
+provider_entry1 = _types.SSHAgentSocketProviderEntry(
+    provider_entry_provider, 'entry1', ('entry1a', 'entry1b', 'entry1c')
+)
+"""A sample [`_types.SSHAgentSocketProviderEntry`][]."""
+
+provider_entry2 = _types.SSHAgentSocketProviderEntry(
+    provider_entry_provider, 'entry2', ('entry2d', 'entry2e')
+)
+"""A sample [`_types.SSHAgentSocketProviderEntry`][]."""
+
+posix_entry = _types.SSHAgentSocketProviderEntry(
+    socketprovider.SocketProvider.resolve('posix'), 'posix', ()
+)
+"""
+The standard [`_types.SSHAgentSocketProviderEntry`][] for the UNIX
+domain socket handler on POSIX systems.
+"""
+
+the_annoying_os_entry = _types.SSHAgentSocketProviderEntry(
+    socketprovider.SocketProvider.resolve('the_annoying_os'),
+    'the_annoying_os',
+    (),
+)
+"""
+The standard [`_types.SSHAgentSocketProviderEntry`][] for the named pipe
+handler on The Annoying Operating System.
+"""
+
+faulty_entry_callable = _types.SSHAgentSocketProviderEntry(
+    (),  # type: ignore[arg-type]
+    'tuple',
+    (),
+)
+"""
+A faulty [`_types.SSHAgentSocketProviderEntry`][]: the indicated handler
+is not a callable.
+"""
+
+faulty_entry_name_exists = _types.SSHAgentSocketProviderEntry(
+    socketprovider.SocketProvider.resolve('the_annoying_os'), 'posix', ()
+)
+"""
+A faulty [`_types.SSHAgentSocketProviderEntry`][]: the indicated handler
+is already registered with a different callable.
+"""
+
+faulty_entry_alias_exists = _types.SSHAgentSocketProviderEntry(
+    socketprovider.SocketProvider.resolve('posix'),
+    'posix',
+    ('unix_domain', 'the_annoying_os'),
+)
+"""
+A faulty [`_types.SSHAgentSocketProviderEntry`][]: the alias is already
+registered with a different callable.
+"""
+
+
+@contextlib.contextmanager
+def faked_entry_point_list(  # noqa: C901
+    additional_entry_points: Sequence[importlib.metadata.EntryPoint],
+    remove_conflicting_entries: bool = False,
+) -> Iterator[Sequence[str]]:
+    """Yield a context where additional entry points are visible.
+
+    Args:
+        additional_entry_points:
+            A sequence of entry point objects that should additionally
+            be visible.
+        remove_conflicting_entries:
+            If true, remove all names provided by the additional entry
+            points, otherwise leave them untouched.
+
+    Yields:
+        A sequence of registry names that are newly available within the
+        context.
+
+    """
+    true_entry_points = importlib.metadata.entry_points()
+    additional_entry_points = list(additional_entry_points)
+
+    if sys.version_info >= (3, 12):
+        new_entry_points = importlib.metadata.EntryPoints(
+            list(true_entry_points) + additional_entry_points
+        )
+
+        @overload
+        def mangled_entry_points(
+            *, group: None = None
+        ) -> importlib.metadata.EntryPoints: ...
+
+        @overload
+        def mangled_entry_points(
+            *, group: str
+        ) -> importlib.metadata.EntryPoints: ...
+
+        def mangled_entry_points(
+            **params: Any,
+        ) -> importlib.metadata.EntryPoints:
+            return new_entry_points.select(**params)
+
+    elif sys.version_info >= (3, 10):
+        # Compatibility concerns within importlib.metadata: depending on
+        # whether the .select() API is used, the result is either the dict
+        # of groups of points (as in < 3.10), or the EntryPoints iterable
+        # (as in >= 3.12).  So our wrapper needs to duplicate that
+        # interface.  FUN.
+        new_entry_points_dict = {
+            k: list(v) for k, v in true_entry_points.items()
+        }
+        for ep in additional_entry_points:
+            new_entry_points_dict.setdefault(ep.group, []).append(ep)
+        new_entry_points = importlib.metadata.EntryPoints([
+            ep for group in new_entry_points_dict.values() for ep in group
+        ])
+
+        @overload
+        def mangled_entry_points(
+            *, group: None = None
+        ) -> dict[
+            str,
+            list[importlib.metadata.EntryPoint]
+            | tuple[importlib.metadata.EntryPoint, ...],
+        ]: ...
+
+        @overload
+        def mangled_entry_points(
+            *, group: str
+        ) -> importlib.metadata.EntryPoints: ...
+
+        def mangled_entry_points(
+            **params: Any,
+        ) -> (
+            importlib.metadata.EntryPoints
+            | dict[
+                str,
+                list[importlib.metadata.EntryPoint]
+                | tuple[importlib.metadata.EntryPoint, ...],
+            ]
+        ):
+            return (
+                new_entry_points.select(**params)
+                if params
+                else new_entry_points_dict
+            )
+
+    else:
+        new_entry_points: dict[
+            str,
+            list[importlib.metadata.EntryPoint]
+            | tuple[importlib.metadata.EntryPoint, ...],
+        ] = {
+            group_name: list(group)
+            for group_name, group in true_entry_points.items()
+        }
+        for ep in additional_entry_points:
+            new_entry_points.setdefault(ep.group, [])
+            new_entry_points[ep.group].append(ep)
+        new_entry_points = {
+            group_name: tuple(group)
+            for group_name, group in new_entry_points.items()
+        }
+
+        @overload
+        def mangled_entry_points(
+            *, group: None = None
+        ) -> dict[str, tuple[importlib.metadata.EntryPoint, ...]]: ...
+
+        @overload
+        def mangled_entry_points(
+            *, group: str
+        ) -> tuple[importlib.metadata.EntryPoint, ...]: ...
+
+        def mangled_entry_points(
+            *, group: str | None = None
+        ) -> (
+            dict[str, tuple[importlib.metadata.EntryPoint, ...]]
+            | tuple[importlib.metadata.EntryPoint, ...]
+        ):
+            return (
+                new_entry_points.get(group, ())
+                if group is not None
+                else new_entry_points
+            )
+
+    registry = socketprovider.SocketProvider.registry
+    new_registry = registry.copy()
+    keys = [ep.load().key for ep in additional_entry_points]
+    aliases = [a for ep in additional_entry_points for a in ep.load().aliases]
+    if remove_conflicting_entries:  # pragma: no cover [unused]
+        for name in [*keys, *aliases]:
+            new_registry.pop(name, None)
+
+    with pytest.MonkeyPatch.context() as monkeypatch:
+        monkeypatch.setattr(
+            socketprovider.SocketProvider, 'registry', new_registry
+        )
+        monkeypatch.setattr(
+            importlib.metadata, 'entry_points', mangled_entry_points
+        )
+        yield (*keys, *aliases)
+
+
 @contextlib.contextmanager
 def isolated_config(
     monkeypatch: pytest.MonkeyPatch,
@@ -2525,10 +2741,11 @@ def make_file_readonly(
     group and other, and ensuring the read permission bit for user is
     set.
 
-    Unfortunately, Windows has its own rules: Set exactly(?) the read
-    permission bit for user to make the file read-only, and set
-    exactly(?) the write permission bit for user to make the file
-    read/write; all other permission bit settings are ignored.
+    Unfortunately, The Annoying OS (a.k.a. Microsoft Windows) has its
+    own rules: Set exactly(?) the read permission bit for user to make
+    the file read-only, and set exactly(?) the write permission bit for
+    user to make the file read/write; all other permission bit settings
+    are ignored.
 
     The cross-platform procedure therefore is:
 

tests/conftest.py

Zeige Datei @ da49d3b

@@ -8,7 +8,6 @@ import base64
 import contextlib
 import datetime
 import importlib
-import operator
 import os
 import shutil
 import socket
@@ -170,10 +169,10 @@ class SpawnFunc(Protocol):
         """
 
 
-def spawn_pageant(  # pragma: no cover [external]
+def spawn_pageant_on_posix(  # pragma: no cover [external]
     executable: str | None, env: dict[str, str]
 ) -> subprocess.Popen[str] | None:
-    """Spawn an isolated Pageant, if possible.
+    """Spawn an isolated (UNIX) Pageant, if possible.
 
     We attempt to detect whether Pageant is usable, i.e. whether Pageant
     has output buffering problems when announcing its authentication
@@ -195,7 +194,7 @@ def spawn_pageant(  # pragma: no cover [external]
         subprocess.
 
     """
-    if executable is None:  # pragma: no cover [external]
+    if executable is None or os.name == 'nt':  # pragma: no cover [external]
         return None
 
     # Apparently, Pageant 0.81 and lower running in debug mode does
@@ -242,10 +241,10 @@ def spawn_pageant(  # pragma: no cover [external]
     )
 
 
-def spawn_openssh_agent(  # pragma: no cover [external]
+def spawn_openssh_agent_on_posix(  # pragma: no cover [external]
     executable: str | None, env: dict[str, str]
 ) -> subprocess.Popen[str] | None:
-    """Spawn an isolated OpenSSH agent, if possible.
+    """Spawn an isolated OpenSSH agent (on UNIX), if possible.
 
     Args:
         executable:
@@ -262,7 +261,7 @@ def spawn_openssh_agent(  # pragma: no cover [external]
         subprocess.
 
     """
-    if executable is None:
+    if executable is None or os.name == 'nt':  # pragma: no cover [external]
         return None
     return subprocess.Popen(
         ['ssh-agent', '-D', '-s'],
@@ -282,11 +281,29 @@ def spawn_noop(  # pragma: no cover [unused]
     """Placeholder function. Does nothing."""
 
 
-spawn_handlers: Sequence[tuple[str, SpawnFunc, tests.KnownSSHAgent]] = [
-    ('pageant', spawn_pageant, tests.KnownSSHAgent.Pageant),
-    ('ssh-agent', spawn_openssh_agent, tests.KnownSSHAgent.OpenSSHAgent),
-    ('(system)', spawn_noop, tests.KnownSSHAgent.UNKNOWN),
-]
+spawn_handlers: dict[str, tuple[str, SpawnFunc, tests.KnownSSHAgent]] = {
+    'pageant': (
+        'pageant',
+        spawn_pageant_on_posix,
+        tests.KnownSSHAgent.Pageant,
+    ),
+    'ssh-agent': (
+        'ssh-agent',
+        spawn_openssh_agent_on_posix,
+        tests.KnownSSHAgent.OpenSSHAgent,
+    ),
+    'stub_agent': (
+        'stub_agent',
+        spawn_noop,
+        tests.KnownSSHAgent.StubbedSSHAgent,
+    ),
+    'stub_agent_with_extensions': (
+        'stub_agent_with_extensions',
+        spawn_noop,
+        tests.KnownSSHAgent.StubbedSSHAgent,
+    ),
+    '(system)': ('(system)', spawn_noop, tests.KnownSSHAgent.UNKNOWN),
+}
 """
 The standard registry of agent spawning functions.
 """
@@ -377,7 +394,12 @@ def spawn_named_agent(
     ssh_auth_sock = agent_env.pop('SSH_AUTH_SOCK', None)
     proc = spawn_func(executable=shutil.which(exec_name), env=agent_env)
     with exit_stack:
-        if spawn_func is spawn_noop:
+        if (
+            spawn_func is spawn_noop
+            and agent_type == tests.KnownSSHAgent.StubbedSSHAgent
+        ):
+            ssh_auth_sock = None
+        elif spawn_func is spawn_noop:
             ssh_auth_sock = os.environ['SSH_AUTH_SOCK']
         elif proc is None:  # pragma: no cover [external]
             err_msg = f'Cannot spawn usable {exec_name}'
@@ -404,11 +426,51 @@ def spawn_named_agent(
                 err_msg = f'Cannot parse agent output: {pid_line!r}'
                 raise CannotSpawnError(err_msg)
         monkeypatch = exit_stack.enter_context(pytest.MonkeyPatch.context())
+        if ssh_auth_sock is not None:
             monkeypatch.setenv('SSH_AUTH_SOCK', ssh_auth_sock)
             client = exit_stack.enter_context(
                 ssh_agent.SSHAgentClient.ensure_agent_subcontext()
             )
+        else:
+            monkeypatch.setenv(
+                'SSH_AUTH_SOCK', tests.StubbedSSHAgentSocketWithAddress.ADDRESS
+            )
+            monkeypatch.setattr(
+                ssh_agent.SSHAgentClient,
+                'SOCKET_PROVIDERS',
+                ['stub_with_address_and_deterministic_dsa']
+                if exec_name == 'stub_agent_with_extensions'
+                else ['stub_with_address'],
+            )
+            client = exit_stack.enter_context(
+                ssh_agent.SSHAgentClient.ensure_agent_subcontext(
+                    tests.StubbedSSHAgentSocketWithAddressAndDeterministicDSA()
+                    if exec_name == 'stub_agent_with_extensions'
+                    else tests.StubbedSSHAgentSocketWithAddress()
+                )
+            )
+        # We sanity-test the connected SSH agent if it is not one of our
+        # test agents, because allowing the user to run the test suite
+        # with a clearly faulty agent would likely and unfairly
+        # misattribute the agent's protocol violations to our test
+        # suite.  On the flip side, for our own test agents, the
+        # correctness tests are part of the test suite, so we don't want
+        # the *setup* code here to already decide that the agent is
+        # unsuitable. Therefore, do a sanity check if and only if the
+        # agent is not one of our test agents, and if the check fails,
+        # skip this agent.
+        if (
+            agent_type != tests.KnownSSHAgent.StubbedSSHAgent
+        ):  # pragma: no cover [external]
+            try:
                 client.list_keys()  # sanity test
+            except (
+                EOFError,
+                OSError,
+                ssh_agent.SSHAgentFailedError,
+            ) as exc:  # pragma: no cover [failsafe]
+                msg = f'agent failed the "list keys" sanity test: {exc!r}'
+                raise CannotSpawnError(msg) from exc
         yield tests.SpawnedSSHAgentInfo(
             agent_type, client, spawn_func is not spawn_noop
         )
@@ -417,9 +479,59 @@ def spawn_named_agent(
     )
 
 
+def is_agent_permitted(
+    agent_type: tests.KnownSSHAgent,
+) -> bool:  # pragma: no cover [external]
+    """May the given SSH agent be spawned by the test harness?
+
+    If the environment variable `PERMITTED_SSH_AGENTS` is given, it
+    names a comma-separated list of known SSH agent names that the test
+    harness may spawn.  Invalid names are silently ignored.  If not
+    given, or empty, then any agent may be spawned.
+
+    (To not allow any agent to be spawned, specify a single comma as the
+    list.  But see below.)
+
+    The stubbed agents cannot be restricted in this manner, as the test
+    suite depends on their availability.
+
+    """
+    if not os.environ.get('PERMITTED_SSH_AGENTS'):
+        return True
+    permitted_agents = {tests.KnownSSHAgent.StubbedSSHAgent}
+    permitted_agents.update({
+        tests.KnownSSHAgent(x)
+        for x in os.environ['PERMITTED_SSH_AGENTS'].split(',')
+        if x in tests.KnownSSHAgent.__members__
+    })
+    return agent_type in permitted_agents
+
+
+spawn_handlers_params: list[Sequence] = []
+"""
+The standard registry of agent spawning functions, annotated as
+a `pytest` parameter set.  (In particular, this includes the conditional
+skip marks.)  Used by some test fixtures.
+"""
+for key, handler in spawn_handlers.items():
+    marks = [
+        pytest.mark.skipif(
+            not is_agent_permitted(handler[2]),
+            reason='agent excluded via PERMITTED_AGENTS environment variable',
+        ),
+    ]
+    if key in {'pageant', 'ssh-agent', '(system)'}:
+        marks.append(
+            pytest.mark.skipif(
+                not hasattr(socket, 'AF_UNIX'),
+                reason='socket module does not support AF_UNIX',
+            )
+        )
+    spawn_handlers_params.append(pytest.param(handler, id=key, marks=marks))
+
+
 @pytest.fixture
 def running_ssh_agent(  # pragma: no cover [external]
-    skip_if_no_af_unix_support: None,
 ) -> Iterator[tests.RunningSSHAgentInfo]:
     """Ensure a running SSH agent, if possible, as a pytest fixture.
 
@@ -441,7 +553,30 @@ def running_ssh_agent(  # pragma: no cover [external]
             If no agent is running or can be spawned, skip this test.
 
     """
-    del skip_if_no_af_unix_support
+
+    def prepare_environment(
+        agent_type: tests.KnownSSHAgent,
+    ) -> Iterator[tests.RunningSSHAgentInfo]:
+        with pytest.MonkeyPatch.context() as monkeypatch:
+            if agent_type == tests.KnownSSHAgent.StubbedSSHAgent:
+                monkeypatch.setattr(
+                    ssh_agent.SSHAgentClient,
+                    'SOCKET_PROVIDERS',
+                    ['stub_with_address'],
+                )
+                monkeypatch.setenv(
+                    'SSH_AUTH_SOCK',
+                    tests.StubbedSSHAgentSocketWithAddress.ADDRESS,
+                )
+                yield tests.RunningSSHAgentInfo(
+                    tests.StubbedSSHAgentSocketWithAddress,
+                    tests.KnownSSHAgent.StubbedSSHAgent,
+                )
+            else:
+                yield tests.RunningSSHAgentInfo(
+                    os.environ['SSH_AUTH_SOCK'],
+                    agent_type,
+                )
 
     with pytest.MonkeyPatch.context() as monkeypatch:
         # pytest's fixture system does not seem to guarantee that
@@ -455,30 +590,30 @@ def running_ssh_agent(  # pragma: no cover [external]
             monkeypatch.setenv('SSH_AUTH_SOCK', startup_ssh_auth_sock)
         else:
             monkeypatch.delenv('SSH_AUTH_SOCK', raising=False)
-        for exec_name, spawn_func, agent_type in spawn_handlers:
+        for exec_name, spawn_func, agent_type in spawn_handlers.values():
+            if not is_agent_permitted(agent_type):
+                continue
             try:
                 for _agent_info in spawn_named_agent(
                     exec_name, spawn_func, agent_type
                 ):
-                    yield tests.RunningSSHAgentInfo(
-                        os.environ['SSH_AUTH_SOCK'], agent_type
-                    )
+                    yield from prepare_environment(agent_type)
             except (KeyError, OSError, CannotSpawnError):
                 continue
             return
         pytest.skip('No SSH agent running or spawnable')
 
 
-@pytest.fixture(params=spawn_handlers, ids=operator.itemgetter(0))
+@pytest.fixture(params=spawn_handlers_params)
 def spawn_ssh_agent(
     request: pytest.FixtureRequest,
-    skip_if_no_af_unix_support: None,
 ) -> Iterator[tests.SpawnedSSHAgentInfo]:  # pragma: no cover [external]
     """Spawn an isolated SSH agent, if possible, as a pytest fixture.
 
     Spawn a new SSH agent isolated from other SSH use by other
-    processes, if possible.  We know how to spawn OpenSSH's agent and
-    PuTTY's Pageant, and the "(system)" fallback agent.
+    processes, if possible.  We know how to spawn OpenSSH's agent (on
+    UNIX) and PuTTY's Pageant (on UNIX), the stubbed agents, and the
+    "(system)" fallback agent.
 
     Yields:
         A [named tuple][collections.namedtuple] containing information
@@ -491,7 +626,7 @@ def spawn_ssh_agent(
             If the agent cannot be spawned, skip this test.
 
     """
-    del skip_if_no_af_unix_support
+
     with pytest.MonkeyPatch.context() as monkeypatch:
         # pytest's fixture system does not seem to guarantee that
         # environment variables are set up correctly if nested and
@@ -550,6 +685,13 @@ def ssh_agent_client_with_test_keys_loaded(  # noqa: C901
     agent_type, client, isolated = spawn_ssh_agent
     successfully_loaded_keys: set[str] = set()
 
+    # This fixture relies on `spawn_ssh_agent`, which in turn uses
+    # `spawn_named_agent` to, well, spawn agents.  `spawn_named_agent`
+    # runs sanity tests on the agents it spawns, via
+    # `SSHAgentClient.list_keys`.  There is thus little point in
+    # repeating the very same sanity test here, on an agent that was
+    # already sanity-tested.
+
     def prepare_payload(
         payload: bytes | bytearray,
         *,

tests/test_derivepassphrase_cli.py

Zeige Datei @ da49d3b

@@ -7,6 +7,7 @@ from __future__ import annotations
 import base64
 import contextlib
 import copy
+import ctypes
 import enum
 import errno
 import io
@@ -39,6 +40,7 @@ from derivepassphrase._internals import (
     cli_machinery,
     cli_messages,
 )
+from derivepassphrase.ssh_agent import socketprovider
 
 if TYPE_CHECKING:
     import multiprocessing
@@ -537,6 +539,258 @@ def zsh_format(item: click.shell_completion.CompletionItem) -> str:
     return f'{item.type}\n{value}\n{help_}'
 
 
+class ListKeysAction(str, enum.Enum):
+    """Test fixture settings for [`ssh_agent.SSHAgentClient.list_keys`][].
+
+    Attributes:
+        EMPTY: Return an empty key list.
+        FAIL: Raise an [`ssh_agent.SSHAgentFailedError`][].
+        FAIL_RUNTIME: Raise an [`ssh_agent.TrailingDataError`][].
+
+    """
+
+    EMPTY = enum.auto()
+    """"""
+    FAIL = enum.auto()
+    """"""
+    FAIL_RUNTIME = enum.auto()
+    """"""
+
+    def __call__(self, *_args: Any, **_kwargs: Any) -> Any:
+        """Execute the respective action."""
+        # TODO(the-13th-letter): Rewrite using structural pattern
+        # matching.
+        # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
+        if self == self.EMPTY:
+            return []
+        if self == self.FAIL:
+            raise ssh_agent.SSHAgentFailedError(
+                _types.SSH_AGENT.FAILURE.value, b''
+            )
+        if self == self.FAIL_RUNTIME:
+            raise ssh_agent.TrailingDataError()
+        raise AssertionError()
+
+
+class SignAction(str, enum.Enum):
+    """Test fixture settings for [`ssh_agent.SSHAgentClient.sign`][].
+
+    Attributes:
+        FAIL: Raise an [`ssh_agent.SSHAgentFailedError`][].
+        FAIL_RUNTIME: Raise an [`ssh_agent.TrailingDataError`][].
+
+    """
+
+    FAIL = enum.auto()
+    """"""
+    FAIL_RUNTIME = enum.auto()
+    """"""
+
+    def __call__(self, *_args: Any, **_kwargs: Any) -> Any:
+        """Execute the respective action."""
+        # TODO(the-13th-letter): Rewrite using structural pattern
+        # matching.
+        # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
+        if self == self.FAIL:
+            raise ssh_agent.SSHAgentFailedError(
+                _types.SSH_AGENT.FAILURE.value, b''
+            )
+        if self == self.FAIL_RUNTIME:
+            raise ssh_agent.TrailingDataError()
+        raise AssertionError()
+
+
+class SocketAddressAction(str, enum.Enum):
+    """Test fixture settings for the SSH agent socket address.
+
+    Attributes:
+        MANGLE_ANNOYING_OS_NAMED_PIPE:
+            Mangle the address for the Annoying OS named pipe endpoint.
+        MANGLE_SSH_AUTH_SOCK:
+            Mangle the address for the UNIX domain socket (the
+            `SSH_AUTH_SOCK` environment variable).
+        UNSET_ANNOYING_OS_NAMED_PIPE:
+            Unset the address for the Annoying OS named pipe endpoint.
+        UNSET_SSH_AUTH_SOCK:
+            Unset the `SSH_AUTH_SOCK` environment variable (the address
+            for the UNIX domain socket).
+
+    """
+
+    MANGLE_ANNOYING_OS_NAMED_PIPE = enum.auto()
+    """"""
+    MANGLE_SSH_AUTH_SOCK = enum.auto()
+    """"""
+    UNSET_ANNOYING_OS_NAMED_PIPE = enum.auto()
+    """"""
+    UNSET_SSH_AUTH_SOCK = enum.auto()
+    """"""
+
+    def __call__(
+        self, monkeypatch: pytest.MonkeyPatch, /, *_args: Any, **_kwargs: Any
+    ) -> None:
+        """Execute the respective action."""
+        # TODO(the-13th-letter): Rewrite using structural pattern
+        # matching.
+        # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
+        if self in {
+            self.MANGLE_ANNOYING_OS_NAMED_PIPE,
+            self.UNSET_ANNOYING_OS_NAMED_PIPE,
+        }:  # pragma: no cover [unused]
+            pass
+        elif self == self.MANGLE_SSH_AUTH_SOCK:
+            monkeypatch.setenv(
+                'SSH_AUTH_SOCK', os.environ['SSH_AUTH_SOCK'] + '~'
+            )
+        elif self == self.UNSET_SSH_AUTH_SOCK:
+            monkeypatch.delenv('SSH_AUTH_SOCK', raising=False)
+        else:
+            raise AssertionError()
+
+
+class SystemSupportAction(str, enum.Enum):
+    """Test fixture settings for [`ssh_agent.SSHAgentClient`][] system support.
+
+    Attributes:
+        UNSET_AF_UNIX:
+            Ensure lack of support for UNIX domain sockets.
+        UNSET_AF_UNIX_AND_ENSURE_USE:
+            Ensure lack of support for UNIX domain sockets, and that the
+            agent will use this socket provider.
+        UNSET_NATIVE:
+            Ensure both `UNSET_AF_UNIX` and `UNSET_WINDLL`.
+        UNSET_NATIVE_AND_ENSURE_USE:
+            Ensure both `UNSET_AF_UNIX` and `UNSET_WINDLL`, and that the
+            agent will use the native socket provider.
+        UNSET_PROVIDER_LIST:
+            Ensure an empty list of SSH agent socket providers.
+        UNSET_WINDLL:
+            Ensure lack of support for The Annoying OS named pipes.
+        UNSET_WINDLL_AND_ENSURE_USE:
+            Ensure lack of support for The Annoying OS named pipes, and
+            that the agent will use this socket provider.
+
+    """
+
+    UNSET_AF_UNIX = enum.auto()
+    """"""
+    UNSET_AF_UNIX_AND_ENSURE_USE = enum.auto()
+    """"""
+    UNSET_NATIVE = enum.auto()
+    """"""
+    UNSET_NATIVE_AND_ENSURE_USE = enum.auto()
+    """"""
+    UNSET_PROVIDER_LIST = enum.auto()
+    """"""
+    UNSET_WINDLL = enum.auto()
+    """"""
+    UNSET_WINDLL_AND_ENSURE_USE = enum.auto()
+    """"""
+
+    def __call__(
+        self, monkeypatch: pytest.MonkeyPatch, /, *_args: Any, **_kwargs: Any
+    ) -> None:
+        """Execute the respective action.
+
+        Args:
+            monkeypatch: The current monkeypatch context.
+
+        """
+        # TODO(the-13th-letter): Rewrite using structural pattern
+        # matching.
+        # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
+        if self == self.UNSET_PROVIDER_LIST:
+            monkeypatch.setattr(
+                ssh_agent.SSHAgentClient, 'SOCKET_PROVIDERS', []
+            )
+        elif self in {self.UNSET_NATIVE, self.UNSET_NATIVE_AND_ENSURE_USE}:
+            self.check_or_ensure_use(
+                'native',
+                monkeypatch=monkeypatch,
+                ensure_use=(self == self.UNSET_NATIVE_AND_ENSURE_USE),
+            )
+            monkeypatch.delattr(socket, 'AF_UNIX', raising=False)
+            monkeypatch.delattr(ctypes, 'WinDLL', raising=False)
+            monkeypatch.delattr(ctypes, 'windll', raising=False)
+        elif self in {self.UNSET_AF_UNIX, self.UNSET_AF_UNIX_AND_ENSURE_USE}:
+            self.check_or_ensure_use(
+                'posix',
+                monkeypatch=monkeypatch,
+                ensure_use=(self == self.UNSET_AF_UNIX_AND_ENSURE_USE),
+            )
+            monkeypatch.delattr(socket, 'AF_UNIX', raising=False)
+        elif self in {self.UNSET_WINDLL, self.UNSET_WINDLL_AND_ENSURE_USE}:
+            self.check_or_ensure_use(
+                'the_annoying_os',
+                monkeypatch=monkeypatch,
+                ensure_use=(self == self.UNSET_WINDLL_AND_ENSURE_USE),
+            )
+            monkeypatch.delattr(ctypes, 'WinDLL', raising=False)
+            monkeypatch.delattr(ctypes, 'windll', raising=False)
+        else:
+            raise AssertionError()
+
+    @staticmethod
+    def check_or_ensure_use(
+        provider: str, /, *, monkeypatch: pytest.MonkeyPatch, ensure_use: bool
+    ) -> None:
+        """Check that the named SSH agent socket provider will be used.
+
+        Either ensure that the socket provider will definitely be used,
+        or, upon detecting that it won't be used, skip the test.
+
+        Args:
+            provider:
+                The provider to check for.
+            ensure_use:
+                If true, ensure that the socket provider will definitely
+                be used.  If false, then check for whether it will be
+                used, and skip this test if not.
+            monkeypatch:
+                The monkeypatch context within which the fixture
+                adjustments should be executed.
+
+        """
+        if ensure_use:
+            monkeypatch.setattr(
+                ssh_agent.SSHAgentClient, 'SOCKET_PROVIDERS', [provider]
+            )
+        else:  # pragma: no cover [external]
+            # This branch operates completely on instrumented or on
+            # externally defined, non-deterministic state.
+            intended: (
+                _types.SSHAgentSocketProvider
+                | socketprovider.NoSuchProviderError
+                | None
+            )
+            try:
+                intended = socketprovider.SocketProvider.resolve(provider)
+            except socketprovider.NoSuchProviderError as exc:
+                intended = exc
+            except NotImplementedError:
+                intended = None
+            actual: (
+                _types.SSHAgentSocketProvider
+                | socketprovider.NoSuchProviderError
+                | None
+            )
+            for name in ssh_agent.SSHAgentClient.SOCKET_PROVIDERS:
+                try:
+                    actual = socketprovider.SocketProvider.resolve(name)
+                except socketprovider.NoSuchProviderError as exc:
+                    actual = exc
+                except NotImplementedError:
+                    continue
+                break
+            else:
+                actual = None
+            if intended != actual:
+                pytest.skip(
+                    f'{provider!r} SSH agent socket provider '
+                    f'is not currently in use'
+                )
+
+
 class Parametrize(types.SimpleNamespace):
     """Common test parametrizations."""
 
@@ -1308,6 +1562,97 @@ class Parametrize(types.SimpleNamespace):
     KEY_INDEX = pytest.mark.parametrize(
         'key_index', [1, 2, 3], ids=lambda i: f'index{i}'
     )
+    KEY_TO_PHRASE_SETTINGS = pytest.mark.parametrize(
+        [
+            'list_keys_action',
+            'address_action',
+            'system_support_action',
+            'sign_action',
+            'pattern',
+        ],
+        [
+            pytest.param(
+                ListKeysAction.EMPTY,
+                None,
+                None,
+                SignAction.FAIL,
+                'not loaded into the agent',
+                id='key-not-loaded',
+            ),
+            pytest.param(
+                ListKeysAction.FAIL,
+                None,
+                None,
+                SignAction.FAIL,
+                'SSH agent failed to or refused to',
+                id='list-keys-refused',
+            ),
+            pytest.param(
+                ListKeysAction.FAIL_RUNTIME,
+                None,
+                None,
+                SignAction.FAIL,
+                'SSH agent failed to or refused to',
+                id='list-keys-protocol-error',
+            ),
+            pytest.param(
+                None,
+                SocketAddressAction.UNSET_SSH_AUTH_SOCK,
+                None,
+                SignAction.FAIL,
+                'Cannot find any running SSH agent',
+                id='agent-address-missing',
+            ),
+            pytest.param(
+                None,
+                SocketAddressAction.MANGLE_SSH_AUTH_SOCK,
+                None,
+                SignAction.FAIL,
+                'Cannot connect to the SSH agent',
+                id='agent-address-mangled',
+            ),
+            pytest.param(
+                None,
+                None,
+                SystemSupportAction.UNSET_NATIVE,
+                SignAction.FAIL,
+                'does not support communicating with it',
+                id='no-agent-support',
+            ),
+            pytest.param(
+                None,
+                None,
+                SystemSupportAction.UNSET_PROVIDER_LIST,
+                SignAction.FAIL,
+                'does not support communicating with it',
+                id='no-agent-support',
+            ),
+            pytest.param(
+                None,
+                None,
+                SystemSupportAction.UNSET_AF_UNIX_AND_ENSURE_USE,
+                SignAction.FAIL,
+                'does not support communicating with it',
+                id='no-agent-support',
+            ),
+            pytest.param(
+                None,
+                None,
+                SystemSupportAction.UNSET_WINDLL_AND_ENSURE_USE,
+                SignAction.FAIL,
+                'does not support communicating with it',
+                id='no-agent-support',
+            ),
+            pytest.param(
+                None,
+                None,
+                None,
+                SignAction.FAIL_RUNTIME,
+                'violates the communication protocol',
+                id='sign-violates-protocol',
+            ),
+        ],
+    )
     UNICODE_NORMALIZATION_ERROR_INPUTS = pytest.mark.parametrize(
         ['main_config', 'command_line', 'input', 'error_message'],
         [
@@ -2267,6 +2612,7 @@ class TestCLI:
         key_index: int,
     ) -> None:
         """A command-line SSH key will override the configured key."""
+        del running_ssh_agent
         runner = tests.CliRunner(mix_stderr=False)
         # TODO(the-13th-letter): Rewrite using parenthesized
         # with-statements.
@@ -2280,7 +2626,6 @@ class TestCLI:
                     vault_config=config,
                 )
             )
-            monkeypatch.setenv('SSH_AUTH_SOCK', running_ssh_agent.socket)
             monkeypatch.setattr(
                 ssh_agent.SSHAgentClient, 'list_keys', tests.list_keys
             )
@@ -2302,6 +2647,7 @@ class TestCLI:
         running_ssh_agent: tests.RunningSSHAgentInfo,
     ) -> None:
         """A command-line passphrase will override the configured key."""
+        del running_ssh_agent
         runner = tests.CliRunner(mix_stderr=False)
         # TODO(the-13th-letter): Rewrite using parenthesized
         # with-statements.
@@ -2323,7 +2669,6 @@ class TestCLI:
                     },
                 )
             )
-            monkeypatch.setenv('SSH_AUTH_SOCK', running_ssh_agent.socket)
             monkeypatch.setattr(
                 ssh_agent.SSHAgentClient, 'list_keys', tests.list_keys
             )
@@ -2352,6 +2697,7 @@ class TestCLI:
         command_line: list[str],
     ) -> None:
         """Configuring a passphrase atop an SSH key works, but warns."""
+        del running_ssh_agent
         runner = tests.CliRunner(mix_stderr=False)
         # TODO(the-13th-letter): Rewrite using parenthesized
         # with-statements.
@@ -2365,7 +2711,6 @@ class TestCLI:
                     vault_config=config,
                 )
             )
-            monkeypatch.setenv('SSH_AUTH_SOCK', running_ssh_agent.socket)
             monkeypatch.setattr(
                 ssh_agent.SSHAgentClient, 'list_keys', tests.list_keys
             )
@@ -3629,8 +3974,10 @@ class TestCLI:
 
     def test_225a_store_config_fail_manual_no_ssh_key_selection(
         self,
+        running_ssh_agent: tests.RunningSSHAgentInfo,
     ) -> None:
         """Not selecting an SSH key during `--config --key` fails."""
+        del running_ssh_agent
         runner = tests.CliRunner(mix_stderr=False)
         # TODO(the-13th-letter): Rewrite using parenthesized
         # with-statements.
@@ -3644,27 +3991,33 @@ class TestCLI:
                     vault_config={'global': {'phrase': 'abc'}, 'services': {}},
                 )
             )
-            custom_error = 'custom error message'
 
-            def raiser(*_args: Any, **_kwargs: Any) -> None:
-                raise RuntimeError(custom_error)
+            def prompt_for_selection(*_args: Any, **_kwargs: Any) -> NoReturn:
+                raise IndexError(cli_helpers.EMPTY_SELECTION)
 
-            monkeypatch.setattr(cli_helpers, 'select_ssh_key', raiser)
+            monkeypatch.setattr(
+                cli_helpers, 'prompt_for_selection', prompt_for_selection
+            )
+            # Also patch the list of suitable SSH keys, lest we be at
+            # the mercy of whatever SSH agent may be running.
+            monkeypatch.setattr(
+                cli_helpers, 'get_suitable_ssh_keys', tests.suitable_ssh_keys
+            )
             result = runner.invoke(
                 cli.derivepassphrase_vault,
                 ['--key', '--config'],
                 catch_exceptions=False,
             )
-        assert result.error_exit(error=custom_error), (
+        assert result.error_exit(error='the user aborted the request'), (
             'expected error exit and known error message'
         )
 
     def test_225b_store_config_fail_manual_no_ssh_agent(
         self,
-        skip_if_no_af_unix_support: None,
+        running_ssh_agent: tests.RunningSSHAgentInfo,
     ) -> None:
         """Not running an SSH agent during `--config --key` fails."""
-        del skip_if_no_af_unix_support
+        del running_ssh_agent
         runner = tests.CliRunner(mix_stderr=False)
         # TODO(the-13th-letter): Rewrite using parenthesized
         # with-statements.
@@ -3693,7 +4046,7 @@ class TestCLI:
         running_ssh_agent: tests.RunningSSHAgentInfo,
     ) -> None:
         """Not running a reachable SSH agent during `--config --key` fails."""
-        del running_ssh_agent
+        running_ssh_agent.require_external_address()
         runner = tests.CliRunner(mix_stderr=False)
         # TODO(the-13th-letter): Rewrite using parenthesized
         # with-statements.
@@ -4273,6 +4626,7 @@ class TestCLI:
 
     def test_400_missing_af_unix_support(
         self,
+        caplog: pytest.LogCaptureFixture,
     ) -> None:
         """Querying the SSH agent without `AF_UNIX` support fails."""
         runner = tests.CliRunner(mix_stderr=False)
@@ -4291,6 +4645,9 @@ class TestCLI:
             monkeypatch.setenv(
                 'SSH_AUTH_SOCK', "the value doesn't even matter"
             )
+            monkeypatch.setattr(
+                ssh_agent.SSHAgentClient, 'SOCKET_PROVIDERS', ['posix']
+            )
             monkeypatch.delattr(socket, 'AF_UNIX', raising=False)
             result = runner.invoke(
                 cli.derivepassphrase_vault,
@@ -4300,6 +4657,10 @@ class TestCLI:
         assert result.error_exit(
             error='does not support communicating with it'
         ), 'expected error exit and known error message'
+        assert tests.warning_emitted(
+            'Cannot connect to an SSH agent via UNIX domain sockets',
+            caplog.record_tuples,
+        ), 'expected known warning message in stderr'
 
 
 class TestCLIUtils:
@@ -5122,20 +5483,26 @@ Will replace with spam, okay? (Please say "y" or "n".): Boo.
     ) -> None:
         """[`cli_helpers.get_suitable_ssh_keys`][] works."""
         with pytest.MonkeyPatch.context() as monkeypatch:
-            monkeypatch.setenv('SSH_AUTH_SOCK', running_ssh_agent.socket)
             monkeypatch.setattr(
                 ssh_agent.SSHAgentClient, 'list_keys', tests.list_keys
             )
-            hint: ssh_agent.SSHAgentClient | socket.socket | None
+            hint: ssh_agent.SSHAgentClient | _types.SSHAgentSocket | None
             # TODO(the-13th-letter): Rewrite using structural pattern
             # matching.
             # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
             if conn_hint == 'client':
                 hint = ssh_agent.SSHAgentClient()
             elif conn_hint == 'socket':
+                if isinstance(
+                    running_ssh_agent.socket, str
+                ):  # pragma: no cover
+                    if not hasattr(socket, 'AF_UNIX'):
+                        pytest.skip('socket module does not support AF_UNIX')
                     # socket.AF_UNIX is not defined everywhere.
                     hint = socket.socket(family=socket.AF_UNIX)  # type: ignore[attr-defined]
                     hint.connect(running_ssh_agent.socket)
+                else:  # pragma: no cover
+                    hint = running_ssh_agent.socket()
             else:
                 assert conn_hint == 'none'
                 hint = None
@@ -5151,10 +5518,15 @@ Will replace with spam, okay? (Please say "y" or "n".): Boo.
                     'exception querying suitable SSH keys'
                 )
 
+    @Parametrize.KEY_TO_PHRASE_SETTINGS
     def test_400_key_to_phrase(
         self,
-        skip_if_no_af_unix_support: None,
         ssh_agent_client_with_test_keys_loaded: ssh_agent.SSHAgentClient,
+        list_keys_action: ListKeysAction | None,
+        system_support_action: SystemSupportAction | None,
+        address_action: SocketAddressAction | None,
+        sign_action: SignAction,
+        pattern: str,
     ) -> None:
         """All errors in [`cli_helpers.key_to_phrase`][] are handled."""
 
@@ -5167,44 +5539,23 @@ Will replace with spam, okay? (Please say "y" or "n".): Boo.
         def err(*args: Any, **_kwargs: Any) -> NoReturn:
             raise ErrCallback(*args, **_kwargs)
 
-        def fail(*_args: Any, **_kwargs: Any) -> Any:
-            raise ssh_agent.SSHAgentFailedError(
-                _types.SSH_AGENT.FAILURE.value,
-                b'',
-            )
-
-        def fail_runtime(*_args: Any, **_kwargs: Any) -> Any:
-            raise ssh_agent.TrailingDataError()
-
-        del skip_if_no_af_unix_support
         with pytest.MonkeyPatch.context() as monkeypatch:
-            monkeypatch.setattr(ssh_agent.SSHAgentClient, 'sign', fail)
             loaded_keys = list(
                 ssh_agent_client_with_test_keys_loaded.list_keys()
             )
             loaded_key = base64.standard_b64encode(loaded_keys[0][0])
-            with monkeypatch.context() as mp:
-                mp.setattr(
-                    ssh_agent.SSHAgentClient,
-                    'list_keys',
-                    lambda *_a, **_kw: [],
+            monkeypatch.setattr(ssh_agent.SSHAgentClient, 'sign', sign_action)
+            if list_keys_action:
+                monkeypatch.setattr(
+                    ssh_agent.SSHAgentClient, 'list_keys', list_keys_action
                 )
-                with pytest.raises(
-                    ErrCallback, match='not loaded into the agent'
-                ):
-                    cli_helpers.key_to_phrase(loaded_key, error_callback=err)
-            with monkeypatch.context() as mp:
-                mp.setattr(ssh_agent.SSHAgentClient, 'list_keys', fail)
-                with pytest.raises(
-                    ErrCallback, match='SSH agent failed to or refused to'
-                ):
-                    cli_helpers.key_to_phrase(loaded_key, error_callback=err)
-            with monkeypatch.context() as mp:
-                mp.setattr(ssh_agent.SSHAgentClient, 'list_keys', fail_runtime)
-                with pytest.raises(
-                    ErrCallback, match='SSH agent failed to or refused to'
-                ) as excinfo:
+            if address_action:
+                address_action(monkeypatch)
+            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)
+            if list_keys_action == ListKeysAction.FAIL_RUNTIME:
                 assert excinfo.value.kwargs
                 assert isinstance(
                     excinfo.value.kwargs['exc_info'],
@@ -5215,30 +5566,6 @@ Will replace with spam, okay? (Please say "y" or "n".): Boo.
                     excinfo.value.kwargs['exc_info'].__context__,
                     ssh_agent.TrailingDataError,
                 )
-            with monkeypatch.context() as mp:
-                mp.delenv('SSH_AUTH_SOCK', raising=True)
-                with pytest.raises(
-                    ErrCallback, match='Cannot find any running SSH agent'
-                ):
-                    cli_helpers.key_to_phrase(loaded_key, error_callback=err)
-            with monkeypatch.context() as mp:
-                mp.setenv('SSH_AUTH_SOCK', os.environ['SSH_AUTH_SOCK'] + '~')
-                with pytest.raises(
-                    ErrCallback, match='Cannot connect to the SSH agent'
-                ):
-                    cli_helpers.key_to_phrase(loaded_key, error_callback=err)
-            with monkeypatch.context() as mp:
-                mp.delattr(socket, 'AF_UNIX', raising=True)
-                with pytest.raises(
-                    ErrCallback, match='does not support communicating with it'
-                ):
-                    cli_helpers.key_to_phrase(loaded_key, error_callback=err)
-            with monkeypatch.context() as mp:
-                mp.setattr(ssh_agent.SSHAgentClient, 'sign', fail_runtime)
-                with pytest.raises(
-                    ErrCallback, match='violates the communication protocol'
-                ):
-                    cli_helpers.key_to_phrase(loaded_key, error_callback=err)
 
 
 # TODO(the-13th-letter): Remove this class in v1.0.
@@ -6654,7 +6981,7 @@ class FakeConfigurationMutexStateMachine(stateful.RuleBasedStateMachine):
         ] = {}
         true_configs: dict[int, _types.VaultConfig] = {}
         true_results: dict[int, tuple[bool, str | None, str | None]] = {}
-        timeout = 5
+        timeout = 30  # Hopefully slow enough to accomodate The Annoying OS.
         actions = self.actions
         mp = multiprocessing.get_context()
         # Coverage tracking writes coverage data to the current working

tests/test_derivepassphrase_ssh_agent.py

Zeige Datei @ da49d3b

@@ -9,11 +9,13 @@ from __future__ import annotations
 import base64
 import contextlib
 import errno
+import importlib.metadata
 import io
 import os
 import pathlib
 import re
 import socket
+import sys
 import types
 from typing import TYPE_CHECKING
 
@@ -21,7 +23,7 @@ import click
 import click.testing
 import hypothesis
 import pytest
-from hypothesis import strategies
+from hypothesis import stateful, strategies
 
 import tests
 from derivepassphrase import _types, ssh_agent, vault
@@ -31,10 +33,83 @@ from derivepassphrase.ssh_agent import socketprovider
 if TYPE_CHECKING:
     from collections.abc import Iterable
 
-    from typing_extensions import Any, Buffer
+    from typing_extensions import Any, Buffer, Literal
+
+if sys.version_info < (3, 11):
+    from exceptiongroup import ExceptionGroup
 
 
 class Parametrize(types.SimpleNamespace):
+    BAD_ENTRY_POINTS = pytest.mark.parametrize(
+        'additional_entry_points',
+        [
+            pytest.param(
+                [
+                    importlib.metadata.EntryPoint(
+                        name=tests.faulty_entry_callable.key,
+                        group=socketprovider.SocketProvider.ENTRY_POINT_GROUP_NAME,
+                        value='tests: faulty_entry_callable',
+                    ),
+                ],
+                id='not-callable',
+            ),
+            pytest.param(
+                [
+                    importlib.metadata.EntryPoint(
+                        name=tests.faulty_entry_name_exists.key,
+                        group=socketprovider.SocketProvider.ENTRY_POINT_GROUP_NAME,
+                        value='tests: faulty_entry_name_exists',
+                    ),
+                ],
+                id='name-already-exists',
+            ),
+            pytest.param(
+                [
+                    importlib.metadata.EntryPoint(
+                        name=tests.faulty_entry_alias_exists.key,
+                        group=socketprovider.SocketProvider.ENTRY_POINT_GROUP_NAME,
+                        value='tests: faulty_entry_alias_exists',
+                    ),
+                ],
+                id='alias-already-exists',
+            ),
+        ],
+    )
+    GOOD_ENTRY_POINTS = pytest.mark.parametrize(
+        'additional_entry_points',
+        [
+            pytest.param(
+                [
+                    importlib.metadata.EntryPoint(
+                        name=tests.posix_entry.key,
+                        group=socketprovider.SocketProvider.ENTRY_POINT_GROUP_NAME,
+                        value='tests: posix_entry',
+                    ),
+                    importlib.metadata.EntryPoint(
+                        name=tests.the_annoying_os_entry.key,
+                        group=socketprovider.SocketProvider.ENTRY_POINT_GROUP_NAME,
+                        value='tests: the_annoying_os_entry',
+                    ),
+                ],
+                id='existing-entries',
+            ),
+            pytest.param(
+                [
+                    importlib.metadata.EntryPoint(
+                        name=tests.provider_entry1.key,
+                        group=socketprovider.SocketProvider.ENTRY_POINT_GROUP_NAME,
+                        value='tests: provider_entry1',
+                    ),
+                    importlib.metadata.EntryPoint(
+                        name=tests.provider_entry2.key,
+                        group=socketprovider.SocketProvider.ENTRY_POINT_GROUP_NAME,
+                        value='tests: provider_entry2',
+                    ),
+                ],
+                id='new-entries',
+            ),
+        ],
+    )
     STUBBED_AGENT_ADDRESSES = pytest.mark.parametrize(
         ['address', 'exception', 'match'],
         [
@@ -60,6 +135,9 @@ class Parametrize(types.SimpleNamespace):
             ),
         ],
     )
+    EXISTING_REGISTRY_ENTRIES = pytest.mark.parametrize(
+        'existing', ['posix', 'the_annoying_os']
+    )
     SSH_STRING_EXCEPTIONS = pytest.mark.parametrize(
         ['input', 'exc_type', 'exc_pattern'],
         [
@@ -414,6 +492,17 @@ class Parametrize(types.SimpleNamespace):
         list(tests.UNSUITABLE_KEYS.items()),
         ids=tests.UNSUITABLE_KEYS.keys(),
     )
+    RESOLVE_CHAINS = pytest.mark.parametrize(
+        ['terminal', 'chain'],
+        [
+            pytest.param('callable', ['a'], id='callable-1'),
+            pytest.param('callable', ['a', 'b', 'c', 'd'], id='callable-4'),
+            pytest.param('alias', ['e'], id='alias-5'),
+            pytest.param('alias', ['e', 'f', 'g', 'h', 'i'], id='alias-5'),
+            pytest.param('unimplemented', ['j'], id='unimplemented-1'),
+            pytest.param('unimplemented', ['j', 'k'], id='unimplemented-2'),
+        ],
+    )
 
 
 class TestTestingMachineryStubbedSSHAgentSocket:
@@ -709,18 +798,19 @@ class TestStaticFunctionality:
             with pytest.raises(ValueError, match='Cannot parse sh line:'):
                 tests.parse_sh_export_line(line, env_name=env_name)
 
-    def test_200_constructor_no_running_agent(
+    def test_200_constructor_posix_no_ssh_auth_sock(
         self,
         skip_if_no_af_unix_support: None,
     ) -> None:
-        """Abort if the running agent cannot be located."""
+        """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'
             ):
-                ssh_agent.SSHAgentClient()
+                posix_handler()
 
     @Parametrize.UINT32_INPUT
     def test_210_uint32(self, input: int, expected: bytes | bytearray) -> None:
@@ -831,6 +921,139 @@ class TestStaticFunctionality:
                 assert canon1(encoded) == canon2(encoded)
                 assert canon1(canon2(encoded)) == canon1(encoded)
 
+    def test_220_registry_resolve(
+        self,
+    ) -> None:
+        """Resolving entries in the socket provider registry works."""
+        registry = socketprovider.SocketProvider.registry
+        resolve = socketprovider.SocketProvider.resolve
+        with pytest.MonkeyPatch.context() as monkeypatch:
+            monkeypatch.setitem(registry, 'stub_agent', None)
+            assert callable(resolve('native'))
+            with pytest.raises(NotImplementedError):
+                resolve('stub_agent')
+
+    @Parametrize.RESOLVE_CHAINS
+    def test_221_registry_resolve_chains(
+        self,
+        terminal: Literal['unimplemented', 'alias', 'callable'],
+        chain: list[str],
+    ) -> None:
+        """Resolving a chain of providers works."""
+        registry = socketprovider.SocketProvider.registry
+        resolve = socketprovider.SocketProvider.resolve
+        try:
+            implementation = resolve('native')
+        except NotImplementedError:  # pragma: no cover
+            pytest.fail('Native SSH agent socket provider is unavailable?!')
+        # TODO(the-13th-letter): Rewrite using structural pattern matching.
+        # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
+        target = (
+            None
+            if terminal == 'unimplemented'
+            else 'native'
+            if terminal == 'alias'
+            else implementation
+        )
+        with pytest.MonkeyPatch.context() as monkeypatch:
+            for link in chain:
+                monkeypatch.setitem(registry, link, target)
+                target = link
+            if terminal == 'unimplemented':
+                with pytest.raises(NotImplementedError):
+                    resolve(chain[-1])
+            else:
+                assert resolve(chain[-1]) == implementation
+
+    @hypothesis.given(
+        terminal=strategies.sampled_from([
+            'unimplemented',
+            'alias',
+            'callable',
+        ]),
+        chain=strategies.lists(
+            strategies.sampled_from([
+                'c1',
+                'c2',
+                'c3',
+                'c4',
+                'c5',
+                'c6',
+                'c7',
+                'c8',
+                'c9',
+                'c10',
+            ]),
+            min_size=1,
+            unique=True,
+        ),
+    )
+    def test_221a_registry_resolve_chains(
+        self,
+        terminal: Literal['unimplemented', 'alias', 'callable'],
+        chain: list[str],
+    ) -> None:
+        """Resolving a chain of providers works."""
+        registry = socketprovider.SocketProvider.registry
+        resolve = socketprovider.SocketProvider.resolve
+        try:
+            implementation = resolve('native')
+        except NotImplementedError:  # pragma: no cover
+            hypothesis.note(f'{registry = }')
+            pytest.fail('Native SSH agent socket provider is unavailable?!')
+        # TODO(the-13th-letter): Rewrite using structural pattern matching.
+        # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
+        target = (
+            None
+            if terminal == 'unimplemented'
+            else 'native'
+            if terminal == 'alias'
+            else implementation
+        )
+        with pytest.MonkeyPatch.context() as monkeypatch:
+            for link in chain:
+                monkeypatch.setitem(registry, link, target)
+                target = link
+            if terminal == 'unimplemented':
+                with pytest.raises(NotImplementedError):
+                    resolve(chain[-1])
+            else:
+                assert resolve(chain[-1]) == implementation
+
+    @Parametrize.GOOD_ENTRY_POINTS
+    def test_230_find_all_socket_providers(
+        self,
+        additional_entry_points: list[importlib.metadata.EntryPoint],
+    ) -> None:
+        """Finding all SSH agent socket providers works."""
+        resolve = socketprovider.SocketProvider.resolve
+        old_registry = socketprovider.SocketProvider.registry
+        with tests.faked_entry_point_list(
+            additional_entry_points, remove_conflicting_entries=False
+        ) as names:
+            socketprovider.SocketProvider._find_all_ssh_agent_socket_providers()
+            for name in names:
+                assert name in socketprovider.SocketProvider.registry
+                assert resolve(name) in {
+                    tests.provider_entry_provider,
+                    *old_registry.values(),
+                }
+
+    @Parametrize.BAD_ENTRY_POINTS
+    def test_231_find_all_socket_providers_errors(
+        self,
+        additional_entry_points: list[importlib.metadata.EntryPoint],
+    ) -> None:
+        """Finding faulty SSH agent socket providers raises errors."""
+        with contextlib.ExitStack() as stack:
+            stack.enter_context(
+                tests.faked_entry_point_list(
+                    additional_entry_points, remove_conflicting_entries=False
+                )
+            )
+            stack.enter_context(pytest.raises(AssertionError))
+            socketprovider.SocketProvider._find_all_ssh_agent_socket_providers()
+
     @Parametrize.UINT32_EXCEPTIONS
     def test_310_uint32_exceptions(
         self, input: int, exc_type: type[Exception], exc_pattern: str
@@ -869,6 +1092,123 @@ class TestStaticFunctionality:
             with pytest.raises(exc_type, match=exc_pattern):
                 unstring_prefix(input)
 
+    def test_320_registry_already_registered(
+        self,
+    ) -> None:
+        """The registry forbids overwriting entries."""
+        registry = socketprovider.SocketProvider.registry.copy()
+        resolve = socketprovider.SocketProvider.resolve
+        register = socketprovider.SocketProvider.register
+        the_annoying_os = resolve('the_annoying_os')
+        posix = resolve('posix')
+        with pytest.MonkeyPatch.context() as monkeypatch:
+            monkeypatch.setattr(
+                socketprovider.SocketProvider, 'registry', registry
+            )
+            register('posix')(posix)
+            register('the_annoying_os')(the_annoying_os)
+            with pytest.raises(ValueError, match='already registered'):
+                register('posix')(the_annoying_os)
+            with pytest.raises(ValueError, match='already registered'):
+                register('the_annoying_os')(posix)
+            with pytest.raises(ValueError, match='already registered'):
+                register('posix', 'the_annoying_os_named_pipe')(posix)
+            with pytest.raises(ValueError, match='already registered'):
+                register('the_annoying_os', 'unix_domain')(the_annoying_os)
+
+    def test_321_registry_resolve_non_existant_entries(
+        self,
+    ) -> None:
+        """Resolving a non-existant entry fails."""
+        new_registry = {
+            'posix': socketprovider.SocketProvider.registry['posix'],
+            'the_annoying_os': socketprovider.SocketProvider.registry[
+                'the_annoying_os'
+            ],
+        }
+        with pytest.MonkeyPatch.context() as monkeypatch:
+            monkeypatch.setattr(
+                socketprovider.SocketProvider, 'registry', new_registry
+            )
+            with pytest.raises(socketprovider.NoSuchProviderError):
+                socketprovider.SocketProvider.resolve('native')
+
+    def test_322_registry_register_new_entry(
+        self,
+    ) -> None:
+        """Registering new entries works."""
+
+        def socket_provider() -> _types.SSHAgentSocket:
+            raise AssertionError
+
+        names = ['spam', 'ham', 'eggs', 'parrot']
+        new_registry = {
+            'posix': socketprovider.SocketProvider.registry['posix'],
+            'the_annoying_os': socketprovider.SocketProvider.registry[
+                'the_annoying_os'
+            ],
+        }
+        with pytest.MonkeyPatch.context() as monkeypatch:
+            monkeypatch.setattr(
+                socketprovider.SocketProvider, 'registry', new_registry
+            )
+            assert not any(
+                map(socketprovider.SocketProvider.registry.__contains__, names)
+            )
+            assert (
+                socketprovider.SocketProvider.register(*names)(socket_provider)
+                is socket_provider
+            )
+            assert all(
+                map(socketprovider.SocketProvider.registry.__contains__, names)
+            )
+            assert all([
+                socketprovider.SocketProvider.resolve(n) is socket_provider
+                for n in names
+            ])
+
+    @Parametrize.EXISTING_REGISTRY_ENTRIES
+    def test_323_registry_register_old_entry(
+        self,
+        existing: str,
+    ) -> None:
+        """Registering old entries works."""
+
+        provider = socketprovider.SocketProvider.resolve(existing)
+        new_registry = {
+            'posix': socketprovider.SocketProvider.registry['posix'],
+            'the_annoying_os': socketprovider.SocketProvider.registry[
+                'the_annoying_os'
+            ],
+            'unix_domain': 'posix',
+            'the_annoying_os_named_pipe': 'the_annoying_os',
+        }
+        names = [
+            k
+            for k, v in socketprovider.SocketProvider.registry.items()
+            if v == existing
+        ]
+        with pytest.MonkeyPatch.context() as monkeypatch:
+            monkeypatch.setattr(
+                socketprovider.SocketProvider, 'registry', new_registry
+            )
+            assert not all(
+                map(socketprovider.SocketProvider.registry.__contains__, names)
+            )
+            assert (
+                socketprovider.SocketProvider.register(existing, *names)(
+                    provider
+                )
+                is provider
+            )
+            assert all(
+                map(socketprovider.SocketProvider.registry.__contains__, names)
+            )
+            assert all([
+                socketprovider.SocketProvider.resolve(n) is provider
+                for n in [existing, *names]
+            ])
+
 
 class TestAgentInteraction:
     """Test actually talking to the SSH agent."""
@@ -1012,7 +1352,7 @@ class TestAgentInteraction:
         @click.command()
         def driver() -> None:
             """Call [`cli_helpers.select_ssh_key`][] directly, as a command."""
-            key = cli_helpers.select_ssh_key()
+            key = cli_helpers.select_ssh_key(client)
             click.echo(base64.standard_b64encode(key).decode('ASCII'))
 
         # TODO(the-13th-letter): (Continued from above.)  Update input
@@ -1033,14 +1373,18 @@ class TestAgentInteraction:
     ) -> None:
         """Fail if the agent address is invalid."""
         with pytest.MonkeyPatch.context() as monkeypatch:
-            monkeypatch.setenv('SSH_AUTH_SOCK', running_ssh_agent.socket + '~')
-            # socket.AF_UNIX is not defined everywhere.
-            sock = socket.socket(family=socket.AF_UNIX)  # type: ignore[attr-defined]
+            new_socket_name = (
+                running_ssh_agent.socket + '~'
+                if isinstance(running_ssh_agent.socket, str)
+                else '<invalid//address>'
+            )
+            monkeypatch.setenv('SSH_AUTH_SOCK', new_socket_name)
             with pytest.raises(OSError):  # noqa: PT011
-                ssh_agent.SSHAgentClient(socket=sock)
+                ssh_agent.SSHAgentClient()
 
     def test_301_constructor_no_af_unix_support(self) -> None:
         """Fail without [`socket.AF_UNIX`][] support."""
+        assert 'posix' in socketprovider.SocketProvider.registry
         with pytest.MonkeyPatch.context() as monkeypatch:
             monkeypatch.setenv('SSH_AUTH_SOCK', "the value doesn't matter")
             monkeypatch.delattr(socket, 'AF_UNIX', raising=False)
@@ -1048,7 +1392,31 @@ class TestAgentInteraction:
                 NotImplementedError,
                 match='UNIX domain sockets',
             ):
-                ssh_agent.SSHAgentClient()
+                ssh_agent.SSHAgentClient(socket='posix')
+
+    def test_302_no_ssh_agent_socket_provider_available(
+        self,
+    ) -> None:
+        """Fail if no SSH agent socket provider is available."""
+        with pytest.MonkeyPatch.context() as monkeypatch:
+            monkeypatch.setitem(
+                socketprovider.SocketProvider.registry, 'stub_agent', None
+            )
+            with pytest.raises(ExceptionGroup) as excinfo:
+                ssh_agent.SSHAgentClient(
+                    socket=['stub_agent', 'stub_agent', 'stub_agent']
+                )
+            assert all([
+                isinstance(e, NotImplementedError)
+                for e in excinfo.value.exceptions
+            ])
+
+    def test_303_explicit_socket(
+        self,
+        spawn_ssh_agent: tests.SpawnedSSHAgentInfo,
+    ) -> None:
+        conn = spawn_ssh_agent.client._connection
+        ssh_agent.SSHAgentClient(socket=conn)
 
     @Parametrize.TRUNCATED_AGENT_RESPONSES
     def test_310_truncated_server_response(
@@ -1294,3 +1662,254 @@ class TestAgentInteraction:
                 match=r'Malformed response|does not match request',
             ):
                 client.query_extensions()
+
+
+def safe_resolve(name: str) -> _types.SSHAgentSocketProvider | None:
+    """Safely resolve an SSH agent socket provider name.
+
+    If the provider is merely reserved, return `None` instead of raising
+    an exception.  Otherwise behave like
+    [`socketprovider.SocketProvider.resolve`][] does.
+
+    Args:
+        name: The name of the provider to resolve.
+
+    Returns:
+        The SSH agent socket provider if registered, `None` if merely
+        reserved, and an error if not found.
+
+    Raises:
+        socketprovider.NoSuchProviderError:
+            No such provider was found in the registry.  This includes
+            entries that are merely reserved.
+
+    """
+    try:
+        return socketprovider.SocketProvider.resolve(name)
+    except NotImplementedError:  # pragma: no cover
+        return None
+
+
+@strategies.composite
+def draw_alias_chain(
+    draw: strategies.DrawFn,
+    *,
+    known_keys_strategy: strategies.SearchStrategy[str],
+    new_keys_strategy: strategies.SearchStrategy[str],
+    chain_size: strategies.SearchStrategy[int] = strategies.integers(  # noqa: B008
+        min_value=1,
+        max_value=5,
+    ),
+    existing: bool = False,
+) -> tuple[str, ...]:
+    """Draw names for alias chains in the SSH agent socket provider registry.
+
+    Depending on arguments, draw a set of names from the new keys bundle
+    that do not yet exist in the registry, to insert as a new alias
+    chain.  Alternatively, draw a non-alias name from the known keys
+    bundle, then draw other names that either don't exist yet in the
+    registry, or that alias the first name directly or indirectly.  The
+    chain length, and whether to target existing registry entries or
+    not, may be set statically, or may be drawn from a respective
+    strategy.
+
+    Args:
+        draw:
+            The `hypothesis` draw function.
+        chain_size:
+            A strategy for determining the correct alias chain length.
+            Must not yield any integers less than 1.
+        existing:
+            If true, target an existing registry entry in the alias
+            chain, and permit rewriting existing aliases of that same
+            entry to the new alias.  Otherwise, draw only new names.
+        known_keys_strategy:
+            A strategy for generating provider registry keys already
+            contained in the registry.  Typically, this is
+            a [Bundle][hypothesis.stateful.Bundle].
+        new_keys_strategy:
+            A strategy for generating provider registry keys not yet
+            contained in the registry with high probability.  Typically,
+            this is a [consuming][hypothesis.stateful.consumes]
+            [Bundle][hypothesis.stateful.Bundle].
+
+    Returns:
+        A tuple of names forming an alias chain, each entry pointing to
+        or intending to point to the previous entry in the tuple.
+
+    """
+    registry = socketprovider.SocketProvider.registry
+
+    def not_an_alias(key: str) -> bool:
+        return key in registry and not isinstance(registry[key], str)
+
+    def is_indirect_alias_of(
+        key: str, target: str
+    ) -> bool:  # pragma: no cover
+        if key == target:
+            return False  # not an alias
+        seen = set()  # loop detection
+        while key not in seen:
+            seen.add(key)
+            if key not in registry:
+                return False
+            if not isinstance(registry[key], str):
+                return False
+            if key == target:
+                return True
+            tmp = registry[key]
+            assert isinstance(tmp, str)
+            key = tmp
+        return False  # loop
+
+    err_msg_chain_size = 'Chain sizes must always be 1 or larger.'
+
+    size = draw(chain_size)
+    if size < 1:  # pragma: no cover
+        raise ValueError(err_msg_chain_size)
+    names: list[str] = []
+    base: str | None = None
+    if existing:
+        names.append(draw(known_keys_strategy.filter(not_an_alias)))
+        base = names[0]
+        size -= 1
+        new_key_strategy = new_keys_strategy.filter(
+            lambda key: key not in registry
+        )
+        old_key_strategy = known_keys_strategy.filter(
+            lambda key: is_indirect_alias_of(key, target=base)
+        )
+        list_strategy_source = strategies.one_of(
+            new_key_strategy, old_key_strategy
+        )
+    else:
+        list_strategy_source = new_keys_strategy.filter(
+            lambda key: key not in registry
+        )
+    list_strategy = strategies.lists(
+        list_strategy_source.filter(lambda candidate: candidate != base),
+        min_size=size,
+        max_size=size,
+        unique=True,
+    )
+    names.extend(draw(list_strategy))
+    return tuple(names)
+
+
+class SSHAgentSocketProviderRegistryStateMachine(
+    stateful.RuleBasedStateMachine
+):
+    """A state machine for the SSH agent socket provider registry.
+
+    Record possible changes to the socket provider registry, keeping track
+    of true entries, aliases, and reservations.
+
+    """
+
+    def __init__(self) -> None:
+        """Initialize self, set up context managers and enter them."""
+        super().__init__()
+        self.exit_stack = contextlib.ExitStack().__enter__()
+        self.monkeypatch = self.exit_stack.enter_context(
+            pytest.MonkeyPatch.context()
+        )
+        self.orig_registry = socketprovider.SocketProvider.registry
+        self.registry: dict[
+            str, _types.SSHAgentSocketProvider | str | None
+        ] = {
+            'posix': self.orig_registry['posix'],
+            'the_annoying_os': self.orig_registry['the_annoying_os'],
+            'native': self.orig_registry['native'],
+            'unix_domain': 'posix',
+            'the_annoying_os_named_pipe': 'the_annoying_os',
+        }
+        self.monkeypatch.setattr(
+            socketprovider.SocketProvider, 'registry', self.registry
+        )
+        self.model: dict[str, _types.SSHAgentSocketProvider | None] = {}
+
+    known_keys: stateful.Bundle[str] = stateful.Bundle('known_keys')
+    """"""
+    new_keys: stateful.Bundle[str] = stateful.Bundle('new_keys')
+    """"""
+
+    def sample_provider(self) -> _types.SSHAgentSocket:
+        raise AssertionError
+
+    @stateful.initialize(
+        target=known_keys,
+    )
+    def get_registry_keys(self) -> stateful.MultipleResults[str]:
+        """Read the standard keys from the registry."""
+        self.model.update({k: safe_resolve(k) for k in self.registry})
+        return stateful.multiple(*self.registry.keys())
+
+    @stateful.rule(
+        target=new_keys,
+        k=strategies.text('abcdefghijklmnopqrstuvwxyz0123456789_').filter(
+            lambda s: s not in socketprovider.SocketProvider.registry
+        ),
+    )
+    def new_key(self, k: str) -> str:
+        return k
+
+    @stateful.invariant()
+    def check_consistency(self) -> None:
+        assert self.registry.keys() == self.model.keys()
+        for k in self.model:
+            resolved = safe_resolve(k)
+            modelled = self.model[k]
+            step1 = self.registry[k]
+            manually = safe_resolve(step1) if isinstance(step1, str) else step1
+            assert resolved == modelled
+            assert resolved == manually
+
+    @stateful.rule(
+        target=known_keys,
+        chain=draw_alias_chain(
+            known_keys_strategy=known_keys,
+            new_keys_strategy=stateful.consumes(new_keys),
+            existing=True,
+        ),
+    )
+    def alias_existing(
+        self, chain: tuple[str, ...]
+    ) -> stateful.MultipleResults[str]:
+        try:
+            provider = socketprovider.SocketProvider.resolve(chain[0])
+        except NotImplementedError:  # pragma: no cover [failsafe]
+            provider = self.sample_provider
+        assert (
+            socketprovider.SocketProvider.register(*chain)(provider)
+            == provider
+        )
+        for k in chain:
+            self.model[k] = provider
+        return stateful.multiple(*chain[1:])
+
+    @stateful.rule(
+        target=known_keys,
+        chain=draw_alias_chain(
+            known_keys_strategy=known_keys,
+            new_keys_strategy=stateful.consumes(new_keys),
+            existing=False,
+        ),
+    )
+    def alias_new(self, chain: list[str]) -> stateful.MultipleResults[str]:
+        provider = self.sample_provider
+        assert (
+            socketprovider.SocketProvider.register(*chain)(provider)
+            == provider
+        )
+        for k in chain:
+            self.model[k] = provider
+        return stateful.multiple(*chain)
+
+    def teardown(self) -> None:
+        """Upon teardown, exit all contexts entered in `__init__`."""
+        self.exit_stack.close()
+
+
+TestSSHAgentSocketProviderRegistry = (
+    SSHAgentSocketProviderRegistryStateMachine.TestCase
+)


...	...	@@ -27,7 +27,7 @@ import click.testing
27	27	import hypothesis
28	28	import pytest
29	29	from hypothesis import strategies
30		-from typing_extensions import NamedTuple, assert_never
	30	+from typing_extensions import NamedTuple, assert_never, overload
31	31
32	32	from derivepassphrase import _types, cli, ssh_agent, vault
33	33	from derivepassphrase._internals import cli_helpers, cli_machinery
...	...	@@ -670,11 +670,19 @@ class RunningSSHAgentInfo(NamedTuple):
670	670
671	671	"""
672	672
673		- socket: str
	673	+ socket: str \| type[_types.SSHAgentSocket]
674	674	""""""
675	675	agent_type: KnownSSHAgent
676	676	""""""
677	677
	678	+ def require_external_address(self) -> str: # pragma: no cover
	679	+ if not isinstance(self.socket, str):
	680	+ pytest.skip(
	681	+ reason='This test requires a real, externally resolvable '
	682	+ 'address for the SSH agent socket.'
	683	+ )
	684	+ return self.socket
	685	+
678	686
679	687	ALL_KEYS: Mapping[str, SSHTestKey] = {
680	688	'ed25519': SSHTestKey(
...	...	@@ -2306,6 +2314,214 @@ def phrase_from_key(
2306	2314	raise KeyError(key) # pragma: no cover
2307	2315
2308	2316
	2317	+def provider_entry_provider() -> _types.SSHAgentSocket: # pragma: no cover
	2318	+ """A pseudo provider for a [`_types.SSHAgentSocketProviderEntry`][]."""
	2319	+ msg = 'We are not supposed to be called!'
	2320	+ raise AssertionError(msg)
	2321	+
	2322	+
	2323	+provider_entry1 = _types.SSHAgentSocketProviderEntry(
	2324	+ provider_entry_provider, 'entry1', ('entry1a', 'entry1b', 'entry1c')
	2325	+)
	2326	+"""A sample [`_types.SSHAgentSocketProviderEntry`][]."""
	2327	+
	2328	+provider_entry2 = _types.SSHAgentSocketProviderEntry(
	2329	+ provider_entry_provider, 'entry2', ('entry2d', 'entry2e')
	2330	+)
	2331	+"""A sample [`_types.SSHAgentSocketProviderEntry`][]."""
	2332	+
	2333	+posix_entry = _types.SSHAgentSocketProviderEntry(
	2334	+ socketprovider.SocketProvider.resolve('posix'), 'posix', ()
	2335	+)
	2336	+"""
	2337	+The standard [`_types.SSHAgentSocketProviderEntry`][] for the UNIX
	2338	+domain socket handler on POSIX systems.
	2339	+"""
	2340	+
	2341	+the_annoying_os_entry = _types.SSHAgentSocketProviderEntry(
	2342	+ socketprovider.SocketProvider.resolve('the_annoying_os'),
	2343	+ 'the_annoying_os',
	2344	+ (),
	2345	+)
	2346	+"""
	2347	+The standard [`_types.SSHAgentSocketProviderEntry`][] for the named pipe
	2348	+handler on The Annoying Operating System.
	2349	+"""
	2350	+
	2351	+faulty_entry_callable = _types.SSHAgentSocketProviderEntry(
	2352	+ (), # type: ignore[arg-type]
	2353	+ 'tuple',
	2354	+ (),
	2355	+)
	2356	+"""
	2357	+A faulty [`_types.SSHAgentSocketProviderEntry`][]: the indicated handler
	2358	+is not a callable.
	2359	+"""
	2360	+
	2361	+faulty_entry_name_exists = _types.SSHAgentSocketProviderEntry(
	2362	+ socketprovider.SocketProvider.resolve('the_annoying_os'), 'posix', ()
	2363	+)
	2364	+"""
	2365	+A faulty [`_types.SSHAgentSocketProviderEntry`][]: the indicated handler
	2366	+is already registered with a different callable.
	2367	+"""
	2368	+
	2369	+faulty_entry_alias_exists = _types.SSHAgentSocketProviderEntry(
	2370	+ socketprovider.SocketProvider.resolve('posix'),
	2371	+ 'posix',
	2372	+ ('unix_domain', 'the_annoying_os'),
	2373	+)
	2374	+"""
	2375	+A faulty [`_types.SSHAgentSocketProviderEntry`][]: the alias is already
	2376	+registered with a different callable.
	2377	+"""
	2378	+
	2379	+
	2380	+@contextlib.contextmanager
	2381	+def faked_entry_point_list( # noqa: C901
	2382	+ additional_entry_points: Sequence[importlib.metadata.EntryPoint],
	2383	+ remove_conflicting_entries: bool = False,
	2384	+) -> Iterator[Sequence[str]]:
	2385	+ """Yield a context where additional entry points are visible.
	2386	+
	2387	+ Args:
	2388	+ additional_entry_points:
	2389	+ A sequence of entry point objects that should additionally
	2390	+ be visible.
	2391	+ remove_conflicting_entries:
	2392	+ If true, remove all names provided by the additional entry
	2393	+ points, otherwise leave them untouched.
	2394	+
	2395	+ Yields:
	2396	+ A sequence of registry names that are newly available within the
	2397	+ context.
	2398	+
	2399	+ """
	2400	+ true_entry_points = importlib.metadata.entry_points()
	2401	+ additional_entry_points = list(additional_entry_points)
	2402	+
	2403	+ if sys.version_info >= (3, 12):
	2404	+ new_entry_points = importlib.metadata.EntryPoints(
	2405	+ list(true_entry_points) + additional_entry_points
	2406	+ )
	2407	+
	2408	+ @overload
	2409	+ def mangled_entry_points(
	2410	+ *, group: None = None
	2411	+ ) -> importlib.metadata.EntryPoints: ...
	2412	+
	2413	+ @overload
	2414	+ def mangled_entry_points(
	2415	+ *, group: str
	2416	+ ) -> importlib.metadata.EntryPoints: ...
	2417	+
	2418	+ def mangled_entry_points(
	2419	+ **params: Any,
	2420	+ ) -> importlib.metadata.EntryPoints:
	2421	+ return new_entry_points.select(**params)
	2422	+
	2423	+ elif sys.version_info >= (3, 10):
	2424	+ # Compatibility concerns within importlib.metadata: depending on
	2425	+ # whether the .select() API is used, the result is either the dict
	2426	+ # of groups of points (as in < 3.10), or the EntryPoints iterable
	2427	+ # (as in >= 3.12). So our wrapper needs to duplicate that
	2428	+ # interface. FUN.
	2429	+ new_entry_points_dict = {
	2430	+ k: list(v) for k, v in true_entry_points.items()
	2431	+ }
	2432	+ for ep in additional_entry_points:
	2433	+ new_entry_points_dict.setdefault(ep.group, []).append(ep)
	2434	+ new_entry_points = importlib.metadata.EntryPoints([
	2435	+ ep for group in new_entry_points_dict.values() for ep in group
	2436	+ ])
	2437	+
	2438	+ @overload
	2439	+ def mangled_entry_points(
	2440	+ *, group: None = None
	2441	+ ) -> dict[
	2442	+ str,
	2443	+ list[importlib.metadata.EntryPoint]
	2444	+ \| tuple[importlib.metadata.EntryPoint, ...],
	2445	+ ]: ...
	2446	+
	2447	+ @overload
	2448	+ def mangled_entry_points(
	2449	+ *, group: str
	2450	+ ) -> importlib.metadata.EntryPoints: ...
	2451	+
	2452	+ def mangled_entry_points(
	2453	+ **params: Any,
	2454	+ ) -> (
	2455	+ importlib.metadata.EntryPoints
	2456	+ \| dict[
	2457	+ str,
	2458	+ list[importlib.metadata.EntryPoint]
	2459	+ \| tuple[importlib.metadata.EntryPoint, ...],
	2460	+ ]
	2461	+ ):
	2462	+ return (
	2463	+ new_entry_points.select(**params)
	2464	+ if params
	2465	+ else new_entry_points_dict
	2466	+ )
	2467	+
	2468	+ else:
	2469	+ new_entry_points: dict[
	2470	+ str,
	2471	+ list[importlib.metadata.EntryPoint]
	2472	+ \| tuple[importlib.metadata.EntryPoint, ...],
	2473	+ ] = {
	2474	+ group_name: list(group)
	2475	+ for group_name, group in true_entry_points.items()
	2476	+ }
	2477	+ for ep in additional_entry_points:
	2478	+ new_entry_points.setdefault(ep.group, [])
	2479	+ new_entry_points[ep.group].append(ep)
	2480	+ new_entry_points = {
	2481	+ group_name: tuple(group)
	2482	+ for group_name, group in new_entry_points.items()
	2483	+ }
	2484	+
	2485	+ @overload
	2486	+ def mangled_entry_points(
	2487	+ *, group: None = None
	2488	+ ) -> dict[str, tuple[importlib.metadata.EntryPoint, ...]]: ...
	2489	+
	2490	+ @overload
	2491	+ def mangled_entry_points(
	2492	+ *, group: str
	2493	+ ) -> tuple[importlib.metadata.EntryPoint, ...]: ...
	2494	+
	2495	+ def mangled_entry_points(
	2496	+ *, group: str \| None = None
	2497	+ ) -> (
	2498	+ dict[str, tuple[importlib.metadata.EntryPoint, ...]]
	2499	+ \| tuple[importlib.metadata.EntryPoint, ...]
	2500	+ ):
	2501	+ return (
	2502	+ new_entry_points.get(group, ())
	2503	+ if group is not None
	2504	+ else new_entry_points
	2505	+ )
	2506	+
	2507	+ registry = socketprovider.SocketProvider.registry
	2508	+ new_registry = registry.copy()
	2509	+ keys = [ep.load().key for ep in additional_entry_points]
	2510	+ aliases = [a for ep in additional_entry_points for a in ep.load().aliases]
	2511	+ if remove_conflicting_entries: # pragma: no cover [unused]
	2512	+ for name in [keys, aliases]:
	2513	+ new_registry.pop(name, None)
	2514	+
	2515	+ with pytest.MonkeyPatch.context() as monkeypatch:
	2516	+ monkeypatch.setattr(
	2517	+ socketprovider.SocketProvider, 'registry', new_registry
	2518	+ )
	2519	+ monkeypatch.setattr(
	2520	+ importlib.metadata, 'entry_points', mangled_entry_points
	2521	+ )
	2522	+ yield (keys, aliases)
	2523	+
	2524	+
2309	2525	@contextlib.contextmanager
2310	2526	def isolated_config(
2311	2527	monkeypatch: pytest.MonkeyPatch,
...	...	@@ -2525,10 +2741,11 @@ def make_file_readonly(
2525	2741	group and other, and ensuring the read permission bit for user is
2526	2742	set.
2527	2743
2528		- Unfortunately, Windows has its own rules: Set exactly(?) the read
2529		- permission bit for user to make the file read-only, and set
2530		- exactly(?) the write permission bit for user to make the file
2531		- read/write; all other permission bit settings are ignored.
	2744	+ Unfortunately, The Annoying OS (a.k.a. Microsoft Windows) has its
	2745	+ own rules: Set exactly(?) the read permission bit for user to make
	2746	+ the file read-only, and set exactly(?) the write permission bit for
	2747	+ user to make the file read/write; all other permission bit settings
	2748	+ are ignored.
2532	2749
2533	2750	The cross-platform procedure therefore is:
2534	2751

...	...	@@ -8,7 +8,6 @@ import base64
8	8	import contextlib
9	9	import datetime
10	10	import importlib
11		-import operator
12	11	import os
13	12	import shutil
14	13	import socket
...	...	@@ -170,10 +169,10 @@ class SpawnFunc(Protocol):
170	169	"""
171	170
172	171
173		-def spawn_pageant( # pragma: no cover [external]
	172	+def spawn_pageant_on_posix( # pragma: no cover [external]
174	173	executable: str \| None, env: dict[str, str]
175	174	) -> subprocess.Popen[str] \| None:
176		- """Spawn an isolated Pageant, if possible.
	175	+ """Spawn an isolated (UNIX) Pageant, if possible.
177	176
178	177	We attempt to detect whether Pageant is usable, i.e. whether Pageant
179	178	has output buffering problems when announcing its authentication
...	...	@@ -195,7 +194,7 @@ def spawn_pageant( # pragma: no cover [external]
195	194	subprocess.
196	195
197	196	"""
198		- if executable is None: # pragma: no cover [external]
	197	+ if executable is None or os.name == 'nt': # pragma: no cover [external]
199	198	return None
200	199
201	200	# Apparently, Pageant 0.81 and lower running in debug mode does
...	...	@@ -242,10 +241,10 @@ def spawn_pageant( # pragma: no cover [external]
242	241	)
243	242
244	243
245		-def spawn_openssh_agent( # pragma: no cover [external]
	244	+def spawn_openssh_agent_on_posix( # pragma: no cover [external]
246	245	executable: str \| None, env: dict[str, str]
247	246	) -> subprocess.Popen[str] \| None:
248		- """Spawn an isolated OpenSSH agent, if possible.
	247	+ """Spawn an isolated OpenSSH agent (on UNIX), if possible.
249	248
250	249	Args:
251	250	executable:
...	...	@@ -262,7 +261,7 @@ def spawn_openssh_agent( # pragma: no cover [external]
262	261	subprocess.
263	262
264	263	"""
265		- if executable is None:
	264	+ if executable is None or os.name == 'nt': # pragma: no cover [external]
266	265	return None
267	266	return subprocess.Popen(
268	267	['ssh-agent', '-D', '-s'],
...	...	@@ -282,11 +281,29 @@ def spawn_noop( # pragma: no cover [unused]
282	281	"""Placeholder function. Does nothing."""
283	282
284	283
285		-spawn_handlers: Sequence[tuple[str, SpawnFunc, tests.KnownSSHAgent]] = [
286		- ('pageant', spawn_pageant, tests.KnownSSHAgent.Pageant),
287		- ('ssh-agent', spawn_openssh_agent, tests.KnownSSHAgent.OpenSSHAgent),
288		- ('(system)', spawn_noop, tests.KnownSSHAgent.UNKNOWN),
289		-]
	284	+spawn_handlers: dict[str, tuple[str, SpawnFunc, tests.KnownSSHAgent]] = {
	285	+ 'pageant': (
	286	+ 'pageant',
	287	+ spawn_pageant_on_posix,
	288	+ tests.KnownSSHAgent.Pageant,
	289	+ ),
	290	+ 'ssh-agent': (
	291	+ 'ssh-agent',
	292	+ spawn_openssh_agent_on_posix,
	293	+ tests.KnownSSHAgent.OpenSSHAgent,
	294	+ ),
	295	+ 'stub_agent': (
	296	+ 'stub_agent',
	297	+ spawn_noop,
	298	+ tests.KnownSSHAgent.StubbedSSHAgent,
	299	+ ),
	300	+ 'stub_agent_with_extensions': (
	301	+ 'stub_agent_with_extensions',
	302	+ spawn_noop,
	303	+ tests.KnownSSHAgent.StubbedSSHAgent,
	304	+ ),
	305	+ '(system)': ('(system)', spawn_noop, tests.KnownSSHAgent.UNKNOWN),
	306	+}
290	307	"""
291	308	The standard registry of agent spawning functions.
292	309	"""
...	...	@@ -377,7 +394,12 @@ def spawn_named_agent(
377	394	ssh_auth_sock = agent_env.pop('SSH_AUTH_SOCK', None)
378	395	proc = spawn_func(executable=shutil.which(exec_name), env=agent_env)
379	396	with exit_stack:
380		- if spawn_func is spawn_noop:
	397	+ if (
	398	+ spawn_func is spawn_noop
	399	+ and agent_type == tests.KnownSSHAgent.StubbedSSHAgent
	400	+ ):
	401	+ ssh_auth_sock = None
	402	+ elif spawn_func is spawn_noop:
381	403	ssh_auth_sock = os.environ['SSH_AUTH_SOCK']
382	404	elif proc is None: # pragma: no cover [external]
383	405	err_msg = f'Cannot spawn usable {exec_name}'
...	...	@@ -404,11 +426,51 @@ def spawn_named_agent(
404	426	err_msg = f'Cannot parse agent output: {pid_line!r}'
405	427	raise CannotSpawnError(err_msg)
406	428	monkeypatch = exit_stack.enter_context(pytest.MonkeyPatch.context())
	429	+ if ssh_auth_sock is not None:
407	430	monkeypatch.setenv('SSH_AUTH_SOCK', ssh_auth_sock)
408	431	client = exit_stack.enter_context(
409	432	ssh_agent.SSHAgentClient.ensure_agent_subcontext()
410	433	)
	434	+ else:
	435	+ monkeypatch.setenv(
	436	+ 'SSH_AUTH_SOCK', tests.StubbedSSHAgentSocketWithAddress.ADDRESS
	437	+ )
	438	+ monkeypatch.setattr(
	439	+ ssh_agent.SSHAgentClient,
	440	+ 'SOCKET_PROVIDERS',
	441	+ ['stub_with_address_and_deterministic_dsa']
	442	+ if exec_name == 'stub_agent_with_extensions'
	443	+ else ['stub_with_address'],
	444	+ )
	445	+ client = exit_stack.enter_context(
	446	+ ssh_agent.SSHAgentClient.ensure_agent_subcontext(
	447	+ tests.StubbedSSHAgentSocketWithAddressAndDeterministicDSA()
	448	+ if exec_name == 'stub_agent_with_extensions'
	449	+ else tests.StubbedSSHAgentSocketWithAddress()
	450	+ )
	451	+ )
	452	+ # We sanity-test the connected SSH agent if it is not one of our
	453	+ # test agents, because allowing the user to run the test suite
	454	+ # with a clearly faulty agent would likely and unfairly
	455	+ # misattribute the agent's protocol violations to our test
	456	+ # suite. On the flip side, for our own test agents, the
	457	+ # correctness tests are part of the test suite, so we don't want
	458	+ # the setup code here to already decide that the agent is
	459	+ # unsuitable. Therefore, do a sanity check if and only if the
	460	+ # agent is not one of our test agents, and if the check fails,
	461	+ # skip this agent.
	462	+ if (
	463	+ agent_type != tests.KnownSSHAgent.StubbedSSHAgent
	464	+ ): # pragma: no cover [external]
	465	+ try:
411	466	client.list_keys() # sanity test
	467	+ except (
	468	+ EOFError,
	469	+ OSError,
	470	+ ssh_agent.SSHAgentFailedError,
	471	+ ) as exc: # pragma: no cover [failsafe]
	472	+ msg = f'agent failed the "list keys" sanity test: {exc!r}'
	473	+ raise CannotSpawnError(msg) from exc
412	474	yield tests.SpawnedSSHAgentInfo(
413	475	agent_type, client, spawn_func is not spawn_noop
414	476	)
...	...	@@ -417,9 +479,59 @@ def spawn_named_agent(
417	479	)
418	480
419	481
	482	+def is_agent_permitted(
	483	+ agent_type: tests.KnownSSHAgent,
	484	+) -> bool: # pragma: no cover [external]
	485	+ """May the given SSH agent be spawned by the test harness?
	486	+
	487	+ If the environment variable `PERMITTED_SSH_AGENTS` is given, it
	488	+ names a comma-separated list of known SSH agent names that the test
	489	+ harness may spawn. Invalid names are silently ignored. If not
	490	+ given, or empty, then any agent may be spawned.
	491	+
	492	+ (To not allow any agent to be spawned, specify a single comma as the
	493	+ list. But see below.)
	494	+
	495	+ The stubbed agents cannot be restricted in this manner, as the test
	496	+ suite depends on their availability.
	497	+
	498	+ """
	499	+ if not os.environ.get('PERMITTED_SSH_AGENTS'):
	500	+ return True
	501	+ permitted_agents = {tests.KnownSSHAgent.StubbedSSHAgent}
	502	+ permitted_agents.update({
	503	+ tests.KnownSSHAgent(x)
	504	+ for x in os.environ['PERMITTED_SSH_AGENTS'].split(',')
	505	+ if x in tests.KnownSSHAgent.__members__
	506	+ })
	507	+ return agent_type in permitted_agents
	508	+
	509	+
	510	+spawn_handlers_params: list[Sequence] = []
	511	+"""
	512	+The standard registry of agent spawning functions, annotated as
	513	+a `pytest` parameter set. (In particular, this includes the conditional
	514	+skip marks.) Used by some test fixtures.
	515	+"""
	516	+for key, handler in spawn_handlers.items():
	517	+ marks = [
	518	+ pytest.mark.skipif(
	519	+ not is_agent_permitted(handler[2]),
	520	+ reason='agent excluded via PERMITTED_AGENTS environment variable',
	521	+ ),
	522	+ ]
	523	+ if key in {'pageant', 'ssh-agent', '(system)'}:
	524	+ marks.append(
	525	+ pytest.mark.skipif(
	526	+ not hasattr(socket, 'AF_UNIX'),
	527	+ reason='socket module does not support AF_UNIX',
	528	+ )
	529	+ )
	530	+ spawn_handlers_params.append(pytest.param(handler, id=key, marks=marks))
	531	+
	532	+
420	533	@pytest.fixture
421	534	def running_ssh_agent( # pragma: no cover [external]
422		- skip_if_no_af_unix_support: None,
423	535	) -> Iterator[tests.RunningSSHAgentInfo]:
424	536	"""Ensure a running SSH agent, if possible, as a pytest fixture.
425	537
...	...	@@ -441,7 +553,30 @@ def running_ssh_agent( # pragma: no cover [external]
441	553	If no agent is running or can be spawned, skip this test.
442	554
443	555	"""
444		- del skip_if_no_af_unix_support
	556	+
	557	+ def prepare_environment(
	558	+ agent_type: tests.KnownSSHAgent,
	559	+ ) -> Iterator[tests.RunningSSHAgentInfo]:
	560	+ with pytest.MonkeyPatch.context() as monkeypatch:
	561	+ if agent_type == tests.KnownSSHAgent.StubbedSSHAgent:
	562	+ monkeypatch.setattr(
	563	+ ssh_agent.SSHAgentClient,
	564	+ 'SOCKET_PROVIDERS',
	565	+ ['stub_with_address'],
	566	+ )
	567	+ monkeypatch.setenv(
	568	+ 'SSH_AUTH_SOCK',
	569	+ tests.StubbedSSHAgentSocketWithAddress.ADDRESS,
	570	+ )
	571	+ yield tests.RunningSSHAgentInfo(
	572	+ tests.StubbedSSHAgentSocketWithAddress,
	573	+ tests.KnownSSHAgent.StubbedSSHAgent,
	574	+ )
	575	+ else:
	576	+ yield tests.RunningSSHAgentInfo(
	577	+ os.environ['SSH_AUTH_SOCK'],
	578	+ agent_type,
	579	+ )
445	580
446	581	with pytest.MonkeyPatch.context() as monkeypatch:
447	582	# pytest's fixture system does not seem to guarantee that
...	...	@@ -455,30 +590,30 @@ def running_ssh_agent( # pragma: no cover [external]
455	590	monkeypatch.setenv('SSH_AUTH_SOCK', startup_ssh_auth_sock)
456	591	else:
457	592	monkeypatch.delenv('SSH_AUTH_SOCK', raising=False)
458		- for exec_name, spawn_func, agent_type in spawn_handlers:
	593	+ for exec_name, spawn_func, agent_type in spawn_handlers.values():
	594	+ if not is_agent_permitted(agent_type):
	595	+ continue
459	596	try:
460	597	for _agent_info in spawn_named_agent(
461	598	exec_name, spawn_func, agent_type
462	599	):
463		- yield tests.RunningSSHAgentInfo(
464		- os.environ['SSH_AUTH_SOCK'], agent_type
465		- )
	600	+ yield from prepare_environment(agent_type)
466	601	except (KeyError, OSError, CannotSpawnError):
467	602	continue
468	603	return
469	604	pytest.skip('No SSH agent running or spawnable')
470	605
471	606
472		-@pytest.fixture(params=spawn_handlers, ids=operator.itemgetter(0))
	607	+@pytest.fixture(params=spawn_handlers_params)
473	608	def spawn_ssh_agent(
474	609	request: pytest.FixtureRequest,
475		- skip_if_no_af_unix_support: None,
476	610	) -> Iterator[tests.SpawnedSSHAgentInfo]: # pragma: no cover [external]
477	611	"""Spawn an isolated SSH agent, if possible, as a pytest fixture.
478	612
479	613	Spawn a new SSH agent isolated from other SSH use by other
480		- processes, if possible. We know how to spawn OpenSSH's agent and
481		- PuTTY's Pageant, and the "(system)" fallback agent.
	614	+ processes, if possible. We know how to spawn OpenSSH's agent (on
	615	+ UNIX) and PuTTY's Pageant (on UNIX), the stubbed agents, and the
	616	+ "(system)" fallback agent.
482	617
483	618	Yields:
484	619	A [named tuple][collections.namedtuple] containing information
...	...	@@ -491,7 +626,7 @@ def spawn_ssh_agent(
491	626	If the agent cannot be spawned, skip this test.
492	627
493	628	"""
494		- del skip_if_no_af_unix_support
	629	+
495	630	with pytest.MonkeyPatch.context() as monkeypatch:
496	631	# pytest's fixture system does not seem to guarantee that
497	632	# environment variables are set up correctly if nested and
...	...	@@ -550,6 +685,13 @@ def ssh_agent_client_with_test_keys_loaded( # noqa: C901
550	685	agent_type, client, isolated = spawn_ssh_agent
551	686	successfully_loaded_keys: set[str] = set()
552	687
	688	+ # This fixture relies on `spawn_ssh_agent`, which in turn uses
	689	+ # `spawn_named_agent` to, well, spawn agents. `spawn_named_agent`
	690	+ # runs sanity tests on the agents it spawns, via
	691	+ # `SSHAgentClient.list_keys`. There is thus little point in
	692	+ # repeating the very same sanity test here, on an agent that was
	693	+ # already sanity-tested.
	694	+
553	695	def prepare_payload(
554	696	payload: bytes \| bytearray,
555	697	*,