src/derivepassphrase/exporter/vault_native.py (fe7b634) - derivepassphrase.git

vault_native.py

# SPDX-FileCopyrightText: 2025 Marco Ricci <software@the13thletter.info>
#
# SPDX-License-Identifier: Zlib

"""Exporter for the vault native configuration format (v0.2 or v0.3).

The vault native formats are the configuration formats used by vault
v0.2 and v0.3.  The configuration is stored as a single encrypted file,
which is encrypted and authenticated.  v0.2 and v0.3 differ in some
details concerning key derivation and expected format of internal
structures, so they are *not* compatible.  v0.2 additionally contains
cryptographic weaknesses (API misuse of a key derivation function, and
a low-entropy method of generating initialization vectors for CBC block
encryption mode) and should thus be avoided if possible.

The public interface is the [`export_vault_native_data`][] function.
Multiple *non-public* classes are additionally documented here for
didactical and educational reasons, but they are not part of the module
API, are subject to change without notice (including removal), and
should *not* be used or relied on.

"""

from __future__ import annotations

import abc
import base64
import json
import logging
import warnings
from typing import TYPE_CHECKING

from derivepassphrase import _cli_msg as _msg
from derivepassphrase import exporter, vault

if TYPE_CHECKING:
    from collections.abc import Sequence
    from typing import Any

from typing_extensions import Buffer

if TYPE_CHECKING:
    from cryptography import exceptions as crypt_exceptions
    from cryptography import utils as crypt_utils
    from cryptography.hazmat.primitives import ciphers, hashes, hmac, padding
    from cryptography.hazmat.primitives.ciphers import algorithms, modes
    from cryptography.hazmat.primitives.kdf import pbkdf2
else:
    try:
        from cryptography import exceptions as crypt_exceptions
        from cryptography import utils as crypt_utils
        from cryptography.hazmat.primitives import (
            ciphers,
            hashes,
            hmac,
            padding,
        )
        from cryptography.hazmat.primitives.ciphers import algorithms, modes
        from cryptography.hazmat.primitives.kdf import pbkdf2
    except ModuleNotFoundError as exc:

class _DummyModule:  # pragma: no cover
            def __init__(self, exc: type[Exception]) -> None:
                self.exc = exc

def __getattr__(self, name: str) -> Any:  # noqa: ANN401
                def func(*args: Any, **kwargs: Any) -> Any:  # noqa: ANN401,ARG001
                    raise self.exc

return func

crypt_exceptions = crypt_utils = _DummyModule(exc)
        ciphers = hashes = hmac = padding = _DummyModule(exc)
        algorithms = modes = pbkdf2 = _DummyModule(exc)
        STUBBED = True
    else:
        STUBBED = False

__all__ = ('export_vault_native_data',)

logger = logging.getLogger(__name__)

def _h(bs: Buffer) -> str:
    return '<{}>'.format(memoryview(bs).hex(' '))

class VaultNativeConfigParser(abc.ABC):
    """A base parser for vault's native configuration format.

Certain details are specific to the respective vault versions, and
    are abstracted out.  This class by itself is not instantiable
    because of this.

"""

def __init__(self, contents: Buffer, password: str | Buffer) -> None:
        """Initialize the parser.

Args:
            contents:
                The binary contents of the encrypted configuration file.

Note: On disk, these are usually stored in
                base64-encoded form, not in the "raw" form as needed
                here.

password:
                The vault master key/master passphrase the file is
                encrypted with.  Must be non-empty.  See
                [`exporter.get_vault_key`][] for details.

If this is a text string, then the UTF-8 encoding of the
                string is used as the binary password.

Raises:
            ValueError:
                The password must not be empty.

Warning:
            Non-public class, provided for didactical and educational
            purposes only. Subject to change without notice, including
            removal.

"""
        if not password:
            msg = 'Password must not be empty'
            raise ValueError(msg)
        self._contents = bytes(contents)
        self._iv_size = 0
        self._mac_size = 0
        self._encryption_key = b''
        self._encryption_key_size = 0
        self._signing_key = b''
        self._signing_key_size = 0
        self._message = b''
        self._message_tag = b''
        self._iv = b''
        self._payload = b''
        self._password = password
        self._sentinel: object = object()
        self._data: Any = self._sentinel

