derivepassphrase.git

@@ -37,6 +37,9 @@ def pytest_configure(config: pytest.Config) -> None:

         (

             "heavy_duty: "

             "mark test as a slow, heavy-duty test (e.g., an integration test)"

+            "\n"

+            "non_reentrant_agent: "

+            "mark the agent (fixture output) as non-reentrant"

         ),

     )

     hypothesis_machinery._hypothesis_settings_setup()

@@ -390,6 +393,58 @@ The standard registry of agent spawning functions.

 """

+def external_agent_restriction(

+    agent_type: data.KnownSSHAgent,

+    env_var: str,

+    *,

+    default: bool = True,

+) -> bool:  # pragma: no cover [external]

+    """Classify an SSH agent according to some environment variable.

+

+    Look up the given SSH agent type in a certain environment variable,

+    and report on whether the agent type is listed in that variable.

+    This can be used to check if the agent is e.g. in a list of

+    permitted agents, or if it is in the list of agents requiring

+    special workarounds.  Invalid agent type names are silently ignored.

+

+    If the environment variable is empty, return a default value.

+

+    By design, the stubbed agents have a fixed classification (the

+    default value), unaffected by the environment variable.

+

+    Args:

+        agent_type:

+            The agent type to classify.

+        env_var:

+            The environment variable in which to look up the agent type

+            name.

+        default:

+            The value to return if the environment variable is unset, or

+            if a stub agent is to be classified.

+

+    Returns:

+        If the agent type is not a stub agent, and the environment

+        variable is non-empty, then `True` if the agent type is

+        mentioned in the environment variable, and `False` otherwise;

+        the default value otherwise.

+

+    See also:

+        [`is_agent_permitted`][] and [`is_agent_non_reentrant`][], which

+        make use of this function internally.

+

+    """

+    if agent_type == data.KnownSSHAgent.StubbedSSHAgent:

+        return default

+    if not os.environ.get(env_var):

+        return default

+    marked_agents = {

+        data.KnownSSHAgent(x)

+        for x in os.environ[env_var].split(",")

+        if x in data.KnownSSHAgent.__members__

+    }

+    return agent_type in marked_agents

+

+

 def is_agent_permitted(

     agent_type: data.KnownSSHAgent,

 ) -> bool:  # pragma: no cover [external]

@@ -407,15 +462,35 @@ def is_agent_permitted(

     suite depends on their availability.

     """

-    if not os.environ.get("PERMITTED_SSH_AGENTS"):

-        return True

-    permitted_agents = {data.KnownSSHAgent.StubbedSSHAgent}

-    permitted_agents.update({

-        data.KnownSSHAgent(x)

-        for x in os.environ["PERMITTED_SSH_AGENTS"].split(",")

-        if x in data.KnownSSHAgent.__members__

-    })

-    return agent_type in permitted_agents

+    return external_agent_restriction(

+        agent_type, "PERMITTED_SSH_AGENTS", default=True

+    )

+

+

+def is_agent_non_reentrant(

+    agent_type: data.KnownSSHAgent,

+) -> bool:  # pragma: no cover [external]

+    """Is the given SSH agent assumed to be non-reentrant?

+

+    If the environment variable `NON_REENTRANT_SSH_AGENTS` is given, it

+    names a comma-separated list of known SSH agent names that are

+    assumed to be non-reentrant, i.e., where the SSH agent cannot

+    tolerate multiple clients connecting to it at the same time, and

+    thus no client for the same agent may be constructed while another

+    one is already connected.  Invalid names are silently ignored.  If

+    not given, or empty, then all agents are considered reentrant, i.e.,

+    all agents are not considered non-reentrant.

+

+    (To consider all agents non-reentrant, specify a single comma as the

+    list.  But see below.)

+

+    The stubbed agents cannot be influenced in this manner, as the test

+    suite depends on their reentrancy.

+

+    """

