derivepassphrase.git

@@ -1143,11 +1143,39 @@ def vault_version_option_callback(

             _types.Feature.SSH_KEY,

         ]

         click.echo()

+

+        from derivepassphrase.ssh_agent import socketprovider  # noqa: PLC0415

+

+        socket_providers: dict[str, bool] = {}

+        for key, names in socketprovider.SocketProvider.grouped().items():

+            if isinstance(key, _types.BuiltinSSHAgentSocketProvider):

+                if not key.is_known_fake_agent():

+                    other_names = set(names) - {key}

+                    formatted_key = (

+                        "{key!s} ({aliases_key!s} {aliases})".format(

+                            key=key,

+                            aliases_key=_msg.TranslatedString(

+                                _msg.Label.FEATURE_ITEM_ALIASES

+                            ),

+                            aliases=", ".join(sorted(other_names, key=str)),

+                        )

+                        if other_names

+                        else str(key)

+                    )

+                    socket_providers[formatted_key] = key.test()

+            elif key:  # pragma: no cover [external]

+                socket_providers[key] = True

         version_info_types: dict[_msg.Label, list[str]] = {

             _msg.Label.SUPPORTED_FEATURES: [k for k in features if k.test()],

             _msg.Label.UNAVAILABLE_FEATURES: [

                 k for k in features if not k.test()

             ],

+            _msg.Label.SUPPORTED_SSH_AGENT_SOCKET_PROVIDERS: sorted(

+                k for k, v in socket_providers.items() if v

+            ),

+            _msg.Label.UNAVAILABLE_SSH_AGENT_SOCKET_PROVIDERS: sorted(

+                k for k, v in socket_providers.items() if not v

+            ),

         }

         print_version_info_types(version_info_types, ctx=ctx)

         ctx.exit()

@@ -1342,6 +1342,16 @@ class Label(enum.Enum):

         "Supported foreign configuration formats:",

     )

     """"""

+    SUPPORTED_SSH_AGENT_SOCKET_PROVIDERS = commented(

+        "This is part of the version output, emitting lists of supported "

+        "SSH agent socket providers.  "

+        "A comma-separated English list of items follows, "

+        "with standard English punctuation.",

+    )(

+        "Label :: Info Message:: Table row header",

+        "Supported SSH agent socket providers:",

+    )

+    """"""

     SUPPORTED_SUBCOMMANDS = commented(

         "This is part of the version output, emitting lists of supported "

         "subcommands.  "

@@ -1382,6 +1392,16 @@ class Label(enum.Enum):

         "Known foreign configuration formats:",

     )

     """"""

+    UNAVAILABLE_SSH_AGENT_SOCKET_PROVIDERS = commented(

+        "This is part of the version output, emitting lists of known, "

+        "unavailable SSH agent socket providers.  "

+        "A comma-separated English list of items follows, "

+        "with standard English punctuation.",

+    )(

+        "Label :: Info Message:: Table row header",

+        "Known SSH agent socket providers:",

+    )