def __call__(self) -> Any:  # noqa: ANN401
        """Return the decrypted and parsed vault configuration.

Raises:
            cryptography.exceptions.InvalidSignature:
                The encrypted configuration does not contain a valid
                signature.
            ValueError:
                The format is invalid, in a non-cryptographic way.  (For
                example, it contains an unsupported version marker, or
                unexpected extra contents, or invalid padding.)

"""
        if self._data is self._sentinel:
            self._parse_contents()
            self._derive_keys()
            self._check_signature()
            self._data = self._decrypt_payload()
        return self._data

@staticmethod
    def _pbkdf2(
        password: str | Buffer, key_size: int, iterations: int
    ) -> bytes:
        if isinstance(password, str):
            password = password.encode('utf-8')
        raw_key = pbkdf2.PBKDF2HMAC(
            algorithm=hashes.SHA1(),
            length=key_size // 2,
            salt=vault.Vault._UUID,  # noqa: SLF001
            iterations=iterations,
        ).derive(bytes(password))
        result_key = raw_key.hex().lower().encode('ASCII')
        logger.debug(
            _msg.TranslatedString(
                _msg.DebugMsgTemplate.VAULT_NATIVE_PBKDF2_CALL,
                password=password,
                salt=vault.Vault._UUID,  # noqa: SLF001
                iterations=iterations,
                key_size=key_size // 2,
                algorithm='sha1',
                raw_result=raw_key,
                result_key=result_key.decode('ASCII'),
            ),
        )
        return result_key

def _parse_contents(self) -> None:
        logger.info(
            _msg.TranslatedString(
                _msg.InfoMsgTemplate.VAULT_NATIVE_PARSING_IV_PAYLOAD_MAC,
            ),
        )

if len(self._contents) < self._iv_size + 16 + self._mac_size:
            msg = 'Invalid vault configuration file: file is truncated'
            raise ValueError(msg)

def cut(buffer: bytes, cutpoint: int) -> tuple[bytes, bytes]:
            return buffer[:cutpoint], buffer[cutpoint:]

cutpos1 = len(self._contents) - self._mac_size
        cutpos2 = self._iv_size

self._message, self._message_tag = cut(self._contents, cutpos1)
        self._iv, self._payload = cut(self._message, cutpos2)

logger.debug(
            _msg.TranslatedString(
                _msg.DebugMsgTemplate.VAULT_NATIVE_PARSE_BUFFER,
                contents=_h(self._contents),
                iv=_h(self._iv),
                payload=_h(self._payload),
                mac=_h(self._message_tag),
            ),
        )

def _derive_keys(self) -> None:
        logger.info(
            _msg.TranslatedString(
                _msg.InfoMsgTemplate.VAULT_NATIVE_DERIVING_KEYS,
            ),
        )
        self._generate_keys()
        assert len(self._encryption_key) == self._encryption_key_size, (
            'Derived encryption key is invalid'
        )
        assert len(self._signing_key) == self._signing_key_size, (
            'Derived signing key is invalid'
        )

@abc.abstractmethod
    def _generate_keys(self) -> None:
        raise AssertionError

def _check_signature(self) -> None:
        logger.info(
            _msg.TranslatedString(
                _msg.InfoMsgTemplate.VAULT_NATIVE_CHECKING_MAC,
            ),
        )
        mac = hmac.HMAC(self._signing_key, hashes.SHA256())
        mac_input = self._hmac_input()
        logger.debug(
            _msg.TranslatedString(
                _msg.DebugMsgTemplate.VAULT_NATIVE_CHECKING_MAC_DETAILS,
                mac_input=_h(mac_input),
                mac=_h(self._message_tag),
            ),
        )
        mac.update(mac_input)
        try:
            mac.verify(self._message_tag)
        except crypt_exceptions.InvalidSignature:
            msg = 'File does not contain a valid signature'
            raise ValueError(msg) from None

@abc.abstractmethod
    def _hmac_input(self) -> bytes:
        raise AssertionError