+    return external_agent_restriction(

+        agent_type, "NON_REENTRANT_SSH_AGENTS", default=False

+    )

 spawn_handlers_params: list[Sequence] = []

@@ -432,6 +507,13 @@ for key, handler in spawn_handlers.items():

             "environment variable.",

         ),

     ]

+    # Allow agents to be marked as non-reentrant.  Workaround for an

+    # issue with GnuPG 2.4.8 on The Annoying OS when `gpg-agent`

+    # masquerades as OpenSSH's `ssh-agent`.

+    if is_agent_non_reentrant(

+        handler.agent_type

+    ):  # pragma: no cover [external]

+        marks.append(pytest.mark.non_reentrant_agent)

     # Mark non-isolated agents as, well, using non-isolated agents.

     # Assume by default that an agent is potentially non-isolated unless

     # we can be absolutely sure it is isolated, by design.  (The "stub"

@@ -34,6 +34,7 @@ from typing_extensions import assert_never, overload

 import tests.data

 import tests.machinery

+from derivepassphrase import _types, ssh_agent

 from derivepassphrase._internals import cli_helpers, cli_machinery

 from derivepassphrase.ssh_agent import socketprovider

@@ -547,3 +548,56 @@ def isolated_vault_exporter_config(

             yield

         finally:

             cli_helpers.config_filename("write lock").unlink(missing_ok=True)

+

+

+def ensure_singleton_client_if_non_reentrant_agent(

+    *,

+    request: pytest.FixtureRequest,

+    monkeypatch: pytest.MonkeyPatch,

+    client: ssh_agent.SSHAgentClient,

+) -> None:

+    """Mangle SSH agent construction if the spawned agent is non-reentrant.

+

+    Monkeypatch a suitable environment to ensure that constructing a new

+    SSH agent client -- usually via the

+    [`ssh_agent.SSHAgentClient.ensure_agent_subcontext`][] context

+    manager -- doesn't really construct a new client, in case the agent

+    is non-reentrant.

+

+    Agent reentrancy is determined by the presence of the

+    `non_reentrant_agent` mark.  We change nothing if the agent is

+    actually reentrant.

+

+    Args:

+        request:

+            The `pytest` request fixture.  Needed to get the active

+            markers on this test function execution, which detail

+            whether the agent is potentially non-reentrant.

+        monkeypatch:

+            A monkeypatching context to enrich.

+        client:

+            The singleton SSH agent client that is to be used.

+

+    """

+    if (

+        request.node.get_closest_marker("non_reentrant_agent") is not None

+    ):  # pragma: no cover [external]

+        real_ensure_agent_subcontext = (

+            ssh_agent.SSHAgentClient.ensure_agent_subcontext

+        )

+

+        def ensure_agent_subcontext(

+            conn: ssh_agent.SSHAgentClient

+            | _types.SSHAgentSocket

+            | Sequence[str]

+            | None = None,

+        ) -> contextlib.AbstractContextManager[ssh_agent.SSHAgentClient]:

+            if isinstance(conn, ssh_agent.SSHAgentClient):

+                return real_ensure_agent_subcontext(conn)

+            return real_ensure_agent_subcontext(client)

+

+        monkeypatch.setattr(

+            ssh_agent.SSHAgentClient,

+            "ensure_agent_subcontext",

+            ensure_agent_subcontext,

+        )

@@ -29,7 +29,6 @@ import click

 import hypothesis

 import pytest

 from hypothesis import strategies

-from typing_extensions import Any

 from derivepassphrase import _types, cli, ssh_agent, vault

 from derivepassphrase._internals import (

@@ -46,6 +45,8 @@ if TYPE_CHECKING:

     from collections.abc import Callable, Iterable, Iterator

     from typing import NoReturn

+    from typing_extensions import Any

+

 DUMMY_SERVICE = data.DUMMY_SERVICE

 DUMMY_PASSPHRASE = data.DUMMY_PASSPHRASE

@@ -480,6 +481,7 @@ class Parametrize(types.SimpleNamespace):

             "sign_action",

             "pattern",

             "warnings_patterns",

+            "tests_construction_failure",

         ],

         [

             pytest.param(

@@ -489,6 +491,7 @@ class Parametrize(types.SimpleNamespace):

                 SignAction.FAIL,

                 "not loaded into the agent",

                 [],

+                False,

                 id="key-not-loaded",

             ),

             pytest.param(

@@ -498,6 +501,7 @@ class Parametrize(types.SimpleNamespace):

                 SignAction.FAIL,

                 "SSH agent failed to or refused to",

                 [],

+                False,

                 id="list-keys-refused",

             ),

             pytest.param(

@@ -507,6 +511,7 @@ class Parametrize(types.SimpleNamespace):

                 SignAction.FAIL,

                 "SSH agent failed to or refused to",

                 [],

+                False,

                 id="list-keys-protocol-error",

             ),

             pytest.param(

@@ -516,6 +521,7 @@ class Parametrize(types.SimpleNamespace):

                 SignAction.FAIL,

                 "Cannot connect to the SSH agent",

                 [],

+                True,

                 id="agent-address-mangled",

             ),

             pytest.param(

@@ -525,6 +531,7 @@ class Parametrize(types.SimpleNamespace):

                 SignAction.FAIL,

                 "Cannot find any running SSH agent",

                 [],

+                True,

                 id="agent-address-missing",

             ),

             pytest.param(

@@ -534,6 +541,7 @@ class Parametrize(types.SimpleNamespace):

                 SignAction.FAIL,

                 "does not support communicating with it",

                 [],

+                True,

                 id="no-native-agent-available",

             ),

             pytest.param(

@@ -543,6 +551,7 @@ class Parametrize(types.SimpleNamespace):

                 SignAction.FAIL,

                 "does not support communicating with it",

                 [],

+                True,

                 id="no-agents-in-agent-provider-list",

             ),

             pytest.param(

@@ -552,6 +561,7 @@ class Parametrize(types.SimpleNamespace):

                 SignAction.FAIL,

                 "does not support communicating with it",

                 ["Cannot connect to an SSH agent via UNIX domain sockets"],

+                True,

                 id="no-unix-domain-sockets",

             ),

             pytest.param(

@@ -561,6 +571,7 @@ class Parametrize(types.SimpleNamespace):

                 SignAction.FAIL,

                 "does not support communicating with it",

                 ["Cannot connect to an SSH agent via Windows named pipes"],

+                True,

                 id="no-windows-named-pipes",

             ),

             pytest.param(

@@ -570,6 +581,7 @@ class Parametrize(types.SimpleNamespace):

                 SignAction.FAIL_RUNTIME,

                 "violates the communication protocol",

                 [],

+                False,

                 id="sign-violates-protocol",

             ),

         ],

@@ -1466,6 +1478,7 @@ class TestMisc:

     @Parametrize.KEY_TO_PHRASE_SETTINGS

     def test_key_to_phrase(

         self,

+        request: pytest.FixtureRequest,

         ssh_agent_client_with_test_keys_loaded: ssh_agent.SSHAgentClient,

         list_keys_action: ListKeysAction | None,

         system_support_action: SystemSupportAction | None,

@@ -1473,6 +1486,7 @@ class TestMisc:

         sign_action: SignAction,

         pattern: str,

         warnings_patterns: list[str],

+        tests_construction_failure: bool,

     ) -> None:

         """All errors in [`cli_helpers.key_to_phrase`][] are handled."""

@@ -1505,6 +1520,14 @@ class TestMisc:

                 address_action(monkeypatch)

             if system_support_action:

                 system_support_action(monkeypatch)

+

+            if not tests_construction_failure:

+                pytest_machinery.ensure_singleton_client_if_non_reentrant_agent(

+                    request=request,

+                    monkeypatch=monkeypatch,

+                    client=ssh_agent_client_with_test_keys_loaded,

+                )

+

             with pytest.raises(ErrCallback, match=pattern) as excinfo:

                 cli_helpers.key_to_phrase(

                     loaded_key, error_callback=err, warning_callback=warn

@@ -27,7 +27,6 @@ from typing import TYPE_CHECKING

 import hypothesis

 import pytest

 from hypothesis import strategies

-from typing_extensions import Any

 from derivepassphrase import _types, cli, ssh_agent

 from derivepassphrase._internals import (

@@ -43,6 +42,8 @@ if TYPE_CHECKING:

     from collections.abc import Iterator

     from typing import NoReturn

+    from typing_extensions import Any

+

 DUMMY_SERVICE = data.DUMMY_SERVICE

 DUMMY_KEY1_B64 = data.DUMMY_KEY1_B64

@@ -879,15 +880,21 @@ class TestStoringConfigurationFailures:

             pass

     def test_fail_because_ssh_agent_has_no_keys_loaded(

-        self, spawn_ssh_agent: data.SpawnedSSHAgentInfo

+        self,

+        request: pytest.FixtureRequest,

+        spawn_ssh_agent: data.SpawnedSSHAgentInfo,

     ) -> None:

         """Not holding any SSH keys during `--config --key` fails."""

-        del spawn_ssh_agent

         with self._test(

             ["--key"],

             error_text="no keys suitable",

             patch_suitable_ssh_keys=False,

         ) as monkeypatch:

+            pytest_machinery.ensure_singleton_client_if_non_reentrant_agent(

+                request=request,

+                monkeypatch=monkeypatch,

+                client=spawn_ssh_agent.client,

+            )

             def func(

                 *_args: Any,

@@ -898,15 +905,21 @@ class TestStoringConfigurationFailures:

             monkeypatch.setattr(ssh_agent.SSHAgentClient, "list_keys", func)

     def test_store_config_fail_manual_ssh_agent_runtime_error(

-        self, spawn_ssh_agent: data.SpawnedSSHAgentInfo

+        self,

+        request: pytest.FixtureRequest,

+        spawn_ssh_agent: data.SpawnedSSHAgentInfo,

     ) -> None:

         """Triggering an error in the SSH agent during `--config --key` leads to failure."""

-        del spawn_ssh_agent

         with self._test(

             ["--key"],

             error_text="violates the communication protocol",

             patch_suitable_ssh_keys=False,

         ) as monkeypatch:

+            pytest_machinery.ensure_singleton_client_if_non_reentrant_agent(

+                request=request,

+                monkeypatch=monkeypatch,

+                client=spawn_ssh_agent.client,

+            )

             def raiser(*_args: Any, **_kwargs: Any) -> None:

                 raise ssh_agent.TrailingDataError()

@@ -914,13 +927,19 @@ class TestStoringConfigurationFailures:

             monkeypatch.setattr(ssh_agent.SSHAgentClient, "list_keys", raiser)

     def test_store_config_fail_manual_ssh_agent_refuses(

-        self, spawn_ssh_agent: data.SpawnedSSHAgentInfo

+        self,

+        request: pytest.FixtureRequest,

+        spawn_ssh_agent: data.SpawnedSSHAgentInfo,

     ) -> None:

         """The SSH agent refusing during `--config --key` leads to failure."""

-        del spawn_ssh_agent

         with self._test(

             ["--key"], error_text="refused to", patch_suitable_ssh_keys=False

         ) as monkeypatch:

+            pytest_machinery.ensure_singleton_client_if_non_reentrant_agent(

+                request=request,

+                monkeypatch=monkeypatch,

+                client=spawn_ssh_agent.client,

+            )

             def func(*_args: Any, **_kwargs: Any) -> NoReturn:

                 raise ssh_agent.SSHAgentFailedError(