"""Strict, current-format records for password-wrapped profile custody."""
from __future__ import annotations
import json
from typing import Annotated, ClassVar, Final, Literal, cast
from uuid import UUID
from pydantic import BaseModel, Field, ValidationError, field_validator
from .....core.errors.hierarchy import pydantic_validation_boundary
from .....core.external_constants import UTF_8_ENCODING as _UTF_8_ENCODING
from .....core.hashing import (
bounded_canonical_json_bytes,
canonical_json_digest,
reject_duplicate_json_members,
reject_json_constant,
validate_prefixed_digest,
)
from .....core.models import STRICT_FROZEN_CONFIG as _STRICT_FROZEN
from .._kdf_salt import KDF_SALT_BYTES
from ..crypto.aes_gcm import GCM_TAG_SIZE, KEY_SIZE, NONCE_SIZE
from ._kdf_records import (
KdfAlgorithm,
KdfIterations,
KdfMemoryMib,
KdfOutputBytes,
KdfParallelism,
KdfVersion,
canonical_b64,
)
from .digest_model import CustodyDigestModel
from .errors import ProfileCustodyRecordError
PROFILE_CUSTODY_ENVELOPE_SCHEMA_VERSION: Final = 1
#: The password envelope's name inside a capsule's ``custody/`` directory.
#: Named beside its size bound and its sentinel/recovery siblings, which
#: already carry one; it was the last capsule member whose filename was a
#: literal repeated at each publication and read site.
PROFILE_CUSTODY_ENVELOPE_FILENAME: Final = "envelope.v1.json"
#: Largest possible canonical v1 envelope: all fixed-width fields, a present
#: predecessor digest, and the ten-digit maximum password generation.
PROFILE_CUSTODY_ENVELOPE_MAX_BYTES: Final = 704
PROFILE_CUSTODY_PASSWORD_GENERATION_MAX: Final = 2_147_483_647
_DEK_EPOCH_BYTES: Final = 16
_KEY_SCHEDULE: Final = "profile-password-dek-wrap/v1"
def _canonical_json_bytes(value: object) -> bytes:
return bounded_canonical_json_bytes(
value,
maximum_bytes=PROFILE_CUSTODY_ENVELOPE_MAX_BYTES,
subject="profile custody envelope",
)
[docs]
class ProfileCustodyKdfParameters(BaseModel):
"""The finite Argon2id parameter grid accepted by custody v1."""
model_config = _STRICT_FROZEN
algorithm: KdfAlgorithm
version: KdfVersion
memory_mib: KdfMemoryMib
iterations: KdfIterations
parallelism: KdfParallelism
salt_b64: str
output_bytes: KdfOutputBytes
@field_validator("salt_b64")
@classmethod
@pydantic_validation_boundary
def _validate_salt(cls, value: str) -> str:
return canonical_b64(value, field_name="salt_b64", expected_bytes=KDF_SALT_BYTES)
[docs]
class ProfileCustodyWrappedDek(BaseModel):
"""Authenticated AES-GCM payload carrying exactly one 32-byte DEK."""
model_config = _STRICT_FROZEN
nonce_b64: str
ciphertext_b64: str
tag_b64: str
@field_validator("nonce_b64")
@classmethod
@pydantic_validation_boundary
def _validate_nonce(cls, value: str) -> str:
return canonical_b64(value, field_name="nonce_b64", expected_bytes=NONCE_SIZE)
@field_validator("ciphertext_b64")
@classmethod
@pydantic_validation_boundary
def _validate_ciphertext(cls, value: str) -> str:
return canonical_b64(value, field_name="ciphertext_b64", expected_bytes=KEY_SIZE)
@field_validator("tag_b64")
@classmethod
@pydantic_validation_boundary
def _validate_tag(cls, value: str) -> str:
return canonical_b64(value, field_name="tag_b64", expected_bytes=GCM_TAG_SIZE)
PasswordGeneration = Annotated[int, Field(ge=1, le=PROFILE_CUSTODY_PASSWORD_GENERATION_MAX)]
"""A profile password generation counter: at least the first, never past the ceiling."""
PostChangePasswordGeneration = Annotated[int, Field(ge=2, le=PROFILE_CUSTODY_PASSWORD_GENERATION_MAX)]
"""The generation a profile reaches after a passphrase change.
Narrower than :obj:`PasswordGeneration` at the bottom only: a profile starts at
generation one, so the result of CHANGING the passphrase is two or more. The
operator-facing result payload carried that lower bound and no ceiling at all,
which let it report a generation the custody envelope itself would refuse to
store. Both halves are declared here, beside the envelope that owns them.
"""
class _ProfileCustodyEnvelopePayload(BaseModel):
"""Validated v1 envelope fields before self-digest construction."""
model_config = _STRICT_FROZEN
schema_version: Literal[1]
profile_id: UUID
password_generation: PasswordGeneration
dek_epoch: str
password_encoding: Literal["utf-8"]
key_schedule: Literal["profile-password-dek-wrap/v1"]
kdf: ProfileCustodyKdfParameters
wrapped_dek: ProfileCustodyWrappedDek
previous_envelope_digest: str | None
@field_validator("dek_epoch")
@classmethod
@pydantic_validation_boundary
def _validate_dek_epoch(cls, value: str) -> str:
return canonical_b64(value, field_name="dek_epoch", expected_bytes=_DEK_EPOCH_BYTES)
@field_validator("previous_envelope_digest")
@classmethod
@pydantic_validation_boundary
def _validate_previous_envelope_digest(cls, value: str | None) -> str | None:
if value is None:
return None
return validate_prefixed_digest(value, field_name="previous_envelope_digest")
[docs]
class ProfileCustodyEnvelope(_ProfileCustodyEnvelopePayload, CustodyDigestModel):
"""Immutable v1 authority for one profile's normal password unlock."""
_digest_maximum_bytes: ClassVar[int] = PROFILE_CUSTODY_ENVELOPE_MAX_BYTES
_digest_subject: ClassVar[str] = "profile custody envelope"
_digest_mismatch_message: ClassVar[str] = "profile custody self_digest does not match its canonical record"
self_digest: str
@field_validator("self_digest")
@classmethod
@pydantic_validation_boundary
def _validate_self_digest(cls, value: str) -> str:
return validate_prefixed_digest(value, field_name="self_digest")
[docs]
@classmethod
def create(
cls,
*,
profile_id: UUID,
password_generation: int,
dek_epoch: str,
kdf: ProfileCustodyKdfParameters,
wrapped_dek: ProfileCustodyWrappedDek,
previous_envelope_digest: str | None = None,
) -> ProfileCustodyEnvelope:
"""Build a validated envelope with its canonical self-digest."""
try:
payload = _ProfileCustodyEnvelopePayload(
schema_version=PROFILE_CUSTODY_ENVELOPE_SCHEMA_VERSION,
profile_id=profile_id,
password_generation=password_generation,
dek_epoch=dek_epoch,
password_encoding=cast(Literal["utf-8"], _UTF_8_ENCODING),
key_schedule=_KEY_SCHEDULE,
kdf=kdf,
wrapped_dek=wrapped_dek,
previous_envelope_digest=previous_envelope_digest,
).model_dump(mode="json")
payload["self_digest"] = canonical_json_digest(
payload,
maximum_bytes=PROFILE_CUSTODY_ENVELOPE_MAX_BYTES,
subject="profile custody envelope",
)
return cls.model_validate_json(_canonical_json_bytes(payload))
except (ValidationError, ValueError, TypeError) as exc:
raise ProfileCustodyRecordError("cannot construct a valid profile custody envelope") from exc
[docs]
def parse_profile_custody_envelope(value: bytes) -> ProfileCustodyEnvelope:
"""Parse one canonical strict UTF-8 current-format custody envelope."""
if len(value) > PROFILE_CUSTODY_ENVELOPE_MAX_BYTES:
raise ProfileCustodyRecordError(
f"profile custody envelope exceeds {PROFILE_CUSTODY_ENVELOPE_MAX_BYTES}-byte canonical limit",
)
try:
text = value.decode(_UTF_8_ENCODING, errors="strict")
parsed = json.loads(
text,
object_pairs_hook=reject_duplicate_json_members,
parse_constant=reject_json_constant,
)
if not isinstance(parsed, dict):
raise ValueError("profile custody record must be a JSON object")
canonical_input = _canonical_json_bytes(cast(dict[str, object], parsed))
envelope = ProfileCustodyEnvelope.model_validate_json(canonical_input)
if value != envelope.canonical_json_bytes():
raise ValueError("profile custody record is not in canonical JSON form")
return envelope
except (UnicodeDecodeError, ValidationError, ValueError, TypeError) as exc:
raise ProfileCustodyRecordError("profile custody envelope is not a valid current-format record") from exc
__all__ = [
"PROFILE_CUSTODY_ENVELOPE_MAX_BYTES",
"PROFILE_CUSTODY_ENVELOPE_SCHEMA_VERSION",
"PROFILE_CUSTODY_PASSWORD_GENERATION_MAX",
"ProfileCustodyEnvelope",
"ProfileCustodyKdfParameters",
"ProfileCustodyWrappedDek",
"parse_profile_custody_envelope",
]