def _decrypt_payload(self) -> Any:  # noqa: ANN401
        logger.info(
            _msg.TranslatedString(
                _msg.InfoMsgTemplate.VAULT_NATIVE_DECRYPTING_CONTENTS,
            ),
        )
        decryptor = self._make_decryptor()
        padded_plaintext = bytearray()
        padded_plaintext.extend(decryptor.update(self._payload))
        padded_plaintext.extend(decryptor.finalize())
        logger.debug(
            _msg.TranslatedString(
                _msg.DebugMsgTemplate.VAULT_NATIVE_PADDED_PLAINTEXT,
                contents=_h(padded_plaintext),
            ),
        )
        unpadder = padding.PKCS7(self._iv_size * 8).unpadder()
        plaintext = bytearray()
        plaintext.extend(unpadder.update(padded_plaintext))
        plaintext.extend(unpadder.finalize())
        logger.debug(
            _msg.TranslatedString(
                _msg.DebugMsgTemplate.VAULT_NATIVE_PLAINTEXT,
                contents=_h(plaintext),
            ),
        )
        return json.loads(plaintext)

@abc.abstractmethod
    def _make_decryptor(self) -> ciphers.CipherContext:
        raise AssertionError

class VaultNativeV03ConfigParser(VaultNativeConfigParser):
    """A parser for vault's native configuration format (v0.3).

This is the modern, pre-storeroom configuration format.

Warning:
        Non-public class, provided for didactical and educational
        purposes only. Subject to change without notice, including
        removal.

"""

KEY_SIZE = 32

def __init__(self, *args: Any, **kwargs: Any) -> None:  # noqa: ANN401
        super().__init__(*args, **kwargs)
        self._iv_size = 16
        self._mac_size = 32

def _generate_keys(self) -> None:
        self._encryption_key = self._pbkdf2(self._password, self.KEY_SIZE, 100)
        self._signing_key = self._pbkdf2(self._password, self.KEY_SIZE, 200)
        self._encryption_key_size = self._signing_key_size = self.KEY_SIZE

def _hmac_input(self) -> bytes:
        return self._message.hex().lower().encode('ASCII')

def _make_decryptor(self) -> ciphers.CipherContext:
        return ciphers.Cipher(
            algorithms.AES256(self._encryption_key), modes.CBC(self._iv)
        ).decryptor()

class VaultNativeV02ConfigParser(VaultNativeConfigParser):
    """A parser for vault's native configuration format (v0.2).

This is the classic configuration format.  Compared to v0.3, it
    contains an (accidental) API misuse for the generation of the master
    keys, a low-entropy method of generating initialization vectors for
    the AES-CBC encryption step, and extra layers of base64 encoding.
    Because of these significantly weakened confidentiality guarantees,
    v0.2 configurations should be upgraded to at least v0.3 as soon as
    possible.

Warning:
        Non-public class, provided for didactical and educational
        purposes only. Subject to change without notice, including
        removal.

"""

def __init__(self, *args: Any, **kwargs: Any) -> None:  # noqa: ANN401
        super().__init__(*args, **kwargs)
        self._iv_size = 16
        self._mac_size = 64

def _parse_contents(self) -> None:
        super()._parse_contents()
        self._payload = base64.standard_b64decode(self._payload)
        self._message_tag = bytes.fromhex(self._message_tag.decode('ASCII'))
        logger.debug(
            _msg.TranslatedString(
                _msg.DebugMsgTemplate.VAULT_NATIVE_V02_PAYLOAD_MAC_POSTPROCESSING,
                payload=_h(self._payload),
                mac=_h(self._message_tag),
            ),
        )

def _generate_keys(self) -> None:
        self._encryption_key = self._pbkdf2(self._password, 8, 16)
        self._signing_key = self._pbkdf2(self._password, 16, 16)
        self._encryption_key_size = 8
        self._signing_key_size = 16

def _hmac_input(self) -> bytes:
        return base64.standard_b64encode(self._message)