+    """"""

     CONFIRM_THIS_CHOICE_PROMPT_TEXT = commented(

         'There is no support for "yes" or "no" in other languages '

         "than English, so it is advised that your translation makes it "

@@ -1122,6 +1122,14 @@ class BuiltinSSHAgentSocketProvider(str, enum.Enum):

                 pass

         return ret

+    def is_known_fake_agent(self, *_args: Any, **_kwargs: Any) -> bool:  # noqa: ANN401

+        """Return true if this SSH agent is a known fake agent."""

+        return self in {

+            BuiltinSSHAgentSocketProvider.STUB_AGENT,

+            BuiltinSSHAgentSocketProvider.STUB_AGENT_WITH_ADDRESS,

+            BuiltinSSHAgentSocketProvider.STUB_AGENT_WITH_ADDRESS_AND_DETERMINISTIC_DSA,

+        }

+

     __str__ = str.__str__

     __format__ = str.__format__  # type: ignore[assignment]

@@ -10,7 +10,10 @@ import collections

 import ctypes

 import enum

 import errno

+import graphlib

 import hashlib

+import itertools

+import operator

 import os

 import socket

 from ctypes.wintypes import (  # type: ignore[attr-defined]

@@ -27,7 +30,8 @@ from typing import TYPE_CHECKING, cast

 from derivepassphrase import _types

 if TYPE_CHECKING:

-    from collections.abc import Callable

+    from collections.abc import Callable, Mapping

+    from collections.abc import Set as AbstractSet

     from typing import ClassVar

     from typing_extensions import (

@@ -903,6 +907,59 @@ class SocketProvider:

             raise NotImplementedError(msg)

         return ret

+    @classmethod

+    def grouped(cls) -> Mapping[str, AbstractSet[str]]:

+        """Calculate a mapping of canonical socket provider entries.

+

+        Specifically, determine the non-alias entries in the socket

+        provider registry, and map each such non-alias entry to its set

+        of aliases.

+

+        Returns:

+            A mapping of non-alias entry names to sets of alias entry

+            names.

+

+        Warning:

+            The results are undefined if the registry has been modified

+            by any means other than the [`register`][] decorator.

+

+        """

+        known_socket_provider_values = frozenset({

+            v for v in cls.registry.values() if not isinstance(v, str)

+        })

+        canonical: dict[str, str] = {}

+        sorter: graphlib.TopologicalSorter[

+            _types.SSHAgentSocketProvider | str | None

+        ] = graphlib.TopologicalSorter()

+        k: _types.SSHAgentSocketProvider | str | None

+        v: _types.SSHAgentSocketProvider | str | None

+        for k, v in cls.registry.items():

+            sorter.add(k, v)

+        sorter.prepare()

+        while sorter:

+            entries = sorter.get_ready()

+            for k in entries:

+                if not isinstance(k, str):

+                    assert k in known_socket_provider_values

+                    sorter.done(k)

+                    continue

+                v = cls.registry[k]

+                canonical[k] = (

+                    canonical[v]  # alias

+                    if isinstance(v, str)

+                    else k  # actual entry

+                )

+                sorter.done(k)

+

+        key_of_entry = operator.itemgetter(0)

+        by_value = operator.itemgetter(1)

+

+        sorted_entries = sorted(canonical.items(), key=by_value)

+        return {

+            k: frozenset(map(key_of_entry, v))

+            for k, v in itertools.groupby(sorted_entries, key=by_value)

+        }

+

     ENTRY_POINT_GROUP_NAME = "derivepassphrase.ssh_agent_socket_providers"

     """

     The group name under which [entry

@@ -36,6 +36,7 @@ class VersionOutputData(NamedTuple):

     extras: frozenset[str]

     subcommands: frozenset[str]

     features: dict[str, bool]

+    ssh_agent_socket_providers: dict[str, bool]

 def _label_text(e: cli_messages.Label, /) -> str:

@@ -60,6 +61,12 @@ class KnownLineType(str, enum.Enum):

     )

     SUPPORTED_FEATURES = _label_text(cli_messages.Label.SUPPORTED_FEATURES)

     UNAVAILABLE_FEATURES = _label_text(cli_messages.Label.UNAVAILABLE_FEATURES)

+    SUPPORTED_SSH_AGENT_SOCKET_PROVIDERS = _label_text(

+        cli_messages.Label.SUPPORTED_SSH_AGENT_SOCKET_PROVIDERS

+    )

+    UNAVAILABLE_SSH_AGENT_SOCKET_PROVIDERS = _label_text(

+        cli_messages.Label.UNAVAILABLE_SSH_AGENT_SOCKET_PROVIDERS

+    )

     ENABLED_EXTRAS = _label_text(cli_messages.Label.ENABLED_PEP508_EXTRAS)

@@ -191,6 +198,7 @@ PEP 508 extras: export.

                     subcommands=frozenset(),

                     features={},

                     extras=frozenset({"export"}),

+                    ssh_agent_socket_providers={},

                 ),

                 id="derivepassphrase-0.4.0-export",

             ),

@@ -215,6 +223,7 @@ No PEP 508 extras are active.

                     subcommands=frozenset({"export", "vault"}),

                     features={},

                     extras=frozenset({}),

+                    ssh_agent_socket_providers={},

                 ),

                 id="derivepassphrase-0.5-plain",

             ),

@@ -237,6 +246,8 @@ Known foreign configuration formats: divination v3.141592,

 Supported subcommands: delete-all-files, dump-core.

 Supported features: delete-while-open.

 Known features: backups-are-nice-to-have.

+Supported SSH agent socket providers: agents-of-shield.

+Known SSH agent socket providers: agent-smith.

 PEP 508 extras: annoying-popups, delete-all-files,

     dump-core-depending-on-the-phase-of-the-moon.

@@ -269,6 +280,10 @@ PEP 508 extras: annoying-popups, delete-all-files,

                         "delete-all-files",

                         "dump-core-depending-on-the-phase-of-the-moon",

                     }),

+                    ssh_agent_socket_providers={

+                        "agents-of-shield": True,

+                        "agent-smith": False,

+                    },

                 ),

                 id="inventpassphrase",

             ),

@@ -305,6 +320,7 @@ Supported subcommands: export, spectre ({aliases!s} master-password, mpw),

                     }),

                     features={},

                     extras=frozenset(),

+                    ssh_agent_socket_providers={},

                 ),

                 id="aliases",

             ),

@@ -476,6 +492,7 @@ def parse_version_output(  # noqa: C901

     subcommands: set[str] = set()

     extras: set[str] = set()

     features: dict[str, bool] = {}

+    ssh_agent_socket_providers: dict[str, bool] = {}

     if len(paragraphs) < 2:  # pragma: no cover

         return VersionOutputData(

             derivation_schemes=schemes,

@@ -483,6 +500,7 @@ def parse_version_output(  # noqa: C901

             subcommands=frozenset(subcommands),

             extras=frozenset(extras),

             features=features,

+            ssh_agent_socket_providers=ssh_agent_socket_providers,

         )

     for line in paragraphs[1]:

         line_type, _, value = line.partition(":")

@@ -506,6 +524,15 @@ def parse_version_output(  # noqa: C901

                 features[item] = True

             elif line_type == KnownLineType.UNAVAILABLE_FEATURES:

                 features[item] = False

+            elif (

+                line_type == KnownLineType.SUPPORTED_SSH_AGENT_SOCKET_PROVIDERS

+            ):

+                ssh_agent_socket_providers[item] = True

+            elif (

+                line_type

+                == KnownLineType.UNAVAILABLE_SSH_AGENT_SOCKET_PROVIDERS

+            ):

+                ssh_agent_socket_providers[item] = False

             else:

                 raise AssertionError(  # noqa: TRY003

                     f"Unknown version info line type: {line_type!r}"  # noqa: EM102

@@ -516,6 +543,7 @@ def parse_version_output(  # noqa: C901

         subcommands=frozenset(subcommands),

         extras=frozenset(extras),

         features=features,

+        ssh_agent_socket_providers=ssh_agent_socket_providers,

     )