Add finished command-line interface, with tests (6e05ad2) - derivepassphrase.git

Add finished command-line interface, with tests

Marco Ricci commited on 2024-06-22 21:19:30
Zeige 9 geänderte Dateien mit 1422 Einfügungen und 50 Löschungen.

Add a finished command-line interface, with interface parity with
vault v0.3.  The implementation has 100% coverage, including the
tests... even if the tests still contain a lot of manual code and data
duplication.  The project settings were updated as well, mostly
expanding the scope of what is covered by type checking and what is
(not) covered by coverage testing.

.gitignore 2f50334..1e6d786
mkdocs.yml cac2b31..d719751
pyproject.toml 3c7f40d..764ea0a
src/derivepassphrase/__main__.py 0fd647a..ef4fb01
src/derivepassphrase/cli.py 5a99b2e..e434a9d
tests/test_derivepassphrase_cli.py 97ad685..c514397
tests/test_passphrase_generating.py
tests/test_key_signing.py 002d1e8..df99b84
tests/test_sequin.py ab8563e..80da068

.gitignore

Zeige Datei @ 6e05ad2

...	...	@@ -3,3 +3,4 @@ html/
3	3	.pytest_cache/
4	4	__pycache__/
5	5	*.swp
	6	+.coverage*

mkdocs.yml

Zeige Datei @ 6e05ad2

@@ -52,6 +52,7 @@ plugins:
         python:
           import:
             - https://docs.python.org/3/objects.inv
+            - https://click.palletsprojects.com/en/8.1.x/objects.inv
           options:
             docstring_options:
               ignore_init_summary: true
@@ -82,6 +83,7 @@ nav:
     - Module derivepassphrase: reference/derivepassphrase.md
     - Module sequin: reference/sequin.md
     - Module ssh_agent_client: reference/ssh_agent_client.md
+    - Coverage: reference/coverage/index.html
   #- explanation.md
 
 markdown_extensions:

pyproject.toml

Zeige Datei @ 6e05ad2

@@ -43,6 +43,7 @@ Source = "https://github.com/the-13th-letter/derivepassphrase"
 derivepassphrase = "derivepassphrase.cli:derivepassphrase"
 
 [tool.mypy]
+files = ['src/**/*.py']
 mypy_path = '$MYPY_CONFIG_FILE_DIR/src'
 explicit_package_bases = true
 implicit_reexport = false