def _make_decryptor(self) -> ciphers.CipherContext:
        def evp_bytestokey_md5_one_iteration_no_salt(
            data: bytes, key_size: int, iv_size: int
        ) -> tuple[bytes, bytes]:
            total_size = key_size + iv_size
            buffer = bytearray()
            last_block = b''
            salt = b''
            logger.debug(
                _msg.TranslatedString(
                    _msg.DebugMsgTemplate.VAULT_NATIVE_EVP_BYTESTOKEY_INIT,
                    data=_h(data),
                    salt=_h(salt),
                    key_size=key_size,
                    iv_size=iv_size,
                    buffer_length=len(buffer),
                    buffer=_h(buffer),
                ),
            )
            while len(buffer) < total_size:
                with warnings.catch_warnings():
                    warnings.simplefilter(
                        'ignore', crypt_utils.CryptographyDeprecationWarning
                    )
                    block = hashes.Hash(hashes.MD5())
                block.update(last_block)
                block.update(data)
                block.update(salt)
                last_block = block.finalize()
                buffer.extend(last_block)
                logger.debug(
                    _msg.TranslatedString(
                        _msg.DebugMsgTemplate.VAULT_NATIVE_EVP_BYTESTOKEY_ROUND,
                        buffer_length=len(buffer),
                        buffer=_h(buffer),
                    ),
                )
            logger.debug(
                _msg.TranslatedString(
                    _msg.DebugMsgTemplate.VAULT_NATIVE_EVP_BYTESTOKEY_RESULT,
                    enc_key=_h(buffer[:key_size]),
                    iv=_h(buffer[key_size:total_size]),
                ),
            )
            return bytes(buffer[:key_size]), bytes(buffer[key_size:total_size])

data = base64.standard_b64encode(self._iv + self._encryption_key)
        encryption_key, iv = evp_bytestokey_md5_one_iteration_no_salt(
            data, key_size=32, iv_size=16
        )
        return ciphers.Cipher(
            algorithms.AES256(encryption_key), modes.CBC(iv)
        ).decryptor()

def export_vault_native_data(
    contents: Buffer | None = None,
    key: str | Buffer | None = None,
    *,
    try_formats: Sequence[str] = ('v0.3', 'v0.2'),
) -> Any:  # noqa: ANN401
    """Export the full configuration stored in vault native format.

Args:
        contents:
            The binary encrypted contents of the vault configuration
            file.  If not given, then query
            [`exporter.get_vault_path`][] for the correct filename and
            read the contents from there.

Note: On disk, these are usually stored in base64-encoded
            form, not in the "raw" form as needed here.
        key:
            Encryption key/password for the configuration file, usually
            the username, or passed via the `VAULT_KEY` environment
            variable.  If not given, then query
            [`exporter.get_vault_key`][] for the value.
        try_formats:
            A sequence of formats to try out, in order.  Each key must
            be one of `v0.2` or `v0.3`.

Returns:
        The vault configuration, as recorded in the configuration file.

This may or may not be a valid configuration according to vault
        or derivepassphrase.

Raises:
        RuntimeError:
            Something went wrong during data collection, e.g. we
            encountered unsupported or corrupted data in the storeroom.
        json.JSONDecodeError:
            An internal JSON data structure failed to parse from disk.
            The storeroom is probably corrupted.
        ValueError:
            The requested formats to try out are invalid, or the
            encrypted contents aren't in any of the attempted
            configuration formats.

"""
    if contents is None:
        with open(exporter.get_vault_path(), 'rb') as infile:
            contents = base64.standard_b64decode(infile.read())
    if key is None:
        key = exporter.get_vault_key()
    stored_exception: Exception | None = None
    for config_format in try_formats:
        # Use match/case here once Python 3.9 becomes unsupported.
        if config_format == 'v0.2':
            try:
                return VaultNativeV02ConfigParser(contents, key)()
            except ValueError as exc:
                exc.__context__ = stored_exception
                stored_exception = exc
        elif config_format == 'v0.3':
            try:
                return VaultNativeV03ConfigParser(contents, key)()
            except ValueError as exc:
                exc.__context__ = stored_exception
                stored_exception = exc
        else:  # pragma: no cover
            msg = (
                f'Invalid vault native configuration format: {config_format!r}'
            )
            raise ValueError(msg)
    msg = (
        f'Not a valid vault native configuration. (We tried: {try_formats!r}.)'
    )
    raise stored_exception or ValueError(msg)

if __name__ == '__main__':
    import os

logging.basicConfig(level=('DEBUG' if os.getenv('DEBUG') else 'WARNING'))
    with open(exporter.get_vault_path(), 'rb') as infile:
        contents = base64.standard_b64decode(infile.read())
    password = exporter.get_vault_key()
    try:
        config = VaultNativeV03ConfigParser(contents, password)()
    except ValueError:
        config = VaultNativeV02ConfigParser(contents, password)()
    print(json.dumps(config, indent=2, sort_keys=True))  # noqa: T201

derivepassphrase.git

Make debug and info messages from exporter subcommands translatable