derivepassphrase.git

@@ -6,7 +6,7 @@ build-backend = "hatchling.build"

 name = "derivepassphrase"

 description = "An almost faithful Python reimplementation of James Coglan's vault."

 readme = "README.md"

-requires-python = ">= 3.10"

+requires-python = ">= 3.9"

 license = "MIT"

 keywords = []

 authors = [

@@ -133,7 +133,7 @@ extra-dependencies = [

 matrix-name-format = '{variable}_{value}'

 [[tool.hatch.envs.hatch-test.matrix]]

-python = ["3.12", "3.11", "3.10", "pypy3.10"]

+python = ["3.12", "3.11", "3.10", "3.9", "pypy3.10", "pypy3.9"]

 cryptography = ["no", "yes"]

 hypothesis-profile = ["user-default"]

@@ -221,18 +221,18 @@ def validate_vault_config(  # noqa: C901,PLR0912,PLR0915

         if not isinstance(o_global, dict):

             raise TypeError(err_not_a_dict(['global']))

         for key, value in o_global.items():

-            match key:

-                case 'key' | 'phrase':

+            # Use match/case here once Python 3.9 becomes unsupported.

+            if key in {'key', 'phrase'}:

                 if not isinstance(value, str):

                     raise TypeError(err_not_a_dict(['global', key]))

-                case 'unicode_normalization_form':

+            elif key == 'unicode_normalization_form':

                 if not isinstance(value, str):

                     raise TypeError(err_not_a_dict(['global', key]))

                 if not allow_derivepassphrase_extensions:

                     raise ValueError(

                         err_derivepassphrase_extension(key, ('global',))

                     )

-                case _ if not allow_unknown_settings:

+            elif not allow_unknown_settings:

                 raise ValueError(err_unknown_setting(key, ('global',)))

         if 'key' in o_global and 'phrase' in o_global:

             raise ValueError(err_key_and_phrase)

@@ -244,17 +244,15 @@ def validate_vault_config(  # noqa: C901,PLR0912,PLR0915

         if not isinstance(service, dict):

             raise TypeError(err_not_a_dict(['services', sv_name]))

         for key, value in service.items():

-            match key:

-                case 'notes' | 'phrase' | 'key':

+            # Use match/case here once Python 3.9 becomes unsupported.

+            if key in {'notes', 'phrase', 'key'}:

                 if not isinstance(value, str):

                     raise TypeError(

                         err_not_a_string(['services', sv_name, key])

                     )

-                case 'length':

+            elif key == 'length':

                 if not isinstance(value, int):

-                        raise TypeError(

-                            err_not_an_int(['services', sv_name, key])

-                        )

+                    raise TypeError(err_not_an_int(['services', sv_name, key]))

                 if value < 1:

                     raise ValueError(

                         err_bad_number(

@@ -263,19 +261,17 @@ def validate_vault_config(  # noqa: C901,PLR0912,PLR0915

                             strictly_positive=True,

                         )

                     )

-                case (

-                    'repeat'

-                    | 'lower'

-                    | 'upper'

-                    | 'number'

-                    | 'space'

-                    | 'dash'

-                    | 'symbol'

-                ):

+            elif key in {

+                'repeat',

+                'lower',

+                'upper',

+                'number',

+                'space',

+                'dash',

+                'symbol',

+            }:

                 if not isinstance(value, int):

-                        raise TypeError(

-                            err_not_an_int(['services', sv_name, key])

-                        )

+                    raise TypeError(err_not_an_int(['services', sv_name, key]))

                 if value < 0:

                     raise ValueError(

                         err_bad_number(

@@ -284,7 +280,7 @@ def validate_vault_config(  # noqa: C901,PLR0912,PLR0915

                             strictly_positive=False,

                         )

                     )

-                case _ if not allow_unknown_settings:

+            elif not allow_unknown_settings:

                 raise ValueError(

                     err_unknown_setting(key, ['services', sv_name])

                 )

@@ -202,8 +202,8 @@ def _load_data(

 ) -> Any:  # noqa: ANN401

     contents: bytes

     module: types.ModuleType

-    match fmt:

-        case 'v0.2':

+    # Use match/case here once Python 3.9 becomes unsupported.

+    if fmt == 'v0.2':

         module = importlib.import_module(

             'derivepassphrase.exporter.vault_native'

         )

@@ -214,7 +214,7 @@ def _load_data(

         return module.export_vault_native_data(

             contents, key, try_formats=['v0.2']

         )

-        case 'v0.3':

+    elif fmt == 'v0.3':  # noqa: RET505

         module = importlib.import_module(

             'derivepassphrase.exporter.vault_native'

         )

@@ -225,14 +225,12 @@ def _load_data(

         return module.export_vault_native_data(

             contents, key, try_formats=['v0.3']

         )

-        case 'storeroom':

-            module = importlib.import_module(

-                'derivepassphrase.exporter.storeroom'

-            )

+    elif fmt == 'storeroom':

+        module = importlib.import_module('derivepassphrase.exporter.storeroom')

         if module.STUBBED:

             raise ModuleNotFoundError

         return module.export_storeroom_data(path, key)

-        case _:  # pragma: no cover

+    else:  # pragma: no cover

         assert_never(fmt)

@@ -373,12 +371,12 @@ def _config_filename(

     path = os.getenv(PROG_NAME.upper() + '_PATH') or click.get_app_dir(

         PROG_NAME, force_posix=True

     )

-    match subsystem:

-        case None:

+    # Use match/case here once Python 3.9 becomes unsupported.

+    if subsystem is None:

         return path

-        case 'vault' | 'settings':

+    elif subsystem in {'vault', 'settings'}:  # noqa: RET505

         filename = f'{subsystem}.json'

-        case _:  # pragma: no cover

+    else:  # pragma: no cover

         msg = f'Unknown configuration subsystem: {subsystem!r}'

         raise AssertionError(msg)

     return os.path.join(path, filename)

@@ -522,14 +520,14 @@ def _get_suitable_ssh_keys(

     """

     client: ssh_agent.SSHAgentClient

     client_context: contextlib.AbstractContextManager[Any]

-    match conn:

-        case ssh_agent.SSHAgentClient():

+    # Use match/case here once Python 3.9 becomes unsupported.

+    if isinstance(conn, ssh_agent.SSHAgentClient):

         client = conn

         client_context = contextlib.nullcontext()

-        case socket.socket() | None:

+    elif isinstance(conn, socket.socket) or conn is None:

         client = ssh_agent.SSHAgentClient(socket=conn)

         client_context = client

-        case _:  # pragma: no cover

+    else:  # pragma: no cover

         assert_never(conn)

         msg = f'invalid connection hint: {conn!r}'

         raise TypeError(msg)  # noqa: DOC501

@@ -1216,18 +1214,18 @@ def derivepassphrase_vault(  # noqa: C901,PLR0912,PLR0913,PLR0914,PLR0915

     for param in ctx.command.params:

         if isinstance(param, click.Option):

             group: type[click.Option]

-            match param:

-                case PasswordGenerationOption():

+            # Use match/case here once Python 3.9 becomes unsupported.

+            if isinstance(param, PasswordGenerationOption):

                 group = PasswordGenerationOption

-                case ConfigurationOption():

+            elif isinstance(param, ConfigurationOption):

                 group = ConfigurationOption

-                case StorageManagementOption():

+            elif isinstance(param, StorageManagementOption):

                 group = StorageManagementOption

-                case OptionGroupOption():

-                    raise AssertionError(  # noqa: DOC501,TRY003

+            elif isinstance(param, OptionGroupOption):

+                raise AssertionError(  # noqa: DOC501,TRY003,TRY004

                     f'Unknown option group for {param!r}'  # noqa: EM102

                 )

-                case _:

+            else:

                 group = click.Option

             options_in_group.setdefault(group, []).append(param)

         params_by_str[param.human_readable_name] = param

@@ -23,7 +23,7 @@ should *not* be used or relied on.

 from __future__ import annotations

 import base64

-import glob

+import fnmatch

 import json

 import logging

 import os

@@ -678,9 +678,15 @@ def export_storeroom_data(  # noqa: C901,PLR0912,PLR0914,PLR0915

     config_structure: dict[str, Any] = {}

     json_contents: dict[str, bytes] = {}

-    for file in glob.glob(

-        '[01][0-9a-f]', root_dir=os.fsdecode(storeroom_path)

-    ):

+    # Use glob.glob(..., root_dir=...) here once Python 3.9 becomes

+    # unsupported.

+    storeroom_path_str = os.fsdecode(storeroom_path)

+    valid_hashdirs = [

+        hashdir_name

+        for hashdir_name in os.listdir(storeroom_path_str)

+        if fnmatch.fnmatch(hashdir_name, '[01][0-9a-f]')

+    ]

+    for file in valid_hashdirs:

         bucket_contents = list(

             decrypt_bucket_file(file, master_keys, root_dir=storeroom_path)

         )

@@ -441,20 +441,20 @@ def export_vault_native_data(

         key = exporter.get_vault_key()

     stored_exception: Exception | None = None

     for config_format in try_formats:

-        match config_format:

-            case 'v0.2':

+        # Use match/case here once Python 3.9 becomes unsupported.

+        if config_format == 'v0.2':

             try:

                 return VaultNativeV02ConfigParser(contents, key)()

             except ValueError as exc:

                 exc.__context__ = stored_exception

                 stored_exception = exc

-            case 'v0.3':

+        elif config_format == 'v0.3':

             try:

                 return VaultNativeV03ConfigParser(contents, key)()

             except ValueError as exc:

                 exc.__context__ = stored_exception

                 stored_exception = exc

-            case _:  # pragma: no cover

+        else:  # pragma: no cover

             msg = (

                 f'Invalid vault native configuration format: '

                 f'{config_format!r}'

@@ -43,14 +43,17 @@ class SSHAgentFailedError(RuntimeError):

     """The SSH agent failed to complete the requested operation."""

     def __str__(self) -> str:

-        match self.args:

-            case (_types.SSH_AGENT.FAILURE.value, b''):  # pragma: no branch

+        # Use match/case here once Python 3.9 becomes unsupported.

+        if self.args == (  # pragma: no branch

+            _types.SSH_AGENT.FAILURE.value,

+            b'',

+        ):

             return 'The SSH agent failed to complete the request'

-            case (_, _msg) if _msg:  # pragma: no cover

+        elif self.args[1]:  # noqa: RET505  # pragma: no cover

             code = self.args[0]

             msg = self.args[1].decode('utf-8', 'surrogateescape')

             return f'[Code {code:d}] {msg:s}'

-            case _:  # pragma: no cover

+        else:  # pragma: no cover

             return repr(self)

     def __repr__(self) -> str:  # pragma: no cover

@@ -348,7 +351,7 @@ class SSHAgentClient:

         """

         if isinstance(  # pragma: no branch

-            response_code, int | _types.SSH_AGENT

+            response_code, (int, _types.SSH_AGENT)

         ):

             response_code = frozenset({response_code})

         if response_code is not None:  # pragma: no branch

@@ -11,13 +11,15 @@ import collections

 import hashlib

 import math

 import types

-from collections.abc import Callable

-from typing import TypeAlias

+from typing import TYPE_CHECKING

-from typing_extensions import assert_type

+from typing_extensions import TypeAlias, assert_type

 from derivepassphrase import sequin, ssh_agent

+if TYPE_CHECKING:

+    from collections.abc import Callable

+

 __author__ = 'Marco Ricci <software@the13thletter.info>'

@@ -458,7 +460,7 @@ class Vault:

             a passphrase deterministically.

         """

-        TestFunc: TypeAlias = Callable[[bytes | bytearray], bool]

+        TestFunc: TypeAlias = 'Callable[[bytes | bytearray], bool]'

         deterministic_signature_types: dict[str, TestFunc]

         deterministic_signature_types = {

             'ssh-ed25519': lambda k: k.startswith(

@@ -806,11 +806,11 @@ def isolated_vault_exporter_config(

         monkeypatch.delenv('USERNAME', raising=False)

         if vault_key is not None:

             monkeypatch.setenv('VAULT_KEY', vault_key)

-        match vault_config:

-            case str():

+        # Use match/case here once Python 3.9 becomes unsupported.

+        if isinstance(vault_config, str):

             with open('.vault', 'w', encoding='UTF-8') as outfile:

                 print(vault_config, file=outfile)

-            case bytes():

+        elif isinstance(vault_config, bytes):

             os.makedirs('.vault', mode=0o700, exist_ok=True)

             with (

                 chdir('.vault'),

@@ -822,9 +822,9 @@ def isolated_vault_exporter_config(

                 tmpzipfile.seek(0, 0)

                 with zipfile.ZipFile(tmpzipfile.file) as zipfileobj:

                     zipfileobj.extractall()

-            case None:

+        elif vault_config is None:

             pass

-            case _:  # pragma: no cover

+        else:  # pragma: no cover

             assert_never(vault_config)

         yield

@@ -935,14 +935,14 @@ class ReadableResult(NamedTuple):

                 code, or an expected exception type.

         """

-        match error:

-            case str():

+        # Use match/case here once Python 3.9 becomes unsupported.

+        if isinstance(error, str):

             return (

                 isinstance(self.exception, SystemExit)

                 and self.exit_code > 0

                 and (not error or error in self.stderr)

             )

-            case _:

+        else:  # noqa: RET505

             return isinstance(self.exception, error)

@@ -262,8 +262,8 @@ def running_ssh_agent() -> Iterator[str]:  # pragma: no cover

         else:  # pragma: no cover

             monkeypatch.delenv('SSH_AUTH_SOCK', raising=False)

         for exec_name, spawn_func, _ in _spawn_handlers:

-            match exec_name:

-                case '(system)':

+            # Use match/case here once Python 3.9 becomes unsupported.

+            if exec_name == '(system)':

                 assert (

                     os.environ.get('SSH_AUTH_SOCK', None)

                     == startup_ssh_auth_sock

@@ -277,8 +277,8 @@ def running_ssh_agent() -> Iterator[str]:  # pragma: no cover

                 assert (

                     os.environ.get('SSH_AUTH_SOCK', None)

                     == startup_ssh_auth_sock

-                    ), 'SSH_AUTH_SOCK mismatch after returning from running agent'  # noqa: E501

-                case _:

+                ), 'SSH_AUTH_SOCK mismatch after returning from running agent'

+            else:

                 assert (

                     os.environ.get('SSH_AUTH_SOCK', None)

                     == startup_ssh_auth_sock

@@ -310,9 +310,7 @@ def running_ssh_agent() -> Iterator[str]:  # pragma: no cover

                         'pid' not in pid_line.lower()

                         and '_pid' not in pid_line.lower()

                     ):  # pragma: no cover

-                            pytest.skip(

-                                f'Cannot parse agent output: {pid_line!r}'

-                            )

+                        pytest.skip(f'Cannot parse agent output: {pid_line!r}')

                     proc2 = _spawn_data_sink(

                         emits_debug_output=emits_debug_output, proc=proc

                     )

@@ -419,11 +417,10 @@ def spawn_ssh_agent(  # noqa: C901

         else:  # pragma: no cover

             monkeypatch.delenv('SSH_AUTH_SOCK', raising=False)

         exec_name, spawn_func, agent_type = request.param

-        match exec_name:

-            case '(system)':

+        # Use match/case here once Python 3.9 becomes unsupported.

+        if exec_name == '(system)':

             assert (

-                    os.environ.get('SSH_AUTH_SOCK', None)

-                    == startup_ssh_auth_sock

+                os.environ.get('SSH_AUTH_SOCK', None) == startup_ssh_auth_sock

             ), 'SSH_AUTH_SOCK mismatch when checking for running agent'

             try:

                 client = ssh_agent.SSHAgentClient()

@@ -439,25 +436,22 @@ def spawn_ssh_agent(  # noqa: C901

                 assert (

                     os.environ.get('SSH_AUTH_SOCK', None)

                     == startup_ssh_auth_sock

-                    ), 'SSH_AUTH_SOCK mismatch before setting up for running agent'  # noqa: E501

+                ), 'SSH_AUTH_SOCK mismatch before setting up for running agent'

                 yield tests.SpawnedSSHAgentInfo(agent_type, client, False)

             assert (

-                    os.environ.get('SSH_AUTH_SOCK', None)

-                    == startup_ssh_auth_sock

+                os.environ.get('SSH_AUTH_SOCK', None) == startup_ssh_auth_sock

             ), 'SSH_AUTH_SOCK mismatch after returning from running agent'

             return

-            case _:

+        else:

             assert (

-                    os.environ.get('SSH_AUTH_SOCK', None)

-                    == startup_ssh_auth_sock

+                os.environ.get('SSH_AUTH_SOCK', None) == startup_ssh_auth_sock

             ), f'SSH_AUTH_SOCK mismatch when checking for spawnable {exec_name}'  # noqa: E501

             spawn_data = spawn_func(

                 executable=shutil.which(exec_name), env=agent_env

             )

             assert (

-                    os.environ.get('SSH_AUTH_SOCK', None)

-                    == startup_ssh_auth_sock

+                os.environ.get('SSH_AUTH_SOCK', None) == startup_ssh_auth_sock

             ), f'SSH_AUTH_SOCK mismatch after spawning {exec_name}'

             if spawn_data is None:  # pragma: no cover

                 pytest.skip(f'Cannot spawn usable {exec_name}')

@@ -483,7 +477,7 @@ def spawn_ssh_agent(  # noqa: C901

                 assert (

                     os.environ.get('SSH_AUTH_SOCK', None)

                     == startup_ssh_auth_sock

-                    ), f'SSH_AUTH_SOCK mismatch before spawning {exec_name} helper'  # noqa: E501

+                ), f'SSH_AUTH_SOCK mismatch before spawning {exec_name} helper'

                 proc2 = _spawn_data_sink(

                     emits_debug_output=emits_debug_output, proc=proc

                 )

@@ -492,7 +486,7 @@ def spawn_ssh_agent(  # noqa: C901

                 assert (

                     os.environ.get('SSH_AUTH_SOCK', None)

                     == startup_ssh_auth_sock

-                    ), f'SSH_AUTH_SOCK mismatch after spawning {exec_name} helper'  # noqa: E501

+                ), f'SSH_AUTH_SOCK mismatch after spawning {exec_name} helper'

                 monkeypatch2 = exit_stack.enter_context(

                     pytest.MonkeyPatch.context()

                 )

@@ -507,8 +501,7 @@ def spawn_ssh_agent(  # noqa: C901

                 exit_stack.enter_context(client)

                 yield tests.SpawnedSSHAgentInfo(agent_type, client, True)

             assert (

-                    os.environ.get('SSH_AUTH_SOCK', None)

-                    == startup_ssh_auth_sock

+                os.environ.get('SSH_AUTH_SOCK', None) == startup_ssh_auth_sock

             ), f'SSH_AUTH_SOCK mismatch after tearing down {exec_name}'

             return

@@ -1568,13 +1568,13 @@ Boo.

                 ssh_agent.SSHAgentClient, 'list_keys', tests.list_keys

             )

             hint: ssh_agent.SSHAgentClient | socket.socket | None

-            match conn_hint:

-                case 'client':

+            # Use match/case here once Python 3.9 becomes unsupported.

+            if conn_hint == 'client':

                 hint = ssh_agent.SSHAgentClient()

-                case 'socket':

+            elif conn_hint == 'socket':

                 hint = socket.socket(family=socket.AF_UNIX)

                 hint.connect(running_ssh_agent)

-                case _:

+            else:

                 assert conn_hint == 'none'

                 hint = None

             exception: Exception | None = None

@@ -430,7 +430,7 @@ class TestAgentInteraction:

             del request_code

             del payload

             if isinstance(  # pragma: no branch

-                response_code, int | _types.SSH_AGENT

+                response_code, (int, _types.SSH_AGENT)

             ):

                 response_code = frozenset({response_code})

             if response_code is not None:  # pragma: no branch

@@ -507,7 +507,7 @@ class TestAgentInteraction:

             del request_code

             del payload

             if isinstance(  # pragma: no branch

-                response_code, int | _types.SSH_AGENT

+                response_code, (int, _types.SSH_AGENT)

             ):

                 response_code = frozenset({response_code})

             if response_code is not None:  # pragma: no branch

@@ -7,11 +7,12 @@

 from __future__ import annotations

 import math

-from typing import TYPE_CHECKING, TypeAlias, TypeVar

+from typing import TYPE_CHECKING

 import hypothesis

 import pytest

 from hypothesis import strategies

+from typing_extensions import TypeAlias, TypeVar

 import derivepassphrase