@@ -84,21 +85,32 @@ extra-dependencies = [
 [tool.hatch.envs.types.scripts]
 check = "mypy --install-types --non-interactive {args:src/derivepassphrase tests}"
 
+[tool.coverage.html]
+directory = "docs/reference/coverage"
+
 [tool.coverage.run]
-source_pkgs = ["derivepassphrase", "tests"]
+source_pkgs = ["derivepassphrase", "sequin", "ssh_agent_client", "tests"]
 branch = true
 parallel = true
 omit = [
-  "src/derivepassphrase/__main__.py",
+  "__main__.py",
 ]
 
 [tool.coverage.paths]
-derivepassphrase = ["src/derivepassphrase", "*/derivepassphrase/src/derivepassphrase"]
-tests = ["tests", "*/derivepassphrase/tests"]
+src = ["src"]
+tests = ["tests"]
 
 [tool.coverage.report]
-exclude_lines = [
-  "no cov",
+skip_covered = false
+skip_empty = true
+precision = 3
+partial_branches = [
+    'pragma: no branch',
+]
+exclude_also = [
   "if __name__ == .__main__.:",
-  "if TYPE_CHECKING:",
+  'if (?:typing\.)?TYPE_CHECKING:',
+  "raise AssertionError",
+  "raise NotImplementedError",
+  'assert False',
 ]

src/derivepassphrase/__main__.py

Zeige Datei @ 6e05ad2

@@ -1,6 +1,8 @@
 # SPDX-FileCopyrightText: 2024 Marco Ricci <m@the13thletter.info>
 #
 # SPDX-License-Identifier: MIT
+"""Run [`derivepassphrase.cli.derivepassphrase`][] on import."""
+
 import sys
 
 if __name__ == "__main__":

src/derivepassphrase/cli.py

Zeige Datei @ 6e05ad2

@@ -8,14 +8,24 @@
 
 from __future__ import annotations
 
+import base64
+import collections
+from collections.abc import MutableMapping
+import contextlib
 import inspect
 import json
+import os
 import pathlib
-from typing import Any, TextIO
+import socket
+from typing import (
+    Any, assert_never, cast, reveal_type, Iterator, Never, NotRequired,
+    Sequence, TextIO, TypedDict, TYPE_CHECKING,
+)
 
 import click
 import derivepassphrase as dpp
 from derivepassphrase import types as dpp_types
+import ssh_agent_client
 
 __author__ = dpp.__author__
 __version__ = dpp.__version__
@@ -92,6 +102,202 @@ def _save_config(config: dpp_types.VaultConfig, /) -> None:
         json.dump(config, fileobj)
 
 
+def _get_suitable_ssh_keys(
+    conn: ssh_agent_client.SSHAgentClient | socket.socket | None = None,
+    /
+) -> Iterator[ssh_agent_client.types.KeyCommentPair]:
+    """Yield all SSH keys suitable for passphrase derivation.
+
+    Suitable SSH keys are queried from the running SSH agent (see
+    [`ssh_agent_client.SSHAgentClient.list_keys`][]).
+
+    Args:
+        conn:
+            An optional connection hint to the SSH agent; specifically,
+            an SSH agent client, or a socket connected to an SSH agent.
+
+            If an existing SSH agent client, then this client will be
+            queried for the SSH keys, and otherwise left intact.
+
+            If a socket, then a one-shot client will be constructed
+            based on the socket to query the agent, and deconstructed
+            afterwards.
+
+            If neither are given, then the agent's socket location is
+            looked up in the `SSH_AUTH_SOCK` environment variable, and
+            used to construct/deconstruct a one-shot client, as in the
+            previous case.
+
+    Yields:
+        :
+            Every SSH key from the SSH agent that is suitable for
+            passphrase derivation.
+
+    Raises:
+        RuntimeError:
+            There was an error communicating with the SSH agent.
+        RuntimeError:
+            No keys usable for passphrase derivation are loaded into the
+            SSH agent.
+
+    """
+    client: ssh_agent_client.SSHAgentClient
+    client_context: contextlib.AbstractContextManager
+    match conn:
+        case ssh_agent_client.SSHAgentClient():
+            client = conn
+            client_context = contextlib.nullcontext()
+        case socket.socket() | None:
+            client = ssh_agent_client.SSHAgentClient(socket=conn)
+            client_context = client
+        case _:  # pragma: no cover
+            assert_never(conn)
+            raise TypeError(f'invalid connection hint: {conn!r}')
+    with client_context:
+        try:
+            all_key_comment_pairs = list(client.list_keys())
+        except EOFError as e:  # pragma: no cover
+            raise RuntimeError(
+                'error communicating with the SSH agent'
+            ) from e
+    suitable_keys = all_key_comment_pairs[:]
+    for pair in all_key_comment_pairs:
+        key, comment = pair
+        if dpp.Vault._is_suitable_ssh_key(key):
+            yield pair
+    if not suitable_keys:  # pragma: no cover
+        raise RuntimeError('No usable SSH keys were found')
+
+
+def _prompt_for_selection(
+    items: Sequence[str | bytes], heading: str = 'Possible choices:',
+    single_choice_prompt: str = 'Confirm this choice?',
+) -> int:
+    """Prompt user for a choice among the given items.
+
+    Print the heading, if any, then present the items to the user.  If
+    there are multiple items, prompt the user for a selection, validate
+    the choice, then return the list index of the selected item.  If
+    there is only a single item, request confirmation for that item
+    instead, and return the correct index.
+
+    Args:
+        heading:
+            A heading for the list of items, to print immediately
+            before.  Defaults to a reasonable standard heading.  If
+            explicitly empty, print no heading.
+        single_choice_prompt:
+            The confirmation prompt if there is only a single possible
+            choice.  Defaults to a reasonable standard prompt.
+
+    Returns:
+        An index into the items sequence, indicating the user's
+        selection.
+
+    Raises:
+        IndexError:
+            The user made an invalid or empty selection, or requested an
+            abort.
+
+    """
+    n = len(items)
+    if heading:
+        click.echo(click.style(heading, bold=True))
+    for i, x in enumerate(items, start=1):
+        click.echo(click.style(f'[{i}]', bold=True), nl=False)
+        click.echo(' ', nl=False)
+        click.echo(x)
+    if n > 1:
+        choices = click.Choice([''] + [str(i) for i in range(1, n + 1)])
+        choice = click.prompt(
+            f'Your selection? (1-{n}, leave empty to abort)',
+            err=True, type=choices, show_choices=False,
+            show_default=False, default='')
+        if not choice:
+            raise IndexError('empty selection')
+        return int(choice) - 1
+    else:
+        prompt_suffix = (' '
+                         if single_choice_prompt.endswith(tuple('?.!'))
+                         else ': ')
+        try:
+            click.confirm(single_choice_prompt,
+                          prompt_suffix=prompt_suffix, err=True,
+                          abort=True, default=False, show_default=False)
+        except click.Abort:
+            raise IndexError('empty selection') from None
+        return 0
+
+
+def _select_ssh_key(
+    conn: ssh_agent_client.SSHAgentClient | socket.socket | None = None,
+    /
+) -> bytes | bytearray:
+    """Interactively select an SSH key for passphrase derivation.
+
+    Suitable SSH keys are queried from the running SSH agent (see
+    [`ssh_agent_client.SSHAgentClient.list_keys`][]), then the user is
+    prompted interactively (see [`click.prompt`][]) for a selection.
+
+    Args:
+        conn:
+            An optional connection hint to the SSH agent; specifically,
+            an SSH agent client, or a socket connected to an SSH agent.
+
+            If an existing SSH agent client, then this client will be
+            queried for the SSH keys, and otherwise left intact.
+
+            If a socket, then a one-shot client will be constructed
+            based on the socket to query the agent, and deconstructed
+            afterwards.
+
+            If neither are given, then the agent's socket location is
+            looked up in the `SSH_AUTH_SOCK` environment variable, and
+            used to construct/deconstruct a one-shot client, as in the
+            previous case.
+
+    Returns:
+        The selected SSH key.
+
+    Raises:
+        IndexError:
+            The user made an invalid or empty selection, or requested an
+            abort.
+        RuntimeError:
+            There was an error communicating with the SSH agent.
+        RuntimeError:
+            No keys usable for passphrase derivation are loaded into the
+            SSH agent.
+    """
+    suitable_keys = list(_get_suitable_ssh_keys(conn))
+    key_listing: list[str] = []
+    unstring_prefix = ssh_agent_client.SSHAgentClient.unstring_prefix
+    for key, comment in suitable_keys:
+        keytype = unstring_prefix(key)[0].decode('ASCII')
+        key_str = base64.standard_b64encode(key).decode('ASCII')
+        key_prefix = key_str if len(key_str) < 30 else key_str[:27] + '...'
+        comment_str = comment.decode('UTF-8', errors='replace')
+        key_listing.append(f'{keytype} {key_prefix} {comment_str}')
+    choice = _prompt_for_selection(
+        key_listing, heading='Suitable SSH keys:',
+        single_choice_prompt='Use this key?')
+    return suitable_keys[choice].key
+
+
+def _prompt_for_passphrase() -> str:
+    """Interactively prompt for the passphrase.
+
+    Calls [`click.prompt`][] internally.  Moved into a separate function
+    mainly for testing/mocking purposes.
+
+    Returns:
+        The user input.
+
+    """
+    return click.prompt('Passphrase', default='', hide_input=True,
+                        show_default=False, err=True)
+
+
 class OptionGroupOption(click.Option):
     """A [`click.Option`][] with an associated group name and group epilog.
 
@@ -240,15 +446,36 @@ def _validate_length(
         raise click.BadParameter('not a positive integer')
     return int_value
 
+DEFAULT_NOTES_TEMPLATE = '''\
+# Enter notes below the line with the cut mark (ASCII scissors and
+# dashes).  Lines above the cut mark (such as this one) will be ignored.
+#
+# If you wish to clear the notes, leave everything beyond the cut mark
+# blank.  However, if you leave the *entire* file blank, also removing
+# the cut mark, then the edit is aborted, and the old notes contents are
+# retained.
+#
+# - - - - - >8 - - - - - >8 - - - - - >8 - - - - - >8 - - - - -
+'''
+DEFAULT_NOTES_MARKER = '# - - - - - >8 - - - - -'
+
+
 @click.command(
     context_settings={"help_option_names": ["-h", "--help"]},
     cls=CommandWithHelpGroups,
-    epilog='''
+    epilog=r'''
         WARNING: There is NO WAY to retrieve the generated passphrases
         if the master passphrase, the SSH key, or the exact passphrase
         settings are lost, short of trying out all possible
         combinations.  You are STRONGLY advised to keep independent
         backups of the settings and the SSH key, if any.
+
+        Configuration is stored in a directory according to the
+        DERIVEPASSPHRASE_PATH variable, which defaults to
+        `~/.derivepassphrase` on UNIX-like systems and
+        `C:\Users\<user>\AppData\Roaming\Derivepassphrase` on Windows.
+        The configuration is NOT encrypted, and you are STRONGLY
+        discouraged from using a stored passphrase.
     ''',
 )
 @click.option('-p', '--phrase', 'use_phrase', is_flag=True,
@@ -513,11 +740,166 @@ def derivepassphrase(
             opt_str = param.opts[0]
             raise click.UsageError(
                 f'{opt_str} does not take a SERVICE argument')
-    #if kwargs['length'] is None:
-    #    kwargs['length'] = dpp.Vault.__init__.__kwdefaults__['length']
-    #if kwargs['repeat'] is None:
-    #    kwargs['repeat'] = dpp.Vault.__init__.__kwdefaults__['repeat']
-    click.echo(repr(ctx.params))
+
+    if edit_notes:
+        assert service is not None
+        configuration = get_config()
+        text = (DEFAULT_NOTES_TEMPLATE +
+                configuration['services']
+                .get(service, cast(dpp_types.VaultConfigServicesSettings, {}))
+                .get('notes', ''))
+        notes_value = click.edit(text=text)
+        if notes_value is not None:
+            notes_lines = collections.deque(notes_value.splitlines(True))
+            while notes_lines:
+                line = notes_lines.popleft()
+                if line.startswith(DEFAULT_NOTES_MARKER):
+                    notes_value = ''.join(notes_lines)
+                    break
+            else:
+                if not notes_value.strip():
+                    ctx.fail('not saving new notes: user aborted request')
+            configuration['services'].setdefault(service, {})['notes'] = (
+                notes_value.strip('\n'))
+            _save_config(configuration)
+    elif delete_service_settings:
+        assert service is not None
+        configuration = get_config()
+        if service in configuration['services']:
+            del configuration['services'][service]
+            _save_config(configuration)
+    elif delete_globals:
+        configuration = get_config()
+        if 'global' in configuration:
+            del configuration['global']
+            _save_config(configuration)
+    elif clear_all_settings:
+        _save_config({'services': {}})
+    elif import_settings:
+        try:
+            # TODO: keep track of auto-close; try os.dup if feasible
+            infile = (cast(TextIO, import_settings)
+                      if hasattr(import_settings, 'close')
+                      else click.open_file(os.fspath(import_settings), 'rt'))
+            with infile:
+                maybe_config = json.load(infile)
+        except json.JSONDecodeError as e:
+            ctx.fail(f'Cannot load config: cannot decode JSON: {e}')
+        except OSError as e:
+            ctx.fail(f'Cannot load config: {e.strerror}')
+        if dpp_types.is_vault_config(maybe_config):
+            _save_config(maybe_config)
+        else:
+            ctx.fail('not a valid config')
+    elif export_settings:
+        configuration = get_config()
+        try:
+            # TODO: keep track of auto-close; try os.dup if feasible
+            outfile = (cast(TextIO, export_settings)
+                       if hasattr(export_settings, 'close')
+                       else click.open_file(os.fspath(export_settings), 'wt'))
+            with outfile:
+                json.dump(configuration, outfile)
+        except OSError as e:
+            ctx.fail('cannot write config: {e.strerror}')
+    else:
+        configuration = get_config()
+        # This block could be type checked more stringently, but this
+        # would probably involve a lot of code repetition.  Since we
+        # have a type guarding function anyway, assert that we didn't
+        # make any mistakes at the end instead.
+        global_keys = {'key', 'phrase'}
+        service_keys = {'key', 'phrase', 'length', 'repeat', 'lower',
+                        'upper', 'number', 'space', 'dash', 'symbol'}
+        settings: collections.ChainMap[str, Any] = collections.ChainMap(
+            {k: v for k, v in locals().items()
+             if k in service_keys and v is not None},
+            cast(dict[str, Any],
+                 configuration['services'].get(service or '', {})),
+            {},
+            cast(dict[str, Any], configuration.get('global', {}))
+        )
+        if use_key:
+            try:
+                key = base64.standard_b64encode(
+                    _select_ssh_key()).decode('ASCII')
+            except IndexError:
+                ctx.fail('no valid SSH key selected')
+            except RuntimeError as e:
+                ctx.fail(str(e))
+        elif use_phrase:
+            maybe_phrase = _prompt_for_passphrase()
+            if not maybe_phrase:
+                ctx.fail('no passphrase given')
+            else:
+                phrase = maybe_phrase
+        if store_config_only:
+            view: collections.ChainMap[str, Any]
+            view = (collections.ChainMap(*settings.maps[:2]) if service
+                    else settings.parents.parents)
+            if use_key:
+                view['key'] = key
+                for m in view.maps:
+                    m.pop('phrase', '')
+            elif use_phrase:
+                view['phrase'] = phrase
+                for m in view.maps:
+                    m.pop('key', '')
+            if service:
+                if not view.maps[0]:
+                    raise click.UsageError('cannot update service settings '
+                                           'without actual settings')
+                else:
+                    configuration['services'].setdefault(
+                        service, {}).update(view)  # type: ignore[typeddict-item]
+            else:
+                if not view.maps[0]:
+                    raise click.UsageError('cannot update global settings '
+                                           'without actual settings')
+                else:
+                    configuration.setdefault(
+                        'global', {}).update(view)  # type: ignore[typeddict-item]
+            assert dpp_types.is_vault_config(configuration), (
+                f'invalid vault configuration: {configuration!r}'
+            )
+            _save_config(configuration)
+        else:
+            if not service:
+                raise click.UsageError(f'SERVICE is required')
+            kwargs: dict[str, Any] = {k: v for k, v in settings.items()
+                                      if k in service_keys and v is not None}
+            # If either --key or --phrase are given, use that setting.
+            # Otherwise, if both key and phrase are set in the config,
+            # one must be global (ignore it) and one must be
+            # service-specific (use that one). Otherwise, if only one of
+            # key and phrase is set in the config, use that one.  In all
+            # these above cases, set the phrase via
+            # derivepassphrase.Vault.phrase_from_key if a key is
+            # given. Finally, if nothing is set, error out.
+            key_to_phrase = lambda key: dpp.Vault.phrase_from_key(
+                base64.standard_b64decode(key))
+            if use_key or use_phrase:
+                if use_key:
+                    kwargs['phrase'] = key_to_phrase(key)
+                else:
+                    kwargs['phrase'] = phrase
+                    kwargs.pop('key', '')
+            elif kwargs.get('phrase') and kwargs.get('key'):
+                if any('key' in m for m in settings.maps[:2]):
+                    kwargs['phrase'] = key_to_phrase(kwargs.pop('key'))
+                else:
+                    kwargs.pop('key')
+            elif kwargs.get('key'):
+                kwargs['phrase'] = key_to_phrase(kwargs.pop('key'))
+            elif kwargs.get('phrase'):
+                pass
+            else:
+                raise click.UsageError(
+                    'no passphrase or key given on command-line '
+                    'or in configuration')
+            vault = dpp.Vault(**kwargs)
+            result = vault.generate(service)
+            click.echo(result.decode('ASCII'))
 
 
 if __name__ == '__main__':

tests/test_derivepassphrase_cli.py

Zeige Datei @ 6e05ad2

@@ -2,17 +2,308 @@
 #
 # SPDX-License-Identifier: MIT
 
+from __future__ import annotations
+
+import base64
+import contextlib
+import errno
+import json
+import os
+from typing import Any, cast, TYPE_CHECKING, NamedTuple
+
 import click.testing
-import derivepassphrase
-import derivepassphrase.cli
+import derivepassphrase as dpp
+import derivepassphrase.cli as cli
+import ssh_agent_client.types
 import pytest
 
 DUMMY_SERVICE = 'service1'
 DUMMY_PASSPHRASE = b'my secret passphrase\n'
+DUMMY_CONFIG_SETTINGS = {"length": 10, "upper": 1, "lower": 1, "repeat": 5,
+                         "number": 1, "space": 1, "dash": 1, "symbol": 1}
+DUMMY_RESULT_PASSPHRASE = b'.2V_QJkd o'
+DUMMY_RESULT_KEY1 = b'E<b<{ -7iG'
+DUMMY_PHRASE_FROM_KEY1_RAW = (
+    b'\x00\x00\x00\x0bssh-ed25519'
+    b'\x00\x00\x00@\xf0\x98\x19\x80l\x1a\x97\xd5&\x03n'
+    b'\xcc\xe3e\x8f\x86f\x07\x13\x19\x13\t!33\xf9\xe46S'
+    b'\x1d\xaf\xfd\r\x08\x1f\xec\xf8s\x9b\x8c_U9\x16|ST,'
+    b'\x1eR\xbb0\xed\x7f\x89\xe2/iQU\xd8\x9e\xa6\x02'
+)
+DUMMY_PHRASE_FROM_KEY1 = b'8JgZgGwal9UmA27M42WPhmYHExkTCSEzM/nkNlMdr/0NCB/s+HObjF9VORZ8U1QsHlK7MO1/ieIvaVFV2J6mAg=='
+
+# See Ed25519 and RSA test keys in test_key_signing.py
+DUMMY_KEY1 = 'AAAAC3NzaC1lZDI1NTE5AAAAIIF4gWgm1gJIXw//Mkhv5MEwidwcakUGCekJD/vCEml2'
+DUMMY_KEY2 = 'AAAAB3NzaC1yc2EAAAADAQABAAABgQCxoe7pezhxWy4NI0mUwKqg9WCYOAS+IjxN9eYcqpfcmQiojcuy9XsiN/xYJ1O94SrsKS5mEia2xHnYA4RUChTyYNcM2v6cnnBQ/N/VQhpGMN7SVxdbhKUXTWFCwbjBgO6rGyHB6WtoH8vd7TOEPt+NgcXwhsWyoaUUdYTA62V+GF9vEmxMaC4ubgDz+B0QkPnauSoNxmkhcIe0lsLNb1pClZyz88PDnKXCX/d0HuN/HJ+sbPg7dCvOyqFYSyKn3uY6bCXqoIdurxXzH3O7z0P8f5sbmKOrGGKNuNxVRbeVl/D/3uDL0nqsbfUc1qvkfwbJwtMXC4IV6kOZMSk2BAsqh7x48gQ+rhYeEVSi8F3CWs4HJQoqrGt7K9a3mCSlMBHP70u3w6ME7eumoryxlUofewTd17ZEkzdX08l2ZlKzZvwQUrc+xQZ2Uw8z2mfW6Ti4gi0pYGaig7Ke4PwuXpo/C5YAWfeXycsvJZ2uaYRjMdZeJGNAnHLUGLkBscw5aI8='
+
+
+class IncompatibleConfiguration(NamedTuple):
+    other_options: list[tuple[str, ...]]
+    needs_service: bool | None
+    input: bytes | None
+
+class SingleConfiguration(NamedTuple):
+    needs_service: bool | None
+    input: bytes | None
+    check_success: bool
+
+class OptionCombination(NamedTuple):
+    options: list[str]
+    incompatible: bool
+    needs_service: bool | None
+    input: bytes | None
+    check_success: bool
+
+PASSWORD_GENERATION_OPTIONS: list[tuple[str, ...]] = [
+    ('--phrase',), ('--key',), ('--length', '20'), ('--repeat', '20'),
+    ('--lower', '1'), ('--upper', '1'), ('--number', '1'),
+    ('--space', '1'), ('--dash', '1'), ('--symbol', '1')
+]
+CONFIGURATION_OPTIONS: list[tuple[str, ...]] = [
+    ('--notes',), ('--config',), ('--delete',), ('--delete-globals',),
+    ('--clear',)
+]
+CONFIGURATION_COMMANDS: list[tuple[str, ...]] = [
+    ('--notes',), ('--delete',), ('--delete-globals',), ('--clear',)
+]
+STORAGE_OPTIONS: list[tuple[str, ...]] = [
+    ('--export', '-'), ('--import', '-')
+]
+INCOMPATIBLE: dict[tuple[str, ...], IncompatibleConfiguration] = {
+    ('--phrase',): IncompatibleConfiguration(
+        [('--key',)] + CONFIGURATION_COMMANDS + STORAGE_OPTIONS,
+        True, DUMMY_PASSPHRASE),
+    ('--key',): IncompatibleConfiguration(
+        CONFIGURATION_COMMANDS + STORAGE_OPTIONS,
+        True, DUMMY_PASSPHRASE),
+    ('--length', '20'): IncompatibleConfiguration(
+        CONFIGURATION_COMMANDS + STORAGE_OPTIONS,
+        True, DUMMY_PASSPHRASE),
+    ('--repeat', '20'): IncompatibleConfiguration(
+        CONFIGURATION_COMMANDS + STORAGE_OPTIONS,
+        True, DUMMY_PASSPHRASE),
+    ('--lower', '1'): IncompatibleConfiguration(
+        CONFIGURATION_COMMANDS + STORAGE_OPTIONS,
+        True, DUMMY_PASSPHRASE),
+    ('--upper', '1'): IncompatibleConfiguration(
+        CONFIGURATION_COMMANDS + STORAGE_OPTIONS,
+        True, DUMMY_PASSPHRASE),
+    ('--number', '1'): IncompatibleConfiguration(
+        CONFIGURATION_COMMANDS + STORAGE_OPTIONS,
+        True, DUMMY_PASSPHRASE),
+    ('--space', '1'): IncompatibleConfiguration(
+        CONFIGURATION_COMMANDS + STORAGE_OPTIONS,
+        True, DUMMY_PASSPHRASE),
+    ('--dash', '1'): IncompatibleConfiguration(
+        CONFIGURATION_COMMANDS + STORAGE_OPTIONS,
+        True, DUMMY_PASSPHRASE),
+    ('--symbol', '1'): IncompatibleConfiguration(
+        CONFIGURATION_COMMANDS + STORAGE_OPTIONS,
+        True, DUMMY_PASSPHRASE),
+    ('--notes',): IncompatibleConfiguration(
+        [('--config',), ('--delete',), ('--delete-globals',),
+         ('--clear',)] + STORAGE_OPTIONS,
+        True, None),
+    ('--config', '-p'): IncompatibleConfiguration(
+        [('--delete',), ('--delete-globals',),
+         ('--clear',)] + STORAGE_OPTIONS,
+        None, DUMMY_PASSPHRASE),
+    ('--delete',): IncompatibleConfiguration(
+        [('--delete-globals',), ('--clear',)] + STORAGE_OPTIONS, True, None),
+    ('--delete-globals',): IncompatibleConfiguration(
+        [('--clear',)] + STORAGE_OPTIONS, False, None),
+    ('--clear',): IncompatibleConfiguration(STORAGE_OPTIONS, False, None),
+    ('--export', '-'): IncompatibleConfiguration(
+        [('--import', '-')], False, None),
+    ('--import', '-'): IncompatibleConfiguration(
+        [], False, None),
+}
+SINGLES: dict[tuple[str, ...], SingleConfiguration] = {
+    ('--phrase',): SingleConfiguration(True, DUMMY_PASSPHRASE, True),
+    ('--key',): SingleConfiguration(True, None, False),
+    ('--length', '20'): SingleConfiguration(True, DUMMY_PASSPHRASE, True),
+    ('--repeat', '20'): SingleConfiguration(True, DUMMY_PASSPHRASE, True),
+    ('--lower', '1'): SingleConfiguration(True, DUMMY_PASSPHRASE, True),
+    ('--upper', '1'): SingleConfiguration(True, DUMMY_PASSPHRASE, True),
+    ('--number', '1'): SingleConfiguration(True, DUMMY_PASSPHRASE, True),
+    ('--space', '1'): SingleConfiguration(True, DUMMY_PASSPHRASE, True),
+    ('--dash', '1'): SingleConfiguration(True, DUMMY_PASSPHRASE, True),
+    ('--symbol', '1'): SingleConfiguration(True, DUMMY_PASSPHRASE, True),
+    ('--notes',): SingleConfiguration(True, None, False),
+    ('--config', '-p'): SingleConfiguration(None, DUMMY_PASSPHRASE, False),
+    ('--delete',): SingleConfiguration(True, None, False),
+    ('--delete-globals',): SingleConfiguration(False, None, True),
+    ('--clear',): SingleConfiguration(False, None, True),
+    ('--export', '-'): SingleConfiguration(False, None, True),
+    ('--import', '-'): SingleConfiguration(False, b'{"services": {}}', True),
+}
+INTERESTING_OPTION_COMBINATIONS: list[OptionCombination] = []
+config: OptionCombination | SingleConfiguration
+for opt, config in INCOMPATIBLE.items():
+    for opt2 in config.other_options:
+        INTERESTING_OPTION_COMBINATIONS.extend([
+            OptionCombination(options=list(opt + opt2), incompatible=True,
+                              needs_service=config.needs_service,
+                              input=config.input, check_success=False),
+            OptionCombination(options=list(opt2 + opt), incompatible=True,
+                              needs_service=config.needs_service,
+                              input=config.input, check_success=False)
+        ])
+for opt, config in SINGLES.items():
+    INTERESTING_OPTION_COMBINATIONS.append(
+        OptionCombination(options=list(opt), incompatible=False,
+                          needs_service=config.needs_service,
+                          input=config.input,
+                          check_success=config.check_success))
+
+@contextlib.contextmanager
+def isolated_config(
+    monkeypatch: Any, runner: click.testing.CliRunner, config: Any,
+):
+    with runner.isolated_filesystem():
+        monkeypatch.setenv('HOME', os.getcwd())
+        monkeypatch.setenv('USERPROFILE', os.getcwd())
+        monkeypatch.delenv(cli.prog_name.replace(' ', '_').upper() + '_PATH',
+                           raising=False)
+        os.makedirs(os.path.dirname(cli._config_filename()), exist_ok=True)
+        with open(cli._config_filename(), 'wt') as outfile:
+            json.dump(config, outfile)
+        yield
+
+
+def test_100_save_bad_config(monkeypatch: Any) -> None:
+    runner = click.testing.CliRunner()
+    with isolated_config(monkeypatch=monkeypatch, runner=runner, config={}):
+        with pytest.raises(ValueError, match='Invalid vault config'):
+            cli._save_config(None)  # type: ignore
+
+
+def test_101_prompt_for_selection_multiple(monkeypatch: Any) -> None:
+    @click.command()
+    @click.option('--heading', default='Our menu:')
+    @click.argument('items', nargs=-1)
+    def driver(heading, items):
+        # from https://montypython.fandom.com/wiki/Spam#The_menu
+        items = items or [
+            'Egg and bacon',
+            'Egg, sausage and bacon',
+            'Egg and spam',
+            'Egg, bacon and spam',
+            'Egg, bacon, sausage and spam',
+            'Spam, bacon, sausage and spam',
+            'Spam, egg, spam, spam, bacon and spam',
+            'Spam, spam, spam, egg and spam',
+            ('Spam, spam, spam, spam, spam, spam, baked beans, '
+             'spam, spam, spam and spam'),
+            ('Lobster thermidor aux crevettes with a mornay sauce '
+             'garnished with truffle paté, brandy '
+             'and a fried egg on top and spam'),
+        ]
+        index = cli._prompt_for_selection(items, heading=heading)
+        click.echo('A fine choice: ', nl=False)
+        click.echo(items[index])
+        click.echo('(Note: Vikings strictly optional.)')
+    runner = click.testing.CliRunner(mix_stderr=True)
+    result = runner.invoke(driver, [], input='9')
+    assert result.exit_code == 0, 'driver program failed'
+    assert result.stdout == '''\
+Our menu:
+[1] Egg and bacon
+[2] Egg, sausage and bacon
+[3] Egg and spam
+[4] Egg, bacon and spam
+[5] Egg, bacon, sausage and spam
+[6] Spam, bacon, sausage and spam
+[7] Spam, egg, spam, spam, bacon and spam
+[8] Spam, spam, spam, egg and spam
+[9] Spam, spam, spam, spam, spam, spam, baked beans, spam, spam, spam and spam
+[10] Lobster thermidor aux crevettes with a mornay sauce garnished with truffle paté, brandy and a fried egg on top and spam
+Your selection? (1-10, leave empty to abort): 9
+A fine choice: Spam, spam, spam, spam, spam, spam, baked beans, spam, spam, spam and spam
+(Note: Vikings strictly optional.)
+''', 'driver program produced unexpected output'
+    result = runner.invoke(driver, ['--heading='], input='',
+                           catch_exceptions=True)
+    assert result.exit_code > 0, 'driver program succeeded?!'
+    assert result.stdout == '''\
+[1] Egg and bacon
+[2] Egg, sausage and bacon
+[3] Egg and spam
+[4] Egg, bacon and spam
+[5] Egg, bacon, sausage and spam
+[6] Spam, bacon, sausage and spam
+[7] Spam, egg, spam, spam, bacon and spam
+[8] Spam, spam, spam, egg and spam
+[9] Spam, spam, spam, spam, spam, spam, baked beans, spam, spam, spam and spam
+[10] Lobster thermidor aux crevettes with a mornay sauce garnished with truffle paté, brandy and a fried egg on top and spam
+Your selection? (1-10, leave empty to abort): \n''', (
+        'driver program produced unexpected output'
+    )
+    assert isinstance(result.exception, IndexError), (
+        'driver program did not raise IndexError?!'
+    )
+
+
+def test_102_prompt_for_selection_single(monkeypatch: Any) -> None:
+    @click.command()
+    @click.option('--item', default='baked beans')
+    @click.argument('prompt')
+    def driver(item, prompt):
+        try:
+            cli._prompt_for_selection([item], heading='',
+                                      single_choice_prompt=prompt)
+        except IndexError as e:
+            click.echo('Boo.')
+            raise e
+        else:
+            click.echo('Great!')
+    runner = click.testing.CliRunner(mix_stderr=True)
+    result = runner.invoke(driver, ['Will replace with spam. Confirm, y/n?'],
+                           input='y')
+    assert result.exit_code == 0, 'driver program failed'
+    assert result.stdout == '''\
+[1] baked beans
+Will replace with spam. Confirm, y/n? y
+Great!
+''', 'driver program produced unexpected output'
+    result = runner.invoke(driver,
+                           ['Will replace with spam, okay? ' +
+                            '(Please say "y" or "n".)'],
+                           input='')
+    assert result.exit_code > 0, 'driver program succeeded?!'
+    assert result.stdout == '''\
+[1] baked beans
+Will replace with spam, okay? (Please say "y" or "n".): 
+Boo.
+''', 'driver program produced unexpected output'
+    assert isinstance(result.exception, IndexError), (
+        'driver program did not raise IndexError?!'
+    )
+
+
+def test_103_prompt_for_passphrase(monkeypatch: Any) -> None:
+    monkeypatch.setattr(click, 'prompt',
+                        lambda *a, **kw: json.dumps({'args': a, 'kwargs': kw}))
+    res = json.loads(cli._prompt_for_passphrase())
+    assert 'args' in res and 'kwargs' in res, (
+        'missing arguments to passphrase prompt'
+    )
+    assert res['args'][:1] == ['Passphrase'], (
+        'missing arguments to passphrase prompt'
+    )
+    assert (res['kwargs'].get('default') == ''
+            and not res['kwargs'].get('show_default', True)), (
+        'missing arguments to passphrase prompt'
+    )
+    assert res['kwargs'].get('err') and res['kwargs'].get('hide_input'), (
+        'missing arguments to passphrase prompt'
+    )
+
 
 def test_200_help_output():
     runner = click.testing.CliRunner(mix_stderr=False)
-    result = runner.invoke(derivepassphrase.cli.derivepassphrase, ['--help'])
+    result = runner.invoke(cli.derivepassphrase, ['--help'],
+                           catch_exceptions=False)
     assert result.exit_code == 0
     assert 'Password generation:\n' in result.output, (
         'Option groups not respected in help text.'
@@ -21,36 +312,20 @@ def test_200_help_output():
         'Option group epilog not printed.'
     )
 
-@pytest.mark.parametrize(['option'],
-                         [('--lower',), ('--upper',), ('--number',),
-                          ('--space',), ('--dash',), ('--symbol',),
-                          ('--repeat',), ('--length',)])
-def test_201_invalid_argument_range(option):
-    runner = click.testing.CliRunner(mix_stderr=False)
-    result = runner.invoke(derivepassphrase.cli.derivepassphrase,
-                           [option, '-42', '-p', DUMMY_SERVICE],
-                           input=DUMMY_PASSPHRASE)
-    assert result.exit_code > 0, (
-        f'program unexpectedly succeeded'
-    )
-    assert result.stderr_bytes, (
-        f'program did not print any error message'
-    )
-    assert b'Error: Invalid value' in result.stderr_bytes, (
-        f'program did not print the expected error message'
-    )
-
 @pytest.mark.parametrize(['charset_name'],
                          [('lower',), ('upper',), ('number',), ('space',),
                           ('dash',), ('symbol',)])
-@pytest.mark.xfail(reason='implementation not written yet')
-def test_202_disable_character_set(charset_name):
+def test_201_disable_character_set(
+    monkeypatch: Any, charset_name: str
+) -> None:
+    monkeypatch.setattr(cli, '_prompt_for_passphrase',
+                        lambda *a, **kw: DUMMY_PASSPHRASE.decode('UTF-8'))
     option = f'--{charset_name}'
-    charset = derivepassphrase.Vault._CHARSETS[charset_name].decode('ascii')
+    charset = dpp.Vault._CHARSETS[charset_name].decode('ascii')
     runner = click.testing.CliRunner(mix_stderr=False)
-    result = runner.invoke(derivepassphrase.cli.derivepassphrase,
+    result = runner.invoke(cli.derivepassphrase,
                            [option, '0', '-p', DUMMY_SERVICE],
-                           input=DUMMY_PASSPHRASE)
+                           input=DUMMY_PASSPHRASE, catch_exceptions=False)
     assert result.exit_code == 0, (
         f'program died unexpectedly with exit code {result.exit_code}'
     )
@@ -63,12 +338,13 @@ def test_202_disable_character_set(charset_name):
             f'{result.stdout!r}'
         )
 
-@pytest.mark.xfail(reason='implementation not written yet')
-def test_203_disable_repetition():
+def test_202_disable_repetition(monkeypatch: Any) -> None:
+    monkeypatch.setattr(cli, '_prompt_for_passphrase',
+                        lambda *a, **kw: DUMMY_PASSPHRASE.decode('UTF-8'))
     runner = click.testing.CliRunner(mix_stderr=False)
-    result = runner.invoke(derivepassphrase.cli.derivepassphrase,
+    result = runner.invoke(cli.derivepassphrase,
                            ['--repeat', '0', '-p', DUMMY_SERVICE],
-                           input=DUMMY_PASSPHRASE)
+                           input=DUMMY_PASSPHRASE, catch_exceptions=False)
     assert result.exit_code == 0, (
         f'program died unexpectedly with exit code {result.exit_code}'
     )
@@ -78,6 +354,544 @@ def test_203_disable_repetition():
     passphrase = result.stdout.rstrip('\r\n')
     for i in range(len(passphrase) - 1):
         assert passphrase[i:i+1] != passphrase[i+1:i+2], (
-            f'derived password contains repeated character at position {i}: '
-            f'{result.stdout!r}'
+            f'derived password contains repeated character '
+            f'at position {i}: {result.stdout!r}'
+        )
+
+@pytest.mark.parametrize(['command_line', 'config', 'result_config'], [
+    (['--delete-globals'],
+     {'global': {'phrase': 'abc'}, 'services': {}}, {'services': {}}),
+    (['--delete', DUMMY_SERVICE],
+     {'global': {'phrase': 'abc'},
+      'services': {DUMMY_SERVICE: {'notes': '...'}}},
+     {'global': {'phrase': 'abc'}, 'services': {}}),
+    (['--clear'],
+     {'global': {'phrase': 'abc'},
+      'services': {DUMMY_SERVICE: {'notes': '...'}}},
+     {'services': {}}),
+])
+def test_203_repeated_config_deletion(
+    monkeypatch: Any, command_line: list[str],
+    config: dpp.types.VaultConfig, result_config: dpp.types.VaultConfig,
+) -> None:
+    runner = click.testing.CliRunner(mix_stderr=False)
+    for start_config in [config, result_config]:
+        with isolated_config(monkeypatch=monkeypatch, runner=runner,
+                             config=start_config):
+            result = runner.invoke(cli.derivepassphrase, command_line,
+                                   catch_exceptions=False)
+            assert (result.exit_code, result.stderr_bytes) == (0, b''), (
+                'program exited with failure'
+            )
+            with open(cli._config_filename(), 'rt') as infile:
+                config_readback = json.load(infile)
+            assert config_readback == result_config
+
+def test_204_phrase_from_key_manually() -> None:
+    assert (
+        dpp.Vault(phrase=DUMMY_PHRASE_FROM_KEY1, **DUMMY_CONFIG_SETTINGS)
+        .generate(DUMMY_SERVICE) == DUMMY_RESULT_KEY1
+    )
+
+@pytest.mark.parametrize(['config'], [
+    pytest.param({'global': {'key': DUMMY_KEY1},
+                  'services': {DUMMY_SERVICE: DUMMY_CONFIG_SETTINGS}},
+                 id='global'),
+    pytest.param({'global': {'phrase': DUMMY_PASSPHRASE.rstrip(b'\n').decode('ASCII')},
+                  'services': {DUMMY_SERVICE: {'key': DUMMY_KEY1,
+                                               **DUMMY_CONFIG_SETTINGS}}},
+                 id='service'),
+])
+def test_204a_key_from_config(
+    monkeypatch: Any, config: dpp.types.VaultConfig,
+) -> None:
+    def phrase_from_key(key: bytes) -> bytes:
+        if key == base64.standard_b64decode(DUMMY_KEY1):  # pragma: no branch
+            return DUMMY_PHRASE_FROM_KEY1
+        raise KeyError(key)  # pragma: no cover
+    runner = click.testing.CliRunner(mix_stderr=False)
+    with isolated_config(monkeypatch=monkeypatch, runner=runner,
+                         config=config):
+        monkeypatch.setattr(dpp.Vault, 'phrase_from_key',
+                            phrase_from_key)
+        result = runner.invoke(cli.derivepassphrase, [DUMMY_SERVICE],
+                               catch_exceptions=False)
+        assert (result.exit_code, result.stderr_bytes) == (0, b''), (
+            'program exited with failure'
+        )
+        assert result.stdout_bytes.rstrip(b'\n') != DUMMY_RESULT_PASSPHRASE, (
+            'program generated unexpected result (phrase instead of key)'
+        )
+        assert result.stdout_bytes.rstrip(b'\n') == DUMMY_RESULT_KEY1, (
+            'program generated unexpected result (wrong settings?)'
+        )
+
+def test_204b_key_from_command_line(monkeypatch: Any) -> None:
+    KeyCommentPair = ssh_agent_client.types.KeyCommentPair
+    key_list = [
+        KeyCommentPair(base64.standard_b64decode(DUMMY_KEY1), b'no comment'),
+        KeyCommentPair(base64.standard_b64decode(DUMMY_KEY2), b'a comment'),
+    ]
+    def _suitable_ssh_keys(conn: Any) -> Iterator[KeyCommentPair]:
+        yield from key_list
+    def phrase_from_key(key: bytes) -> bytes:
+        if key == base64.standard_b64decode(DUMMY_KEY1):  # pragma: no branch
+            return DUMMY_PHRASE_FROM_KEY1
+        raise KeyError(key)  # pragma: no cover
+    runner = click.testing.CliRunner(mix_stderr=False)
+    with isolated_config(monkeypatch=monkeypatch, runner=runner,
+                         config={'services': {DUMMY_SERVICE: DUMMY_CONFIG_SETTINGS}}):
+        monkeypatch.setattr(cli, '_get_suitable_ssh_keys', _suitable_ssh_keys)
+        monkeypatch.setattr(dpp.Vault, 'phrase_from_key',
+                            phrase_from_key)
+        result = runner.invoke(cli.derivepassphrase,
+                               ['-k', DUMMY_SERVICE],
+                               input=b'1\n', catch_exceptions=False)
+        assert result.exit_code == 0, 'program exited with failure'
+        assert result.stdout_bytes, 'program output expected'
+        last_line = result.stdout_bytes.splitlines(True)[-1]
+        assert last_line.rstrip(b'\n') != DUMMY_RESULT_PASSPHRASE, (
+            'program generated unexpected result (phrase instead of key)'
+        )
+        assert last_line.rstrip(b'\n') == DUMMY_RESULT_KEY1, (
+            'program generated unexpected result (wrong settings?)'
+        )
+
+def test_205_service_phrase_if_key_in_global_config(monkeypatch: Any) -> None:
+    runner = click.testing.CliRunner(mix_stderr=False)
+    with isolated_config(
+        monkeypatch=monkeypatch, runner=runner,
+        config={
+            'global': {'key': DUMMY_KEY1},
+            'services': {
+                DUMMY_SERVICE: {
+                    'phrase': DUMMY_PASSPHRASE.rstrip(b'\n').decode('ASCII'),
+                    **DUMMY_CONFIG_SETTINGS}}}
+    ):
+        result = runner.invoke(cli.derivepassphrase, [DUMMY_SERVICE],
+                               catch_exceptions=False)
+        assert result.exit_code == 0, 'program exited with failure'
+        assert result.stdout_bytes, 'program output expected'
+        last_line = result.stdout_bytes.splitlines(True)[-1]
+        assert last_line.rstrip(b'\n') != DUMMY_RESULT_KEY1, (
+            'program generated unexpected result (key instead of phrase)'
+        )
+        assert last_line.rstrip(b'\n') == DUMMY_RESULT_PASSPHRASE, (
+            'program generated unexpected result (wrong settings?)'
+        )
+
+@pytest.mark.parametrize(['option'],
+                         [('--lower',), ('--upper',), ('--number',),
+                          ('--space',), ('--dash',), ('--symbol',),
+                          ('--repeat',), ('--length',)])
+def test_210_invalid_argument_range(option: str) -> None:
+    runner = click.testing.CliRunner(mix_stderr=False)
+    value: str | int
+    for value in '-42', 'invalid':
+        result = runner.invoke(cli.derivepassphrase,
+                               [option, cast(str, value), '-p', DUMMY_SERVICE],
+                               input=DUMMY_PASSPHRASE, catch_exceptions=False)
+        assert result.exit_code > 0, (
+            'program unexpectedly succeeded'
+        )
+        assert result.stderr_bytes, (
+            'program did not print any error message'
+        )
+        assert b'Error: Invalid value' in result.stderr_bytes, (
+            'program did not print the expected error message'
+        )
+
+@pytest.mark.parametrize(['vfunc', 'input'], [
+    (cli._validate_occurrence_constraint, 20),
+    (cli._validate_length, 20),
+])
+def test_210a_validate_constraints_manually(
+    vfunc: Callable[[click.Context, click.Parameter, Any], int | None],
+    input: int,
+) -> None:
+    ctx = cli.derivepassphrase.make_context(cli.prog_name, [])
+    param = cli.derivepassphrase.params[0]
+    assert vfunc(ctx, param, input) == input
+
+@pytest.mark.parametrize(
+    ['options', 'service', 'input', 'check_success'],
+    [(o.options, o.needs_service, o.input, o.check_success)
+     for o in INTERESTING_OPTION_COMBINATIONS if not o.incompatible],
+)
+def test_211_service_needed(
+    monkeypatch: Any, options: list[str],
+    service: bool | None, input: bytes | None, check_success: bool,
+) -> None:
+    def _prompt1(*args, **kwargs):
+        """Needed for --config handling."""
+        return DUMMY_PASSPHRASE.decode('UTF-8')
+    monkeypatch.setattr(cli, '_prompt_for_passphrase', _prompt1)
+    runner = click.testing.CliRunner(mix_stderr=False)
+    with isolated_config(monkeypatch=monkeypatch, runner=runner,
+                         config={'global': {'phrase': 'abc'},
+                                 'services': {}}):
+        result = runner.invoke(cli.derivepassphrase,
+                               options if service
+                               else options + [DUMMY_SERVICE],
+                               input=input, catch_exceptions=False)
+        if service is not None:
+            assert result.exit_code > 0, (
+                'program unexpectedly succeeded'
+            )
+            assert result.stderr_bytes, (
+                'program did not print any error message'
+            )
+            err_msg = (b' requires a SERVICE' if service
+                       else b' does not take a SERVICE argument')
+            assert err_msg in result.stderr_bytes, (
+                'program did not print the expected error message'
+            )
+        else:
+            assert (result.exit_code, result.stderr_bytes) == (0, b''), (
+                'program unexpectedly failed'
+            )
+    if check_success:
+        with isolated_config(monkeypatch=monkeypatch, runner=runner,
+                             config={'global': {'phrase': 'abc'},
+                                     'services': {}}):
+            def _prompt2(*args, **kwargs):
+                return DUMMY_PASSPHRASE.decode('UTF-8')
+            monkeypatch.setattr(cli, '_prompt_for_passphrase', _prompt2)
+            result = runner.invoke(cli.derivepassphrase,
+                                   options + [DUMMY_SERVICE]
+                                   if service else options,
+                                   input=input, catch_exceptions=False)
+            assert (result.exit_code, result.stderr_bytes) == (0, b''), (
+                'program unexpectedly failed'
+            )
+
+@pytest.mark.parametrize(
+    ['options', 'service', 'input'],
+    [(o.options, o.needs_service, o.input)
+     for o in INTERESTING_OPTION_COMBINATIONS if o.incompatible],
+)
+def test_212_incompatible_options(
+    options: list[str], service: bool | None, input: bytes | None,
+) -> None:
+    runner = click.testing.CliRunner(mix_stderr=False)
+    result = runner.invoke(cli.derivepassphrase,
+                           options + [DUMMY_SERVICE] if service else options,
+                           input=DUMMY_PASSPHRASE, catch_exceptions=False)
+    assert result.exit_code > 0, (
+        'program unexpectedly succeeded'
+    )
+    assert result.stderr_bytes, (
+        'program did not print any error message'
+    )
+    assert b'mutually exclusive with ' in result.stderr_bytes, (
+        'program did not print the expected error message'
+    )
+
+def test_213_import_bad_config_not_vault_config(monkeypatch: Any) -> None:
+    runner = click.testing.CliRunner(mix_stderr=False)
+    with isolated_config(monkeypatch=monkeypatch, runner=runner,
+                         config={'services': {}}):
+        result = runner.invoke(cli.derivepassphrase, ['--import', '-'],
+                               input=b'null', catch_exceptions=False)
+        assert result.exit_code > 0, (
+            'program unexpectedly succeeded'
+        )
+        assert result.stderr_bytes, (
+            'program did not print any error message'
+        )
+        assert b'not a valid config' in result.stderr_bytes, (
+            'program did not print the expected error message'
+        )
+
+def test_213a_import_bad_config_not_json_data(monkeypatch: Any) -> None:
+    runner = click.testing.CliRunner(mix_stderr=False)
+    with isolated_config(monkeypatch=monkeypatch, runner=runner,
+                         config={'services': {}}):
+        result = runner.invoke(cli.derivepassphrase, ['--import', '-'],
+                               input=b'This string is not valid JSON.',
+                               catch_exceptions=False)
+        assert result.exit_code > 0, (
+            'program unexpectedly succeeded'
+        )
+        assert result.stderr_bytes, (
+            'program did not print any error message'
+        )
+        assert b'cannot decode JSON' in result.stderr_bytes, (
+            'program did not print the expected error message'
+        )
+
+def test_213b_import_bad_config_not_a_file(monkeypatch: Any) -> None:
+    runner = click.testing.CliRunner(mix_stderr=False)
+    with isolated_config(monkeypatch=monkeypatch, runner=runner,
+                         config={'services': {}}):
+        with open(cli._config_filename(), 'wt') as outfile:
+            print('This string is not valid JSON.', file=outfile)
+        result = runner.invoke(
+            cli.derivepassphrase,
+            ['--import', os.path.dirname(cli._config_filename())],
+            catch_exceptions=False)
+        assert result.exit_code > 0, (
+            'program unexpectedly succeeded'
+        )
+        assert result.stderr_bytes, (
+            'program did not print any error message'
+        )
+        # Don't test the actual error message, because it is subject to
+        # locale settings.  TODO: find a way anyway.
+
+def test_214_export_settings_no_stored_settings(monkeypatch: Any) -> None:
+    runner = click.testing.CliRunner(mix_stderr=False)
+    with isolated_config(monkeypatch=monkeypatch, runner=runner,
+                         config={'services': {}}):
+        try:
+            os.remove(cli._config_filename())
+        except FileNotFoundError:  # pragma: no cover
+            pass
+        result = runner.invoke(cli.derivepassphrase, ['--export', '-'],
+                               catch_exceptions=False)
+        assert (result.exit_code, result.stderr_bytes) == (0, b''), (
+            'program exited with failure'
+        )
+
+def test_214a_export_settings_bad_stored_config(monkeypatch: Any) -> None:
+    runner = click.testing.CliRunner(mix_stderr=False)
+    with isolated_config(monkeypatch=monkeypatch, runner=runner,
+                         config={}):
+        result = runner.invoke(cli.derivepassphrase, ['--export', '-'],
+                               input=b'null', catch_exceptions=False)
+        assert result.exit_code > 0, (
+            'program unexpectedly succeeded'
+        )
+        assert result.stderr_bytes, (
+            'program did not print any error message'
+        )
+        assert b'cannot load config' in result.stderr_bytes, (
+            'program did not print the expected error message'
+        )
+
+def test_214b_export_settings_not_a_file(monkeypatch: Any) -> None:
+    runner = click.testing.CliRunner(mix_stderr=False)
+    with isolated_config(monkeypatch=monkeypatch, runner=runner,
+                         config={'services': {}}):
+        try:
+            os.remove(cli._config_filename())
+        except FileNotFoundError:  # pragma: no cover
+            pass
+        os.makedirs(cli._config_filename())
+        result = runner.invoke(cli.derivepassphrase, ['--export', '-'],
+                               input=b'null', catch_exceptions=False)
+        assert result.exit_code > 0, (
+            'program unexpectedly succeeded'
+        )
+        assert result.stderr_bytes, (
+            'program did not print any error message'
+        )
+        assert b'cannot load config' in result.stderr_bytes, (
+            'program did not print the expected error message'
+        )
+
+def test_214c_export_settings_target_not_a_file(monkeypatch: Any) -> None:
+    runner = click.testing.CliRunner(mix_stderr=False)
+    with isolated_config(monkeypatch=monkeypatch, runner=runner,
+                         config={'services': {}}):
+        dname = os.path.dirname(cli._config_filename())
+        result = runner.invoke(cli.derivepassphrase,
+                               ['--export', dname],
+                               input=b'null', catch_exceptions=False)
+        assert result.exit_code > 0, (
+            'program unexpectedly succeeded'
+        )
+        assert result.stderr_bytes, (
+            'program did not print any error message'
+        )
+        assert b'cannot write config' in result.stderr_bytes, (
+            'program did not print the expected error message'
+        )
+
+def test_220_edit_notes_successfully(monkeypatch: Any) -> None:
+    edit_result = '''
+
+# - - - - - >8 - - - - - >8 - - - - - >8 - - - - - >8 - - - - -
+contents go here
+'''
+    runner = click.testing.CliRunner(mix_stderr=False)
+    with isolated_config(monkeypatch=monkeypatch, runner=runner,
+                         config={'global': {'phrase': 'abc'},
+                                 'services': {}}):
+        monkeypatch.setattr(click, 'edit',
+                            lambda *a, **kw: edit_result)
+        result = runner.invoke(cli.derivepassphrase, ['--notes', 'sv'],
+                               catch_exceptions=False)
+        assert (result.exit_code, result.stderr_bytes) == (0, b''), (
+            'program exited with failure'
+        )
+        with open(cli._config_filename(), 'rt') as infile:
+            config = json.load(infile)
+        assert config == {'global': {'phrase': 'abc'},
+                          'services': {'sv': {'notes': 'contents go here'}}}
+
+def test_221_edit_notes_noop(monkeypatch: Any) -> None:
+    runner = click.testing.CliRunner(mix_stderr=False)
+    with isolated_config(monkeypatch=monkeypatch, runner=runner,
+                         config={'global': {'phrase': 'abc'},
+                                 'services': {}}):
+        monkeypatch.setattr(click, 'edit', lambda *a, **kw: None)
+        result = runner.invoke(cli.derivepassphrase, ['--notes', 'sv'],
+                               catch_exceptions=False)
+        assert (result.exit_code, result.stderr_bytes) == (0, b''), (
+            'program exited with failure'
+        )
+        with open(cli._config_filename(), 'rt') as infile:
+            config = json.load(infile)
+        assert config == {'global': {'phrase': 'abc'}, 'services': {}}
+
+def test_222_edit_notes_marker_removed(monkeypatch: Any) -> None:
+    runner = click.testing.CliRunner(mix_stderr=False)
+    with isolated_config(monkeypatch=monkeypatch, runner=runner,
+                         config={'global': {'phrase': 'abc'},
+                                 'services': {}}):
+        monkeypatch.setattr(click, 'edit', lambda *a, **kw: 'long\ntext')
+        result = runner.invoke(cli.derivepassphrase, ['--notes', 'sv'],
+                               catch_exceptions=False)
+        assert (result.exit_code, result.stderr_bytes) == (0, b''), (
+            'program exited with failure'
+        )
+        with open(cli._config_filename(), 'rt') as infile:
+            config = json.load(infile)
+        assert config == {'global': {'phrase': 'abc'},
+                          'services': {'sv': {'notes': 'long\ntext'}}}
+
+def test_223_edit_notes_abort(monkeypatch: Any) -> None:
+    runner = click.testing.CliRunner(mix_stderr=False)
+    with isolated_config(monkeypatch=monkeypatch, runner=runner,
+                         config={'global': {'phrase': 'abc'},
+                                 'services': {}}):
+        monkeypatch.setattr(click, 'edit', lambda *a, **kw: '\n\n')
+        result = runner.invoke(cli.derivepassphrase, ['--notes', 'sv'],
+                               catch_exceptions=False)
+        assert result.exit_code != 0, 'program unexpectedly succeeded'
+        assert b'user aborted request' in result.stderr_bytes, (
+            'expected error message missing'
+        )
+        with open(cli._config_filename(), 'rt') as infile:
+            config = json.load(infile)
+        assert config == {'global': {'phrase': 'abc'}, 'services': {}}
+
+@pytest.mark.parametrize(['command_line', 'input', 'result_config'], [
+    (
+        ['--phrase'],
+        b'my passphrase\n',
+        {'global': {'phrase': 'my passphrase'}, 'services': {}},
+    ),
+    (
+        ['--key'],
+        b'1\n',
+        {'global': {'key': DUMMY_KEY1}, 'services': {}},
+    ),
+    (
+        ['--phrase', 'sv'],
+        b'my passphrase\n',
+        {'global': {'phrase': 'abc'},
+         'services': {'sv': {'phrase': 'my passphrase'}}},
+    ),
+    (
+        ['--key', 'sv'],
+        b'1\n',
+        {'global': {'phrase': 'abc'},
+         'services': {'sv': {'key': DUMMY_KEY1}}},
+    ),
+    (
+        ['--key', '--length', '15', 'sv'],
+        b'1\n',
+        {'global': {'phrase': 'abc'},
+         'services': {'sv': {'key': DUMMY_KEY1, 'length': 15}}},
+    ),
+])
+def test_224_store_config_good(
+    monkeypatch: Any, command_line: list[str], input: bytes,
+    result_config: Any,
+) -> None:
+    KeyCommentPair = ssh_agent_client.types.KeyCommentPair
+    key_list = [
+        KeyCommentPair(base64.standard_b64decode(DUMMY_KEY1), b'no comment'),
+        KeyCommentPair(base64.standard_b64decode(DUMMY_KEY2), b'a comment'),
+    ]
+    def _suitable_ssh_keys(conn: Any) -> Iterator[KeyCommentPair]:
+        yield from key_list
+    runner = click.testing.CliRunner(mix_stderr=False)
+    with isolated_config(monkeypatch=monkeypatch, runner=runner,
+                         config={'global': {'phrase': 'abc'},
+                                 'services': {}}):
+        monkeypatch.setattr(cli, '_get_suitable_ssh_keys', _suitable_ssh_keys)
+        result = runner.invoke(cli.derivepassphrase,
+                               ['--config'] + command_line,
+                               catch_exceptions=False, input=input)
+        assert result.exit_code == 0, 'program exited with failure'
+        with open(cli._config_filename(), 'rt') as infile:
+            config = json.load(infile)
+        assert config == result_config, (
+            'stored config does not match expectation'
+        )
+
+@pytest.mark.parametrize(['command_line', 'input', 'err_text'], [
+    ([], b'', b'cannot update global settings without actual settings'),
+    (['sv'], b'', b'cannot update service settings without actual settings'),
+    (['--phrase', 'sv'], b'', b'no passphrase given'),
+    (['--key'], b'', b'no valid SSH key selected'),
+])
+def test_225_store_config_fail(
+    monkeypatch: Any, command_line: list[str], input: bytes, err_text: str,
+) -> None:
+    KeyCommentPair = ssh_agent_client.types.KeyCommentPair
+    key_list = [
+        KeyCommentPair(base64.standard_b64decode(DUMMY_KEY1), b'no comment'),
+        KeyCommentPair(base64.standard_b64decode(DUMMY_KEY2), b'a comment'),
+    ]
+    def _suitable_ssh_keys(conn: Any) -> Iterator[KeyCommentPair]:
+        yield from key_list
+    runner = click.testing.CliRunner(mix_stderr=False)
+    with isolated_config(monkeypatch=monkeypatch, runner=runner,
+                         config={'global': {'phrase': 'abc'},
+                                 'services': {}}):
+        monkeypatch.setattr(cli, '_get_suitable_ssh_keys', _suitable_ssh_keys)
+        result = runner.invoke(cli.derivepassphrase,
+                               ['--config'] + command_line,
+                               catch_exceptions=False, input=input)
+        assert result.exit_code != 0, 'program unexpectedly succeeded?!'
+        assert err_text in result.stderr_bytes, (
+            'expected error message missing'
+        )
+
+def test_225a_store_config_fail_manual_no_ssh_key_selection(
+    monkeypatch: Any,
+) -> None:
+    runner = click.testing.CliRunner(mix_stderr=False)
+    with isolated_config(monkeypatch=monkeypatch, runner=runner,
+                         config={'global': {'phrase': 'abc'},
+                                 'services': {}}):
+        def raiser():
+            raise RuntimeError('custom error message')
+        monkeypatch.setattr(cli, '_select_ssh_key', raiser)
+        result = runner.invoke(cli.derivepassphrase, ['--key', '--config'],
+                               catch_exceptions=False)
+        assert result.exit_code != 0, 'program unexpectedly succeeded'
+        assert b'custom error message' in result.stderr_bytes, (
+            'expected error message missing'
+        )
+
+def test_226_no_arguments() -> None:
+    runner = click.testing.CliRunner(mix_stderr=False)
+    result = runner.invoke(cli.derivepassphrase, [], catch_exceptions=False)
+    assert result.exit_code != 0, 'program unexpectedly succeeded'
+    assert b'SERVICE is required' in result.stderr_bytes, (
+        'expected error message missing'
+    )
+
+def test_226a_no_passphrase_or_key() -> None:
+    runner = click.testing.CliRunner(mix_stderr=False)
+    result = runner.invoke(cli.derivepassphrase, [DUMMY_SERVICE],
+                           catch_exceptions=False)
+    assert result.exit_code != 0, 'program unexpectedly succeeded'
+    assert b'no passphrase or key given' in result.stderr_bytes, (
+        'expected error message missing'
     )

tests/test_passphrase_generating.py

Zeige Datei @ 6e05ad2

tests/test_key_signing.py

Zeige Datei @ 6e05ad2

@@ -4,16 +4,25 @@
 
 """Test OpenSSH key loading and signing."""
 
+from __future__ import annotations
+
+import click
 import pytest
 
 import derivepassphrase
+import derivepassphrase.cli
 import ssh_agent_client
 
 import base64
+import errno
+import io
 import os
 import socket
 import subprocess
 
+if not os.environ.get('SSH_AUTH_SOCK'):  # pragma: no cover
+    pytest.skip('no running SSH agent detected', allow_module_level=True)
+
 SUPPORTED = {
     'ed25519': {
         'private_key': rb'''-----BEGIN OPENSSH PRIVATE KEY-----
@@ -327,6 +336,14 @@ def test_client_string(input, expected):
     string = ssh_agent_client.SSHAgentClient.string
     assert bytes(string(input)) == expected
 
+@pytest.mark.parametrize(['input', 'exc_type', 'exc_pattern'], [
+    ('some string', TypeError, 'invalid payload type'),
+])
+def test_client_string_exceptions(input, exc_type, exc_pattern):
+    string = ssh_agent_client.SSHAgentClient.string
+    with pytest.raises(exc_type, match=exc_pattern):
+        string(input)
+
 @pytest.mark.parametrize(['input', 'expected'], [
     (b'\x00\x00\x00\x07ssh-rsa', b'ssh-rsa'),
     (
@@ -396,6 +413,7 @@ def test_sign_data_via_agent(keytype, data_dict):
                              for k, c in client.list_keys()}
         public_key_data = data_dict['public_key_data']
         expected_signature = data_dict['expected_signature']
+        derived_passphrase = data_dict['derived_passphrase']
         if public_key_data not in key_comment_pairs:  # pragma: no cover
             pytest.skip('prerequisite SSH key not loaded')
         signature = bytes(client.sign(
@@ -405,8 +423,8 @@ def test_sign_data_via_agent(keytype, data_dict):
             payload=derivepassphrase.Vault._UUID, key=public_key_data))
         assert signature2 == expected_signature, 'SSH signature mismatch'
         assert (
-            derivepassphrase.Vault.phrase_from_signature(public_key_data) ==
-            expected_signature
+            derivepassphrase.Vault.phrase_from_key(public_key_data) ==
+            derived_passphrase
         ), 'SSH signature mismatch'
 
 @pytest.mark.parametrize(['keytype', 'data_dict'], list(UNSUITABLE.items()))
@@ -439,4 +457,142 @@ def test_sign_data_via_agent_unsupported(keytype, data_dict):
             payload=derivepassphrase.Vault._UUID, key=public_key_data))
         assert signature != signature2, 'SSH signature repeatable?!'
         with pytest.raises(ValueError, match='unsuitable SSH key'):
-            derivepassphrase.Vault.phrase_from_signature(public_key_data)
+            derivepassphrase.Vault.phrase_from_key(public_key_data)
+
+@pytest.mark.parametrize(['this_data', 'all_data'],
+                         [(v, tuple(SUPPORTED.values()))
+                          for v in SUPPORTED.values()])
+def test_ssh_key_selector(monkeypatch, this_data, all_data):
+    successfully_uploaded_keys: list[bytes] = []
+    for data in all_data:
+        private_key = data['private_key']
+        try:
+            result = subprocess.run(['ssh-add', '-t', '60', '-q', '-'],
+                                    input=private_key, check=True,
+                                    capture_output=True)
+        except subprocess.CalledProcessError as e:
+            if data == this_data:
+                pytest.skip(
+                    f"uploading non-optional test key: {e!r}, "
+                    f"stdout={e.stdout!r}, stderr={e.stderr!r}"
+                )
+        else:
+            successfully_uploaded_keys.append(data['public_key_data'])
+    index = 1 + successfully_uploaded_keys.index(this_data['public_key_data'])
+    b64_key = base64.standard_b64encode(
+        this_data['public_key_data']).decode('ASCII')
+    n = len(successfully_uploaded_keys)
+    text = (f'Your selection? (1-{n}, leave empty to abort): {index}\n'
+            if n > 1 else 'Use this key? yes\n')
+
+    @click.command()
+    def driver():
+        key = derivepassphrase.cli._select_ssh_key()
+        click.echo(base64.standard_b64encode(key).decode('ASCII'))
+
+    runner = click.testing.CliRunner(mix_stderr=True)
+    result = runner.invoke(driver, [],
+                           input=(f'{index}\n' if n > 1 else f'yes\n'),
+                           catch_exceptions=True)
+    assert result.exit_code == 0, 'driver program failed?!'
+    assert result.stdout.startswith('Suitable SSH keys:\n'), (
+        'missing expected output'
+    )
+    assert text in result.stdout, 'missing expected output'
+    assert result.stdout.endswith(f'\n{b64_key}\n'), 'missing expected output'
+
+@pytest.mark.parametrize(['conn_hint'], [('none',), ('socket',), ('client',)])
+def test_get_suitable_ssh_keys(conn_hint):
+    hint: ssh_agent_client.SSHAgentClient | socket.socket | None
+    match conn_hint:
+        case 'client':
+            hint = ssh_agent_client.SSHAgentClient()
+        case 'socket':
+            hint = socket.socket(family=socket.AF_UNIX)
+            hint.connect(os.environ['SSH_AUTH_SOCK'])
+        case _:
+            assert conn_hint == 'none'
+            hint = None
+    exception: type[Exception] | None = None
+    try:
+        list(derivepassphrase.cli._get_suitable_ssh_keys(hint))
+    except RuntimeError:  # pragma: no cover
+        pass
+    except Exception as e:  # pragma: no cover
+        exception = e
+    finally:
+        assert exception == None, 'exception querying suitable SSH keys'
+
+def test_constructor_no_running_agent(monkeypatch):
+    monkeypatch.delenv('SSH_AUTH_SOCK', raising=False)
+    sock = socket.socket(family=socket.AF_UNIX)
+    with pytest.raises(RuntimeError, match='missing SSH_AUTH_SOCK'):
+        ssh_agent_client.SSHAgentClient(socket=sock)
+
+def test_constructor_bad_running_agent(monkeypatch):
+    monkeypatch.setenv('SSH_AUTH_SOCK', os.environ['SSH_AUTH_SOCK'] + '~')
+    sock = socket.socket(family=socket.AF_UNIX)
+    with pytest.raises(RuntimeError, match='unusable SSH_AUTH_SOCK'):
+        ssh_agent_client.SSHAgentClient(socket=sock)
+
+@pytest.mark.parametrize(['response'], [
+    (b'\x00\x00',),
+    (b'\x00\x00\x00\x1f some bytes missing',),
+])
+def test_truncated_server_response(monkeypatch, response):
+    client = ssh_agent_client.SSHAgentClient()
+    response_stream = io.BytesIO(response)
+    class PseudoSocket(object):
+        pass
+    pseudo_socket = PseudoSocket()
+    pseudo_socket.sendall = lambda *a, **kw: None
+    pseudo_socket.recv = response_stream.read
+    monkeypatch.setattr(client, '_connection', pseudo_socket)
+    with pytest.raises(EOFError):
+        client.request(255, b'')
+
+@pytest.mark.parametrize(
+    ['response_code', 'response', 'exc_type', 'exc_pattern'],
+    [
+        (255, b'', RuntimeError, 'error return from SSH agent:'),
+        (12, b'\x00\x00\x00\x01', EOFError, 'truncated response'),
+        (12, b'\x00\x00\x00\x00abc', RuntimeError, 'overlong response'),
+    ]
+)
+def test_list_keys_error_responses(monkeypatch, response_code, response,
+                                   exc_type, exc_pattern):
+    client = ssh_agent_client.SSHAgentClient()
+    monkeypatch.setattr(client, 'request',
+                        lambda *a, **kw: (response_code, response))
+    with pytest.raises(exc_type, match=exc_pattern):
+        client.list_keys()
+
+@pytest.mark.parametrize(
+    ['key', 'check', 'response', 'exc_type', 'exc_pattern'],
+    [
+        (
+            b'invalid-key',
+            True,
+            (255, b''),
+            RuntimeError,
+            'target SSH key not loaded into agent',
+        ),
+        (
+            SUPPORTED['ed25519']['public_key_data'],
+            True,
+            (255, b''),
+            RuntimeError,
+            'signing data failed:',
+        )
+    ]
+)
+def test_sign_error_responses(monkeypatch, key, check, response, exc_type,
+                              exc_pattern):
+    client = ssh_agent_client.SSHAgentClient()
+    monkeypatch.setattr(client, 'request', lambda a, b: response)
+    KeyCommentPair = ssh_agent_client.types.KeyCommentPair
+    loaded_keys = [KeyCommentPair(v['public_key_data'], b'no comment')
+                   for v in SUPPORTED.values()]
+    monkeypatch.setattr(client, 'list_keys', lambda: loaded_keys)
+    with pytest.raises(exc_type, match=exc_pattern):
+        client.sign(key, b'abc', check_if_key_loaded=check)

tests/test_sequin.py

Zeige Datei @ 6e05ad2

@@ -25,6 +25,7 @@ def test_big_endian_number(sequence, base, expected):
 @pytest.mark.parametrize(['exc_type', 'exc_pattern', 'sequence' , 'base'], [
     (ValueError, 'invalid base 3 digit:', [-1], 3),
     (ValueError, 'invalid base:', [0], 1),
+    (TypeError, 'not an integer:', [0.0, 1.0, 0.0, 1.0], 2),
 ])
 def test_big_endian_number_exceptions(exc_type, exc_pattern, sequence, base):
     with pytest.raises(exc_type, match=exc_pattern):
@@ -103,3 +104,5 @@ def test_internal_generating():
     assert seq._generate_inner(1) == 0
     with pytest.raises(ValueError, match='invalid target range'):
         seq._generate_inner(0)
+    with pytest.raises(ValueError, match='invalid base:'):
+        seq._generate_inner(16, base=1)


...	...	@@ -52,6 +52,7 @@ plugins:
52	52	python:
53	53	import:
54	54	- https://docs.python.org/3/objects.inv
	55	+ - https://click.palletsprojects.com/en/8.1.x/objects.inv
55	56	options:
56	57	docstring_options:
57	58	ignore_init_summary: true
...	...	@@ -82,6 +83,7 @@ nav:
82	83	- Module derivepassphrase: reference/derivepassphrase.md
83	84	- Module sequin: reference/sequin.md
84	85	- Module ssh_agent_client: reference/ssh_agent_client.md
	86	+ - Coverage: reference/coverage/index.html
85	87	#- explanation.md
86	88
87	89	markdown_extensions:

...	...	@@ -43,6 +43,7 @@ Source = "https://github.com/the-13th-letter/derivepassphrase"
43	43	derivepassphrase = "derivepassphrase.cli:derivepassphrase"
44	44
45	45	[tool.mypy]
	46	+files = ['src/*/.py']
46	47	mypy_path = '$MYPY_CONFIG_FILE_DIR/src'
47	48	explicit_package_bases = true
48	49	implicit_reexport = false
...	...	@@ -84,21 +85,32 @@ extra-dependencies = [
84	85	[tool.hatch.envs.types.scripts]
85	86	check = "mypy --install-types --non-interactive {args:src/derivepassphrase tests}"
86	87
	88	+[tool.coverage.html]
	89	+directory = "docs/reference/coverage"
	90	+
87	91	[tool.coverage.run]
88		-source_pkgs = ["derivepassphrase", "tests"]
	92	+source_pkgs = ["derivepassphrase", "sequin", "ssh_agent_client", "tests"]
89	93	branch = true
90	94	parallel = true
91	95	omit = [
92		- "src/derivepassphrase/__main__.py",
	96	+ "__main__.py",
93	97	]
94	98
95	99	[tool.coverage.paths]
96		-derivepassphrase = ["src/derivepassphrase", "*/derivepassphrase/src/derivepassphrase"]
97		-tests = ["tests", "*/derivepassphrase/tests"]
	100	+src = ["src"]
	101	+tests = ["tests"]
98	102
99	103	[tool.coverage.report]
100		-exclude_lines = [
101		- "no cov",
	104	+skip_covered = false
	105	+skip_empty = true
	106	+precision = 3
	107	+partial_branches = [
	108	+ 'pragma: no branch',
	109	+]
	110	+exclude_also = [
102	111	"if __name__ == .__main__.:",
103		- "if TYPE_CHECKING:",
	112	+ 'if (?:typing\.)?TYPE_CHECKING:',
	113	+ "raise AssertionError",
	114	+ "raise NotImplementedError",
	115	+ 'assert False',
104	116	]

...	...	@@ -1,6 +1,8 @@
1	1	# SPDX-FileCopyrightText: 2024 Marco Ricci <m@the13thletter.info>
2	2	#
3	3	# SPDX-License-Identifier: MIT
	4	+"""Run [`derivepassphrase.cli.derivepassphrase`][] on import."""
	5	+
4	6	import sys
5	7
6	8	if __name__ == "__main__":

...	...	@@ -8,14 +8,24 @@
8	8
9	9	from __future__ import annotations
10	10
	11	+import base64
	12	+import collections
	13	+from collections.abc import MutableMapping
	14	+import contextlib
11	15	import inspect
12	16	import json
	17	+import os
13	18	import pathlib
14		-from typing import Any, TextIO
	19	+import socket
	20	+from typing import (
	21	+ Any, assert_never, cast, reveal_type, Iterator, Never, NotRequired,
	22	+ Sequence, TextIO, TypedDict, TYPE_CHECKING,
	23	+)
15	24
16	25	import click
17	26	import derivepassphrase as dpp
18	27	from derivepassphrase import types as dpp_types
	28	+import ssh_agent_client
19	29
20	30	__author__ = dpp.__author__
21	31	__version__ = dpp.__version__
...	...	@@ -92,6 +102,202 @@ def _save_config(config: dpp_types.VaultConfig, /) -> None:
92	102	json.dump(config, fileobj)
93	103
94	104
	105	+def _get_suitable_ssh_keys(
	106	+ conn: ssh_agent_client.SSHAgentClient \| socket.socket \| None = None,
	107	+ /
	108	+) -> Iterator[ssh_agent_client.types.KeyCommentPair]:
	109	+ """Yield all SSH keys suitable for passphrase derivation.
	110	+
	111	+ Suitable SSH keys are queried from the running SSH agent (see
	112	+ [`ssh_agent_client.SSHAgentClient.list_keys`][]).
	113	+
	114	+ Args:
	115	+ conn:
	116	+ An optional connection hint to the SSH agent; specifically,
	117	+ an SSH agent client, or a socket connected to an SSH agent.
	118	+
	119	+ If an existing SSH agent client, then this client will be
	120	+ queried for the SSH keys, and otherwise left intact.
	121	+
	122	+ If a socket, then a one-shot client will be constructed
	123	+ based on the socket to query the agent, and deconstructed
	124	+ afterwards.
	125	+
	126	+ If neither are given, then the agent's socket location is
	127	+ looked up in the `SSH_AUTH_SOCK` environment variable, and
	128	+ used to construct/deconstruct a one-shot client, as in the
	129	+ previous case.
	130	+
	131	+ Yields:
	132	+ :
	133	+ Every SSH key from the SSH agent that is suitable for
	134	+ passphrase derivation.
	135	+
	136	+ Raises:
	137	+ RuntimeError:
	138	+ There was an error communicating with the SSH agent.
	139	+ RuntimeError:
	140	+ No keys usable for passphrase derivation are loaded into the
	141	+ SSH agent.
	142	+
	143	+ """
	144	+ client: ssh_agent_client.SSHAgentClient
	145	+ client_context: contextlib.AbstractContextManager
	146	+ match conn:
	147	+ case ssh_agent_client.SSHAgentClient():
	148	+ client = conn
	149	+ client_context = contextlib.nullcontext()
	150	+ case socket.socket() \| None:
	151	+ client = ssh_agent_client.SSHAgentClient(socket=conn)
	152	+ client_context = client
	153	+ case _: # pragma: no cover
	154	+ assert_never(conn)
	155	+ raise TypeError(f'invalid connection hint: {conn!r}')
	156	+ with client_context:
	157	+ try:
	158	+ all_key_comment_pairs = list(client.list_keys())
	159	+ except EOFError as e: # pragma: no cover
	160	+ raise RuntimeError(
	161	+ 'error communicating with the SSH agent'
	162	+ ) from e
	163	+ suitable_keys = all_key_comment_pairs[:]
	164	+ for pair in all_key_comment_pairs:
	165	+ key, comment = pair
	166	+ if dpp.Vault._is_suitable_ssh_key(key):
	167	+ yield pair
	168	+ if not suitable_keys: # pragma: no cover
	169	+ raise RuntimeError('No usable SSH keys were found')
	170	+
	171	+
	172	+def _prompt_for_selection(
	173	+ items: Sequence[str \| bytes], heading: str = 'Possible choices:',
	174	+ single_choice_prompt: str = 'Confirm this choice?',
	175	+) -> int:
	176	+ """Prompt user for a choice among the given items.
	177	+
	178	+ Print the heading, if any, then present the items to the user. If
	179	+ there are multiple items, prompt the user for a selection, validate
	180	+ the choice, then return the list index of the selected item. If
	181	+ there is only a single item, request confirmation for that item
	182	+ instead, and return the correct index.
	183	+
	184	+ Args:
	185	+ heading:
	186	+ A heading for the list of items, to print immediately
	187	+ before. Defaults to a reasonable standard heading. If
	188	+ explicitly empty, print no heading.
	189	+ single_choice_prompt:
	190	+ The confirmation prompt if there is only a single possible
	191	+ choice. Defaults to a reasonable standard prompt.
	192	+
	193	+ Returns:
	194	+ An index into the items sequence, indicating the user's
	195	+ selection.
	196	+
	197	+ Raises:
	198	+ IndexError:
	199	+ The user made an invalid or empty selection, or requested an
	200	+ abort.
	201	+
	202	+ """
	203	+ n = len(items)
	204	+ if heading:
	205	+ click.echo(click.style(heading, bold=True))
	206	+ for i, x in enumerate(items, start=1):
	207	+ click.echo(click.style(f'[{i}]', bold=True), nl=False)
	208	+ click.echo(' ', nl=False)
	209	+ click.echo(x)
	210	+ if n > 1:
	211	+ choices = click.Choice([''] + [str(i) for i in range(1, n + 1)])
	212	+ choice = click.prompt(
	213	+ f'Your selection? (1-{n}, leave empty to abort)',
	214	+ err=True, type=choices, show_choices=False,
	215	+ show_default=False, default='')
	216	+ if not choice:
	217	+ raise IndexError('empty selection')
	218	+ return int(choice) - 1
	219	+ else:
	220	+ prompt_suffix = (' '
	221	+ if single_choice_prompt.endswith(tuple('?.!'))
	222	+ else ': ')
	223	+ try:
	224	+ click.confirm(single_choice_prompt,
	225	+ prompt_suffix=prompt_suffix, err=True,
	226	+ abort=True, default=False, show_default=False)
	227	+ except click.Abort:
	228	+ raise IndexError('empty selection') from None
	229	+ return 0
	230	+
	231	+
	232	+def _select_ssh_key(
	233	+ conn: ssh_agent_client.SSHAgentClient \| socket.socket \| None = None,
	234	+ /
	235	+) -> bytes \| bytearray:
	236	+ """Interactively select an SSH key for passphrase derivation.
	237	+
	238	+ Suitable SSH keys are queried from the running SSH agent (see
	239	+ [`ssh_agent_client.SSHAgentClient.list_keys`][]), then the user is
	240	+ prompted interactively (see [`click.prompt`][]) for a selection.
	241	+
	242	+ Args:
	243	+ conn:
	244	+ An optional connection hint to the SSH agent; specifically,
	245	+ an SSH agent client, or a socket connected to an SSH agent.
	246	+
	247	+ If an existing SSH agent client, then this client will be
	248	+ queried for the SSH keys, and otherwise left intact.
	249	+
	250	+ If a socket, then a one-shot client will be constructed
	251	+ based on the socket to query the agent, and deconstructed
	252	+ afterwards.
	253	+
	254	+ If neither are given, then the agent's socket location is
	255	+ looked up in the `SSH_AUTH_SOCK` environment variable, and
	256	+ used to construct/deconstruct a one-shot client, as in the
	257	+ previous case.
	258	+
	259	+ Returns:
	260	+ The selected SSH key.
	261	+
	262	+ Raises:
	263	+ IndexError:
	264	+ The user made an invalid or empty selection, or requested an
	265	+ abort.
	266	+ RuntimeError:
	267	+ There was an error communicating with the SSH agent.
	268	+ RuntimeError:
	269	+ No keys usable for passphrase derivation are loaded into the
	270	+ SSH agent.
	271	+ """
	272	+ suitable_keys = list(_get_suitable_ssh_keys(conn))
	273	+ key_listing: list[str] = []
	274	+ unstring_prefix = ssh_agent_client.SSHAgentClient.unstring_prefix
	275	+ for key, comment in suitable_keys:
	276	+ keytype = unstring_prefix(key)[0].decode('ASCII')
	277	+ key_str = base64.standard_b64encode(key).decode('ASCII')
	278	+ key_prefix = key_str if len(key_str) < 30 else key_str[:27] + '...'
	279	+ comment_str = comment.decode('UTF-8', errors='replace')
	280	+ key_listing.append(f'{keytype} {key_prefix} {comment_str}')
	281	+ choice = _prompt_for_selection(
	282	+ key_listing, heading='Suitable SSH keys:',
	283	+ single_choice_prompt='Use this key?')
	284	+ return suitable_keys[choice].key
	285	+
	286	+
	287	+def _prompt_for_passphrase() -> str:
	288	+ """Interactively prompt for the passphrase.
	289	+
	290	+ Calls [`click.prompt`][] internally. Moved into a separate function
	291	+ mainly for testing/mocking purposes.
	292	+
	293	+ Returns:
	294	+ The user input.
	295	+
	296	+ """
	297	+ return click.prompt('Passphrase', default='', hide_input=True,
	298	+ show_default=False, err=True)
	299	+
	300	+
95	301	class OptionGroupOption(click.Option):
96	302	"""A [`click.Option`][] with an associated group name and group epilog.
97	303
...	...	@@ -240,15 +446,36 @@ def _validate_length(
240	446	raise click.BadParameter('not a positive integer')
241	447	return int_value
242	448
	449	+DEFAULT_NOTES_TEMPLATE = '''\
	450	+# Enter notes below the line with the cut mark (ASCII scissors and
	451	+# dashes). Lines above the cut mark (such as this one) will be ignored.
	452	+#
	453	+# If you wish to clear the notes, leave everything beyond the cut mark
	454	+# blank. However, if you leave the entire file blank, also removing
	455	+# the cut mark, then the edit is aborted, and the old notes contents are
	456	+# retained.
	457	+#
	458	+# - - - - - >8 - - - - - >8 - - - - - >8 - - - - - >8 - - - - -
	459	+'''
	460	+DEFAULT_NOTES_MARKER = '# - - - - - >8 - - - - -'
	461	+
	462	+
243	463	@click.command(
244	464	context_settings={"help_option_names": ["-h", "--help"]},
245	465	cls=CommandWithHelpGroups,
246		- epilog='''
	466	+ epilog=r'''
247	467	WARNING: There is NO WAY to retrieve the generated passphrases
248	468	if the master passphrase, the SSH key, or the exact passphrase
249	469	settings are lost, short of trying out all possible
250	470	combinations. You are STRONGLY advised to keep independent
251	471	backups of the settings and the SSH key, if any.
	472	+
	473	+ Configuration is stored in a directory according to the
	474	+ DERIVEPASSPHRASE_PATH variable, which defaults to
	475	+ `~/.derivepassphrase` on UNIX-like systems and
	476	+ `C:\Users\<user>\AppData\Roaming\Derivepassphrase` on Windows.
	477	+ The configuration is NOT encrypted, and you are STRONGLY
	478	+ discouraged from using a stored passphrase.
252	479	''',
253	480	)
254	481	@click.option('-p', '--phrase', 'use_phrase', is_flag=True,
...	...	@@ -513,11 +740,166 @@ def derivepassphrase(
513	740	opt_str = param.opts[0]
514	741	raise click.UsageError(
515	742	f'{opt_str} does not take a SERVICE argument')
516		- #if kwargs['length'] is None:
517		- # kwargs['length'] = dpp.Vault.__init__.__kwdefaults__['length']
518		- #if kwargs['repeat'] is None:
519		- # kwargs['repeat'] = dpp.Vault.__init__.__kwdefaults__['repeat']
520		- click.echo(repr(ctx.params))
	743	+
	744	+ if edit_notes:
	745	+ assert service is not None
	746	+ configuration = get_config()
	747	+ text = (DEFAULT_NOTES_TEMPLATE +
	748	+ configuration['services']
	749	+ .get(service, cast(dpp_types.VaultConfigServicesSettings, {}))
	750	+ .get('notes', ''))
	751	+ notes_value = click.edit(text=text)
	752	+ if notes_value is not None:
	753	+ notes_lines = collections.deque(notes_value.splitlines(True))
	754	+ while notes_lines:
	755	+ line = notes_lines.popleft()
	756	+ if line.startswith(DEFAULT_NOTES_MARKER):
	757	+ notes_value = ''.join(notes_lines)
	758	+ break
	759	+ else:
	760	+ if not notes_value.strip():
	761	+ ctx.fail('not saving new notes: user aborted request')
	762	+ configuration['services'].setdefault(service, {})['notes'] = (
	763	+ notes_value.strip('\n'))
	764	+ _save_config(configuration)
	765	+ elif delete_service_settings:
	766	+ assert service is not None
	767	+ configuration = get_config()
	768	+ if service in configuration['services']:
	769	+ del configuration['services'][service]
	770	+ _save_config(configuration)
	771	+ elif delete_globals:
	772	+ configuration = get_config()
	773	+ if 'global' in configuration:
	774	+ del configuration['global']
	775	+ _save_config(configuration)
	776	+ elif clear_all_settings:
	777	+ _save_config({'services': {}})
	778	+ elif import_settings:
	779	+ try:
	780	+ # TODO: keep track of auto-close; try os.dup if feasible
	781	+ infile = (cast(TextIO, import_settings)
	782	+ if hasattr(import_settings, 'close')
	783	+ else click.open_file(os.fspath(import_settings), 'rt'))
	784	+ with infile:
	785	+ maybe_config = json.load(infile)
	786	+ except json.JSONDecodeError as e:
	787	+ ctx.fail(f'Cannot load config: cannot decode JSON: {e}')
	788	+ except OSError as e:
	789	+ ctx.fail(f'Cannot load config: {e.strerror}')
	790	+ if dpp_types.is_vault_config(maybe_config):
	791	+ _save_config(maybe_config)
	792	+ else:
	793	+ ctx.fail('not a valid config')
	794	+ elif export_settings:
	795	+ configuration = get_config()
	796	+ try:
	797	+ # TODO: keep track of auto-close; try os.dup if feasible
	798	+ outfile = (cast(TextIO, export_settings)
	799	+ if hasattr(export_settings, 'close')
	800	+ else click.open_file(os.fspath(export_settings), 'wt'))
	801	+ with outfile:
	802	+ json.dump(configuration, outfile)
	803	+ except OSError as e:
	804	+ ctx.fail('cannot write config: {e.strerror}')
	805	+ else:
	806	+ configuration = get_config()
	807	+ # This block could be type checked more stringently, but this
	808	+ # would probably involve a lot of code repetition. Since we
	809	+ # have a type guarding function anyway, assert that we didn't
	810	+ # make any mistakes at the end instead.
	811	+ global_keys = {'key', 'phrase'}
	812	+ service_keys = {'key', 'phrase', 'length', 'repeat', 'lower',
	813	+ 'upper', 'number', 'space', 'dash', 'symbol'}
	814	+ settings: collections.ChainMap[str, Any] = collections.ChainMap(
	815	+ {k: v for k, v in locals().items()
	816	+ if k in service_keys and v is not None},
	817	+ cast(dict[str, Any],
	818	+ configuration['services'].get(service or '', {})),
	819	+ {},
	820	+ cast(dict[str, Any], configuration.get('global', {}))
	821	+ )
	822	+ if use_key:
	823	+ try:
	824	+ key = base64.standard_b64encode(
	825	+ _select_ssh_key()).decode('ASCII')
	826	+ except IndexError:
	827	+ ctx.fail('no valid SSH key selected')
	828	+ except RuntimeError as e:
	829	+ ctx.fail(str(e))
	830	+ elif use_phrase:
	831	+ maybe_phrase = _prompt_for_passphrase()
	832	+ if not maybe_phrase:
	833	+ ctx.fail('no passphrase given')
	834	+ else:
	835	+ phrase = maybe_phrase
	836	+ if store_config_only:
	837	+ view: collections.ChainMap[str, Any]
	838	+ view = (collections.ChainMap(*settings.maps[:2]) if service
	839	+ else settings.parents.parents)
	840	+ if use_key:
	841	+ view['key'] = key
	842	+ for m in view.maps:
	843	+ m.pop('phrase', '')
	844	+ elif use_phrase:
	845	+ view['phrase'] = phrase
	846	+ for m in view.maps:
	847	+ m.pop('key', '')
	848	+ if service:
	849	+ if not view.maps[0]:
	850	+ raise click.UsageError('cannot update service settings '
	851	+ 'without actual settings')
	852	+ else:
	853	+ configuration['services'].setdefault(
	854	+ service, {}).update(view) # type: ignore[typeddict-item]
	855	+ else:
	856	+ if not view.maps[0]:
	857	+ raise click.UsageError('cannot update global settings '
	858	+ 'without actual settings')
	859	+ else:
	860	+ configuration.setdefault(
	861	+ 'global', {}).update(view) # type: ignore[typeddict-item]
	862	+ assert dpp_types.is_vault_config(configuration), (
	863	+ f'invalid vault configuration: {configuration!r}'
	864	+ )
	865	+ _save_config(configuration)
	866	+ else:
	867	+ if not service:
	868	+ raise click.UsageError(f'SERVICE is required')
	869	+ kwargs: dict[str, Any] = {k: v for k, v in settings.items()
	870	+ if k in service_keys and v is not None}
	871	+ # If either --key or --phrase are given, use that setting.
	872	+ # Otherwise, if both key and phrase are set in the config,
	873	+ # one must be global (ignore it) and one must be
	874	+ # service-specific (use that one). Otherwise, if only one of
	875	+ # key and phrase is set in the config, use that one. In all
	876	+ # these above cases, set the phrase via
	877	+ # derivepassphrase.Vault.phrase_from_key if a key is
	878	+ # given. Finally, if nothing is set, error out.
	879	+ key_to_phrase = lambda key: dpp.Vault.phrase_from_key(
	880	+ base64.standard_b64decode(key))
	881	+ if use_key or use_phrase:
	882	+ if use_key:
	883	+ kwargs['phrase'] = key_to_phrase(key)
	884	+ else:
	885	+ kwargs['phrase'] = phrase
	886	+ kwargs.pop('key', '')
	887	+ elif kwargs.get('phrase') and kwargs.get('key'):
	888	+ if any('key' in m for m in settings.maps[:2]):
	889	+ kwargs['phrase'] = key_to_phrase(kwargs.pop('key'))
	890	+ else:
	891	+ kwargs.pop('key')
	892	+ elif kwargs.get('key'):
	893	+ kwargs['phrase'] = key_to_phrase(kwargs.pop('key'))
	894	+ elif kwargs.get('phrase'):
	895	+ pass
	896	+ else:
	897	+ raise click.UsageError(
	898	+ 'no passphrase or key given on command-line '
	899	+ 'or in configuration')
	900	+ vault = dpp.Vault(**kwargs)
	901	+ result = vault.generate(service)
	902	+ click.echo(result.decode('ASCII'))
521	903
522	904
523	905	if __name__ == '__main__':

...	...	@@ -4,16 +4,25 @@
4	4
5	5	"""Test OpenSSH key loading and signing."""
6	6
	7	+from __future__ import annotations
	8	+
	9	+import click
7	10	import pytest
8	11
9	12	import derivepassphrase
	13	+import derivepassphrase.cli
10	14	import ssh_agent_client
11	15
12	16	import base64
	17	+import errno
	18	+import io
13	19	import os
14	20	import socket
15	21	import subprocess
16	22
	23	+if not os.environ.get('SSH_AUTH_SOCK'): # pragma: no cover
	24	+ pytest.skip('no running SSH agent detected', allow_module_level=True)
	25	+
17	26	SUPPORTED = {
18	27	'ed25519': {
19	28	'private_key': rb'''-----BEGIN OPENSSH PRIVATE KEY-----
...	...	@@ -327,6 +336,14 @@ def test_client_string(input, expected):
327	336	string = ssh_agent_client.SSHAgentClient.string
328	337	assert bytes(string(input)) == expected
329	338
	339	+@pytest.mark.parametrize(['input', 'exc_type', 'exc_pattern'], [
	340	+ ('some string', TypeError, 'invalid payload type'),
	341	+])
	342	+def test_client_string_exceptions(input, exc_type, exc_pattern):
	343	+ string = ssh_agent_client.SSHAgentClient.string
	344	+ with pytest.raises(exc_type, match=exc_pattern):
	345	+ string(input)
	346	+
330	347	@pytest.mark.parametrize(['input', 'expected'], [
331	348	(b'\x00\x00\x00\x07ssh-rsa', b'ssh-rsa'),
332	349	(
...	...	@@ -396,6 +413,7 @@ def test_sign_data_via_agent(keytype, data_dict):
396	413	for k, c in client.list_keys()}
397	414	public_key_data = data_dict['public_key_data']
398	415	expected_signature = data_dict['expected_signature']
	416	+ derived_passphrase = data_dict['derived_passphrase']
399	417	if public_key_data not in key_comment_pairs: # pragma: no cover
400	418	pytest.skip('prerequisite SSH key not loaded')
401	419	signature = bytes(client.sign(
...	...	@@ -405,8 +423,8 @@ def test_sign_data_via_agent(keytype, data_dict):
405	423	payload=derivepassphrase.Vault._UUID, key=public_key_data))
406	424	assert signature2 == expected_signature, 'SSH signature mismatch'
407	425	assert (
408		- derivepassphrase.Vault.phrase_from_signature(public_key_data) ==
409		- expected_signature
	426	+ derivepassphrase.Vault.phrase_from_key(public_key_data) ==
	427	+ derived_passphrase
410	428	), 'SSH signature mismatch'
411	429
412	430	@pytest.mark.parametrize(['keytype', 'data_dict'], list(UNSUITABLE.items()))
...	...	@@ -439,4 +457,142 @@ def test_sign_data_via_agent_unsupported(keytype, data_dict):
439	457	payload=derivepassphrase.Vault._UUID, key=public_key_data))
440	458	assert signature != signature2, 'SSH signature repeatable?!'
441	459	with pytest.raises(ValueError, match='unsuitable SSH key'):
442		- derivepassphrase.Vault.phrase_from_signature(public_key_data)
	460	+ derivepassphrase.Vault.phrase_from_key(public_key_data)
	461	+
	462	+@pytest.mark.parametrize(['this_data', 'all_data'],
	463	+ [(v, tuple(SUPPORTED.values()))
	464	+ for v in SUPPORTED.values()])
	465	+def test_ssh_key_selector(monkeypatch, this_data, all_data):
	466	+ successfully_uploaded_keys: list[bytes] = []
	467	+ for data in all_data:
	468	+ private_key = data['private_key']
	469	+ try:
	470	+ result = subprocess.run(['ssh-add', '-t', '60', '-q', '-'],
	471	+ input=private_key, check=True,
	472	+ capture_output=True)
	473	+ except subprocess.CalledProcessError as e:
	474	+ if data == this_data:
	475	+ pytest.skip(
	476	+ f"uploading non-optional test key: {e!r}, "
	477	+ f"stdout={e.stdout!r}, stderr={e.stderr!r}"
	478	+ )
	479	+ else:
	480	+ successfully_uploaded_keys.append(data['public_key_data'])
	481	+ index = 1 + successfully_uploaded_keys.index(this_data['public_key_data'])
	482	+ b64_key = base64.standard_b64encode(
	483	+ this_data['public_key_data']).decode('ASCII')
	484	+ n = len(successfully_uploaded_keys)
	485	+ text = (f'Your selection? (1-{n}, leave empty to abort): {index}\n'
	486	+ if n > 1 else 'Use this key? yes\n')
	487	+
	488	+ @click.command()
	489	+ def driver():
	490	+ key = derivepassphrase.cli._select_ssh_key()
	491	+ click.echo(base64.standard_b64encode(key).decode('ASCII'))
	492	+
	493	+ runner = click.testing.CliRunner(mix_stderr=True)
	494	+ result = runner.invoke(driver, [],
	495	+ input=(f'{index}\n' if n > 1 else f'yes\n'),
	496	+ catch_exceptions=True)
	497	+ assert result.exit_code == 0, 'driver program failed?!'
	498	+ assert result.stdout.startswith('Suitable SSH keys:\n'), (
	499	+ 'missing expected output'
	500	+ )
	501	+ assert text in result.stdout, 'missing expected output'
	502	+ assert result.stdout.endswith(f'\n{b64_key}\n'), 'missing expected output'
	503	+
	504	+@pytest.mark.parametrize(['conn_hint'], [('none',), ('socket',), ('client',)])
	505	+def test_get_suitable_ssh_keys(conn_hint):
	506	+ hint: ssh_agent_client.SSHAgentClient \| socket.socket \| None
	507	+ match conn_hint:
	508	+ case 'client':
	509	+ hint = ssh_agent_client.SSHAgentClient()
	510	+ case 'socket':
	511	+ hint = socket.socket(family=socket.AF_UNIX)
	512	+ hint.connect(os.environ['SSH_AUTH_SOCK'])
	513	+ case _:
	514	+ assert conn_hint == 'none'
	515	+ hint = None
	516	+ exception: type[Exception] \| None = None
	517	+ try:
	518	+ list(derivepassphrase.cli._get_suitable_ssh_keys(hint))
	519	+ except RuntimeError: # pragma: no cover
	520	+ pass
	521	+ except Exception as e: # pragma: no cover
	522	+ exception = e
	523	+ finally:
	524	+ assert exception == None, 'exception querying suitable SSH keys'
	525	+
	526	+def test_constructor_no_running_agent(monkeypatch):
	527	+ monkeypatch.delenv('SSH_AUTH_SOCK', raising=False)
	528	+ sock = socket.socket(family=socket.AF_UNIX)
	529	+ with pytest.raises(RuntimeError, match='missing SSH_AUTH_SOCK'):
	530	+ ssh_agent_client.SSHAgentClient(socket=sock)
	531	+
	532	+def test_constructor_bad_running_agent(monkeypatch):
	533	+ monkeypatch.setenv('SSH_AUTH_SOCK', os.environ['SSH_AUTH_SOCK'] + '~')
	534	+ sock = socket.socket(family=socket.AF_UNIX)
	535	+ with pytest.raises(RuntimeError, match='unusable SSH_AUTH_SOCK'):
	536	+ ssh_agent_client.SSHAgentClient(socket=sock)
	537	+
	538	+@pytest.mark.parametrize(['response'], [
	539	+ (b'\x00\x00',),
	540	+ (b'\x00\x00\x00\x1f some bytes missing',),
	541	+])
	542	+def test_truncated_server_response(monkeypatch, response):
	543	+ client = ssh_agent_client.SSHAgentClient()
	544	+ response_stream = io.BytesIO(response)
	545	+ class PseudoSocket(object):
	546	+ pass
	547	+ pseudo_socket = PseudoSocket()
	548	+ pseudo_socket.sendall = lambda a, *kw: None
	549	+ pseudo_socket.recv = response_stream.read
	550	+ monkeypatch.setattr(client, '_connection', pseudo_socket)
	551	+ with pytest.raises(EOFError):
	552	+ client.request(255, b'')
	553	+
	554	+@pytest.mark.parametrize(
	555	+ ['response_code', 'response', 'exc_type', 'exc_pattern'],
	556	+ [
	557	+ (255, b'', RuntimeError, 'error return from SSH agent:'),
	558	+ (12, b'\x00\x00\x00\x01', EOFError, 'truncated response'),
	559	+ (12, b'\x00\x00\x00\x00abc', RuntimeError, 'overlong response'),
	560	+ ]
	561	+)
	562	+def test_list_keys_error_responses(monkeypatch, response_code, response,
	563	+ exc_type, exc_pattern):
	564	+ client = ssh_agent_client.SSHAgentClient()
	565	+ monkeypatch.setattr(client, 'request',
	566	+ lambda a, *kw: (response_code, response))
	567	+ with pytest.raises(exc_type, match=exc_pattern):
	568	+ client.list_keys()
	569	+
	570	+@pytest.mark.parametrize(
	571	+ ['key', 'check', 'response', 'exc_type', 'exc_pattern'],
	572	+ [
	573	+ (
	574	+ b'invalid-key',
	575	+ True,
	576	+ (255, b''),
	577	+ RuntimeError,
	578	+ 'target SSH key not loaded into agent',
	579	+ ),
	580	+ (
	581	+ SUPPORTED['ed25519']['public_key_data'],
	582	+ True,
	583	+ (255, b''),
	584	+ RuntimeError,
	585	+ 'signing data failed:',
	586	+ )
	587	+ ]
	588	+)
	589	+def test_sign_error_responses(monkeypatch, key, check, response, exc_type,
	590	+ exc_pattern):
	591	+ client = ssh_agent_client.SSHAgentClient()
	592	+ monkeypatch.setattr(client, 'request', lambda a, b: response)
	593	+ KeyCommentPair = ssh_agent_client.types.KeyCommentPair
	594	+ loaded_keys = [KeyCommentPair(v['public_key_data'], b'no comment')
	595	+ for v in SUPPORTED.values()]
	596	+ monkeypatch.setattr(client, 'list_keys', lambda: loaded_keys)
	597	+ with pytest.raises(exc_type, match=exc_pattern):
	598	+ client.sign(key, b'abc', check_if_key_loaded=check)

...	...	@@ -25,6 +25,7 @@ def test_big_endian_number(sequence, base, expected):
25	25	@pytest.mark.parametrize(['exc_type', 'exc_pattern', 'sequence' , 'base'], [
26	26	(ValueError, 'invalid base 3 digit:', [-1], 3),
27	27	(ValueError, 'invalid base:', [0], 1),
	28	+ (TypeError, 'not an integer:', [0.0, 1.0, 0.0, 1.0], 2),
28	29	])
29	30	def test_big_endian_number_exceptions(exc_type, exc_pattern, sequence, base):
30	31	with pytest.raises(exc_type, match=exc_pattern):
...	...	@@ -103,3 +104,5 @@ def test_internal_generating():
103	104	assert seq._generate_inner(1) == 0
104	105	with pytest.raises(ValueError, match='invalid target range'):
105	106	seq._generate_inner(0)
	107	+ with pytest.raises(ValueError, match='invalid base:'):
	108	+ seq._generate_inner(16, base=1)
106	109