from __future__ import annotations import fcntl import hashlib import json import math import os import stat import threading import time from contextlib import contextmanager from dataclasses import dataclass, field from datetime import datetime, timezone from enum import Enum from pathlib import Path from typing import Any, Iterator, Mapping UNKNOWN_TEXT = "전송 결과를 확인할 수 없습니다. 다시 보내지 마세요. 조정이 필요합니다." AUDIT_PENDING_TEXT = "고객 전송 영수증은 확인됐습니다. 재전송하지 말고 감사 기록을 복구해 주세요." DUPLICATE_TEXT = "이미 처리된 전송입니다." DIAGNOSTIC_JOURNAL_SCHEMA = "diagnostic_authority_journal_v1" DIAGNOSTIC_JOURNAL_FILENAME = "delivery-authority.jsonl" DIAGNOSTIC_LOCK_FILENAME = "delivery-authority.lock" _DIGEST_ZERO = "0" * 64 _DIAGNOSTIC_CANDIDATE_FACTORY_TOKEN = object() _DIAGNOSTIC_HOST_ADMISSION_TOKEN = object() _DIAGNOSTIC_LIVE_LOADER_FACTORY_TOKEN = object() _DIAGNOSTIC_LIVE_SOURCES_CONSTRUCTOR_TOKEN = object() _LIVE_AUTHORITY_SCHEMA = "live_diagnostic_authority_v1" _LIVE_AUTHORITY_STATE = "activated" _LIVE_SOURCE_STATES = frozenset({"active", "enabled", "granted", "approved", "activated"}) _LIVE_SNAPSHOT_FIELDS = frozenset( { "schema_version", "customer_key_digest", "session_id", "session_generation", "owner_digest", "registry_digest", "consent_digest", "activation_receipt_digest", "proposal_digest", "revision", "revision_digest", "rendered_body_digest", "destination_digest", "config_digest", "policy_digest", "catalog_digest", "meal_constraints_digest", "source_digest", "registration_digest", "epoch_digest", "diagnostic_transport_binding_digest", "state", "expires_at_kst", "revoked", } ) _LIVE_PIN_FIELDS = ( "customer_key_digest", "owner_digest", "registry_digest", "consent_digest", "activation_receipt_digest", "proposal_digest", "revision_digest", "rendered_body_digest", "destination_digest", "config_digest", "policy_digest", "catalog_digest", "meal_constraints_digest", "source_digest", "registration_digest", "epoch_digest", "diagnostic_transport_binding_digest", ) class _DiagnosticAdmissionToken: __slots__ = ("authority",) def __init__(self, authority: object) -> None: self.authority = authority _ACTIVATED_DELIVERY_SCHEMA = "diagnostic_activated_delivery_v1" _DIAGNOSTIC_SPEC_SCHEMA = "diagnostic_isolation_spec_v1" _DIAGNOSTIC_TRANSPORT_SCHEMA = "diagnostic_transport_binding_v1" _DIAGNOSTIC_TRANSPORT_ADAPTER = "telegram-test-bot" _DIAGNOSTIC_METHOD_VERSION = "send_diagnostic_customer_v1" _DIAGNOSTIC_ROLE_NAMES = frozenset({"operator", "customer"}) _MAX_VALIDATED_TEXT_LENGTH = 128 _PIN_NAMES = ( "customer_key_digest", "proposal_digest", "revision_digest", "rendered_body_digest", "destination_digest", "registry_digest", "activation_receipt_digest", "config_digest", "policy_digest", "catalog_digest", "meal_constraints_digest", ) _ACTIVATION_DIGEST_NAMES = ( "customer_key_digest", "proposal_digest", "rendered_body_digest", "destination_digest", "diagnostic_transport_binding_digest", "registry_digest", "activation_receipt_digest", "config_digest", "policy_digest", "catalog_digest", "meal_constraints_digest", ) DIAGNOSTIC_ACTIVATION_FILENAME = "activated-delivery.json" _ACTIVATION_RECORD_FIELDS = frozenset( { "schema_version", "customer_key_digest", "proposal_digest", "revision", "revision_digest", "rendered_body", "rendered_body_digest", "destination", "destination_digest", "session_id", "session_generation", "diagnostic_transport_binding_digest", "registry_digest", "activation_receipt_digest", "config_digest", "policy_digest", "catalog_digest", "meal_constraints_digest", "expires_at_kst", } ) def _canonical(value: object) -> bytes: return json.dumps( value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False, ).encode("utf-8") def _digest(value: object) -> str: return hashlib.sha256(_canonical(value)).hexdigest() def _is_digest(value: object) -> bool: return type(value) is str and len(value) == 64 and all( char in "0123456789abcdef" for char in value ) def _require_digest(value: object, label: str, *, allow_empty: bool = False) -> None: if allow_empty and type(value) is str and value == "": return if not _is_digest(value): raise ValueError(f"{label} must be a lowercase SHA256 digest") def _require_bounded_text(value: object, label: str) -> str: if ( type(value) is not str or not value or len(value) > _MAX_VALIDATED_TEXT_LENGTH or not value.strip() ): raise ValueError(f"{label} must be a bounded non-empty string") return value def _parse_timestamp(value: object, label: str) -> datetime: text = _require_bounded_text(value, label) if "T" not in text: raise ValueError(f"{label} must be an ISO-8601 timestamp") try: parsed = datetime.fromisoformat( text[:-1] + "+00:00" if text.endswith("Z") else text ) except (TypeError, ValueError) as exc: raise ValueError(f"{label} is not a valid ISO-8601 timestamp") from exc if parsed.tzinfo is None or parsed.utcoffset() is None: raise ValueError(f"{label} must include a timezone offset") return parsed def _path_digest(path: Path) -> str: return hashlib.sha256(str(path).encode("utf-8")).hexdigest() class DiagnosticIsolationError(RuntimeError): pass class DiagnosticAuthorityError(DiagnosticIsolationError): pass class DiagnosticReservationConflict(DiagnosticIsolationError): pass class DiagnosticSessionState(str, Enum): PREPARED = "prepared" ACTIVATING = "activating" ACTIVE = "active" RESTART_UNVALIDATED = "restart_unvalidated" DETACHING = "detaching" EXPIRED = "expired" REVOKED = "revoked" CLOSED = "closed" RECOVERY_REQUIRED = "recovery_required" @dataclass(frozen=True) class DiagnosticRoleRoute: user_id: int chat_id: int topic_id: int role: str generation: int def __post_init__(self) -> None: if type(self) is not DiagnosticRoleRoute: raise DiagnosticAuthorityError("diagnostic role route class is sealed") for name in ("user_id", "chat_id", "topic_id", "generation"): value = getattr(self, name, None) if type(value) is not int: raise DiagnosticAuthorityError("diagnostic role route identifiers must be integers") if getattr(self, "generation", 0) < 1: raise DiagnosticAuthorityError("diagnostic role route generation is invalid") role = getattr(self, "role", None) if type(role) is not str or role not in _DIAGNOSTIC_ROLE_NAMES: raise DiagnosticAuthorityError("diagnostic role route role is invalid") def matches(self, *, user_id: int, chat_id: int, topic_id: int, generation: int) -> bool: return (self.user_id, self.chat_id, self.topic_id, self.generation) == ( user_id, chat_id, topic_id, generation, ) @dataclass(frozen=True) class DiagnosticIsolationSpecV1: owner_digest: str test_bot_digest: str customer_destination_digest: str operator_destination_digest: str profile_digest: str max_provider_timeout_seconds: int expires_at: str approved_by: str approved_at: str supersedes_digest: str = "" schema_version: str = _DIAGNOSTIC_SPEC_SCHEMA spec_core_digest: str = field(init=False) diagnostic_transport_binding_digest: str = field(init=False) spec_digest: str = field(init=False) authority_digest: str = field(init=False) def __post_init__(self) -> None: try: _validate_spec_fields(self) except (AttributeError, TypeError, ValueError) as exc: raise DiagnosticIsolationError("diagnostic isolation spec is invalid") from exc core = { "schema_version": self.schema_version, "owner_digest": self.owner_digest, "test_bot_digest": self.test_bot_digest, "customer_destination_digest": self.customer_destination_digest, "operator_destination_digest": self.operator_destination_digest, "profile_digest": self.profile_digest, "max_provider_timeout_seconds": self.max_provider_timeout_seconds, "expires_at": self.expires_at, } core_digest = _digest(core) binding = _digest( { "schema_version": _DIAGNOSTIC_TRANSPORT_SCHEMA, "adapter_kind": _DIAGNOSTIC_TRANSPORT_ADAPTER, "test_bot_digest": self.test_bot_digest, "customer_destination_digest": self.customer_destination_digest, "max_provider_timeout_seconds": self.max_provider_timeout_seconds, "method_version": _DIAGNOSTIC_METHOD_VERSION, "spec_core_digest": core_digest, } ) spec_digest = _digest( { "spec_core_digest": core_digest, "diagnostic_transport_binding_digest": binding, } ) authority = _digest( { "spec_digest": spec_digest, "approved_by": self.approved_by, "approved_at": self.approved_at, "supersedes_digest": self.supersedes_digest, } ) object.__setattr__(self, "spec_core_digest", core_digest) object.__setattr__(self, "diagnostic_transport_binding_digest", binding) object.__setattr__(self, "spec_digest", spec_digest) object.__setattr__(self, "authority_digest", authority) def _validate_spec_fields(spec: object) -> None: if type(spec) is not DiagnosticIsolationSpecV1: raise TypeError("diagnostic isolation spec must be exact") if type(spec.schema_version) is not str or spec.schema_version != _DIAGNOSTIC_SPEC_SCHEMA: raise ValueError("diagnostic isolation spec schema is unsupported") for name in ( "owner_digest", "test_bot_digest", "customer_destination_digest", "operator_destination_digest", "profile_digest", ): _require_digest(getattr(spec, name), f"spec {name}") _require_digest(spec.supersedes_digest, "spec supersedes_digest", allow_empty=True) if type(spec.max_provider_timeout_seconds) is not int or not ( 1 <= spec.max_provider_timeout_seconds <= 30 ): raise ValueError("provider timeout must be 1..30") approved_at = _parse_timestamp(spec.approved_at, "spec approved_at") expires_at = _parse_timestamp(spec.expires_at, "spec expires_at") if approved_at >= expires_at: raise ValueError("spec approval must precede expiry") if expires_at <= datetime.now(timezone.utc): raise ValueError("spec expiry must be in the future") _require_bounded_text(spec.approved_by, "spec approved_by") _require_bounded_text(spec.approved_at, "spec approved_at") _require_bounded_text(spec.expires_at, "spec expires_at") destinations = ( spec.customer_destination_digest, spec.operator_destination_digest, ) if len(set(destinations)) != 2: raise ValueError("diagnostic role destinations must be distinct") @dataclass(frozen=True) class DiagnosticSession: session_id: str state: DiagnosticSessionState generation: int boot_epoch: str spec_digest: str authority_digest: str transport_binding_digest: str expires_at: str def __post_init__(self) -> None: try: _validate_session_fields(self) except (AttributeError, TypeError, ValueError) as exc: raise DiagnosticAuthorityError("diagnostic session is invalid") from exc def require_active( self, *, generation: int, boot_epoch: str, now: datetime | None = None, ) -> None: now = now or datetime.now(timezone.utc) if ( type(now) is not datetime or now.tzinfo is None or now.utcoffset() is None ): raise DiagnosticAuthorityError("diagnostic now must be timezone-aware") try: expiry = _parse_timestamp(self.expires_at, "diagnostic session expiry") except (TypeError, ValueError) as exc: raise DiagnosticAuthorityError("invalid diagnostic expiry") from exc if self.state is not DiagnosticSessionState.ACTIVE: raise DiagnosticAuthorityError("diagnostic session is not active") if generation != self.generation or boot_epoch != self.boot_epoch: raise DiagnosticAuthorityError("diagnostic generation is stale") if now >= expiry: raise DiagnosticAuthorityError("diagnostic session expired") def _validate_session_fields(session: object) -> None: if type(session) is not DiagnosticSession: raise TypeError("diagnostic session must be exact") _require_bounded_text(session.session_id, "diagnostic session id") if type(session.state) is not DiagnosticSessionState: raise ValueError("diagnostic session state is invalid") if type(session.generation) is not int or session.generation < 1: raise ValueError("diagnostic session generation is invalid") _require_bounded_text(session.boot_epoch, "diagnostic session boot epoch") for name in ("spec_digest", "authority_digest", "transport_binding_digest"): _require_digest(getattr(session, name), f"diagnostic session {name}") expiry = _parse_timestamp(session.expires_at, "diagnostic session expiry") if expiry <= datetime.now(timezone.utc): raise ValueError("diagnostic session expiry must be in the future") @dataclass(frozen=True, slots=True) class ActivatedDiagnosticDelivery: """One immutable, persisted activation admitted to the diagnostic host.""" schema_version: str customer_key_digest: str proposal_digest: str revision: int rendered_body: bytes rendered_body_digest: str destination: tuple[str, str] destination_digest: str session_id: str session_generation: int diagnostic_transport_binding_digest: str registry_digest: str activation_receipt_digest: str config_digest: str policy_digest: str catalog_digest: str meal_constraints_digest: str expires_at_kst: str def __post_init__(self) -> None: if type(self) is not ActivatedDiagnosticDelivery: raise DiagnosticAuthorityError("diagnostic activation class is sealed") try: if ( type(self.schema_version) is not str or self.schema_version != _ACTIVATED_DELIVERY_SCHEMA ): raise ValueError("diagnostic activation schema is invalid") for name in _ACTIVATION_DIGEST_NAMES: value = getattr(self, name) _require_digest(value, f"diagnostic activation {name}") if value == _DIGEST_ZERO: raise ValueError(f"diagnostic activation pin is invalid: {name}") if type(self.revision) is not int or self.revision < 1: raise ValueError("diagnostic activation revision is invalid") if type(self.rendered_body) is not bytes or not self.rendered_body: raise ValueError("diagnostic rendered body is invalid") if type(self.destination) is not tuple or len(self.destination) != 2: raise ValueError("diagnostic activation destination is invalid") for value in self.destination: _require_bounded_text(value, "diagnostic activation destination") _require_bounded_text(self.session_id, "diagnostic activation session") if type(self.session_generation) is not int or self.session_generation < 1: raise ValueError("diagnostic activation generation is invalid") expiry = _parse_timestamp( self.expires_at_kst, "diagnostic activation expiry" ) if expiry <= datetime.now(timezone.utc): raise ValueError("diagnostic activation is expired") except (AttributeError, TypeError, ValueError) as exc: raise DiagnosticAuthorityError("diagnostic activation is invalid") from exc if hashlib.sha256(self.rendered_body).hexdigest() != self.rendered_body_digest: raise DiagnosticAuthorityError("diagnostic rendered body digest mismatch") if _digest(self.destination) != self.destination_digest: raise DiagnosticAuthorityError("diagnostic destination digest mismatch") @classmethod def from_persisted( cls, *, schema_version: str, customer_key_digest: str, proposal_digest: str, revision: int, rendered_body: bytes, rendered_body_digest: str, destination: tuple[str, str], destination_digest: str, session_id: str, session_generation: int, diagnostic_transport_binding_digest: str, registry_digest: str, activation_receipt_digest: str, config_digest: str, policy_digest: str, catalog_digest: str, meal_constraints_digest: str, expires_at_kst: str, ) -> "ActivatedDiagnosticDelivery": """Build a delivery only from the closed persisted activation shape.""" if cls is not ActivatedDiagnosticDelivery: raise DiagnosticAuthorityError("diagnostic activation class is sealed") return cls( schema_version, customer_key_digest, proposal_digest, revision, rendered_body, rendered_body_digest, destination, destination_digest, session_id, session_generation, diagnostic_transport_binding_digest, registry_digest, activation_receipt_digest, config_digest, policy_digest, catalog_digest, meal_constraints_digest, expires_at_kst, ) @property def revision_digest(self) -> str: return _digest(self.revision) @property def pins(self) -> Mapping[str, str]: return { "customer_key_digest": self.customer_key_digest, "proposal_digest": self.proposal_digest, "revision_digest": _digest(self.revision), "rendered_body_digest": self.rendered_body_digest, "destination_digest": self.destination_digest, "registry_digest": self.registry_digest, "activation_receipt_digest": self.activation_receipt_digest, "config_digest": self.config_digest, "policy_digest": self.policy_digest, "catalog_digest": self.catalog_digest, "meal_constraints_digest": self.meal_constraints_digest, } @property def dedupe_key(self) -> str: return ":".join((self.customer_key_digest, self.proposal_digest, _digest(self.revision))) def to_persisted_record(self) -> dict[str, object]: """Return the closed JSON record shape used by the profile loader.""" try: body = self.rendered_body.decode("utf-8") except UnicodeDecodeError as exc: raise DiagnosticAuthorityError("diagnostic rendered body is not UTF-8") from exc return { "schema_version": self.schema_version, "customer_key_digest": self.customer_key_digest, "proposal_digest": self.proposal_digest, "revision": self.revision, "revision_digest": self.revision_digest, "rendered_body": body, "rendered_body_digest": self.rendered_body_digest, "destination": list(self.destination), "destination_digest": self.destination_digest, "session_id": self.session_id, "session_generation": self.session_generation, "diagnostic_transport_binding_digest": self.diagnostic_transport_binding_digest, "registry_digest": self.registry_digest, "activation_receipt_digest": self.activation_receipt_digest, "config_digest": self.config_digest, "policy_digest": self.policy_digest, "catalog_digest": self.catalog_digest, "meal_constraints_digest": self.meal_constraints_digest, "expires_at_kst": self.expires_at_kst, } @property def record_digest(self) -> str: return _digest(self.to_persisted_record()) class DurableDiagnosticActivationLoader: """Read the one profile-owned activated delivery record under authority lock.""" def __init__(self, authority: "DiagnosticDeliveryAuthority") -> None: if type(authority) is not DiagnosticDeliveryAuthority: raise DiagnosticAuthorityError("diagnostic activation authority is invalid") self._authority = authority self._path = authority.activation_path self._record_identity: tuple[int, int] | None = None @property def path(self) -> Path: return self._path @property def authority(self) -> "DiagnosticDeliveryAuthority": return self._authority def load_activated_delivery( self, session_id: str, *, lock_token: object | None = None, ) -> ActivatedDiagnosticDelivery: if ( type(session_id) is not str or not session_id or len(session_id) > _MAX_VALIDATED_TEXT_LENGTH or not session_id.strip() or session_id != self._authority.session.session_id ): raise DiagnosticAuthorityError("diagnostic activation session is stale") if lock_token is not None: self._authority._require_host_admission(lock_token) with self._authority._authority_lock(lock_token): record, identity = _read_activation_record( self._path, profile_root=self._authority.profile_root, ) if self._record_identity is None: self._record_identity = identity elif self._record_identity != identity: raise DiagnosticAuthorityError("diagnostic activation record was replaced") activated = self._from_record(record) session = self._authority.session if activated.session_id != session.session_id: raise DiagnosticAuthorityError("diagnostic activation session is stale") if activated.session_generation != session.generation: raise DiagnosticAuthorityError("diagnostic activation generation is stale") if activated.diagnostic_transport_binding_digest != session.transport_binding_digest: raise DiagnosticAuthorityError("diagnostic activation binding is stale") return activated def record_digest(self, activated: ActivatedDiagnosticDelivery) -> str: if type(activated) is not ActivatedDiagnosticDelivery: raise DiagnosticAuthorityError("diagnostic activation is invalid") return activated.record_digest @staticmethod def _from_record(record: Mapping[str, object]) -> ActivatedDiagnosticDelivery: if type(record) is not dict: raise DiagnosticAuthorityError("diagnostic activation record is not exact") if set(record) != _ACTIVATION_RECORD_FIELDS: raise DiagnosticAuthorityError("diagnostic activation record schema mismatch") body = record["rendered_body"] destination = record["destination"] if type(body) is not str or not body: raise DiagnosticAuthorityError("diagnostic activation body is invalid") if type(destination) is not list or len(destination) != 2: raise DiagnosticAuthorityError("diagnostic activation destination is invalid") try: for value in destination: _require_bounded_text(value, "diagnostic activation destination") body_bytes = body.encode("utf-8") if type(record["schema_version"]) is not str or ( record["schema_version"] != _ACTIVATED_DELIVERY_SCHEMA ): raise ValueError("diagnostic activation schema is invalid") for name in _ACTIVATION_DIGEST_NAMES + ("revision_digest",): _require_digest(record[name], f"diagnostic activation {name}") revision = record["revision"] if type(revision) is not int or revision < 1: raise ValueError("diagnostic activation revision is invalid") _require_bounded_text(record["session_id"], "diagnostic activation session") if ( type(record["session_generation"]) is not int or record["session_generation"] < 1 ): raise ValueError("diagnostic activation generation is invalid") expiry = _parse_timestamp( record["expires_at_kst"], "diagnostic activation expiry" ) except (KeyError, TypeError, UnicodeError, ValueError) as exc: raise DiagnosticAuthorityError( "diagnostic activation record is invalid" ) from exc if expiry <= datetime.now(timezone.utc): raise DiagnosticAuthorityError("diagnostic activation is expired") if record["revision_digest"] != _digest(revision): raise DiagnosticAuthorityError("diagnostic activation revision pin mismatch") try: return ActivatedDiagnosticDelivery.from_persisted( schema_version=record["schema_version"], customer_key_digest=record["customer_key_digest"], proposal_digest=record["proposal_digest"], revision=revision, rendered_body=body_bytes, rendered_body_digest=record["rendered_body_digest"], destination=(destination[0], destination[1]), destination_digest=record["destination_digest"], session_id=record["session_id"], session_generation=record["session_generation"], diagnostic_transport_binding_digest=record[ "diagnostic_transport_binding_digest" ], registry_digest=record["registry_digest"], activation_receipt_digest=record["activation_receipt_digest"], config_digest=record["config_digest"], policy_digest=record["policy_digest"], catalog_digest=record["catalog_digest"], meal_constraints_digest=record["meal_constraints_digest"], expires_at_kst=record["expires_at_kst"], ) except (KeyError, TypeError, ValueError, DiagnosticAuthorityError) as exc: if isinstance(exc, DiagnosticAuthorityError): raise raise DiagnosticAuthorityError("diagnostic activation record is invalid") from exc def _validate_live_digest(name: str, value: object) -> str: if not _is_digest(value) or value == _DIGEST_ZERO: raise DiagnosticAuthorityError(f"diagnostic live {name} pin is invalid") return value def _validate_live_state(name: str, value: object) -> str: if type(value) is not str or value not in _LIVE_SOURCE_STATES: raise DiagnosticAuthorityError(f"diagnostic live {name} state is invalid") return value def _validate_live_expiry(value: object) -> str: if type(value) is not str: raise DiagnosticAuthorityError("diagnostic live expiry is invalid") try: expiry = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError as exc: raise DiagnosticAuthorityError("diagnostic live expiry is invalid") from exc if expiry.tzinfo is None or expiry <= datetime.now(timezone.utc): raise DiagnosticAuthorityError("diagnostic live expiry is invalid") return value @dataclass(frozen=True, slots=True) class DiagnosticOwnerSnapshot: owner_digest: str state: str def __post_init__(self) -> None: if type(self) is not DiagnosticOwnerSnapshot: raise DiagnosticAuthorityError("diagnostic owner snapshot is sealed") _validate_live_digest("owner", self.owner_digest) _validate_live_state("owner", self.state) @dataclass(frozen=True, slots=True) class DiagnosticRegistrySnapshot: customer_key_digest: str registry_digest: str source_digest: str registration_digest: str state: str def __post_init__(self) -> None: if type(self) is not DiagnosticRegistrySnapshot: raise DiagnosticAuthorityError("diagnostic registry snapshot is sealed") for name in ( "customer_key_digest", "registry_digest", "source_digest", "registration_digest", ): _validate_live_digest(name, getattr(self, name)) _validate_live_state("registry", self.state) @dataclass(frozen=True, slots=True) class DiagnosticConsentActivationSnapshot: consent_digest: str activation_receipt_digest: str session_id: str session_generation: int expires_at_kst: str state: str revoked: bool def __post_init__(self) -> None: if type(self) is not DiagnosticConsentActivationSnapshot: raise DiagnosticAuthorityError("diagnostic consent snapshot is sealed") _validate_live_digest("consent", self.consent_digest) _validate_live_digest("activation receipt", self.activation_receipt_digest) if type(self.session_id) is not str or not self.session_id or len(self.session_id) > 128: raise DiagnosticAuthorityError("diagnostic live session is invalid") if type(self.session_generation) is not int or self.session_generation < 1: raise DiagnosticAuthorityError("diagnostic live generation is invalid") _validate_live_expiry(self.expires_at_kst) _validate_live_state("consent/activation", self.state) if type(self.revoked) is not bool: raise DiagnosticAuthorityError("diagnostic live revocation is invalid") @dataclass(frozen=True, slots=True) class DiagnosticProposalSnapshot: proposal_digest: str revision: int revision_digest: str rendered_body_digest: str destination_digest: str state: str expires_at_kst: str def __post_init__(self) -> None: if type(self) is not DiagnosticProposalSnapshot: raise DiagnosticAuthorityError("diagnostic proposal snapshot is sealed") for name in ( "proposal_digest", "revision_digest", "rendered_body_digest", "destination_digest", ): _validate_live_digest(name, getattr(self, name)) if type(self.revision) is not int or self.revision < 1: raise DiagnosticAuthorityError("diagnostic live revision is invalid") if self.revision_digest != _digest(self.revision): raise DiagnosticAuthorityError("diagnostic live revision digest mismatch") _validate_live_state("proposal", self.state) _validate_live_expiry(self.expires_at_kst) @dataclass(frozen=True, slots=True) class DiagnosticConfigSnapshot: config_digest: str epoch_digest: str diagnostic_transport_binding_digest: str state: str def __post_init__(self) -> None: if type(self) is not DiagnosticConfigSnapshot: raise DiagnosticAuthorityError("diagnostic config snapshot is sealed") _validate_live_digest("config", self.config_digest) _validate_live_digest("epoch", self.epoch_digest) _validate_live_digest("transport binding", self.diagnostic_transport_binding_digest) _validate_live_state("config", self.state) @dataclass(frozen=True, slots=True) class DiagnosticArtifactSnapshot: policy_digest: str catalog_digest: str meal_constraints_digest: str state: str def __post_init__(self) -> None: if type(self) is not DiagnosticArtifactSnapshot: raise DiagnosticAuthorityError("diagnostic artifact snapshot is sealed") _validate_live_digest("policy", self.policy_digest) _validate_live_digest("catalog", self.catalog_digest) _validate_live_digest("meal constraints", self.meal_constraints_digest) _validate_live_state("artifact", self.state) @dataclass(frozen=True, slots=True) class LiveDiagnosticAuthoritySnapshot: schema_version: str customer_key_digest: str session_id: str session_generation: int owner_digest: str registry_digest: str consent_digest: str activation_receipt_digest: str proposal_digest: str revision: int revision_digest: str rendered_body_digest: str destination_digest: str config_digest: str policy_digest: str catalog_digest: str meal_constraints_digest: str source_digest: str registration_digest: str epoch_digest: str diagnostic_transport_binding_digest: str state: str expires_at_kst: str revoked: bool def __post_init__(self) -> None: if type(self) is not LiveDiagnosticAuthoritySnapshot: raise DiagnosticAuthorityError("diagnostic live authority snapshot is sealed") if self.schema_version != _LIVE_AUTHORITY_SCHEMA: raise DiagnosticAuthorityError("diagnostic live authority schema is invalid") for name in _LIVE_PIN_FIELDS: _validate_live_digest(name, getattr(self, name)) if type(self.session_id) is not str or not self.session_id or len(self.session_id) > 128: raise DiagnosticAuthorityError("diagnostic live session is invalid") if type(self.session_generation) is not int or self.session_generation < 1: raise DiagnosticAuthorityError("diagnostic live generation is invalid") if type(self.revision) is not int or self.revision < 1: raise DiagnosticAuthorityError("diagnostic live revision is invalid") if self.revision_digest != _digest(self.revision): raise DiagnosticAuthorityError("diagnostic live revision digest mismatch") if self.state != _LIVE_AUTHORITY_STATE: raise DiagnosticAuthorityError("diagnostic live authority is not activated") if type(self.revoked) is not bool or self.revoked: raise DiagnosticAuthorityError("diagnostic live authority is revoked") _validate_live_expiry(self.expires_at_kst) @property def pin_values(self) -> tuple[str, ...]: return tuple(getattr(self, name) for name in _LIVE_PIN_FIELDS) def to_closed_record(self) -> dict[str, object]: return {name: getattr(self, name) for name in _LIVE_SNAPSHOT_FIELDS} class _DiagnosticLiveSourceLoader: __slots__ = ("_authority", "_spec", "_snapshot", "_snapshot_type") def __setattr__(self, name: str, value: object) -> None: if name in {"_authority", "_spec", "_snapshot", "_snapshot_type"} and hasattr(self, name): raise DiagnosticAuthorityError("diagnostic live loader is sealed") object.__setattr__(self, name, value) def __init__( self, authority: "DiagnosticDeliveryAuthority | None", spec: DiagnosticIsolationSpecV1, snapshot: object, snapshot_type: type[object], *, factory_token: object, ) -> None: if type(self) not in ( DiagnosticOwnerLiveLoader, DiagnosticRegistryLiveLoader, DiagnosticConsentActivationLiveLoader, DiagnosticProposalLiveLoader, DiagnosticConfigLiveLoader, DiagnosticArtifactLiveLoader, ): raise DiagnosticAuthorityError("diagnostic live loader class is sealed") if factory_token is not _DIAGNOSTIC_LIVE_LOADER_FACTORY_TOKEN: raise DiagnosticAuthorityError("diagnostic live loader construction is sealed") if authority is not None and type(authority) is not DiagnosticDeliveryAuthority: raise DiagnosticAuthorityError("diagnostic live loader authority is invalid") if type(spec) is not DiagnosticIsolationSpecV1: raise DiagnosticAuthorityError("diagnostic live loader spec is invalid") if type(snapshot) is not snapshot_type: raise DiagnosticAuthorityError("diagnostic live loader snapshot is invalid") self._authority = authority self._spec = spec self._snapshot = snapshot self._snapshot_type = snapshot_type @property def authority(self) -> "DiagnosticDeliveryAuthority": return self._authority @property def spec(self) -> DiagnosticIsolationSpecV1: return self._spec def _bind_authority( self, authority: "DiagnosticDeliveryAuthority", *, factory_token: object, ) -> None: if factory_token is not _DIAGNOSTIC_LIVE_LOADER_FACTORY_TOKEN: raise DiagnosticAuthorityError("diagnostic live loader binding is sealed") if type(authority) is not DiagnosticDeliveryAuthority: raise DiagnosticAuthorityError("diagnostic live loader authority is invalid") if self._authority is not None and self._authority is not authority: raise DiagnosticAuthorityError("diagnostic live loader authority changed") object.__setattr__(self, "_authority", authority) def load_snapshot( self, activated: ActivatedDiagnosticDelivery, *, lock_token: object, ) -> object: if self._authority is None: raise DiagnosticAuthorityError("diagnostic live loader is unbound") self._authority._require_host_admission(lock_token) if type(activated) is not ActivatedDiagnosticDelivery: raise DiagnosticAuthorityError("diagnostic live activation is invalid") return self._snapshot def replace_snapshot(self, snapshot: object, *, factory_token: object) -> None: if factory_token is not _DIAGNOSTIC_LIVE_LOADER_FACTORY_TOKEN: raise DiagnosticAuthorityError("diagnostic live loader mutation is sealed") if type(snapshot) is not self._snapshot_type: raise DiagnosticAuthorityError("diagnostic live loader snapshot is invalid") object.__setattr__(self, "_snapshot", snapshot) class DiagnosticOwnerLiveLoader(_DiagnosticLiveSourceLoader): __slots__ = () def __init__( self, authority: "DiagnosticDeliveryAuthority", spec: DiagnosticIsolationSpecV1, snapshot: DiagnosticOwnerSnapshot, *, factory_token: object, ) -> None: super().__init__( authority, spec, snapshot, DiagnosticOwnerSnapshot, factory_token=factory_token, ) class DiagnosticRegistryLiveLoader(_DiagnosticLiveSourceLoader): __slots__ = () def __init__( self, authority: "DiagnosticDeliveryAuthority", spec: DiagnosticIsolationSpecV1, snapshot: DiagnosticRegistrySnapshot, *, factory_token: object, ) -> None: super().__init__( authority, spec, snapshot, DiagnosticRegistrySnapshot, factory_token=factory_token, ) class DiagnosticConsentActivationLiveLoader(_DiagnosticLiveSourceLoader): __slots__ = () def __init__( self, authority: "DiagnosticDeliveryAuthority", spec: DiagnosticIsolationSpecV1, snapshot: DiagnosticConsentActivationSnapshot, *, factory_token: object, ) -> None: super().__init__( authority, spec, snapshot, DiagnosticConsentActivationSnapshot, factory_token=factory_token, ) class DiagnosticProposalLiveLoader(_DiagnosticLiveSourceLoader): __slots__ = () def __init__( self, authority: "DiagnosticDeliveryAuthority", spec: DiagnosticIsolationSpecV1, snapshot: DiagnosticProposalSnapshot, *, factory_token: object, ) -> None: super().__init__( authority, spec, snapshot, DiagnosticProposalSnapshot, factory_token=factory_token, ) class DiagnosticConfigLiveLoader(_DiagnosticLiveSourceLoader): __slots__ = () def __init__( self, authority: "DiagnosticDeliveryAuthority", spec: DiagnosticIsolationSpecV1, snapshot: DiagnosticConfigSnapshot, *, factory_token: object, ) -> None: super().__init__( authority, spec, snapshot, DiagnosticConfigSnapshot, factory_token=factory_token, ) class DiagnosticArtifactLiveLoader(_DiagnosticLiveSourceLoader): __slots__ = () def __init__( self, authority: "DiagnosticDeliveryAuthority", spec: DiagnosticIsolationSpecV1, snapshot: DiagnosticArtifactSnapshot, *, factory_token: object, ) -> None: super().__init__( authority, spec, snapshot, DiagnosticArtifactSnapshot, factory_token=factory_token, ) class DiagnosticLiveAuthoritySources: __slots__ = ( "_authority", "_spec", "owner_loader", "registry_loader", "consent_activation_loader", "proposal_loader", "config_loader", "artifact_loader", "_record_path", ) def __setattr__(self, name: str, value: object) -> None: if name in { "_authority", "_spec", "owner_loader", "registry_loader", "consent_activation_loader", "proposal_loader", "config_loader", "artifact_loader", "_record_path", } and hasattr(self, name): raise DiagnosticAuthorityError("diagnostic live source is sealed") object.__setattr__(self, name, value) def __init__( self, authority: "DiagnosticDeliveryAuthority | None", spec: DiagnosticIsolationSpecV1, owner_loader: DiagnosticOwnerLiveLoader, registry_loader: DiagnosticRegistryLiveLoader, consent_activation_loader: DiagnosticConsentActivationLiveLoader, proposal_loader: DiagnosticProposalLiveLoader, config_loader: DiagnosticConfigLiveLoader, artifact_loader: DiagnosticArtifactLiveLoader, record_path: Path | None = None, *, factory_token: object, ) -> None: if type(self) is not DiagnosticLiveAuthoritySources: raise DiagnosticAuthorityError("diagnostic live source class is sealed") if factory_token is not _DIAGNOSTIC_LIVE_SOURCES_CONSTRUCTOR_TOKEN: raise DiagnosticAuthorityError("diagnostic live source construction is sealed") if authority is not None and type(authority) is not DiagnosticDeliveryAuthority: raise DiagnosticAuthorityError("diagnostic live source authority is invalid") if type(spec) is not DiagnosticIsolationSpecV1: raise DiagnosticAuthorityError("diagnostic live source spec is invalid") expected = ( (owner_loader, DiagnosticOwnerLiveLoader), (registry_loader, DiagnosticRegistryLiveLoader), (consent_activation_loader, DiagnosticConsentActivationLiveLoader), (proposal_loader, DiagnosticProposalLiveLoader), (config_loader, DiagnosticConfigLiveLoader), (artifact_loader, DiagnosticArtifactLiveLoader), ) if any(type(loader) is not loader_type for loader, loader_type in expected): raise DiagnosticAuthorityError("diagnostic live source loader is invalid") if any(loader.spec is not spec for loader, _ in expected): raise DiagnosticAuthorityError("diagnostic live source spec changed") if any( loader.authority is not None and loader.authority is not authority for loader, _ in expected ): raise DiagnosticAuthorityError("diagnostic live source binding changed") if record_path is not None: if type(record_path) is not type(Path()): raise DiagnosticAuthorityError("diagnostic live record path is invalid") record_path = record_path.absolute() self._authority = authority self._spec = spec self.owner_loader = owner_loader self.registry_loader = registry_loader self.consent_activation_loader = consent_activation_loader self.proposal_loader = proposal_loader self.config_loader = config_loader self.artifact_loader = artifact_loader self._record_path = record_path @classmethod def from_verified_loaders( cls, *, authority: "DiagnosticDeliveryAuthority | None", spec: DiagnosticIsolationSpecV1, owner_loader: DiagnosticOwnerLiveLoader, registry_loader: DiagnosticRegistryLiveLoader, consent_activation_loader: DiagnosticConsentActivationLiveLoader, proposal_loader: DiagnosticProposalLiveLoader, config_loader: DiagnosticConfigLiveLoader, artifact_loader: DiagnosticArtifactLiveLoader, factory_token: object, ) -> "DiagnosticLiveAuthoritySources": if cls is not DiagnosticLiveAuthoritySources: raise DiagnosticAuthorityError("diagnostic live source factory is sealed") if factory_token is not _DIAGNOSTIC_LIVE_LOADER_FACTORY_TOKEN: raise DiagnosticAuthorityError("diagnostic live source factory is sealed") return cls( authority, spec, owner_loader, registry_loader, consent_activation_loader, proposal_loader, config_loader, artifact_loader, factory_token=_DIAGNOSTIC_LIVE_SOURCES_CONSTRUCTOR_TOKEN, record_path=None, ) @classmethod def from_profile_record( cls, *, profile_root: Path, spec: DiagnosticIsolationSpecV1, factory_token: object, ) -> "DiagnosticLiveAuthoritySources": if cls is not DiagnosticLiveAuthoritySources: raise DiagnosticAuthorityError("diagnostic live source factory is sealed") if factory_token is not _DIAGNOSTIC_LIVE_LOADER_FACTORY_TOKEN: raise DiagnosticAuthorityError("diagnostic live source factory is sealed") if type(profile_root) is not type(Path()): raise DiagnosticAuthorityError("diagnostic live profile root is invalid") record_path = ( profile_root.absolute() / "data" / "diagnostic-isolation" / "live-authority.json" ) record, _ = _read_activation_record( record_path, profile_root=profile_root, ) if set(record) != set(_LIVE_SNAPSHOT_FIELDS): raise DiagnosticAuthorityError("diagnostic live authority record is invalid") snapshot = LiveDiagnosticAuthoritySnapshot(**record) loaders = ( DiagnosticOwnerLiveLoader( None, spec, DiagnosticOwnerSnapshot(snapshot.owner_digest, "active"), factory_token=factory_token, ), DiagnosticRegistryLiveLoader( None, spec, DiagnosticRegistrySnapshot( snapshot.customer_key_digest, snapshot.registry_digest, snapshot.source_digest, snapshot.registration_digest, "enabled", ), factory_token=factory_token, ), DiagnosticConsentActivationLiveLoader( None, spec, DiagnosticConsentActivationSnapshot( snapshot.consent_digest, snapshot.activation_receipt_digest, snapshot.session_id, snapshot.session_generation, snapshot.expires_at_kst, "activated", snapshot.revoked, ), factory_token=factory_token, ), DiagnosticProposalLiveLoader( None, spec, DiagnosticProposalSnapshot( snapshot.proposal_digest, snapshot.revision, snapshot.revision_digest, snapshot.rendered_body_digest, snapshot.destination_digest, "approved", snapshot.expires_at_kst, ), factory_token=factory_token, ), DiagnosticConfigLiveLoader( None, spec, DiagnosticConfigSnapshot( snapshot.config_digest, snapshot.epoch_digest, snapshot.diagnostic_transport_binding_digest, "active", ), factory_token=factory_token, ), DiagnosticArtifactLiveLoader( None, spec, DiagnosticArtifactSnapshot( snapshot.policy_digest, snapshot.catalog_digest, snapshot.meal_constraints_digest, "approved", ), factory_token=factory_token, ), ) return cls( None, spec, *loaders, record_path=record_path, factory_token=_DIAGNOSTIC_LIVE_SOURCES_CONSTRUCTOR_TOKEN, ) @property def record_path(self) -> Path | None: return self._record_path def _bind_authority( self, authority: "DiagnosticDeliveryAuthority", *, factory_token: object, ) -> None: if factory_token is not _DIAGNOSTIC_LIVE_LOADER_FACTORY_TOKEN: raise DiagnosticAuthorityError("diagnostic live source binding is sealed") if type(authority) is not DiagnosticDeliveryAuthority: raise DiagnosticAuthorityError("diagnostic live source authority is invalid") if self._authority is not None and self._authority is not authority: raise DiagnosticAuthorityError("diagnostic live source authority changed") for loader in ( self.owner_loader, self.registry_loader, self.consent_activation_loader, self.proposal_loader, self.config_loader, self.artifact_loader, ): loader._bind_authority(authority, factory_token=factory_token) object.__setattr__(self, "_authority", authority) @property def authority(self) -> "DiagnosticDeliveryAuthority": return self._authority @property def spec(self) -> DiagnosticIsolationSpecV1: return self._spec class DiagnosticLiveAuthorityLoader: __slots__ = ("_profile_root", "_spec", "_authority", "_sources") def __setattr__(self, name: str, value: object) -> None: if name in {"_profile_root", "_spec", "_authority", "_sources"} and hasattr(self, name): raise DiagnosticAuthorityError("diagnostic live authority loader is sealed") object.__setattr__(self, name, value) def __init__( self, profile_root: Path, spec: DiagnosticIsolationSpecV1, authority: "DiagnosticDeliveryAuthority", sources: DiagnosticLiveAuthoritySources, *, factory_token: object, ) -> None: if type(self) is not DiagnosticLiveAuthorityLoader: raise DiagnosticAuthorityError("diagnostic live authority loader class is sealed") if factory_token is not _DIAGNOSTIC_LIVE_LOADER_FACTORY_TOKEN: raise DiagnosticAuthorityError("diagnostic live authority loader construction is sealed") if type(profile_root) is not type(Path()): raise DiagnosticAuthorityError("diagnostic live profile root is invalid") if type(spec) is not DiagnosticIsolationSpecV1: raise DiagnosticAuthorityError("diagnostic live spec is invalid") if type(authority) is not DiagnosticDeliveryAuthority: raise DiagnosticAuthorityError("diagnostic live authority is invalid") if type(sources) is not DiagnosticLiveAuthoritySources: raise DiagnosticAuthorityError("diagnostic live sources are invalid") if sources.authority is not authority or sources.spec is not spec: raise DiagnosticAuthorityError("diagnostic live source binding changed") if authority.profile_root.absolute() != profile_root.absolute(): raise DiagnosticAuthorityError("diagnostic live profile root changed") self._profile_root = profile_root.absolute() self._spec = spec self._authority = authority self._sources = sources @classmethod def for_authority( cls, profile_root: Path, verified_spec: DiagnosticIsolationSpecV1, authority: "DiagnosticDeliveryAuthority", sources: DiagnosticLiveAuthoritySources, *, factory_token: object, ) -> "DiagnosticLiveAuthorityLoader": if cls is not DiagnosticLiveAuthorityLoader: raise DiagnosticAuthorityError("diagnostic live loader factory is sealed") return cls( profile_root, verified_spec, authority, sources, factory_token=factory_token, ) @property def authority(self) -> "DiagnosticDeliveryAuthority": return self._authority @property def sources(self) -> DiagnosticLiveAuthoritySources: return self._sources @property def spec(self) -> DiagnosticIsolationSpecV1: return self._spec @property def profile_root(self) -> Path: return self._profile_root def load_live_diagnostic_authority( self, activated: ActivatedDiagnosticDelivery, *, lock_token: object, ) -> LiveDiagnosticAuthoritySnapshot: self._authority._require_host_admission(lock_token) if type(activated) is not ActivatedDiagnosticDelivery: raise DiagnosticAuthorityError("diagnostic live activation is invalid") record_path = self._sources.record_path if record_path is not None: record, _ = _read_activation_record( record_path, profile_root=self._profile_root, ) if set(record) != set(_LIVE_SNAPSHOT_FIELDS): raise DiagnosticAuthorityError("diagnostic live authority record is invalid") return LiveDiagnosticAuthoritySnapshot(**record) owner = self._sources.owner_loader.load_snapshot(activated, lock_token=lock_token) registry = self._sources.registry_loader.load_snapshot(activated, lock_token=lock_token) consent = self._sources.consent_activation_loader.load_snapshot( activated, lock_token=lock_token, ) proposal = self._sources.proposal_loader.load_snapshot(activated, lock_token=lock_token) config = self._sources.config_loader.load_snapshot(activated, lock_token=lock_token) artifacts = self._sources.artifact_loader.load_snapshot(activated, lock_token=lock_token) if ( type(owner) is not DiagnosticOwnerSnapshot or type(registry) is not DiagnosticRegistrySnapshot or type(consent) is not DiagnosticConsentActivationSnapshot or type(proposal) is not DiagnosticProposalSnapshot or type(config) is not DiagnosticConfigSnapshot or type(artifacts) is not DiagnosticArtifactSnapshot ): raise DiagnosticAuthorityError("diagnostic live source snapshot is invalid") if ( owner.state != "active" or registry.state != "enabled" or consent.state != "activated" or proposal.state != "approved" or config.state != "active" or artifacts.state != "approved" or proposal.expires_at_kst != consent.expires_at_kst ): raise DiagnosticAuthorityError("diagnostic live source state is stale") return LiveDiagnosticAuthoritySnapshot( _LIVE_AUTHORITY_SCHEMA, registry.customer_key_digest, consent.session_id, consent.session_generation, owner.owner_digest, registry.registry_digest, consent.consent_digest, consent.activation_receipt_digest, proposal.proposal_digest, proposal.revision, proposal.revision_digest, proposal.rendered_body_digest, proposal.destination_digest, config.config_digest, artifacts.policy_digest, artifacts.catalog_digest, artifacts.meal_constraints_digest, registry.source_digest, registry.registration_digest, config.epoch_digest, config.diagnostic_transport_binding_digest, _LIVE_AUTHORITY_STATE, consent.expires_at_kst, consent.revoked, ) def load_snapshot( self, activated: ActivatedDiagnosticDelivery, *, lock_token: object, ) -> LiveDiagnosticAuthoritySnapshot: return self.load_live_diagnostic_authority(activated, lock_token=lock_token) @dataclass(frozen=True) class DiagnosticDeliveryCandidate: dedupe_key: str session_digest: str generation: int body_bytes: bytes body_digest: str destination_digest: str transport_binding_digest: str pins: Mapping[str, str] destination: tuple[str, str] = field(default=("", ""), kw_only=True) activated: ActivatedDiagnosticDelivery | None = field( default=None, kw_only=True, repr=False, compare=False ) _factory_token: object | None = field( default=None, kw_only=True, repr=False, compare=False ) def __post_init__(self) -> None: if type(self) is not DiagnosticDeliveryCandidate: raise DiagnosticAuthorityError("diagnostic candidate class is sealed") if self._factory_token is not _DIAGNOSTIC_CANDIDATE_FACTORY_TOKEN: raise DiagnosticAuthorityError("diagnostic candidate factory is sealed") if type(self.activated) is not ActivatedDiagnosticDelivery: raise DiagnosticAuthorityError("diagnostic candidate activation is invalid") if self.destination != self.activated.destination: raise DiagnosticAuthorityError("diagnostic candidate destination is invalid") if self.body_bytes != self.activated.rendered_body: raise DiagnosticAuthorityError("diagnostic candidate body is invalid") if self.body_digest != self.activated.rendered_body_digest: raise DiagnosticAuthorityError("diagnostic candidate body digest is invalid") if self.destination_digest != self.activated.destination_digest: raise DiagnosticAuthorityError("diagnostic candidate destination digest is invalid") if self.transport_binding_digest != self.activated.diagnostic_transport_binding_digest: raise DiagnosticAuthorityError("diagnostic candidate binding is invalid") if dict(self.pins) != dict(self.activated.pins): raise DiagnosticAuthorityError("diagnostic candidate pins are invalid") @classmethod def from_activated( cls, activated: ActivatedDiagnosticDelivery, *, factory_token: object, session_digest: str | None = None, ) -> "DiagnosticDeliveryCandidate": if cls is not DiagnosticDeliveryCandidate: raise DiagnosticAuthorityError("diagnostic candidate class is sealed") if factory_token is not _DIAGNOSTIC_CANDIDATE_FACTORY_TOKEN: raise DiagnosticAuthorityError("diagnostic candidate factory is sealed") if type(activated) is not ActivatedDiagnosticDelivery: raise DiagnosticAuthorityError("diagnostic activation is invalid") return cls( activated.dedupe_key, session_digest if session_digest is not None else _digest({"session_id": activated.session_id}), activated.session_generation, activated.rendered_body, activated.rendered_body_digest, activated.destination_digest, activated.diagnostic_transport_binding_digest, activated.pins, destination=activated.destination, activated=activated, _factory_token=factory_token, ) @dataclass(frozen=True) class VerifiedDiagnosticReservation: reservation_id: str candidate: DiagnosticDeliveryCandidate reservation_digest: str provider_authority: bool = True _authority_token: object | None = field( default=None, kw_only=True, repr=False, compare=False ) @dataclass(frozen=True) class DiagnosticUnknownNoSend: status: str = "delivery_unknown" text: str = UNKNOWN_TEXT reconciliation_available: bool = False provider_authority: bool = False @dataclass(frozen=True) class DiagnosticAuditPendingNoSend: status: str = "audit_pending" text: str = AUDIT_PENDING_TEXT reconciliation_available: bool = True provider_authority: bool = False @dataclass(frozen=True) class DiagnosticDuplicateNoSend: status: str = "duplicate" text: str = DUPLICATE_TEXT reconciliation_available: bool = False provider_authority: bool = False DiagnosticReservationDecision = ( VerifiedDiagnosticReservation | DiagnosticUnknownNoSend | DiagnosticAuditPendingNoSend | DiagnosticDuplicateNoSend ) _JOURNAL_FIELDS = frozenset( { "schema_version", "sequence", "status", "prev_digest", "row_digest", "profile_root_digest", "session_id", "generation", "boot_epoch", "session_digest", "authority_digest", "transport_binding_digest", "dedupe_key", "reservation_id", "reservation_digest", "body_digest", "destination_digest", "receipt_digest", "pins", "reason", } ) _SESSION_STATUS = "session_bound" _DELIVERY_STATUSES = frozenset( { "delivery_attempt_started", "delivery_unknown", "provider_receipt", "audit_pending", "sent_audited", } ) _TERMINAL_STATUSES = frozenset({"delivery_unknown", "audit_pending", "sent_audited"}) def _row_without_digest(row: Mapping[str, object]) -> dict[str, object]: return {key: value for key, value in row.items() if key != "row_digest"} def _reservation_digest(candidate: DiagnosticDeliveryCandidate, reservation_id: str) -> str: return _digest( { "reservation_id": reservation_id, "dedupe_key": candidate.dedupe_key, "body_digest": candidate.body_digest, "destination_digest": candidate.destination_digest, "session_digest": candidate.session_digest, "generation": candidate.generation, "transport_binding_digest": candidate.transport_binding_digest, "pins": dict(candidate.pins), } ) def _ensure_private_directory(path: Path) -> None: try: if path.exists(): if path.is_symlink() or not path.is_dir(): raise DiagnosticAuthorityError("diagnostic profile root is not a private directory") if path.stat().st_uid != os.geteuid() or (path.stat().st_mode & 0o777) != 0o700: raise DiagnosticAuthorityError("diagnostic profile root is not private") return path.mkdir(mode=0o700, parents=True, exist_ok=False) except FileExistsError: _ensure_private_directory(path) except OSError as exc: raise DiagnosticAuthorityError("diagnostic profile root is unavailable") from exc def _verify_private_fd(fd: int, path: Path, *, mode: int) -> tuple[int, int]: try: opened = os.fstat(fd) current = path.lstat() except OSError as exc: raise DiagnosticAuthorityError("diagnostic journal inode is unavailable") from exc if ( not os.path.isfile(path) or opened.st_dev != current.st_dev or opened.st_ino != current.st_ino or opened.st_uid != os.geteuid() or opened.st_nlink != 1 or (opened.st_mode & 0o777) != mode ): raise DiagnosticAuthorityError("diagnostic journal inode is not private") return opened.st_dev, opened.st_ino def _fsync_directory(path: Path) -> None: flags = os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_CLOEXEC", 0) try: fd = os.open(path, flags) except OSError as exc: raise DiagnosticAuthorityError("diagnostic journal directory is unavailable") from exc try: os.fsync(fd) except OSError as exc: raise DiagnosticAuthorityError("diagnostic journal directory is not durable") from exc finally: os.close(fd) @contextmanager def _exclusive_lock(path: Path) -> Iterator[None]: flags = os.O_RDWR | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0) try: fd = os.open(path, flags, 0o600) except OSError as exc: raise DiagnosticAuthorityError("diagnostic journal lock is unavailable") from exc try: identity = _verify_private_fd(fd, path, mode=0o600) fcntl.flock(fd, fcntl.LOCK_EX) if identity != _verify_private_fd(fd, path, mode=0o600): raise DiagnosticAuthorityError("diagnostic journal lock was replaced") yield finally: try: fcntl.flock(fd, fcntl.LOCK_UN) finally: os.close(fd) def _read_journal(path: Path) -> list[dict[str, object]]: flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0) try: fd = os.open(path, flags) except FileNotFoundError: return [] except OSError as exc: raise DiagnosticAuthorityError("diagnostic journal cannot be opened") from exc try: identity = _verify_private_fd(fd, path, mode=0o600) chunks: list[bytes] = [] while True: chunk = os.read(fd, 1024 * 1024) if not chunk: break chunks.append(chunk) if identity != _verify_private_fd(fd, path, mode=0o600): raise DiagnosticAuthorityError("diagnostic journal was replaced") except OSError as exc: raise DiagnosticAuthorityError("diagnostic journal cannot be read") from exc finally: os.close(fd) raw = b"".join(chunks) if not raw: return [] if not raw.endswith(b"\n"): raise DiagnosticAuthorityError("diagnostic journal has a torn final row") rows: list[dict[str, object]] = [] for line in raw.splitlines(): if not line: raise DiagnosticAuthorityError("diagnostic journal contains a blank row") try: value = json.loads(line.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise DiagnosticAuthorityError("diagnostic journal contains invalid JSON") from exc if not isinstance(value, dict): raise DiagnosticAuthorityError("diagnostic journal row is not an object") rows.append(value) return rows def _open_absolute_directory_chain(path: Path) -> list[int]: absolute = path.absolute() if not absolute.is_absolute(): raise DiagnosticAuthorityError("diagnostic root path is not absolute") flags = ( os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0) ) descriptors: list[int] = [] try: current = os.open("/", flags) descriptors.append(current) for component in absolute.parts[1:]: current = os.open(component, flags, dir_fd=current) descriptors.append(current) return descriptors except OSError as exc: for descriptor in reversed(descriptors): os.close(descriptor) raise DiagnosticAuthorityError( "diagnostic root ancestor cannot be opened" ) from exc @contextmanager def _open_profile_directory( profile_root: Path, directory: Path, ) -> Iterator[int]: root = profile_root.absolute() target = directory.absolute() try: relative = target.relative_to(root) except ValueError as exc: raise DiagnosticAuthorityError( "diagnostic directory escapes the profile root" ) from exc if any(part in {"", ".", ".."} for part in relative.parts): raise DiagnosticAuthorityError("diagnostic directory path is invalid") flags = ( os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0) ) descriptors: list[int] = [] try: descriptors.extend(_open_absolute_directory_chain(root)) current = descriptors[-1] for component in relative.parts: info = os.fstat(current) if ( not stat.S_ISDIR(info.st_mode) or info.st_uid != os.geteuid() or info.st_mode & 0o022 ): raise DiagnosticAuthorityError( "diagnostic directory ancestor is unsafe" ) current = os.open(component, flags, dir_fd=current) descriptors.append(current) info = os.fstat(current) if ( not stat.S_ISDIR(info.st_mode) or info.st_uid != os.geteuid() or info.st_mode & 0o022 ): raise DiagnosticAuthorityError("diagnostic directory ancestor is unsafe") yield current except DiagnosticAuthorityError: raise except OSError as exc: raise DiagnosticAuthorityError( "diagnostic directory cannot be opened" ) from exc finally: for descriptor in reversed(descriptors): os.close(descriptor) @contextmanager def _open_profile_private_file( profile_root: Path, path: Path, ) -> Iterator[tuple[int, tuple[int, int]]]: root = profile_root.absolute() target = path.absolute() try: relative = target.relative_to(root) except ValueError as exc: raise DiagnosticAuthorityError( "diagnostic record escapes the profile root" ) from exc if not relative.parts or any(part in {"", ".", ".."} for part in relative.parts): raise DiagnosticAuthorityError("diagnostic record path is invalid") directory_flags = ( os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0) ) descriptors: list[int] = [] try: descriptors.extend(_open_absolute_directory_chain(root)) current = descriptors[-1] for component in relative.parts[:-1]: info = os.fstat(current) if ( not stat.S_ISDIR(info.st_mode) or info.st_uid != os.geteuid() or info.st_mode & 0o022 ): raise DiagnosticAuthorityError( "diagnostic record ancestor is unsafe" ) current = os.open(component, directory_flags, dir_fd=current) descriptors.append(current) parent_info = os.fstat(current) if ( not stat.S_ISDIR(parent_info.st_mode) or parent_info.st_uid != os.geteuid() or parent_info.st_mode & 0o022 ): raise DiagnosticAuthorityError("diagnostic record ancestor is unsafe") file_flags = ( os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0) ) descriptor = os.open(relative.parts[-1], file_flags, dir_fd=current) descriptors.append(descriptor) opened = os.fstat(descriptor) current_entry = os.stat( relative.parts[-1], dir_fd=current, follow_symlinks=False, ) if ( not stat.S_ISREG(opened.st_mode) or opened.st_dev != current_entry.st_dev or opened.st_ino != current_entry.st_ino or opened.st_uid != os.geteuid() or opened.st_nlink != 1 or opened.st_mode & 0o777 != 0o600 ): raise DiagnosticAuthorityError("diagnostic record inode is not private") yield descriptor, (opened.st_dev, opened.st_ino) final = os.fstat(descriptor) current_entry = os.stat( relative.parts[-1], dir_fd=current, follow_symlinks=False, ) if ( (final.st_dev, final.st_ino) != (opened.st_dev, opened.st_ino) or (current_entry.st_dev, current_entry.st_ino) != (opened.st_dev, opened.st_ino) ): raise DiagnosticAuthorityError("diagnostic record was replaced") except DiagnosticAuthorityError: raise except OSError as exc: raise DiagnosticAuthorityError("diagnostic record cannot be opened") from exc finally: for descriptor in reversed(descriptors): os.close(descriptor) def _read_activation_record( path: Path, *, profile_root: Path | None = None, ) -> tuple[dict[str, object], tuple[int, int]]: chunks: list[bytes] = [] if profile_root is not None: with _open_profile_private_file(profile_root, path) as (fd, identity): while True: chunk = os.read(fd, 1024 * 1024) if not chunk: break chunks.append(chunk) else: flags = ( os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0) ) try: fd = os.open(path, flags) except FileNotFoundError as exc: raise DiagnosticAuthorityError( "diagnostic activation record is unavailable" ) from exc except OSError as exc: raise DiagnosticAuthorityError( "diagnostic activation record cannot be opened" ) from exc try: identity = _verify_private_fd(fd, path, mode=0o600) while True: chunk = os.read(fd, 1024 * 1024) if not chunk: break chunks.append(chunk) if identity != _verify_private_fd(fd, path, mode=0o600): raise DiagnosticAuthorityError( "diagnostic activation record was replaced" ) except OSError as exc: raise DiagnosticAuthorityError( "diagnostic activation record cannot be read" ) from exc finally: os.close(fd) try: value = json.loads(b"".join(chunks).decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise DiagnosticAuthorityError("diagnostic activation record is invalid JSON") from exc if not isinstance(value, dict): raise DiagnosticAuthorityError("diagnostic activation record is not an object") return value, identity def _append_journal_row(path: Path, row: Mapping[str, object]) -> None: encoded = _canonical(row) + b"\n" flags = os.O_WRONLY | os.O_APPEND | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0) try: fd = os.open(path, flags, 0o600) except OSError as exc: raise DiagnosticAuthorityError("diagnostic journal cannot be opened for append") from exc try: identity = _verify_private_fd(fd, path, mode=0o600) offset = 0 while offset < len(encoded): offset += os.write(fd, encoded[offset:]) os.fsync(fd) if identity != _verify_private_fd(fd, path, mode=0o600): raise DiagnosticAuthorityError("diagnostic journal was replaced") _fsync_directory(path.parent) except OSError as exc: raise DiagnosticAuthorityError("diagnostic journal append is not durable") from exc finally: os.close(fd) def _validate_journal(rows: list[dict[str, object]], *, profile_root_digest: str) -> dict[str, list[dict[str, object]]]: previous = _DIGEST_ZERO session_ids: set[str] = set() states: dict[str, list[dict[str, object]]] = {} for expected_sequence, row in enumerate(rows, start=1): if set(row) != _JOURNAL_FIELDS: raise DiagnosticAuthorityError("diagnostic journal schema mismatch") if row.get("schema_version") != DIAGNOSTIC_JOURNAL_SCHEMA: raise DiagnosticAuthorityError("diagnostic journal schema version mismatch") if type(row.get("sequence")) is not int or row["sequence"] != expected_sequence: raise DiagnosticAuthorityError("diagnostic journal sequence is not contiguous") if row.get("prev_digest") != previous or not _is_digest(row.get("prev_digest")): raise DiagnosticAuthorityError("diagnostic journal hash chain predecessor mismatch") if not _is_digest(row.get("row_digest")) or _digest(_row_without_digest(row)) != row["row_digest"]: raise DiagnosticAuthorityError("diagnostic journal row digest mismatch") if row.get("profile_root_digest") != profile_root_digest: raise DiagnosticAuthorityError("diagnostic journal profile binding mismatch") if not isinstance(row.get("session_id"), str) or not row["session_id"] or len(row["session_id"] ) > 128: raise DiagnosticAuthorityError("diagnostic journal session binding is invalid") if not isinstance(row.get("boot_epoch"), str) or not row["boot_epoch"] or len(row["boot_epoch"]) > 128: raise DiagnosticAuthorityError("diagnostic journal boot binding is invalid") if type(row.get("generation")) is not int or row["generation"] < 1: raise DiagnosticAuthorityError("diagnostic journal generation is invalid") for key in ("session_digest", "authority_digest", "transport_binding_digest"): if not _is_digest(row.get(key)): raise DiagnosticAuthorityError("diagnostic journal authority digest is invalid") status = row.get("status") if status == _SESSION_STATUS: _validate_session_row(row) if row["session_id"] in session_ids: raise DiagnosticAuthorityError("diagnostic journal repeats a session binding") session_ids.add(row["session_id"]) elif status in _DELIVERY_STATUSES: _validate_delivery_row(row) if row["session_id"] not in session_ids: raise DiagnosticAuthorityError("diagnostic delivery precedes its session binding") dedupe_key = row["dedupe_key"] assert isinstance(dedupe_key, str) chain = states.setdefault(dedupe_key, []) if not chain: if status != "delivery_attempt_started": raise DiagnosticAuthorityError("diagnostic delivery chain has no reservation") chain.append(row) else: _validate_transition(chain, row) chain.append(row) else: raise DiagnosticAuthorityError("diagnostic journal state is invalid") previous = row["row_digest"] return states def _validate_session_row(row: Mapping[str, object]) -> None: delivery_fields = ( "dedupe_key", "reservation_id", "reservation_digest", "body_digest", "destination_digest", "receipt_digest", "reason", ) if any(row.get(key) is not None for key in delivery_fields) or row.get("pins") != {}: raise DiagnosticAuthorityError("diagnostic session row contains delivery state") def _validate_delivery_row(row: Mapping[str, object]) -> None: required = ( "dedupe_key", "reservation_id", "reservation_digest", "body_digest", "destination_digest", ) if any(not isinstance(row.get(key), str) or not row[key] for key in required): raise DiagnosticAuthorityError("diagnostic delivery identity is invalid") for key in ("reservation_digest", "body_digest", "destination_digest"): if not _is_digest(row.get(key)): raise DiagnosticAuthorityError("diagnostic delivery digest is invalid") if not isinstance(row.get("pins"), dict): raise DiagnosticAuthorityError("diagnostic delivery pins are invalid") for key, value in row["pins"].items(): if not isinstance(key, str) or not key or not _is_digest(value): raise DiagnosticAuthorityError("diagnostic delivery pins are invalid") status = row["status"] receipt_digest = row.get("receipt_digest") if status == "provider_receipt": if not _is_digest(receipt_digest) or row.get("reason") is not None: raise DiagnosticAuthorityError("diagnostic provider receipt row is invalid") elif status == "delivery_unknown": if receipt_digest is not None or not isinstance(row.get("reason"), str) or not row["reason"]: raise DiagnosticAuthorityError("diagnostic unknown row is invalid") elif status in {"delivery_attempt_started", "audit_pending", "sent_audited"}: if receipt_digest is not None or row.get("reason") is not None: raise DiagnosticAuthorityError("diagnostic audit row is invalid") def _validate_transition(chain: list[dict[str, object]], row: Mapping[str, object]) -> None: previous = chain[-1] if row["status"] == "delivery_unknown": if previous["status"] != "delivery_attempt_started": raise DiagnosticAuthorityError("diagnostic unknown transition is invalid") elif row["status"] == "provider_receipt": if previous["status"] != "delivery_attempt_started": raise DiagnosticAuthorityError("diagnostic receipt transition is invalid") elif row["status"] == "audit_pending": if previous["status"] != "provider_receipt": raise DiagnosticAuthorityError("diagnostic audit transition is invalid") elif row["status"] == "sent_audited": if previous["status"] not in {"provider_receipt", "audit_pending"}: raise DiagnosticAuthorityError("diagnostic audited transition is invalid") else: raise DiagnosticAuthorityError("diagnostic reservation transition is invalid") for key in ( "session_id", "generation", "boot_epoch", "session_digest", "authority_digest", "transport_binding_digest", "dedupe_key", "reservation_id", "reservation_digest", "body_digest", "destination_digest", "pins", ): if row.get(key) != previous.get(key): raise DiagnosticAuthorityError("diagnostic delivery identity changed in place") class DiagnosticDeliveryAuthority: """Host-owned, fail-closed reservation authority backed by a private JSONL journal.""" def __init__( self, session: DiagnosticSession, *, boot_epoch: str, profile_root: Path, spec: DiagnosticIsolationSpecV1, live_sources: DiagnosticLiveAuthoritySources, ) -> None: if type(session) is not DiagnosticSession: raise DiagnosticAuthorityError("diagnostic session binding is invalid") if type(spec) is not DiagnosticIsolationSpecV1: raise DiagnosticAuthorityError("diagnostic isolation spec is required") if type(live_sources) is not DiagnosticLiveAuthoritySources: raise DiagnosticAuthorityError("diagnostic live sources are required") if live_sources.spec is not spec: raise DiagnosticAuthorityError("diagnostic live source spec binding mismatch") if live_sources.authority is not None: raise DiagnosticAuthorityError("diagnostic live sources are already bound") try: _validate_session_fields(session) _validate_spec_fields(spec) for name in ( "spec_core_digest", "diagnostic_transport_binding_digest", "spec_digest", "authority_digest", ): _require_digest(getattr(spec, name), f"diagnostic spec {name}") except (AttributeError, TypeError, ValueError) as exc: raise DiagnosticAuthorityError("diagnostic authority inputs are invalid") from exc if session.expires_at != spec.expires_at: raise DiagnosticAuthorityError("diagnostic session expiry binding mismatch") if ( spec.spec_digest != session.spec_digest or spec.authority_digest != session.authority_digest or spec.diagnostic_transport_binding_digest != session.transport_binding_digest ): raise DiagnosticAuthorityError("diagnostic authority spec binding mismatch") try: _require_bounded_text(boot_epoch, "diagnostic boot epoch") except (TypeError, ValueError) as exc: raise DiagnosticAuthorityError("diagnostic boot epoch is required") from exc if profile_root is None: raise DiagnosticAuthorityError("isolated diagnostic profile root is required") if type(profile_root) is not type(Path()): raise DiagnosticAuthorityError("isolated diagnostic profile root must be a path") if profile_root.is_symlink(): raise DiagnosticAuthorityError("diagnostic profile root symlinks are not allowed") live_sources._bind_authority( self, factory_token=_DIAGNOSTIC_LIVE_LOADER_FACTORY_TOKEN, ) _ensure_private_directory(profile_root) self._profile_root = profile_root.absolute() self._live_sources = live_sources self._live_loader = DiagnosticLiveAuthorityLoader.for_authority( self._profile_root, spec, self, live_sources, factory_token=_DIAGNOSTIC_LIVE_LOADER_FACTORY_TOKEN, ) self._journal_dir = self._profile_root / "diagnostic" _ensure_private_directory(self._journal_dir) self._journal_path = self._journal_dir / DIAGNOSTIC_JOURNAL_FILENAME self._lock_path = self._journal_dir / DIAGNOSTIC_LOCK_FILENAME self._terminal_path = self._journal_dir / "diagnostic-session-terminal.json" self._session = session self._spec = spec self._boot_epoch = boot_epoch self._profile_root_digest = _path_digest(self._profile_root) self._boot_epoch_revalidated = False self._authority_lock_owner: int | None = None self._authority_lock_depth = 0 self._admission_token: _DiagnosticAdmissionToken | None = None self._restart_unvalidated = self._initialize_durable_session() @classmethod def from_profile_record( cls, session: DiagnosticSession, *, boot_epoch: str, profile_root: Path, spec: DiagnosticIsolationSpecV1, ) -> "DiagnosticDeliveryAuthority": """Create a record-backed authority for dormant start or restart recovery.""" if cls is not DiagnosticDeliveryAuthority: raise DiagnosticAuthorityError("diagnostic authority factory is sealed") if type(session) is not DiagnosticSession: raise DiagnosticAuthorityError("diagnostic session binding is invalid") if type(profile_root) is not type(Path()): raise DiagnosticAuthorityError("isolated diagnostic profile root must be a path") if profile_root.is_symlink(): raise DiagnosticAuthorityError("diagnostic profile root symlinks are not allowed") if type(spec) is not DiagnosticIsolationSpecV1: raise DiagnosticAuthorityError("diagnostic isolation spec is required") if session.state not in { DiagnosticSessionState.PREPARED, DiagnosticSessionState.ACTIVE, }: raise DiagnosticAuthorityError( "diagnostic factory requires a dormant or active session" ) live_sources = DiagnosticLiveAuthoritySources.from_profile_record( profile_root=profile_root, spec=spec, factory_token=_DIAGNOSTIC_LIVE_LOADER_FACTORY_TOKEN, ) return cls( session, boot_epoch=boot_epoch, profile_root=profile_root, spec=spec, live_sources=live_sources, ) @property def journal_path(self) -> Path: return self._journal_path @property def lock_path(self) -> Path: return self._lock_path @property def session(self) -> DiagnosticSession: return self._session @property def profile_root(self) -> Path: return self._profile_root @property def activation_path(self) -> Path: return self._journal_dir / DIAGNOSTIC_ACTIVATION_FILENAME @property def live_sources(self) -> DiagnosticLiveAuthoritySources: return self._live_sources @property def live_loader(self) -> DiagnosticLiveAuthorityLoader: return self._live_loader @property def spec(self) -> DiagnosticIsolationSpecV1 | None: return self._spec @property def restart_unvalidated(self) -> bool: return self._restart_unvalidated @contextmanager def delivery_admission(self, host_token: object) -> Iterator[object]: """Hold the shared authority lock through provider admission and invocation.""" if host_token is not _DIAGNOSTIC_HOST_ADMISSION_TOKEN: raise DiagnosticAuthorityError("diagnostic host admission token is required") with self._authority_lock(): if self._admission_token is not None: raise DiagnosticAuthorityError("diagnostic delivery admission is already active") token = _DiagnosticAdmissionToken(self) self._admission_token = token try: self._require_not_detached() yield token finally: self._admission_token = None @contextmanager def _authority_lock(self, lock_token: object | None = None) -> Iterator[None]: owner = threading.get_ident() if self._authority_lock_owner == owner: self._authority_lock_depth += 1 try: yield finally: self._authority_lock_depth -= 1 if self._authority_lock_depth == 0: self._authority_lock_owner = None return with _exclusive_lock(self._lock_path): self._authority_lock_owner = owner self._authority_lock_depth = 1 try: yield finally: self._authority_lock_depth = 0 self._authority_lock_owner = None def _admission_is_active(self, lock_token: object | None) -> bool: return ( lock_token is self._admission_token and isinstance(lock_token, _DiagnosticAdmissionToken) and lock_token.authority is self and self._authority_lock_owner == threading.get_ident() ) def activate_session(self, *, generation: int) -> DiagnosticSession: """Atomically consume the current dormant session generation.""" if type(generation) is not int or generation < 1: raise DiagnosticAuthorityError("diagnostic activation generation is invalid") if self._restart_unvalidated: raise DiagnosticAuthorityError("diagnostic restart requires revalidation") if self._session.state is not DiagnosticSessionState.PREPARED: raise DiagnosticAuthorityError("diagnostic session is not dormant") if generation != self._session.generation: raise DiagnosticAuthorityError("diagnostic activation generation is stale") with self._authority_lock(): rows = _read_journal(self._journal_path) _validate_journal(rows, profile_root_digest=self._profile_root_digest) bound = [ row for row in rows if row.get("status") == _SESSION_STATUS and row.get("session_id") == self._session.session_id ] if len(bound) != 1 or not self._session_row_matches(bound[0]): raise DiagnosticAuthorityError("diagnostic session journal binding mismatch") self._session = DiagnosticSession( self._session.session_id, DiagnosticSessionState.ACTIVE, self._session.generation, self._session.boot_epoch, self._session.spec_digest, self._session.authority_digest, self._session.transport_binding_digest, self._session.expires_at, ) return self._session def verify_route_active(self, *, generation: int) -> None: if type(generation) is not int or generation < 1: raise DiagnosticAuthorityError("diagnostic route generation is invalid") with self._authority_lock(): self._require_not_detached_locked() if self._restart_unvalidated: raise DiagnosticAuthorityError( "diagnostic restart requires revalidation" ) self._session.require_active( generation=generation, boot_epoch=self._boot_epoch, ) def rows(self, dedupe_key: str) -> tuple[Mapping[str, Any], ...]: with self._authority_lock(): rows = _read_journal(self._journal_path) _validate_journal(rows, profile_root_digest=self._profile_root_digest) return tuple(dict(row) for row in rows if row.get("dedupe_key") == dedupe_key) def _load_live_snapshot( self, activated: ActivatedDiagnosticDelivery, *, candidate: DiagnosticDeliveryCandidate, lock_token: object | None, ) -> LiveDiagnosticAuthoritySnapshot: self._require_host_admission(lock_token) if type(activated) is not ActivatedDiagnosticDelivery: raise DiagnosticAuthorityError("diagnostic activation is invalid") loader = self._live_loader if type(loader) is not DiagnosticLiveAuthorityLoader or loader.authority is not self: raise DiagnosticAuthorityError("diagnostic live authority loader is invalid") try: snapshot = loader.load_live_diagnostic_authority( activated, lock_token=lock_token, ) except (AttributeError, TypeError, ValueError) as exc: raise DiagnosticAuthorityError("diagnostic live authority snapshot is invalid") from exc if type(snapshot) is not LiveDiagnosticAuthoritySnapshot: raise DiagnosticAuthorityError("diagnostic live authority snapshot is invalid") sentinel = object() activation_fields = ( "customer_key_digest", "proposal_digest", "revision_digest", "rendered_body_digest", "destination_digest", "registry_digest", "activation_receipt_digest", "config_digest", "policy_digest", "catalog_digest", "meal_constraints_digest", "diagnostic_transport_binding_digest", ) for name in activation_fields: actual = getattr(activated, name, sentinel) if actual is sentinel or actual != getattr(snapshot, name): raise DiagnosticAuthorityError(f"diagnostic live activation pin is stale: {name}") if activated.revision != snapshot.revision: raise DiagnosticAuthorityError("diagnostic live activation revision is stale") if ( activated.session_id != snapshot.session_id or activated.session_generation != snapshot.session_generation or activated.expires_at_kst != snapshot.expires_at_kst ): raise DiagnosticAuthorityError("diagnostic live activation session is stale") if snapshot.session_id != self._session.session_id: raise DiagnosticAuthorityError("diagnostic live session is stale") if snapshot.session_generation != self._session.generation: raise DiagnosticAuthorityError("diagnostic live generation is stale") if snapshot.diagnostic_transport_binding_digest != self._session.transport_binding_digest: raise DiagnosticAuthorityError("diagnostic live transport binding is stale") if candidate.activated is None or candidate.activated.session_id != snapshot.session_id: raise DiagnosticAuthorityError("diagnostic live candidate session is stale") for name in _LIVE_PIN_FIELDS: actual = getattr(candidate, name, sentinel) if actual is sentinel or actual != getattr(snapshot, name): raise DiagnosticAuthorityError(f"diagnostic live candidate pin is stale: {name}") if getattr(candidate, "revision", sentinel) != snapshot.revision: raise DiagnosticAuthorityError("diagnostic live candidate revision is stale") if candidate.generation != snapshot.session_generation: raise DiagnosticAuthorityError("diagnostic live candidate generation is stale") return snapshot def reserve_and_verify( self, candidate: DiagnosticDeliveryCandidate, *, lock_token: object | None = None, ) -> DiagnosticReservationDecision: self._require_host_admission(lock_token) self._validate_candidate(candidate) self._session.require_active( generation=candidate.generation, boot_epoch=self._boot_epoch, ) if candidate.session_digest != self._session.spec_digest: raise DiagnosticAuthorityError("diagnostic session digest mismatch") if candidate.transport_binding_digest != self._session.transport_binding_digest: raise DiagnosticAuthorityError("diagnostic transport binding mismatch") with self._authority_lock(lock_token): self._require_not_detached() if self._restart_unvalidated: raise DiagnosticAuthorityError("diagnostic restart requires revalidation") activated = candidate.activated if activated is None: raise DiagnosticAuthorityError("diagnostic candidate activation is required") self._load_live_snapshot( activated, candidate=candidate, lock_token=lock_token, ) rows = _read_journal(self._journal_path) states = _validate_journal(rows, profile_root_digest=self._profile_root_digest) chain = states.get(candidate.dedupe_key) if chain: first = chain[0] self._ensure_candidate_matches_row(candidate, first) status = chain[-1]["status"] if status == "sent_audited": return DiagnosticDuplicateNoSend() if status == "audit_pending": return DiagnosticAuditPendingNoSend() if status == "provider_receipt": self._append_event_locked( rows, status="audit_pending", source=first, ) return DiagnosticAuditPendingNoSend() if status == "delivery_unknown": return DiagnosticUnknownNoSend() if status == "delivery_attempt_started": self._append_event_locked( rows, status="delivery_unknown", source=first, reason="diagnostic_started_without_provider_receipt", ) return DiagnosticUnknownNoSend() raise DiagnosticAuthorityError("diagnostic reservation state is invalid") reservation_id = hashlib.sha256( f"{candidate.dedupe_key}:{candidate.body_digest}:{candidate.session_digest}".encode("utf-8") ).hexdigest()[:32] reservation_digest = _reservation_digest(candidate, reservation_id) self._append_event_locked( rows, status="delivery_attempt_started", candidate=candidate, reservation_id=reservation_id, reservation_digest=reservation_digest, ) return VerifiedDiagnosticReservation( reservation_id, candidate, reservation_digest, _authority_token=_DIAGNOSTIC_HOST_ADMISSION_TOKEN, ) def verify_activated_delivery( self, candidate: DiagnosticDeliveryCandidate, activated: ActivatedDiagnosticDelivery, *, session_id: str, activation_loader: object, lock_token: object | None, ) -> ActivatedDiagnosticDelivery: """Reload and compare every persisted activation pin under host admission.""" self._require_host_admission(lock_token) self._validate_candidate(candidate) if type(activated) is not ActivatedDiagnosticDelivery: raise DiagnosticAuthorityError("diagnostic activation is invalid") if type(session_id) is not str or session_id != self._session.session_id: raise DiagnosticAuthorityError("diagnostic activation session is stale") if type(activation_loader) is not DurableDiagnosticActivationLoader: raise DiagnosticAuthorityError("diagnostic activation loader is sealed") if activation_loader.authority is not self: raise DiagnosticAuthorityError("diagnostic activation authority changed") loader = activation_loader.load_activated_delivery reloaded = loader(session_id, lock_token=lock_token) if type(reloaded) is not ActivatedDiagnosticDelivery or reloaded != activated: raise DiagnosticAuthorityError("diagnostic activation changed") try: expiry = _parse_timestamp( reloaded.expires_at_kst, "diagnostic activation expiry" ) except (TypeError, ValueError) as exc: raise DiagnosticAuthorityError("diagnostic activation expiry is invalid") from exc if ( reloaded.expires_at_kst != self._session.expires_at or expiry <= datetime.now(timezone.utc) ): raise DiagnosticAuthorityError("diagnostic activation is expired") if reloaded.session_id != self._session.session_id: raise DiagnosticAuthorityError("diagnostic activation session is stale") if reloaded.session_generation != self._session.generation: raise DiagnosticAuthorityError("diagnostic activation generation is stale") if reloaded.diagnostic_transport_binding_digest != self._session.transport_binding_digest: raise DiagnosticAuthorityError("diagnostic activation binding is stale") if candidate.activated != reloaded or candidate.generation != reloaded.session_generation: raise DiagnosticAuthorityError("diagnostic activation candidate is stale") if candidate.session_digest != self._session.spec_digest: raise DiagnosticAuthorityError("diagnostic session digest mismatch") self._load_live_snapshot( reloaded, candidate=candidate, lock_token=lock_token, ) return reloaded def verify_provider_start( self, verified: VerifiedDiagnosticReservation, *, deadline_monotonic: float, lock_token: object | None, activated: ActivatedDiagnosticDelivery, activation_loader: DurableDiagnosticActivationLoader, ) -> VerifiedDiagnosticReservation: self._require_host_admission(lock_token, require_context=True) if not isinstance(verified, VerifiedDiagnosticReservation): raise DiagnosticReservationConflict("reservation is not verified") if ( verified.provider_authority is not True or verified._authority_token is not _DIAGNOSTIC_HOST_ADMISSION_TOKEN ): raise DiagnosticReservationConflict("reservation authority is invalid") if isinstance(deadline_monotonic, bool) or not isinstance( deadline_monotonic, (int, float) ): raise DiagnosticAuthorityError("diagnostic deadline is invalid") try: deadline = float(deadline_monotonic) except (OverflowError, ValueError) as exc: raise DiagnosticAuthorityError("diagnostic deadline is invalid") from exc if not math.isfinite(deadline) or deadline <= time.monotonic(): raise DiagnosticAuthorityError("diagnostic deadline elapsed") candidate = verified.candidate self._validate_candidate(candidate) self._session.require_active( generation=candidate.generation, boot_epoch=self._boot_epoch, ) if candidate.session_digest != self._session.spec_digest: raise DiagnosticAuthorityError("diagnostic session digest mismatch") if candidate.transport_binding_digest != self._session.transport_binding_digest: raise DiagnosticAuthorityError("diagnostic transport binding mismatch") self.verify_activated_delivery( candidate, activated, session_id=activated.session_id, activation_loader=activation_loader, lock_token=lock_token, ) with self._authority_lock(lock_token): self._require_not_detached() if self._restart_unvalidated: raise DiagnosticAuthorityError("diagnostic restart requires revalidation") rows = _read_journal(self._journal_path) states = _validate_journal(rows, profile_root_digest=self._profile_root_digest) bound = [ row for row in rows if row.get("status") == _SESSION_STATUS and row.get("session_id") == self._session.session_id ] if ( len(bound) != 1 or self._session.boot_epoch != self._boot_epoch or ( bound[0].get("boot_epoch") != self._session.boot_epoch and not self._boot_epoch_revalidated ) or bound[0].get("session_id") != self._session.session_id or bound[0].get("generation") != self._session.generation or bound[0].get("session_digest") != self._session.spec_digest or bound[0].get("authority_digest") != self._session.authority_digest or bound[0].get("transport_binding_digest") != self._session.transport_binding_digest ): raise DiagnosticAuthorityError("diagnostic session journal binding mismatch") chain = states.get(candidate.dedupe_key) if not chain: raise DiagnosticReservationConflict("reservation is not current") first = chain[0] if first.get("reservation_id") != verified.reservation_id: raise DiagnosticReservationConflict("reservation is not current") self._ensure_candidate_matches_row(candidate, first) if first.get("reservation_digest") != verified.reservation_digest: raise DiagnosticReservationConflict("reservation digest is stale") if chain[-1]["status"] != "delivery_attempt_started": raise DiagnosticReservationConflict("reservation is not current") return verified def record_terminal( self, verified: VerifiedDiagnosticReservation, *, receipt: Mapping[str, Any] | None, audited: bool, lock_token: object | None = None, ) -> Mapping[str, Any]: """Finalize a reservation; provider admission is lease-gated.""" if receipt is not None or audited: self._require_host_admission(lock_token, require_context=True) elif lock_token is not None: self._require_host_admission(lock_token) if ( not isinstance(verified, VerifiedDiagnosticReservation) or verified._authority_token is not _DIAGNOSTIC_HOST_ADMISSION_TOKEN ): raise DiagnosticReservationConflict("reservation is not verified") self._validate_candidate(verified.candidate) with self._authority_lock(lock_token): rows = _read_journal(self._journal_path) states = _validate_journal(rows, profile_root_digest=self._profile_root_digest) chain = states.get(verified.candidate.dedupe_key) if not chain or chain[0].get("reservation_id") != verified.reservation_id: raise DiagnosticReservationConflict("reservation is not current") first = chain[0] self._ensure_candidate_matches_row(verified.candidate, first) if first.get("reservation_digest") != verified.reservation_digest: raise DiagnosticReservationConflict("reservation digest is stale") status = chain[-1]["status"] if status == "delivery_unknown": return self._terminal_result(chain[-1]) if status == "sent_audited": return self._terminal_result(chain[-1]) receipt_digest = _digest(receipt) if receipt is not None else None if status == "audit_pending": if audited: provider = self._provider_row(chain) if receipt_digest is not None and receipt_digest != provider.get("receipt_digest"): raise DiagnosticReservationConflict("diagnostic receipt changed") row = self._append_event_locked(rows, status="sent_audited", source=first) return self._terminal_result(row) return self._terminal_result(chain[-1]) if status == "provider_receipt": provider = chain[-1] if receipt_digest is not None and receipt_digest != provider.get("receipt_digest"): raise DiagnosticReservationConflict("diagnostic receipt changed") target = "sent_audited" if audited else "audit_pending" row = self._append_event_locked(rows, status=target, source=first) return self._terminal_result(row) if status != "delivery_attempt_started": raise DiagnosticReservationConflict("reservation is not current") if receipt is None: row = self._append_event_locked( rows, status="delivery_unknown", source=first, reason="provider_receipt_unavailable", ) return self._terminal_result(row) provider = self._append_event_locked( rows, status="provider_receipt", source=first, receipt_digest=receipt_digest, ) target = "sent_audited" if audited else "audit_pending" row = self._append_event_locked(rows, status=target, source=first) return self._terminal_result(row) def revalidate_after_restart(self, *, new_boot_epoch: str) -> DiagnosticSession: """Bind a reconstructed authority to a fresh process epoch explicitly.""" if not self._restart_unvalidated: raise DiagnosticAuthorityError("diagnostic authority is not awaiting restart validation") if not isinstance(new_boot_epoch, str) or not new_boot_epoch or new_boot_epoch == self._boot_epoch: raise DiagnosticAuthorityError("fresh diagnostic boot epoch is required") self._boot_epoch = new_boot_epoch self._session = DiagnosticSession( self._session.session_id, DiagnosticSessionState.ACTIVE, self._session.generation, new_boot_epoch, self._session.spec_digest, self._session.authority_digest, self._session.transport_binding_digest, self._session.expires_at, ) self._boot_epoch_revalidated = True self._restart_unvalidated = False return self._session def detach(self, *, generation: int, state: str) -> DiagnosticSession: """Durably fence the current generation before routes or provider admission.""" if state not in {"detaching", "expired", "revoked", "closed", "recovery_required"}: raise DiagnosticAuthorityError("diagnostic terminal state is invalid") if generation != self._session.generation: raise DiagnosticAuthorityError("diagnostic detach generation is stale") if self._admission_token is not None and self._authority_lock_owner == threading.get_ident(): raise DiagnosticAuthorityError("diagnostic detach is blocked by an active delivery admission") expected = { "schema_version": "diagnostic_session_terminal_v1", "session_id": self._session.session_id, "generation": generation + 1, "state": state, "spec_digest": self._session.spec_digest, "authority_digest": self._session.authority_digest, "transport_binding_digest": self._session.transport_binding_digest, } expected["digest"] = _digest(expected) with self._authority_lock(): with _open_profile_directory( self._profile_root, self._terminal_path.parent, ) as parent_fd: try: existing = os.stat( self._terminal_path.name, dir_fd=parent_fd, follow_symlinks=False, ) except FileNotFoundError: existing = None if existing is not None: value, _ = _read_activation_record( self._terminal_path, profile_root=self._profile_root, ) if set(value) != set(expected) or value != expected: raise DiagnosticAuthorityError( "diagnostic terminal fence conflicts" ) else: temporary_name = ( f".{self._terminal_path.name}." f"{os.getpid()}.{threading.get_ident()}.tmp" ) descriptor = os.open( temporary_name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0), 0o600, dir_fd=parent_fd, ) try: encoded = _canonical(expected) offset = 0 while offset < len(encoded): written = os.write(descriptor, encoded[offset:]) if written <= 0: raise DiagnosticAuthorityError( "diagnostic terminal fence write was incomplete" ) offset += written os.fsync(descriptor) except BaseException: try: os.unlink(temporary_name, dir_fd=parent_fd) except FileNotFoundError: pass raise finally: os.close(descriptor) try: try: os.link( temporary_name, self._terminal_path.name, src_dir_fd=parent_fd, dst_dir_fd=parent_fd, follow_symlinks=False, ) except FileExistsError: value, _ = _read_activation_record( self._terminal_path, profile_root=self._profile_root, ) if set(value) != set(expected) or value != expected: raise DiagnosticAuthorityError( "diagnostic terminal fence conflicts" ) os.fsync(parent_fd) finally: try: os.unlink(temporary_name, dir_fd=parent_fd) except FileNotFoundError: pass return DiagnosticSession( self._session.session_id, DiagnosticSessionState(state), generation + 1, self._session.boot_epoch, self._session.spec_digest, self._session.authority_digest, self._session.transport_binding_digest, self._session.expires_at, ) def _require_host_admission( self, lock_token: object | None, *, require_context: bool = False, ) -> None: if lock_token is _DIAGNOSTIC_HOST_ADMISSION_TOKEN: raise DiagnosticAuthorityError( "diagnostic delivery admission context is required" ) if self._admission_is_active(lock_token): return raise DiagnosticAuthorityError("diagnostic host admission token is required") def _require_not_detached(self) -> None: with self._authority_lock(): self._require_not_detached_locked() def _require_not_detached_locked(self) -> None: if self._authority_lock_owner != threading.get_ident(): raise DiagnosticAuthorityError("diagnostic authority lock is required") with _open_profile_directory( self._profile_root, self._terminal_path.parent, ) as parent_fd: try: os.stat( self._terminal_path.name, dir_fd=parent_fd, follow_symlinks=False, ) except FileNotFoundError: return value, _ = _read_activation_record( self._terminal_path, profile_root=self._profile_root, ) expected_keys = { "schema_version", "session_id", "generation", "state", "spec_digest", "authority_digest", "transport_binding_digest", "digest", } if set(value) != expected_keys: raise DiagnosticAuthorityError("diagnostic terminal fence is corrupt") digest = value.get("digest") unsigned = {key: item for key, item in value.items() if key != "digest"} if not _is_digest(digest) or _digest(unsigned) != digest: raise DiagnosticAuthorityError("diagnostic terminal fence digest mismatch") if ( value.get("schema_version") != "diagnostic_session_terminal_v1" or value.get("session_id") != self._session.session_id or value.get("generation") != self._session.generation + 1 or value.get("state") not in {"detaching", "expired", "revoked", "closed", "recovery_required"} or value.get("spec_digest") != self._session.spec_digest or value.get("authority_digest") != self._session.authority_digest or value.get("transport_binding_digest") != self._session.transport_binding_digest ): raise DiagnosticAuthorityError("diagnostic terminal fence conflicts") raise DiagnosticAuthorityError("diagnostic session is durably detached") def _initialize_durable_session(self) -> bool: with self._authority_lock(): rows = _read_journal(self._journal_path) states = _validate_journal(rows, profile_root_digest=self._profile_root_digest) bound = [row for row in rows if row.get("status") == _SESSION_STATUS and row.get("session_id") == self._session.session_id] if bound: row = bound[0] if not self._session_row_matches(row): raise DiagnosticAuthorityError("diagnostic session journal binding mismatch") restart_unvalidated = True else: self._append_event_locked(rows, status=_SESSION_STATUS, session=self._session) rows = _read_journal(self._journal_path) states = _validate_journal(rows, profile_root_digest=self._profile_root_digest) restart_unvalidated = False if self._session.state is not DiagnosticSessionState.PREPARED: self._recover_incomplete_locked(rows, states) return restart_unvalidated def _recover_incomplete_locked( self, rows: list[dict[str, object]], states: dict[str, list[dict[str, object]]], ) -> None: for chain in tuple(states.values()): last = chain[-1] if last["status"] == "delivery_attempt_started": self._append_event_locked( rows, status="delivery_unknown", source=chain[0], reason="diagnostic_restart_without_terminal_receipt", ) elif last["status"] == "provider_receipt": self._append_event_locked( rows, status="audit_pending", source=chain[0], ) def _session_row_matches(self, row: Mapping[str, object]) -> bool: return ( row.get("session_id") == self._session.session_id and row.get("generation") == self._session.generation and row.get("boot_epoch") == self._session.boot_epoch and row.get("session_digest") == self._session.spec_digest and row.get("authority_digest") == self._session.authority_digest and row.get("transport_binding_digest") == self._session.transport_binding_digest ) @staticmethod def _provider_row(chain: list[dict[str, object]]) -> dict[str, object]: for row in reversed(chain): if row["status"] == "provider_receipt": return row raise DiagnosticAuthorityError("diagnostic provider receipt row is unavailable") @staticmethod def _terminal_result(row: Mapping[str, object]) -> dict[str, object]: result: dict[str, object] = { "status": row["status"], "reservation_id": row["reservation_id"], } if row.get("receipt_digest") is not None: result["receipt_digest"] = row["receipt_digest"] return result def _append_event_locked( self, rows: list[dict[str, object]], *, status: str, session: DiagnosticSession | None = None, candidate: DiagnosticDeliveryCandidate | None = None, reservation_id: str | None = None, reservation_digest: str | None = None, source: Mapping[str, object] | None = None, receipt_digest: str | None = None, reason: str | None = None, ) -> dict[str, object]: if rows: previous = rows[-1] prev_digest = previous["row_digest"] sequence = int(previous["sequence"]) + 1 else: prev_digest = _DIGEST_ZERO sequence = 1 bound_session = session or self._session if candidate is not None: row_values: dict[str, object] = { "dedupe_key": candidate.dedupe_key, "reservation_id": reservation_id, "reservation_digest": reservation_digest, "body_digest": candidate.body_digest, "destination_digest": candidate.destination_digest, "pins": dict(candidate.pins), } session_digest = candidate.session_digest generation = candidate.generation transport_binding_digest = candidate.transport_binding_digest elif source is not None: row_values = { "dedupe_key": source["dedupe_key"], "reservation_id": source["reservation_id"], "reservation_digest": source["reservation_digest"], "body_digest": source["body_digest"], "destination_digest": source["destination_digest"], "pins": dict(source["pins"]), } session_digest = source["session_digest"] generation = source["generation"] transport_binding_digest = source["transport_binding_digest"] else: row_values = { "dedupe_key": None, "reservation_id": None, "reservation_digest": None, "body_digest": None, "destination_digest": None, "pins": {}, } session_digest = bound_session.spec_digest generation = bound_session.generation transport_binding_digest = bound_session.transport_binding_digest row: dict[str, object] = { "schema_version": DIAGNOSTIC_JOURNAL_SCHEMA, "sequence": sequence, "status": status, "prev_digest": prev_digest, "row_digest": "", "profile_root_digest": self._profile_root_digest, "session_id": bound_session.session_id if source is None else source["session_id"], "generation": generation, "boot_epoch": bound_session.boot_epoch if source is None else source["boot_epoch"], "session_digest": session_digest, "authority_digest": bound_session.authority_digest if source is None else source["authority_digest"], "transport_binding_digest": transport_binding_digest, "dedupe_key": row_values["dedupe_key"], "reservation_id": row_values["reservation_id"], "reservation_digest": row_values["reservation_digest"], "body_digest": row_values["body_digest"], "destination_digest": row_values["destination_digest"], "receipt_digest": receipt_digest, "pins": row_values["pins"], "reason": reason, } row["row_digest"] = _digest(_row_without_digest(row)) _append_journal_row(self._journal_path, row) rows.append(row) return row @staticmethod def _validate_candidate(candidate: DiagnosticDeliveryCandidate) -> None: if type(candidate) is not DiagnosticDeliveryCandidate: raise DiagnosticAuthorityError("diagnostic candidate is invalid") if candidate._factory_token is not _DIAGNOSTIC_CANDIDATE_FACTORY_TOKEN: raise DiagnosticAuthorityError("diagnostic candidate factory is sealed") if type(candidate.activated) is not ActivatedDiagnosticDelivery: raise DiagnosticAuthorityError("diagnostic candidate activation is invalid") if candidate.destination != candidate.activated.destination: raise DiagnosticAuthorityError("diagnostic candidate destination is invalid") if candidate.body_bytes != candidate.activated.rendered_body: raise DiagnosticAuthorityError("diagnostic candidate body is invalid") if candidate.body_digest != candidate.activated.rendered_body_digest: raise DiagnosticAuthorityError("diagnostic candidate body digest is invalid") if candidate.destination_digest != candidate.activated.destination_digest: raise DiagnosticAuthorityError("diagnostic candidate destination digest is invalid") if candidate.transport_binding_digest != candidate.activated.diagnostic_transport_binding_digest: raise DiagnosticAuthorityError("diagnostic candidate binding is invalid") if dict(candidate.pins) != dict(candidate.activated.pins): raise DiagnosticAuthorityError("diagnostic candidate pins are invalid") if not isinstance(candidate.dedupe_key, str) or not candidate.dedupe_key or len(candidate.dedupe_key) > 256: raise DiagnosticAuthorityError("diagnostic dedupe key is invalid") if type(candidate.generation) is not int or candidate.generation < 1: raise DiagnosticAuthorityError("diagnostic candidate generation is invalid") if not isinstance(candidate.body_bytes, bytes) or ( hashlib.sha256(candidate.body_bytes).hexdigest() != candidate.body_digest ): raise DiagnosticAuthorityError("diagnostic body digest mismatch") for value in ( candidate.session_digest, candidate.body_digest, candidate.destination_digest, candidate.transport_binding_digest, ): if not _is_digest(value): raise DiagnosticAuthorityError("diagnostic candidate digest is invalid") if not isinstance(candidate.pins, Mapping) or not candidate.pins: raise DiagnosticAuthorityError("diagnostic candidate pins are required") for key, value in candidate.pins.items(): if not isinstance(key, str) or not key or not _is_digest(value): raise DiagnosticAuthorityError("diagnostic candidate pins are invalid") @staticmethod def _ensure_candidate_matches_row( candidate: DiagnosticDeliveryCandidate, row: Mapping[str, object], ) -> None: identity = ( candidate.body_digest, candidate.destination_digest, candidate.session_digest, candidate.generation, candidate.transport_binding_digest, dict(candidate.pins), ) existing = ( row["body_digest"], row["destination_digest"], row["session_digest"], row["generation"], row["transport_binding_digest"], row["pins"], ) if identity != existing: raise DiagnosticReservationConflict("diagnostic dedupe conflict")