derivepassphrase.git

@@ -11,6 +11,8 @@ import collections

 import hashlib

 import math

 import unicodedata

+from collections.abc import Callable

+from typing import TypeAlias

 from typing_extensions import assert_type

@@ -23,7 +25,8 @@ __version__ = '0.1.2'

 class AmbiguousByteRepresentationError(ValueError):

     """The object has an ambiguous byte representation."""

-    def __init__(self):

+

+    def __init__(self) -> None:

         super().__init__('text string has ambiguous byte representation')

@@ -425,6 +428,8 @@ class Vault:

             a passphrase deterministically.

         """

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

+        deterministic_signature_types: dict[str, TestFunc]

         deterministic_signature_types = {

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

                 b'\x00\x00\x00\x0bssh-ed25519'

@@ -28,6 +28,7 @@ from typing_extensions import (

 import derivepassphrase as dpp

 import ssh_agent_client

+import ssh_agent_client.types

 from derivepassphrase import types as dpp_types

 if TYPE_CHECKING:

@@ -159,7 +160,7 @@ def _get_suitable_ssh_keys(

     """

     client: ssh_agent_client.SSHAgentClient

-    client_context: contextlib.AbstractContextManager

+    client_context: contextlib.AbstractContextManager[Any]

     match conn:

         case ssh_agent_client.SSHAgentClient():

             client = conn

@@ -324,8 +325,15 @@ def _prompt_for_passphrase() -> str:

         The user input.

     """

-    return click.prompt(

-        'Passphrase', default='', hide_input=True, show_default=False, err=True

+    return cast(

+        str,

+        click.prompt(

+            'Passphrase',

+            default='',

+            hide_input=True,

+            show_default=False,

+            err=True,

+        ),

     )

@@ -349,8 +357,8 @@ class OptionGroupOption(click.Option):

     option_group_name: str = ''

     epilog: str = ''

-    def __init__(self, *args, **kwargs):  # type: ignore

-        if self.__class__ == __class__:

+    def __init__(self, *args: Any, **kwargs: Any) -> None:

+        if self.__class__ == __class__:  # type: ignore[name-defined]

             raise NotImplementedError

         super().__init__(*args, **kwargs)

@@ -809,7 +817,7 @@ def derivepassphrase(

         for name in param.opts + param.secondary_opts:

             params_by_str[name] = param

-    def is_param_set(param: click.Parameter):

+    def is_param_set(param: click.Parameter) -> bool:

         return bool(ctx.params.get(param.human_readable_name))

     def check_incompatible_options(

@@ -89,7 +89,7 @@ class Sequin:

         """

         msg = 'sequence item out of range'

-        def uint8_to_bits(value):

+        def uint8_to_bits(value: int) -> Iterator[int]:

             """Yield individual bits of an 8-bit number, MSB first."""

             for i in (0x80, 0x40, 0x20, 0x10, 0x08, 0x04, 0x02, 0x01):

                 yield 1 if value | i == value else 0

@@ -388,5 +388,5 @@ class SequinExhaustedError(Exception):

     """

-    def __init__(self):

+    def __init__(self) -> None:

         super().__init__('Sequin is exhausted')

@@ -34,7 +34,7 @@ _socket = socket

 class TrailingDataError(RuntimeError):

     """The result contained trailing data."""

-    def __init__(self):

+    def __init__(self) -> None:

         super().__init__('Overlong response from SSH agent')

@@ -274,7 +274,7 @@ class SSHAgentClient:

             raise EOFError(msg)

         return response[0], response[1:]

-    def list_keys(self) -> Sequence[ssh_types.KeyCommentPair]:

+    def list_keys(self) -> Sequence[types.KeyCommentPair]:

         """Request a list of keys known to the SSH agent.

         Returns:

@@ -290,9 +290,9 @@ class SSHAgentClient:

         """

         response_code, response = self.request(

-            ssh_types.SSH_AGENTC.REQUEST_IDENTITIES.value, b''

+            types.SSH_AGENTC.REQUEST_IDENTITIES.value, b''

         )

-        if response_code != ssh_types.SSH_AGENT.IDENTITIES_ANSWER.value:

+        if response_code != types.SSH_AGENT.IDENTITIES_ANSWER.value:

             msg = (

                 f'error return from SSH agent: '

                 f'{response_code = }, {response = }'

@@ -313,7 +313,7 @@ class SSHAgentClient:

             return bytes(buf)

         key_count = int.from_bytes(shift(4), 'big')

-        keys: collections.deque[ssh_types.KeyCommentPair]

+        keys: collections.deque[types.KeyCommentPair]

         keys = collections.deque()

         for _ in range(key_count):

             key_size = int.from_bytes(shift(4), 'big')

@@ -321,7 +321,7 @@ class SSHAgentClient:

             comment_size = int.from_bytes(shift(4), 'big')

             comment = shift(comment_size)

             # Both `key` and `comment` are not wrapped as SSH strings.

-            keys.append(ssh_types.KeyCommentPair(key, comment))

+            keys.append(types.KeyCommentPair(key, comment))

         if response_stream:

             raise TrailingDataError

         return keys

@@ -379,9 +379,9 @@ class SSHAgentClient:

         request_data.extend(self.string(payload))

         request_data.extend(self.uint32(flags))

         response_code, response = self.request(

-            ssh_types.SSH_AGENTC.SIGN_REQUEST.value, request_data

+            types.SSH_AGENTC.SIGN_REQUEST.value, request_data

         )

-        if response_code != ssh_types.SSH_AGENT.SIGN_RESPONSE.value:

+        if response_code != types.SSH_AGENT.SIGN_RESPONSE.value:

             msg = f'signing data failed: {response_code = }, {response = }'

             raise RuntimeError(msg)

         return self.unstring(response)

@@ -402,7 +402,7 @@ def isolated_config(

     monkeypatch: Any,

     runner: click.testing.CliRunner,

     config: Any,

-):

+) -> Iterator[None]:

     prog_name = derivepassphrase.cli.PROG_NAME

     env_name = prog_name.replace(' ', '_').upper() + '_PATH'

     with runner.isolated_filesystem():

@@ -28,31 +28,33 @@ class TestVault:

             ('twitter', twitter_phrase),

         ],

     )

-    def test_200_basic_configuration(self, service, expected):

+    def test_200_basic_configuration(

+        self, service: bytes | str, expected: bytes

+    ) -> None:

         assert Vault(phrase=self.phrase).generate(service) == expected

-    def test_201_phrase_dependence(self):

+    def test_201_phrase_dependence(self) -> None:

         assert (

             Vault(phrase=(self.phrase + b'X')).generate('google')

             == b'n+oIz6sL>K*lTEWYRO%7'

         )

-    def test_202_reproducibility_and_bytes_service_name(self):

+    def test_202_reproducibility_and_bytes_service_name(self) -> None:

         assert Vault(phrase=self.phrase).generate(b'google') == Vault(

             phrase=self.phrase

         ).generate('google')

-    def test_203_reproducibility_and_bytearray_service_name(self):

+    def test_203_reproducibility_and_bytearray_service_name(self) -> None:

         assert Vault(phrase=self.phrase).generate(b'google') == Vault(

             phrase=self.phrase

         ).generate(bytearray(b'google'))

-    def test_210_nonstandard_length(self):

+    def test_210_nonstandard_length(self) -> None:

         assert (

             Vault(phrase=self.phrase, length=4).generate('google') == b'xDFu'

         )

-    def test_211_repetition_limit(self):

+    def test_211_repetition_limit(self) -> None:

         assert (

             Vault(

                 phrase=b'', length=24, symbol=0, number=0, repeat=1

@@ -60,37 +62,37 @@ class TestVault:

             == b'IVTDzACftqopUXqDHPkuCIhV'

         )

-    def test_212_without_symbols(self):

+    def test_212_without_symbols(self) -> None:

         assert (

             Vault(phrase=self.phrase, symbol=0).generate('google')

             == b'XZ4wRe0bZCazbljCaMqR'

         )

-    def test_213_no_numbers(self):

+    def test_213_no_numbers(self) -> None:

         assert (

             Vault(phrase=self.phrase, number=0).generate('google')

             == b'_*$TVH.%^aZl(LUeOT?>'

         )

-    def test_214_no_lowercase_letters(self):

+    def test_214_no_lowercase_letters(self) -> None:

         assert (

             Vault(phrase=self.phrase, lower=0).generate('google')

             == b':{?)+7~@OA:L]!0E$)(+'

         )

-    def test_215_at_least_5_digits(self):

+    def test_215_at_least_5_digits(self) -> None:

         assert (

             Vault(phrase=self.phrase, length=8, number=5).generate('songkick')

             == b'i0908.7['

         )

-    def test_216_lots_of_spaces(self):

+    def test_216_lots_of_spaces(self) -> None:

         assert (

             Vault(phrase=self.phrase, space=12).generate('songkick')

             == b' c   6 Bq  % 5fR    '

         )

-    def test_217_all_character_classes(self):

+    def test_217_all_character_classes(self) -> None:

         assert (

             Vault(

                 phrase=self.phrase,

@@ -104,7 +106,7 @@ class TestVault:

             == b': : fv_wqt>a-4w1S  R'

         )

-    def test_218_only_numbers_and_very_high_repetition_limit(self):

+    def test_218_only_numbers_and_very_high_repetition_limit(self) -> None:

         generated = Vault(

             phrase=b'',

             length=40,

@@ -130,13 +132,13 @@ class TestVault:

         for substring in forbidden_substrings:

             assert substring not in generated

-    def test_219_very_limited_character_set(self):

+    def test_219_very_limited_character_set(self) -> None:

         generated = Vault(

             phrase=b'', length=24, lower=0, upper=0, space=0, symbol=0

         ).generate('testing')

         assert generated == b'763252593304946694588866'

-    def test_220_character_set_subtraction(self):

+    def test_220_character_set_subtraction(self) -> None:

         assert Vault._subtract(b'be', b'abcdef') == bytearray(b'acdf')

     @pytest.mark.parametrize(

@@ -230,13 +232,13 @@ class TestVault:

             assert binstr(s) == bytes(s)

             assert binstr(binstr(s)) == bytes(s)

-    def test_310_too_many_symbols(self):

+    def test_310_too_many_symbols(self) -> None:

         with pytest.raises(

             ValueError, match='requested passphrase length too short'

         ):

             Vault(phrase=self.phrase, symbol=100)

-    def test_311_no_viable_characters(self):

+    def test_311_no_viable_characters(self) -> None:

         with pytest.raises(ValueError, match='no allowed characters left'):

             Vault(

                 phrase=self.phrase,

@@ -248,7 +250,7 @@ class TestVault:

                 symbol=0,

             )

-    def test_320_character_set_subtraction_duplicate(self):

+    def test_320_character_set_subtraction_duplicate(self) -> None:

         with pytest.raises(ValueError, match='duplicate characters'):

             Vault._subtract(b'abcdef', b'aabbccddeeff')

         with pytest.raises(ValueError, match='duplicate characters'):

@@ -8,7 +8,7 @@ import contextlib

 import json

 import os

 import socket

-from typing import TYPE_CHECKING, cast

+from typing import TYPE_CHECKING

 import click.testing

 import pytest

@@ -196,7 +196,7 @@ for opt, config in SINGLES.items():

 class TestCLI:

-    def test_200_help_output(self):

+    def test_200_help_output(self) -> None:

         runner = click.testing.CliRunner(mix_stderr=False)

         result = runner.invoke(

             cli.derivepassphrase, ['--help'], catch_exceptions=False

@@ -390,11 +390,10 @@ class TestCLI:

     )

     def test_210_invalid_argument_range(self, 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],

+                [option, value, '-p', DUMMY_SERVICE],

                 input=DUMMY_PASSPHRASE,

                 catch_exceptions=False,

             )

@@ -872,7 +871,7 @@ contents go here

         ):

             custom_error = 'custom error message'

-            def raiser():

+            def raiser() -> None:

                 raise RuntimeError(custom_error)

             monkeypatch.setattr(cli, '_select_ssh_key', raiser)

@@ -925,7 +924,7 @@ class TestCLIUtils:

         @click.command()

         @click.option('--heading', default='Our menu:')

         @click.argument('items', nargs=-1)

-        def driver(heading, items):

+        def driver(heading: str, items: list[str]) -> None:

             # from https://montypython.fandom.com/wiki/Spam#The_menu

             items = items or [

                 'Egg and bacon',

@@ -1001,7 +1000,7 @@ Your selection? (1-10, leave empty to abort):\x20

         @click.command()

         @click.option('--item', default='baked beans')

         @click.argument('prompt')

-        def driver(item, prompt):

+        def driver(item: str, prompt: str) -> None:

             try:

                 cli._prompt_for_selection(

                     [item], heading='', single_choice_prompt=prompt

@@ -27,7 +27,9 @@ class TestStaticFunctionality:

             ([1, 7, 5, 5], 8, 0o1755),

         ],

     )

-    def test_200_big_endian_number(self, sequence, base, expected):

+    def test_200_big_endian_number(

+        self, sequence: list[int], base: int, expected: int

+    ) -> None:

         assert (

             sequin.Sequin._big_endian_number(sequence, base=base)

         ) == expected

@@ -41,8 +43,12 @@ class TestStaticFunctionality:

         ],

     )

     def test_300_big_endian_number_exceptions(

-        self, exc_type, exc_pattern, sequence, base

-    ):

+        self,

+        exc_type: type[Exception],

+        exc_pattern: str,

+        sequence: list[int],

+        base: int,

+    ) -> None:

         with pytest.raises(exc_type, match=exc_pattern):

             sequin.Sequin._big_endian_number(sequence, base=base)

@@ -61,11 +67,16 @@ class TestSequin:

             ('OK', False, bitseq('0100111101001011')),

         ],

     )

-    def test_200_constructor(self, sequence, is_bitstring, expected):

+    def test_200_constructor(

+        self,

+        sequence: str | bytes | bytearray | list[int],

+        is_bitstring: bool,

+        expected: list[int],

+    ) -> None:

         seq = sequin.Sequin(sequence, is_bitstring=is_bitstring)

         assert seq.bases == {2: collections.deque(expected)}

-    def test_201_generating(self):

+    def test_201_generating(self) -> None:

         seq = sequin.Sequin(

             [1, 1, 0, 1, 0, 1, 0, 1, 1, 1, 1, 1, 0, 0, 1], is_bitstring=True

         )

@@ -83,7 +94,7 @@ class TestSequin:

         with pytest.raises(ValueError, match='invalid target range'):

             seq.generate(0)

-    def test_210_internal_generating(self):

+    def test_210_internal_generating(self) -> None:

         seq = sequin.Sequin(

             [1, 1, 0, 1, 0, 1, 0, 1, 1, 1, 1, 1, 0, 0, 1], is_bitstring=True

         )

@@ -101,7 +112,7 @@ class TestSequin:

         with pytest.raises(ValueError, match='invalid base:'):

             seq._generate_inner(16, base=1)

-    def test_211_shifting(self):

+    def test_211_shifting(self) -> None:

         seq = sequin.Sequin([1, 0, 1, 0, 0, 1, 0, 0, 0, 1], is_bitstring=True)

         assert seq.bases == {

             2: collections.deque([1, 0, 1, 0, 0, 1, 0, 0, 0, 1])

@@ -130,7 +141,11 @@ class TestSequin:

         ],

     )

     def test_300_constructor_exceptions(

-        self, sequence, is_bitstring, exc_type, exc_pattern

-    ):

+        self,

+        sequence: list[int] | str,

+        is_bitstring: bool,

+        exc_type: type[Exception],

+        exc_pattern: str,

+    ) -> None:

         with pytest.raises(exc_type, match=exc_pattern):

             sequin.Sequin(sequence, is_bitstring=is_bitstring)

@@ -11,6 +11,7 @@ import io

 import os

 import socket

 import subprocess

+from typing import TYPE_CHECKING

 import click

 import click.testing

@@ -22,6 +23,9 @@ import derivepassphrase.cli

 import ssh_agent_client

 import tests

+if TYPE_CHECKING:

+    from collections.abc import Iterator

+

 class TestStaticFunctionality:

     @pytest.mark.parametrize(

@@ -31,13 +35,15 @@ class TestStaticFunctionality:

             for val in tests.SUPPORTED_KEYS.values()

         ],

     )

-    def test_100_key_decoding(self, public_key, public_key_data):

+    def test_100_key_decoding(

+        self, public_key: bytes, public_key_data: bytes

+    ) -> None:

         keydata = base64.b64decode(public_key.split(None, 2)[1])

         assert (

             keydata == public_key_data

         ), "recorded public key data doesn't match"

-    def test_200_constructor_no_running_agent(self, monkeypatch):

+    def test_200_constructor_no_running_agent(self, monkeypatch: Any) -> None:

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

         sock = socket.socket(family=socket.AF_UNIX)

         with pytest.raises(

@@ -51,7 +57,7 @@ class TestStaticFunctionality:

             (16777216, b'\x01\x00\x00\x00'),

         ],

     )

-    def test_210_uint32(self, input, expected):

+    def test_210_uint32(self, input: int, expected: bytes | bytearray) -> None:

         uint32 = ssh_agent_client.SSHAgentClient.uint32

         assert uint32(input) == expected

@@ -66,7 +72,9 @@ class TestStaticFunctionality:

             ),

         ],

     )

-    def test_211_string(self, input, expected):

+    def test_211_string(

+        self, input: bytes | bytearray, expected: bytes | bytearray

+    ) -> None:

         string = ssh_agent_client.SSHAgentClient.string

         assert bytes(string(input)) == expected

@@ -80,7 +88,9 @@ class TestStaticFunctionality:

             ),

         ],

     )

-    def test_212_unstring(self, input, expected):

+    def test_212_unstring(

+        self, input: bytes | bytearray, expected: bytes | bytearray

+    ) -> None:

         unstring = ssh_agent_client.SSHAgentClient.unstring

         unstring_prefix = ssh_agent_client.SSHAgentClient.unstring_prefix

         assert bytes(unstring(input)) == expected

@@ -96,7 +106,9 @@ class TestStaticFunctionality:

             (-1, OverflowError, "can't convert negative int to unsigned"),

         ],

     )

-    def test_310_uint32_exceptions(self, value, exc_type, exc_pattern):

+    def test_310_uint32_exceptions(

+        self, value: int, exc_type: type[Exception], exc_pattern: str

+    ) -> None:

         uint32 = ssh_agent_client.SSHAgentClient.uint32

         with pytest.raises(exc_type, match=exc_pattern):

             uint32(value)

@@ -107,7 +119,9 @@ class TestStaticFunctionality:

             ('some string', TypeError, 'invalid payload type'),

         ],

     )

-    def test_311_string_exceptions(self, input, exc_type, exc_pattern):

+    def test_311_string_exceptions(

+        self, input: Any, exc_type: type[Exception], exc_pattern: str

+    ) -> None:

         string = ssh_agent_client.SSHAgentClient.string

         with pytest.raises(exc_type, match=exc_pattern):

             string(input)

@@ -133,8 +147,13 @@ class TestStaticFunctionality:

         ],

     )

     def test_312_unstring_exceptions(

-        self, input, exc_type, exc_pattern, has_trailer, parts

-    ):

+        self,

+        input: bytes | bytearray,

+        exc_type: type[Exception],

+        exc_pattern: str,

+        has_trailer: bool,

+        parts: tuple[bytes | bytearray, bytes | bytearray] | None,

+    ) -> None:

         unstring = ssh_agent_client.SSHAgentClient.unstring

         unstring_prefix = ssh_agent_client.SSHAgentClient.unstring_prefix

         with pytest.raises(exc_type, match=exc_pattern):

@@ -151,7 +170,9 @@ class TestAgentInteraction:

     @pytest.mark.parametrize(

         ['keytype', 'data_dict'], list(tests.SUPPORTED_KEYS.items())

     )

-    def test_200_sign_data_via_agent(self, keytype, data_dict):

+    def test_200_sign_data_via_agent(

+        self, keytype: str, data_dict: tests.SSHTestKey

+    ) -> None:

         del keytype  # Unused.

         private_key = data_dict['private_key']

         try:

@@ -200,7 +221,9 @@ class TestAgentInteraction:

     @pytest.mark.parametrize(

         ['keytype', 'data_dict'], list(tests.UNSUITABLE_KEYS.items())

     )

-    def test_201_sign_data_via_agent_unsupported(self, keytype, data_dict):

+    def test_201_sign_data_via_agent_unsupported(

+        self, keytype: str, data_dict: tests.SSHTestKey

+    ) -> None:

         del keytype  # Unused.

         private_key = data_dict['private_key']

         try:

@@ -243,7 +266,7 @@ class TestAgentInteraction:

                 derivepassphrase.Vault.phrase_from_key(public_key_data)

     @staticmethod

-    def _params():

+    def _params() -> Iterator[tuple[bytes, bool]]:

         for value in tests.SUPPORTED_KEYS.values():

             key = value['public_key_data']

             yield (key, False)

@@ -254,8 +277,10 @@ class TestAgentInteraction:

                 yield (key, True)

     @pytest.mark.parametrize(['key', 'single'], list(_params()))

-    def test_210_ssh_key_selector(self, monkeypatch, key, single):

-        def key_is_suitable(key: bytes):

+    def test_210_ssh_key_selector(

+        self, monkeypatch: Any, key: bytes, single: bool

+    ) -> None:

+        def key_is_suitable(key: bytes) -> bool:

             return key in {

                 v['public_key_data'] for v in tests.SUPPORTED_KEYS.values()

             }

@@ -288,7 +313,7 @@ class TestAgentInteraction:

         b64_key = base64.standard_b64encode(key).decode('ASCII')

         @click.command()

-        def driver():

+        def driver() -> None:

             key = derivepassphrase.cli._select_ssh_key()

             click.echo(base64.standard_b64encode(key).decode('ASCII'))

@@ -310,7 +335,7 @@ class TestAgentInteraction:

     del _params

-    def test_300_constructor_bad_running_agent(self, monkeypatch):

+    def test_300_constructor_bad_running_agent(self, monkeypatch: Any) -> None:

         monkeypatch.setenv('SSH_AUTH_SOCK', os.environ['SSH_AUTH_SOCK'] + '~')

         sock = socket.socket(family=socket.AF_UNIX)

         with pytest.raises(OSError):  # noqa: PT011

@@ -323,7 +348,9 @@ class TestAgentInteraction:

             b'\x00\x00\x00\x1f some bytes missing',

         ],

     )

-    def test_310_truncated_server_response(self, monkeypatch, response):

+    def test_310_truncated_server_response(

+        self, monkeypatch: Any, response: bytes

+    ) -> None:

         client = ssh_agent_client.SSHAgentClient()

         response_stream = io.BytesIO(response)

@@ -354,8 +381,13 @@ class TestAgentInteraction:

         ],

     )

     def test_320_list_keys_error_responses(

-        self, monkeypatch, response_code, response, exc_type, exc_pattern

-    ):

+        self,

+        monkeypatch: Any,

+        response_code: int,

+        response: bytes | bytearray,

+        exc_type: type[Exception],

+        exc_pattern: str,

+    ) -> None:

         client = ssh_agent_client.SSHAgentClient()

         monkeypatch.setattr(

             client,

@@ -386,8 +418,14 @@ class TestAgentInteraction:

         ],

     )

     def test_330_sign_error_responses(

-        self, monkeypatch, key, check, response, exc_type, exc_pattern

-    ):

+        self,

+        monkeypatch: Any,

+        key: bytes | bytearray,

+        check: bool,

+        response: tuple[int, bytes | bytearray],

+        exc_type: type[Exception],

+        exc_pattern: str,

+    ) -> None:

         client = ssh_agent_client.SSHAgentClient()

         monkeypatch.setattr(client, 'request', lambda a, b: response)  # noqa: ARG005

         KeyCommentPair = ssh_agent_client.types.KeyCommentPair  # noqa: N806

@@ -0,0 +1 @@

+Fix typing issues according to `mypy`'s strict mode.