250 lines
8.6 KiB
Python
250 lines
8.6 KiB
Python
"""Offline, passphrase-protected recovery bundle codec.
|
|
|
|
The binary format is deliberately small and versioned so validation can reject
|
|
unsupported inputs before attempting expensive password derivation. Every
|
|
failure while parsing or authenticating a bundle is reported as the same error
|
|
so callers cannot distinguish a malformed bundle from a wrong passphrase.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import stat
|
|
import struct
|
|
from collections.abc import Mapping
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from argon2.low_level import Type, hash_secret_raw
|
|
from cryptography.exceptions import InvalidTag
|
|
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
|
|
|
|
|
class RecoveryBundleError(ValueError):
|
|
"""A non-disclosing recovery bundle validation failure."""
|
|
|
|
|
|
class RecoveryBundlePathError(ValueError):
|
|
"""A requested recovery bundle path cannot be used safely."""
|
|
|
|
|
|
_MAGIC = b"BTREC"
|
|
_VERSION = 1
|
|
_KDF_ARGON2ID = 1
|
|
_SALT_BYTES = 16
|
|
_NONCE_BYTES = 12
|
|
_KEY_BYTES = 32
|
|
_TAG_BYTES = 16
|
|
_TIME_COST = 3
|
|
_MEMORY_COST_KIB = 65_536
|
|
_PARALLELISM = 1
|
|
_MAX_PASSPHRASE_BYTES = 4_096
|
|
_MAX_PLAINTEXT_BYTES = 8 * 1024 * 1024
|
|
# magic, version, KDF id, Argon2 time/memory/parallelism, salt/nonce lengths,
|
|
# and the AES-GCM ciphertext (including tag) length.
|
|
_HEADER = struct.Struct(">5sBBIIHBBQ")
|
|
_MAX_BUNDLE_BYTES = _HEADER.size + _SALT_BYTES + _NONCE_BYTES + _MAX_PLAINTEXT_BYTES + _TAG_BYTES
|
|
_ERROR = "recovery bundle is invalid"
|
|
|
|
|
|
def _invalid() -> RecoveryBundleError:
|
|
return RecoveryBundleError(_ERROR)
|
|
|
|
|
|
def _canonical_json(payload: Mapping[str, Any]) -> bytes:
|
|
try:
|
|
encoded = json.dumps(
|
|
payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True
|
|
).encode("utf-8")
|
|
except (TypeError, ValueError) as error:
|
|
raise _invalid() from error
|
|
if not encoded or len(encoded) > _MAX_PLAINTEXT_BYTES:
|
|
raise _invalid()
|
|
return encoded
|
|
|
|
|
|
def _passphrase(value: bytes) -> bytes:
|
|
if not isinstance(value, bytes) or not value or len(value) > _MAX_PASSPHRASE_BYTES:
|
|
raise _invalid()
|
|
return value
|
|
|
|
|
|
def _derive_key(passphrase: bytes, salt: bytes) -> bytes:
|
|
return hash_secret_raw(
|
|
secret=passphrase,
|
|
salt=salt,
|
|
time_cost=_TIME_COST,
|
|
memory_cost=_MEMORY_COST_KIB,
|
|
parallelism=_PARALLELISM,
|
|
hash_len=_KEY_BYTES,
|
|
type=Type.ID,
|
|
)
|
|
|
|
|
|
def encrypt_bundle(payload: Mapping[str, Any], passphrase: bytes) -> bytes:
|
|
"""Serialize and encrypt a canonical recovery payload as a BTREC v1 bundle."""
|
|
plaintext = _canonical_json(payload)
|
|
secret = _passphrase(passphrase)
|
|
salt = os.urandom(_SALT_BYTES)
|
|
nonce = os.urandom(_NONCE_BYTES)
|
|
ciphertext_length = len(plaintext) + _TAG_BYTES
|
|
header = _HEADER.pack(
|
|
_MAGIC,
|
|
_VERSION,
|
|
_KDF_ARGON2ID,
|
|
_TIME_COST,
|
|
_MEMORY_COST_KIB,
|
|
_PARALLELISM,
|
|
_SALT_BYTES,
|
|
_NONCE_BYTES,
|
|
ciphertext_length,
|
|
)
|
|
ciphertext = AESGCM(_derive_key(secret, salt)).encrypt(nonce, plaintext, header)
|
|
return header + salt + nonce + ciphertext
|
|
|
|
|
|
def decrypt_bundle(encoded: bytes, passphrase: bytes) -> dict[str, Any]:
|
|
"""Authenticate and decode a BTREC v1 bundle without disclosing failure cause."""
|
|
try:
|
|
if not isinstance(encoded, bytes) or len(encoded) > _MAX_BUNDLE_BYTES:
|
|
raise _invalid()
|
|
if len(encoded) < _HEADER.size + _SALT_BYTES + _NONCE_BYTES + _TAG_BYTES:
|
|
raise _invalid()
|
|
(
|
|
magic,
|
|
version,
|
|
kdf_id,
|
|
time_cost,
|
|
memory_cost,
|
|
parallelism,
|
|
salt_length,
|
|
nonce_length,
|
|
ciphertext_length,
|
|
) = _HEADER.unpack(encoded[: _HEADER.size])
|
|
if (
|
|
magic != _MAGIC
|
|
or version != _VERSION
|
|
or kdf_id != _KDF_ARGON2ID
|
|
or time_cost != _TIME_COST
|
|
or memory_cost != _MEMORY_COST_KIB
|
|
or parallelism != _PARALLELISM
|
|
or salt_length != _SALT_BYTES
|
|
or nonce_length != _NONCE_BYTES
|
|
or ciphertext_length < _TAG_BYTES
|
|
or ciphertext_length > _MAX_PLAINTEXT_BYTES + _TAG_BYTES
|
|
or len(encoded) != _HEADER.size + salt_length + nonce_length + ciphertext_length
|
|
):
|
|
raise _invalid()
|
|
secret = _passphrase(passphrase)
|
|
salt_start = _HEADER.size
|
|
nonce_start = salt_start + salt_length
|
|
ciphertext_start = nonce_start + nonce_length
|
|
plaintext = AESGCM(_derive_key(secret, encoded[salt_start:nonce_start])).decrypt(
|
|
encoded[nonce_start:ciphertext_start],
|
|
encoded[ciphertext_start:],
|
|
encoded[: _HEADER.size],
|
|
)
|
|
if not plaintext or len(plaintext) > _MAX_PLAINTEXT_BYTES:
|
|
raise _invalid()
|
|
payload = json.loads(plaintext.decode("utf-8"))
|
|
if not isinstance(payload, dict):
|
|
raise _invalid()
|
|
# Reject non-canonical encodings to make catalog serialization deterministic.
|
|
if _canonical_json(payload) != plaintext:
|
|
raise _invalid()
|
|
return payload
|
|
except (
|
|
InvalidTag,
|
|
UnicodeDecodeError,
|
|
json.JSONDecodeError,
|
|
struct.error,
|
|
ValueError,
|
|
) as error:
|
|
if isinstance(error, RecoveryBundleError):
|
|
raise error
|
|
raise _invalid() from error
|
|
|
|
|
|
def _check_path_components(path: Path) -> None:
|
|
if not path.is_absolute() or path.name in {"", ".", ".."}:
|
|
raise RecoveryBundlePathError("recovery bundle path is unsafe")
|
|
current = Path(path.anchor)
|
|
for component in path.parts[1:-1]:
|
|
current /= component
|
|
try:
|
|
info = current.lstat()
|
|
except OSError as error:
|
|
raise RecoveryBundlePathError("recovery bundle path is unsafe") from error
|
|
if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode):
|
|
raise RecoveryBundlePathError("recovery bundle path is unsafe")
|
|
|
|
|
|
def write_bundle_exclusive(path: Path, encoded: bytes) -> None:
|
|
"""Write a bundle once with restrictive permissions and no symlink following."""
|
|
if not isinstance(encoded, bytes) or not encoded or len(encoded) > _MAX_BUNDLE_BYTES:
|
|
raise RecoveryBundlePathError("recovery bundle output is unsafe")
|
|
_check_path_components(path)
|
|
flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
|
|
if hasattr(os, "O_NOFOLLOW"):
|
|
flags |= os.O_NOFOLLOW
|
|
try:
|
|
descriptor = os.open(path, flags, 0o600)
|
|
with os.fdopen(descriptor, "wb") as handle:
|
|
handle.write(encoded)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
except OSError as error:
|
|
raise RecoveryBundlePathError("recovery bundle output is unsafe") from error
|
|
try:
|
|
info = path.lstat()
|
|
if (
|
|
stat.S_ISLNK(info.st_mode)
|
|
or not stat.S_ISREG(info.st_mode)
|
|
or stat.S_IMODE(info.st_mode) != 0o600
|
|
):
|
|
path.unlink(missing_ok=True)
|
|
raise RecoveryBundlePathError("recovery bundle output is unsafe")
|
|
except OSError as error:
|
|
raise RecoveryBundlePathError("recovery bundle output is unsafe") from error
|
|
|
|
|
|
def read_bundle_file(path: Path) -> bytes:
|
|
"""Read a regular, non-symlink bundle with a bounded size."""
|
|
_check_path_components(path)
|
|
flags = os.O_RDONLY
|
|
if hasattr(os, "O_NOFOLLOW"):
|
|
flags |= os.O_NOFOLLOW
|
|
try:
|
|
descriptor = os.open(path, flags)
|
|
with os.fdopen(descriptor, "rb") as handle:
|
|
info = os.fstat(handle.fileno())
|
|
if (
|
|
not stat.S_ISREG(info.st_mode)
|
|
or info.st_size <= 0
|
|
or info.st_size > _MAX_BUNDLE_BYTES
|
|
):
|
|
raise RecoveryBundlePathError("recovery bundle input is unsafe")
|
|
return handle.read()
|
|
except RecoveryBundlePathError:
|
|
raise
|
|
except OSError as error:
|
|
raise RecoveryBundlePathError("recovery bundle input is unsafe") from error
|
|
|
|
|
|
def read_passphrase_fd(fd: int) -> bytes:
|
|
"""Read one newline-terminated passphrase from an inherited file descriptor."""
|
|
if not isinstance(fd, int) or fd < 0:
|
|
raise RecoveryBundleError("recovery passphrase is unavailable")
|
|
try:
|
|
value = os.read(fd, _MAX_PASSPHRASE_BYTES + 2)
|
|
except OSError as error:
|
|
raise RecoveryBundleError("recovery passphrase is unavailable") from error
|
|
if value.endswith(b"\r\n"):
|
|
value = value[:-2]
|
|
elif value.endswith(b"\n"):
|
|
value = value[:-1]
|
|
if not value or len(value) > _MAX_PASSPHRASE_BYTES:
|
|
raise RecoveryBundleError("recovery passphrase is unavailable")
|
|
return value
|