derivepassphrase.git

@@ -34,7 +34,7 @@ if TYPE_CHECKING:

 __all__ = (

     'SSH_AGENT',

     'SSH_AGENTC',

-    'KeyCommentPair',

+    'SSHKeyCommentPair',

     'VaultConfig',

     'is_vault_config',

 )

@@ -577,7 +577,7 @@ def clean_up_falsy_vault_config_values(  # noqa: C901,PLR0912

     return cleanup_completed

-class KeyCommentPair(NamedTuple):

+class SSHKeyCommentPair(NamedTuple):

     """SSH key plus comment pair.  For typing purposes.

     Attributes:

@@ -660,3 +660,44 @@ class SSH_AGENT(enum.Enum):  # noqa: N801

     """"""

     EXTENSION_RESPONSE: int = 29

     """"""

+

+

+class StoreroomKeyPair(NamedTuple):

+    """A pair of AES256 keys, one for encryption and one for signing.

+

+    Attributes:

+        encryption_key:

+            AES256 key, used for encryption with AES256-CBC (with PKCS#7

+            padding).

+        signing_key:

+            AES256 key, used for signing with HMAC-SHA256.

+

+    """

+

+    encryption_key: bytes

+    """"""

+    signing_key: bytes

+    """"""

+

+

+class StoreroomMasterKeys(NamedTuple):

+    """A triple of AES256 keys, for encryption, signing and hashing.

+

+    Attributes:

+        hashing_key:

+            AES256 key, used for hashing with HMAC-SHA256 to derive

+            a hash table slot for an item.

+        encryption_key:

+            AES256 key, used for encryption with AES256-CBC (with PKCS#7

+            padding).

+        signing_key:

+            AES256 key, used for signing with HMAC-SHA256.

+

+    """

+

+    hashing_key: bytes

+    """"""

+    encryption_key: bytes

+    """"""

+    signing_key: bytes

+    """"""

@@ -1794,7 +1794,7 @@ def _load_user_config() -> dict[str, Any]:

 def _get_suitable_ssh_keys(

     conn: ssh_agent.SSHAgentClient | socket.socket | None = None, /

-) -> Iterator[_types.KeyCommentPair]:

+) -> Iterator[_types.SSHKeyCommentPair]:

     """Yield all SSH keys suitable for passphrase derivation.

     Suitable SSH keys are queried from the running SSH agent (see

@@ -32,10 +32,10 @@ import logging

 import os

 import os.path

 import struct

-from typing import TYPE_CHECKING, Any, TypedDict

+from typing import TYPE_CHECKING, Any

 from derivepassphrase import _cli_msg as _msg

-from derivepassphrase import exporter

+from derivepassphrase import _types, exporter

 if TYPE_CHECKING:

     from collections.abc import Iterator

@@ -90,51 +90,10 @@ def _h(bs: Buffer) -> str:

     return '<{}>'.format(memoryview(bs).hex(' '))

-class KeyPair(TypedDict):

-    """A pair of AES256 keys, one for encryption and one for signing.

-

-    Attributes:

-        encryption_key:

-            AES256 key, used for encryption with AES256-CBC (with PKCS#7

-            padding).

-        signing_key:

-            AES256 key, used for signing with HMAC-SHA256.

-

-    """

-

-    encryption_key: bytes

-    """"""

-    signing_key: bytes

-    """"""

-

-

-class MasterKeys(TypedDict):

-    """A triple of AES256 keys, for encryption, signing and hashing.

-

-    Attributes:

-        hashing_key:

-            AES256 key, used for hashing with HMAC-SHA256 to derive

-            a hash table slot for an item.

-        encryption_key:

-            AES256 key, used for encryption with AES256-CBC (with PKCS#7

-            padding).

-        signing_key:

-            AES256 key, used for signing with HMAC-SHA256.

-

-    """

-

-    hashing_key: bytes

-    """"""

-    encryption_key: bytes

-    """"""

-    signing_key: bytes

-    """"""

-

-

 def derive_master_keys_keys(

     password: str | Buffer,

     iterations: int,

-) -> KeyPair:

+) -> _types.StoreroomKeyPair:

     """Derive encryption and signing keys for the master keys data.

     The master password is run through a key derivation function to

@@ -185,16 +144,16 @@ def derive_master_keys_keys(

             iterations=iterations,

         ),

     )

-    return {

-        'encryption_key': encryption_key,

-        'signing_key': signing_key,

-    }

+    return _types.StoreroomKeyPair(

+        encryption_key=encryption_key,

+        signing_key=signing_key,

+    )

 def decrypt_master_keys_data(

     data: Buffer,

-    keys: KeyPair,

-) -> MasterKeys:

+    keys: _types.StoreroomKeyPair,

+) -> _types.StoreroomMasterKeys:

     r"""Decrypt the master keys data.

     The master keys data contains:

@@ -245,12 +204,12 @@ def decrypt_master_keys_data(

     ciphertext, claimed_mac = struct.unpack(

         f'{len(data) - MAC_SIZE}s {MAC_SIZE}s', data

     )

-    actual_mac = hmac.HMAC(keys['signing_key'], hashes.SHA256())

+    actual_mac = hmac.HMAC(keys.signing_key, hashes.SHA256())

     actual_mac.update(ciphertext)

     logger.debug(

         _msg.TranslatedString(

             _msg.DebugMsgTemplate.MASTER_KEYS_DATA_MAC_INFO,

-            sign_key=_h(keys['signing_key']),

+            sign_key=_h(keys.signing_key),

             ciphertext=_h(ciphertext),

             claimed_mac=_h(claimed_mac),

             actual_mac=_h(actual_mac.copy().finalize()),

@@ -263,7 +222,7 @@ def decrypt_master_keys_data(

             f'{IV_SIZE}s {len(ciphertext) - IV_SIZE}s', ciphertext

         )

         decryptor = ciphers.Cipher(

-            algorithms.AES256(keys['encryption_key']), modes.CBC(iv)

+            algorithms.AES256(keys.encryption_key), modes.CBC(iv)

         ).decryptor()

         padded_plaintext = bytearray()

         padded_plaintext.extend(decryptor.update(payload))

@@ -278,17 +237,17 @@ def decrypt_master_keys_data(

     except (ValueError, struct.error) as exc:

         msg = 'Invalid encrypted master keys payload'

         raise ValueError(msg) from exc

-    return {

-        'hashing_key': hashing_key,

-        'encryption_key': encryption_key,

-        'signing_key': signing_key,

-    }

+    return _types.StoreroomMasterKeys(

+        hashing_key=hashing_key,

+        encryption_key=encryption_key,

+        signing_key=signing_key,

+    )

 def decrypt_session_keys(

     data: Buffer,

-    master_keys: MasterKeys,

-) -> KeyPair:

+    master_keys: _types.StoreroomMasterKeys,

+) -> _types.StoreroomKeyPair:

     r"""Decrypt the bucket item's session keys.

     The bucket item's session keys are single-use keys for encrypting

@@ -337,12 +296,12 @@ def decrypt_session_keys(

     ciphertext, claimed_mac = struct.unpack(

         f'{len(data) - MAC_SIZE}s {MAC_SIZE}s', data

     )

-    actual_mac = hmac.HMAC(master_keys['signing_key'], hashes.SHA256())

+    actual_mac = hmac.HMAC(master_keys.signing_key, hashes.SHA256())

     actual_mac.update(ciphertext)

     logger.debug(

         _msg.TranslatedString(

             _msg.DebugMsgTemplate.DECRYPT_BUCKET_ITEM_SESSION_KEYS_MAC_INFO,

-            sign_key=_h(master_keys['signing_key']),

+            sign_key=_h(master_keys.signing_key),

             ciphertext=_h(ciphertext),

             claimed_mac=_h(claimed_mac),

             actual_mac=_h(actual_mac.copy().finalize()),

@@ -355,7 +314,7 @@ def decrypt_session_keys(

             f'{IV_SIZE}s {len(ciphertext) - IV_SIZE}s', ciphertext

         )

         decryptor = ciphers.Cipher(

-            algorithms.AES256(master_keys['encryption_key']), modes.CBC(iv)

+            algorithms.AES256(master_keys.encryption_key), modes.CBC(iv)

         ).decryptor()

         padded_plaintext = bytearray()

         padded_plaintext.extend(decryptor.update(payload))

@@ -371,23 +330,23 @@ def decrypt_session_keys(

         msg = 'Invalid encrypted session keys payload'

         raise ValueError(msg) from exc

-    session_keys: KeyPair = {

-        'encryption_key': session_encryption_key,

-        'signing_key': session_signing_key,

-    }

+    session_keys = _types.StoreroomKeyPair(

+        encryption_key=session_encryption_key,

+        signing_key=session_signing_key,

+    )

     logger.debug(

         _msg.TranslatedString(

             _msg.DebugMsgTemplate.DECRYPT_BUCKET_ITEM_SESSION_KEYS_INFO,

-            enc_key=_h(master_keys['encryption_key']),

+            enc_key=_h(master_keys.encryption_key),

             iv=_h(iv),

             ciphertext=_h(payload),

             plaintext=_h(plaintext),

             code=_msg.TranslatedString(

-                '{{"encryption_key": bytes.fromhex({enc_key!r}), '

-                '"signing_key": bytes.fromhex({sign_key!r})}}',

-                enc_key=session_keys['encryption_key'].hex(' '),

-                sign_key=session_keys['signing_key'].hex(' '),

+                'StoreroomKeyPair(encryption_key=bytes.fromhex({enc_key!r}), '

+                'signing_key=bytes.fromhex({sign_key!r}))',

+                enc_key=session_keys.encryption_key.hex(' '),

+                sign_key=session_keys.signing_key.hex(' '),

             ),

         ),

     )

@@ -397,7 +356,7 @@ def decrypt_session_keys(

 def decrypt_contents(

     data: Buffer,

-    session_keys: KeyPair,

+    session_keys: _types.StoreroomKeyPair,

 ) -> Buffer:

     """Decrypt the bucket item's contents.

@@ -443,12 +402,12 @@ def decrypt_contents(

     ciphertext, claimed_mac = struct.unpack(

         f'{len(data) - MAC_SIZE}s {MAC_SIZE}s', data

     )

-    actual_mac = hmac.HMAC(session_keys['signing_key'], hashes.SHA256())

+    actual_mac = hmac.HMAC(session_keys.signing_key, hashes.SHA256())

     actual_mac.update(ciphertext)

     logger.debug(

         _msg.TranslatedString(

             _msg.DebugMsgTemplate.DECRYPT_BUCKET_ITEM_MAC_INFO,

-            sign_key=_h(session_keys['signing_key']),

+            sign_key=_h(session_keys.signing_key),

             ciphertext=_h(ciphertext),

             claimed_mac=_h(claimed_mac),

             actual_mac=_h(actual_mac.copy().finalize()),

@@ -460,7 +419,7 @@ def decrypt_contents(

         f'{IV_SIZE}s {len(ciphertext) - IV_SIZE}s', ciphertext

     )

     decryptor = ciphers.Cipher(

-        algorithms.AES256(session_keys['encryption_key']), modes.CBC(iv)

+        algorithms.AES256(session_keys.encryption_key), modes.CBC(iv)

     ).decryptor()

     padded_plaintext = bytearray()

     padded_plaintext.extend(decryptor.update(payload))

@@ -473,7 +432,7 @@ def decrypt_contents(

     logger.debug(

         _msg.TranslatedString(

             _msg.DebugMsgTemplate.DECRYPT_BUCKET_ITEM_INFO,

-            enc_key=_h(session_keys['encryption_key']),

+            enc_key=_h(session_keys.encryption_key),

             iv=_h(iv),

             ciphertext=_h(payload),

             plaintext=_h(plaintext),

@@ -485,7 +444,7 @@ def decrypt_contents(

 def decrypt_bucket_item(

     bucket_item: Buffer,

-    master_keys: MasterKeys,

+    master_keys: _types.StoreroomMasterKeys,

 ) -> Buffer:

     """Decrypt a bucket item.

@@ -519,8 +478,8 @@ def decrypt_bucket_item(

         _msg.TranslatedString(

             _msg.DebugMsgTemplate.DECRYPT_BUCKET_ITEM_KEY_INFO,

             plaintext=_h(bucket_item),

-            enc_key=_h(master_keys['encryption_key']),

-            sign_key=_h(master_keys['signing_key']),

+            enc_key=_h(master_keys.encryption_key),

+            sign_key=_h(master_keys.signing_key),

         ),

     )

     data_version, encrypted_session_keys, data_contents = struct.unpack(

@@ -539,7 +498,7 @@ def decrypt_bucket_item(

 def decrypt_bucket_file(

     filename: str,

-    master_keys: MasterKeys,

+    master_keys: _types.StoreroomMasterKeys,

     *,

     root_dir: str | bytes | os.PathLike = '.',

 ) -> Iterator[Buffer]:

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

             raise SSHAgentFailedError(response[0], response[1:])

         return response[1:]

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

+    def list_keys(self) -> Sequence[_types.SSHKeyCommentPair]:

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

         Returns:

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

             return bytes(buf)

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

-        keys: collections.deque[_types.KeyCommentPair]

+        keys: collections.deque[_types.SSHKeyCommentPair]

         keys = collections.deque()

         for _ in range(key_count):

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

@@ -529,7 +529,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(_types.KeyCommentPair(key, comment))

+            keys.append(_types.SSHKeyCommentPair(key, comment))

         if response_stream:

             raise TrailingDataError

         return keys

@@ -1364,9 +1364,9 @@ def hypothesis_settings_coverage_compatible_with_caplog(

     return settings if f is None else settings(f)

-def list_keys(self: Any = None) -> list[_types.KeyCommentPair]:

+def list_keys(self: Any = None) -> list[_types.SSHKeyCommentPair]:

     del self  # Unused.

-    Pair = _types.KeyCommentPair  # noqa: N806

+    Pair = _types.SSHKeyCommentPair  # noqa: N806

     list1 = [

         Pair(value['public_key_data'], f'{key} test key'.encode('ASCII'))

         for key, value in SUPPORTED_KEYS.items()

@@ -1390,9 +1390,9 @@ def sign(

     raise AssertionError

-def list_keys_singleton(self: Any = None) -> list[_types.KeyCommentPair]:

+def list_keys_singleton(self: Any = None) -> list[_types.SSHKeyCommentPair]:

     del self  # Unused.

-    Pair = _types.KeyCommentPair  # noqa: N806

+    Pair = _types.SSHKeyCommentPair  # noqa: N806

     list1 = [

         Pair(value['public_key_data'], f'{key} test key'.encode('ASCII'))

         for key, value in SUPPORTED_KEYS.items()

@@ -1400,9 +1400,9 @@ def list_keys_singleton(self: Any = None) -> list[_types.KeyCommentPair]:

     return list1[:1]

-def suitable_ssh_keys(conn: Any) -> Iterator[_types.KeyCommentPair]:

+def suitable_ssh_keys(conn: Any) -> Iterator[_types.SSHKeyCommentPair]:

     del conn  # Unused.

-    Pair = _types.KeyCommentPair  # noqa: N806

+    Pair = _types.SSHKeyCommentPair  # noqa: N806

     yield from [

         Pair(DUMMY_KEY1, b'no comment'),

         Pair(DUMMY_KEY2, b'a comment'),

@@ -1563,7 +1563,7 @@ contents go here

             def func(

                 *_args: Any,

                 **_kwargs: Any,

-            ) -> list[_types.KeyCommentPair]:

+            ) -> list[_types.SSHKeyCommentPair]:

                 return []

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

@@ -297,11 +297,11 @@ class TestStoreroom:

         bucket_item = (

             b'\xff' + bytes(storeroom.ENCRYPTED_KEYPAIR_SIZE) + bytes(3)

         )

-        master_keys: storeroom.MasterKeys = {

-            'encryption_key': bytes(storeroom.KEY_SIZE),

-            'signing_key': bytes(storeroom.KEY_SIZE),

-            'hashing_key': bytes(storeroom.KEY_SIZE),

-        }

+        master_keys = _types.StoreroomMasterKeys(

+            encryption_key=bytes(storeroom.KEY_SIZE),

+            signing_key=bytes(storeroom.KEY_SIZE),

+            hashing_key=bytes(storeroom.KEY_SIZE),

+        )

         with pytest.raises(ValueError, match='Cannot handle version 255'):

             storeroom.decrypt_bucket_item(bucket_item, master_keys)

@@ -312,11 +312,11 @@ class TestStoreroom:

         config: str,

     ) -> None:

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

-        master_keys: storeroom.MasterKeys = {

-            'encryption_key': bytes(storeroom.KEY_SIZE),

-            'signing_key': bytes(storeroom.KEY_SIZE),

-            'hashing_key': bytes(storeroom.KEY_SIZE),

-        }

+        master_keys = _types.StoreroomMasterKeys(

+            encryption_key=bytes(storeroom.KEY_SIZE),

+            signing_key=bytes(storeroom.KEY_SIZE),

+            hashing_key=bytes(storeroom.KEY_SIZE),

+        )

         with tests.isolated_vault_exporter_config(

             monkeypatch=monkeypatch,

             runner=runner,

@@ -443,7 +443,8 @@ class TestStoreroom:

             match=r'Invalid encrypted master keys payload',

         ):

             storeroom.decrypt_master_keys_data(

-                data, {'encryption_key': key, 'signing_key': key}

+                data,

+                _types.StoreroomKeyPair(encryption_key=key, signing_key=key),

             )

         with pytest.raises(

             ValueError,

@@ -451,11 +452,9 @@ class TestStoreroom:

         ):

             storeroom.decrypt_session_keys(

                 data,

-                {

-                    'hashing_key': key,

-                    'encryption_key': key,

-                    'signing_key': key,

-                },

+                _types.StoreroomMasterKeys(

+                    hashing_key=key, encryption_key=key, signing_key=key

+                ),

             )

     @tests.hypothesis_settings_coverage_compatible

@@ -472,16 +471,15 @@ class TestStoreroom:

         # such random sampling is astronomically tiny.

         with pytest.raises(cryptography.exceptions.InvalidSignature):

             storeroom.decrypt_master_keys_data(

-                data, {'encryption_key': key, 'signing_key': key}

+                data,

+                _types.StoreroomKeyPair(encryption_key=key, signing_key=key),

             )

         with pytest.raises(cryptography.exceptions.InvalidSignature):

             storeroom.decrypt_session_keys(

                 data,

-                {

-                    'hashing_key': key,

-                    'encryption_key': key,

-                    'signing_key': key,

-                },

+                _types.StoreroomMasterKeys(

+                    hashing_key=key, encryption_key=key, signing_key=key

+                ),

             )

@@ -560,9 +560,9 @@ class TestAgentInteraction:

         with monkeypatch.context() as monkeypatch2:

             client = ssh_agent.SSHAgentClient()

             monkeypatch2.setattr(client, 'request', request)

-            KeyCommentPair = _types.KeyCommentPair  # noqa: N806

+            SSHKeyCommentPair = _types.SSHKeyCommentPair  # noqa: N806

             loaded_keys = [

-                KeyCommentPair(v['public_key_data'], b'no comment')

+                SSHKeyCommentPair(v['public_key_data'], b'no comment')

                 for v in tests.SUPPORTED_KEYS.values()

             ]

             monkeypatch2.setattr(client, 'list_keys', lambda: loaded_keys)