from __future__ import annotations import fcntl import hashlib import hmac import json import os import re import stat from contextlib import contextmanager from dataclasses import dataclass from enum import Enum from pathlib import Path from typing import Callable, Iterator, Mapping, Sequence, TypeAlias class DiagnosticEvidenceError(RuntimeError): """Raised when bounded diagnostic evidence cannot be read safely.""" class DiagnosticEvidenceSource(str, Enum): """The fixed writer-lock domains that can be used for evidence.""" WIZARD_DRAFT = "wizard_draft" CANONICAL_EVENT_SEQUENCE = "canonical_event_sequence" ADAPTIVE_JOURNALS = "adaptive_journals" OVERLAY = "overlay" SCHEDULE = "schedule" PROFILE_AUTHORITY = "profile_authority" # A descriptor is deliberately not constructible by callers. In particular, # a caller cannot pair an arbitrary data file with a different lock and then # claim that the resulting bytes came from one of the profile writer domains. _DESCRIPTOR_TOKEN = object() _DESCRIPTOR_VALIDATOR: TypeAlias = Callable[["DiagnosticSourceDescriptor"], None] @dataclass(frozen=True, slots=True, init=False) class DiagnosticSourceDescriptor: """Sealed description of one profile-owned evidence source. ``data_paths`` and ``lock_path`` are derived by :func:`diagnostic_source`. The private constructor token prevents callers from manufacturing a descriptor with an unrelated lock inode. """ kind: DiagnosticEvidenceSource data_paths: tuple[Path, ...] lock_path: Path root: Path profile_root: Path validator: _DESCRIPTOR_VALIDATOR _seal: object def __init__( self, kind: DiagnosticEvidenceSource, data_paths: tuple[Path, ...], lock_path: Path, root: Path, profile_root: Path, validator: _DESCRIPTOR_VALIDATOR, *, _token: object | None = None, ) -> None: if _token is not _DESCRIPTOR_TOKEN: raise TypeError("diagnostic source descriptors are factory-created") if not isinstance(kind, DiagnosticEvidenceSource): raise TypeError("diagnostic source kind is invalid") if not isinstance(data_paths, tuple) or not all(isinstance(item, Path) for item in data_paths): raise TypeError("diagnostic source paths are invalid") if not isinstance(lock_path, Path) or not isinstance(root, Path) or not isinstance(profile_root, Path): raise TypeError("diagnostic source roots are invalid") if not callable(validator): raise TypeError("diagnostic source validator is invalid") object.__setattr__(self, "kind", kind) object.__setattr__(self, "data_paths", data_paths) object.__setattr__(self, "lock_path", lock_path) object.__setattr__(self, "root", root) object.__setattr__(self, "profile_root", profile_root) object.__setattr__(self, "validator", validator) object.__setattr__(self, "_seal", _DESCRIPTOR_TOKEN) @classmethod def _create( cls, kind: DiagnosticEvidenceSource, data_paths: tuple[Path, ...], lock_path: Path, root: Path, profile_root: Path, ) -> "DiagnosticSourceDescriptor": return cls( kind, data_paths, lock_path, root, profile_root, _validate_descriptor, _token=_DESCRIPTOR_TOKEN, ) @dataclass(frozen=True) class ReadOnlySnapshot: schema_version: str kind: DiagnosticEvidenceSource | str row_count: int content_digest: str revision_token: str @property def source_kind(self) -> str: """Compatibility spelling for older bounded fixture consumers.""" return self.kind.value if isinstance(self.kind, DiagnosticEvidenceSource) else str(self.kind) def to_dict(self) -> dict[str, object]: return { "schema_version": self.schema_version, "kind": self.source_kind, "row_count": self.row_count, "content_digest": self.content_digest, "revision_token": self.revision_token, } @contextmanager def existing_read_lock(lock_path: Path) -> Iterator[int]: """Acquire ``LOCK_SH`` on an existing writer lock without mutation.""" path = Path(lock_path) flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0) descriptor: int | None = None try: before = path.lstat() descriptor = os.open(path, flags) opened = os.fstat(descriptor) after = path.lstat() if ( not stat.S_ISREG(opened.st_mode) or stat.S_IMODE(opened.st_mode) != 0o600 or opened.st_nlink != 1 or opened.st_uid != os.geteuid() or (opened.st_dev, opened.st_ino) != (before.st_dev, before.st_ino) or (opened.st_dev, opened.st_ino) != (after.st_dev, after.st_ino) ): raise DiagnosticEvidenceError("corrupt_state") fcntl.flock(descriptor, fcntl.LOCK_SH) current = path.lstat() if (opened.st_dev, opened.st_ino) != (current.st_dev, current.st_ino): raise DiagnosticEvidenceError("snapshot_unstable") yield descriptor except DiagnosticEvidenceError: raise except (FileNotFoundError, OSError) as exc: raise DiagnosticEvidenceError("snapshot_unstable") from exc finally: if descriptor is not None: try: fcntl.flock(descriptor, fcntl.LOCK_UN) finally: os.close(descriptor) def _canonical(value: object) -> bytes: return json.dumps( value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False, ).encode("utf-8") def _sha256(raw: bytes) -> str: return hashlib.sha256(raw).hexdigest() def _bounded_row(row: Mapping[str, object]) -> dict[str, object]: allowed = {"event_type", "status", "reason_code", "schema_version", "epoch", "count"} bounded: dict[str, object] = {} for key, value in row.items(): if key in allowed and isinstance(value, (str, int, bool)): bounded[key] = value elif key.endswith("_digest") and isinstance(value, str) and len(value) == 64: bounded[key] = value return bounded def _resolve_directory(path: Path, label: str) -> Path: candidate = Path(path) try: if candidate.is_symlink() or not candidate.exists() or not candidate.is_dir(): raise DiagnosticEvidenceError(f"{label} is unavailable") resolved = candidate.resolve(strict=True) except (OSError, RuntimeError) as exc: raise DiagnosticEvidenceError(f"{label} is unavailable") from exc if resolved.is_symlink(): raise DiagnosticEvidenceError(f"{label} is unsafe") return resolved def _reject_symlink_components(path: Path, *, stop: Path) -> None: """Reject symlink components between ``stop`` and ``path``.""" current = stop try: relative = path.relative_to(stop) except ValueError as exc: raise DiagnosticEvidenceError("source path escapes its root") from exc for part in relative.parts: current = current / part if current.is_symlink(): raise DiagnosticEvidenceError("source path symlink is not allowed") def _require_contained(path: Path, root: Path, *, allow_missing: bool = True) -> None: try: relative = path.relative_to(root) except ValueError as exc: raise DiagnosticEvidenceError("source path escapes its root") from exc if not relative.parts or any(part in {"", ".", ".."} for part in relative.parts): raise DiagnosticEvidenceError("source path is not relative to its root") _reject_symlink_components(path, stop=root) if path.is_symlink() or (not allow_missing and not path.exists()): raise DiagnosticEvidenceError("source path symlink or absence is unsafe") def _registry_path(profile: Path) -> Path: candidates = (profile / "customers" / "registry.json", profile / "registry.json") for candidate in candidates: if not candidate.exists() and not candidate.is_symlink(): continue if candidate.is_symlink(): raise DiagnosticEvidenceError("registered customer registry symlink is not allowed") try: resolved = candidate.resolve(strict=True) except (OSError, RuntimeError) as exc: raise DiagnosticEvidenceError("registered customer registry is unavailable") from exc _require_contained(resolved, profile, allow_missing=False) if not resolved.is_file(): raise DiagnosticEvidenceError("registered customer registry is unavailable") return resolved raise DiagnosticEvidenceError("registered customer registry is unavailable") def _customer_root(profile: Path, customer_runtime: object) -> Path: # Importing here avoids making diagnostic evidence part of the customer # registry import cycle while still requiring the sealed runtime type. from checkin_cli.customer_coaching import ( CustomerRegistryError, CustomerRuntime, CustomerSpec, RegisteredCustomerBinding, load_customer_registry, ) if type(customer_runtime) is not CustomerRuntime: raise DiagnosticEvidenceError("registered customer runtime is required") try: spec = customer_runtime.spec raw_root = customer_runtime.data_root binding = customer_runtime.registered_binding except (AttributeError, CustomerRegistryError, OSError, RuntimeError) as exc: raise DiagnosticEvidenceError("registered customer runtime is invalid") from exc if type(spec) is not CustomerSpec or type(binding) is not RegisteredCustomerBinding: raise DiagnosticEvidenceError("registered customer runtime is invalid") if not isinstance(raw_root, Path): raise DiagnosticEvidenceError("customer data root is not canonical") try: canonical_root = raw_root.resolve() except (OSError, RuntimeError) as exc: raise DiagnosticEvidenceError("customer data root is not canonical") from exc if not raw_root.is_absolute() or raw_root != canonical_root: raise DiagnosticEvidenceError("customer data root is not canonical") if raw_root.is_symlink(): raise DiagnosticEvidenceError("customer data root symlink is not allowed") root = _resolve_directory(raw_root, "customer data root") _require_contained(root, profile, allow_missing=False) expected_root = (profile / "data" / "customers" / spec.customer_key).resolve() if root != expected_root: raise DiagnosticEvidenceError("customer data root does not match the profile registry") registry_path = _registry_path(profile) try: registry_bytes = registry_path.read_bytes() registry = load_customer_registry(registry_path, profile) except (CustomerRegistryError, OSError, RuntimeError, TypeError, ValueError) as exc: raise DiagnosticEvidenceError("registered customer registry is invalid") from exc matches = tuple( runtime for runtime in registry.customers if type(runtime) is CustomerRuntime and runtime.spec.customer_key == spec.customer_key ) if len(matches) != 1: raise DiagnosticEvidenceError("registered customer runtime is not in the profile registry") registered = matches[0] try: registered_binding = registered.registered_binding binding_values = ( binding.customer_key_digest, binding.data_root_digest, binding.registry_digest, binding.registry_version, binding.activation_digest, binding.mode, binding.binding_digest, ) expected_binding_digest = _sha256( _canonical( { "customer_key_digest": binding.customer_key_digest, "data_root_digest": binding.data_root_digest, "registry_digest": binding.registry_digest, "registry_version": binding.registry_version, "activation_digest": binding.activation_digest, "mode": binding.mode, } ) ) except (AttributeError, TypeError, ValueError) as exc: raise DiagnosticEvidenceError("registered customer binding is invalid") from exc if ( customer_runtime.spec != registered.spec or root != registered.customer_root or binding != registered_binding or not all(isinstance(value, str) for value in binding_values) or binding.data_root_digest != _sha256(str(root).encode("utf-8")) or binding.customer_key_digest != _sha256(_canonical({"customer_key": spec.customer_key})) or binding.registry_digest != _sha256(registry_bytes) or binding.binding_digest != expected_binding_digest ): raise DiagnosticEvidenceError("registered customer binding does not match the profile registry") return root def diagnostic_source( profile_root: Path, customer_runtime: object | None, source_kind: DiagnosticEvidenceSource, ) -> DiagnosticSourceDescriptor: """Derive a sealed descriptor for one fixed writer-lock domain.""" if not isinstance(source_kind, DiagnosticEvidenceSource): raise TypeError("source_kind must be DiagnosticEvidenceSource") profile = _resolve_directory(Path(profile_root), "profile root") if source_kind in { DiagnosticEvidenceSource.WIZARD_DRAFT, DiagnosticEvidenceSource.CANONICAL_EVENT_SEQUENCE, DiagnosticEvidenceSource.ADAPTIVE_JOURNALS, DiagnosticEvidenceSource.OVERLAY, }: root = _customer_root(profile, customer_runtime) wizard = root / "wizard" adaptive = root / "nutrition-plans" if source_kind is DiagnosticEvidenceSource.WIZARD_DRAFT: data_paths = (wizard / "drafts",) lock_path = wizard / ".wizard.lock" elif source_kind is DiagnosticEvidenceSource.CANONICAL_EVENT_SEQUENCE: data_paths = (wizard / "events.jsonl", adaptive / "canonical-sequence.jsonl") lock_path = wizard / ".events.lock" elif source_kind is DiagnosticEvidenceSource.ADAPTIVE_JOURNALS: data_paths = tuple( adaptive / name for name in ( "events.jsonl", "source-days.jsonl", "source-day-intents.jsonl", "authority-mirror-intents.jsonl", "config-epoch-journal.jsonl", ) ) lock_path = adaptive / ".adaptive.lock" else: data_paths = (adaptive / "adaptive-overlays.jsonl",) lock_path = adaptive / ".authority-transition.lock" elif source_kind is DiagnosticEvidenceSource.SCHEDULE: root = profile data = profile / "data" data_paths = ( data / "scheduled-deliveries.jsonl", data / "scheduled-deliveries-fence.json", data / "customer-schedule-claims", ) lock_path = data / ".scheduled-deliveries.lock" else: root = profile registry = _registry_path(profile) data_paths = ( registry, profile / "data" / "customer-activation-journal.json", profile / "data" / "customer-activation-audit.jsonl", ) lock_path = profile / "data" / ".adaptive-authority.lock" descriptor = DiagnosticSourceDescriptor._create( source_kind, tuple(Path(path) for path in data_paths), Path(lock_path), Path(root), profile, ) _validate_descriptor(descriptor) return descriptor def _validate_descriptor(descriptor: DiagnosticSourceDescriptor) -> None: if type(descriptor) is not DiagnosticSourceDescriptor: raise DiagnosticEvidenceError("diagnostic source descriptor is invalid") try: seal = descriptor._seal validator = descriptor.validator profile_root = Path(descriptor.profile_root) root = Path(descriptor.root) data_paths = descriptor.data_paths lock_path = descriptor.lock_path except (AttributeError, TypeError, ValueError) as exc: raise DiagnosticEvidenceError("diagnostic source descriptor is invalid") from exc if seal is not _DESCRIPTOR_TOKEN or validator is not _validate_descriptor: raise DiagnosticEvidenceError("diagnostic source descriptor is not sealed") if not isinstance(descriptor.kind, DiagnosticEvidenceSource): raise DiagnosticEvidenceError("diagnostic source kind is invalid") if ( not profile_root.is_absolute() or profile_root != profile_root.resolve() or profile_root.is_symlink() or not profile_root.exists() or not profile_root.is_dir() ): raise DiagnosticEvidenceError("diagnostic source profile root is unavailable") if not root.is_absolute() or root != root.resolve(): raise DiagnosticEvidenceError("diagnostic source root is not canonical") if root.is_symlink() or not root.exists() or not root.is_dir(): raise DiagnosticEvidenceError("diagnostic source root is unavailable") if not isinstance(data_paths, tuple) or len(set(data_paths)) != len(data_paths): raise DiagnosticEvidenceError("diagnostic source paths are invalid") for path in data_paths: if not isinstance(path, Path) or not path.is_absolute(): raise DiagnosticEvidenceError("diagnostic source path is not canonical") _require_contained(path, root) if not isinstance(lock_path, Path) or not lock_path.is_absolute() or lock_path != lock_path.resolve(): raise DiagnosticEvidenceError("diagnostic source lock is not canonical") _require_contained(lock_path, root) kind = descriptor.kind expected_lock: Path allowed_paths: set[Path] customer_kinds = { DiagnosticEvidenceSource.WIZARD_DRAFT, DiagnosticEvidenceSource.CANONICAL_EVENT_SEQUENCE, DiagnosticEvidenceSource.ADAPTIVE_JOURNALS, DiagnosticEvidenceSource.OVERLAY, } if kind is DiagnosticEvidenceSource.WIZARD_DRAFT: expected_lock = root / "wizard" / ".wizard.lock" allowed_paths = {root / "wizard" / "drafts"} elif kind is DiagnosticEvidenceSource.CANONICAL_EVENT_SEQUENCE: expected_lock = root / "wizard" / ".events.lock" allowed_paths = {root / "wizard" / "events.jsonl", root / "nutrition-plans" / "canonical-sequence.jsonl"} elif kind is DiagnosticEvidenceSource.ADAPTIVE_JOURNALS: expected_lock = root / "nutrition-plans" / ".adaptive.lock" allowed_paths = { root / "nutrition-plans" / name for name in ( "events.jsonl", "source-days.jsonl", "source-day-intents.jsonl", "authority-mirror-intents.jsonl", "config-epoch-journal.jsonl", ) } elif kind is DiagnosticEvidenceSource.OVERLAY: expected_lock = root / "nutrition-plans" / ".authority-transition.lock" allowed_paths = {root / "nutrition-plans" / "adaptive-overlays.jsonl"} elif kind is DiagnosticEvidenceSource.SCHEDULE: expected_lock = root / "data" / ".scheduled-deliveries.lock" allowed_paths = { root / "data" / "scheduled-deliveries.jsonl", root / "data" / "scheduled-deliveries-fence.json", root / "data" / "customer-schedule-claims", } else: expected_lock = root / "data" / ".adaptive-authority.lock" allowed_paths = { root / "customers" / "registry.json", root / "registry.json", root / "data" / "customer-activation-journal.json", root / "data" / "customer-activation-audit.jsonl", } if kind in customer_kinds: try: if root.parent != profile_root / "data" / "customers": raise DiagnosticEvidenceError("customer source root is not registry-contained") except (OSError, RuntimeError) as exc: raise DiagnosticEvidenceError("customer source root is not registry-contained") from exc elif root != profile_root: raise DiagnosticEvidenceError("profile source root is not canonical") if lock_path != expected_lock: raise DiagnosticEvidenceError("diagnostic source lock is unrelated") if not set(data_paths).issubset(allowed_paths): raise DiagnosticEvidenceError("diagnostic source contains an unrelated path") def _inventory_directory(path: Path, *, kind: DiagnosticEvidenceSource) -> tuple[Path, ...]: if not path.exists(): return () if path.is_symlink() or not path.is_dir(): raise DiagnosticEvidenceError("diagnostic source directory is unsafe") result: list[Path] = [] try: if kind is DiagnosticEvidenceSource.WIZARD_DRAFT: for item in sorted(path.iterdir(), key=lambda value: value.name): if item.is_symlink() or not item.is_file() or re.fullmatch(r"[a-f0-9]{32}\.json", item.name) is None: raise DiagnosticEvidenceError("wizard draft inventory is unsafe") result.append(item) else: # Schedule claims are bounded to customer/date directories and # regular tombstone files. No names enter the returned receipt. for customer in sorted(path.iterdir(), key=lambda value: value.name): if customer.is_symlink() or not customer.is_dir() or re.fullmatch(r"[a-z0-9][a-z0-9_-]{2,63}", customer.name) is None: raise DiagnosticEvidenceError("schedule claim inventory is unsafe") for day in sorted(customer.iterdir(), key=lambda value: value.name): if day.is_symlink() or not day.is_dir() or re.fullmatch(r"\d{4}-\d{2}-\d{2}", day.name) is None: raise DiagnosticEvidenceError("schedule claim inventory is unsafe") for item in sorted(day.iterdir(), key=lambda value: value.name): if item.is_symlink() or not item.is_file() or item.name.startswith("."): raise DiagnosticEvidenceError("schedule claim inventory is unsafe") result.append(item) except OSError as exc: raise DiagnosticEvidenceError("diagnostic source inventory unavailable") from exc if len(result) > 4096: raise DiagnosticEvidenceError("diagnostic source inventory is too large") return tuple(result) def _read_regular_bytes(path: Path, *, require_private: bool = True) -> bytes: if path.is_symlink(): raise DiagnosticEvidenceError("diagnostic source file symlink is not allowed") flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0) descriptor: int | None = None try: descriptor = os.open(path, flags) before = os.fstat(descriptor) named = os.stat(path, follow_symlinks=False) if ( not stat.S_ISREG(before.st_mode) or before.st_nlink != 1 or before.st_uid != os.geteuid() or (require_private and stat.S_IMODE(before.st_mode) != 0o600) or (before.st_dev, before.st_ino) != (named.st_dev, named.st_ino) or before.st_size > 8 * 1024 * 1024 ): raise DiagnosticEvidenceError("corrupt_state") chunks: list[bytes] = [] remaining = 8 * 1024 * 1024 while remaining: chunk = os.read(descriptor, remaining) if not chunk: break chunks.append(chunk) remaining -= len(chunk) after = os.fstat(descriptor) if (after.st_dev, after.st_ino, after.st_size) != (before.st_dev, before.st_ino, before.st_size): raise DiagnosticEvidenceError("snapshot_unstable") return b"".join(chunks) except DiagnosticEvidenceError: raise except (FileNotFoundError, OSError) as exc: raise DiagnosticEvidenceError("snapshot_unstable") from exc finally: if descriptor is not None: os.close(descriptor) def _expand_descriptor_paths(descriptor: DiagnosticSourceDescriptor) -> tuple[Path, ...]: result: list[Path] = [] for path in descriptor.data_paths: _require_contained(path, descriptor.root) if path.name in {"drafts", "customer-schedule-claims"}: result.extend(_inventory_directory(path, kind=descriptor.kind)) elif path.exists(): if path.is_symlink() or not path.is_file(): raise DiagnosticEvidenceError("diagnostic source file is unsafe") result.append(path) return tuple(result) def _bounded_payload(raw: bytes, path: Path) -> tuple[int, object]: if not raw: return 0, [] if path.suffix == ".jsonl" or path.name.endswith(".jsonl"): rows: list[dict[str, object]] = [] for line in raw.splitlines(): if not line.strip(): continue try: value = json.loads(line.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise DiagnosticEvidenceError("corrupt_state") from exc if not isinstance(value, Mapping): raise DiagnosticEvidenceError("corrupt_state") rows.append(_bounded_row(value)) return len(rows), rows try: value = json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError): # Fixed profile config formats not represented by JSON are still # represented only by a count; their raw bytes remain in the private # revision token and never enter the bounded receipt. return 1, [{"schema_version": "opaque_document"}] if isinstance(value, Mapping): return 1, [_bounded_row(value)] if isinstance(value, list): bounded = [_bounded_row(item) for item in value if isinstance(item, Mapping)] return len(bounded), bounded raise DiagnosticEvidenceError("corrupt_state") def _snapshot_schedule_source(source: DiagnosticSourceDescriptor) -> ReadOnlySnapshot: from checkin_cli.customer_schedule import ( _validate_schedule_delivery_read_snapshot_locked, schedule_delivery_read_lock, ) from checkin_cli.weekly_operations_schedule_host_models_r4 import CustomerScheduleError try: with schedule_delivery_read_lock(source.root) as (ledger, fence, claims_root): schedule = _validate_schedule_delivery_read_snapshot_locked( ledger, fence, claims_root, ) except DiagnosticEvidenceError: raise except CustomerScheduleError as exc: raise DiagnosticEvidenceError("corrupt_state") from exc except (OSError, RuntimeError, TypeError, ValueError) as exc: raise DiagnosticEvidenceError("corrupt_state") from exc bounded = schedule.to_dict() content = { "schema_version": bounded["schema_version"], "row_count": bounded["row_count"], "tombstone_count": bounded["tombstone_count"], "ledger_digest": bounded["ledger_digest"], "tombstone_inventory_digest": bounded["tombstone_inventory_digest"], "fence_digest": bounded["fence_digest"], } return ReadOnlySnapshot( schema_version="diagnostic_snapshot_v1", kind=source.kind, row_count=schedule.row_count, content_digest=_sha256(_canonical(content)), revision_token=schedule.revision_token, ) def snapshot_diagnostic_source(source: DiagnosticSourceDescriptor) -> ReadOnlySnapshot: """Read one typed source under its existing shared writer lock.""" if type(source) is not DiagnosticSourceDescriptor: raise TypeError("snapshot requires DiagnosticSourceDescriptor") try: _validate_descriptor(source) except DiagnosticEvidenceError: raise except (AttributeError, OSError, RuntimeError, TypeError, ValueError) as exc: raise DiagnosticEvidenceError("diagnostic source descriptor is invalid") from exc if source.kind is DiagnosticEvidenceSource.SCHEDULE: return _snapshot_schedule_source(source) expanded = _expand_descriptor_paths(source) total_bytes = bytearray() bounded_documents: list[object] = [] row_count = 0 with existing_read_lock(source.lock_path): # Inventory is repeated after taking the lock so files created or # replaced concurrently cannot be silently omitted. expanded = _expand_descriptor_paths(source) for path in expanded: raw = _read_regular_bytes(path) total_bytes.extend(len(raw).to_bytes(8, "big")) total_bytes.extend(raw) count, bounded = _bounded_payload(raw, path) row_count += count bounded_documents.append(bounded) after = _expand_descriptor_paths(source) if expanded != after: raise DiagnosticEvidenceError("snapshot_unstable") return ReadOnlySnapshot( schema_version="diagnostic_snapshot_v1", kind=source.kind, row_count=row_count, content_digest=_sha256(_canonical(bounded_documents)), revision_token=_sha256(bytes(total_bytes)), ) def coherent_diagnostic_snapshot( sources: Sequence[DiagnosticSourceDescriptor], *, max_attempts: int = 3, ) -> tuple[ReadOnlySnapshot, ...]: """Read typed domains separately and fence the complete revision vector.""" if max_attempts < 1 or max_attempts > 5: raise ValueError("max_attempts must be 1..5") if not all(type(source) is DiagnosticSourceDescriptor for source in sources): raise TypeError("coherent snapshots require typed source descriptors") typed = tuple(sources) for _ in range(max_attempts): before = tuple(snapshot_diagnostic_source(source) for source in typed) after = tuple(snapshot_diagnostic_source(source) for source in typed) if tuple(item.revision_token for item in before) == tuple(item.revision_token for item in after): return after raise DiagnosticEvidenceError("snapshot_unstable") def synthetic_replay_fixture(snapshots: Sequence[ReadOnlySnapshot]) -> dict[str, object]: return { "schema_version": "diagnostic_synthetic_replay_v1", "provenance": "synthetic_diagnostic_only", "source_digests": [snapshot.content_digest for snapshot in snapshots], "rows": [ { "synthetic": True, "kind": snapshot.source_kind, "count": snapshot.row_count, } for snapshot in snapshots ], } def require_synthetic_provenance(value: Mapping[str, object]) -> None: if value.get("schema_version") != "diagnostic_synthetic_replay_v1" or value.get("provenance") != "synthetic_diagnostic_only": raise DiagnosticEvidenceError("production replay provenance rejected") _MANIFEST_SCHEMA_VERSION = "diagnostic_promotion_manifest_v1" _MANIFEST_KEYS = frozenset({"kind", "relative_path", "sha256", "schema_version"}) _BANNED_PATH_PARTS = frozenset( { ".env", ".git", "auth", "build", "capability", "credential", "customer", "data", "dist", "generated", "ledger", "replay", "runtime", "secret", "session", "snapshot", "token", } ) _HEX_DIGEST = re.compile(r"^[0-9a-f]{64}$") def _manifest_roots(approved_roots: Mapping[str, Path], policy: Mapping[str, object]) -> dict[str, Path]: if not isinstance(approved_roots, Mapping) or set(approved_roots) != {"profile", "hermes"}: raise DiagnosticEvidenceError("approved promotion roots are closed") result: dict[str, Path] = {} for name, marker_field in (("profile", "profile_markers"), ("hermes", "hermes_markers")): value = approved_roots[name] if not isinstance(value, Path): raise TypeError("approved promotion roots must be Path values") root = _resolve_directory(value, f"{name} promotion root") markers = policy.get(marker_field) if not isinstance(markers, list) or not markers: raise DiagnosticEvidenceError("promotion policy markers are invalid") for marker in markers: if not isinstance(marker, str) or not marker or marker.startswith("/") or "\\" in marker or ".." in marker.split("/"): raise DiagnosticEvidenceError("promotion policy marker is unsafe") marker_path = root / marker _require_contained(marker_path, root, allow_missing=False) if marker.endswith("/"): if not marker_path.is_dir(): raise DiagnosticEvidenceError("promotion root marker is not a directory") elif not marker_path.is_file(): raise DiagnosticEvidenceError("promotion root marker is not a regular file") result[name] = root return result def _manifest_candidate(root: Path, relative_path: str) -> Path: if ( not isinstance(relative_path, str) or not relative_path or relative_path.startswith("/") or relative_path.endswith("/") or "\\" in relative_path or "\x00" in relative_path ): raise DiagnosticEvidenceError("promotion path is not relative") parts = relative_path.split("/") if any(not part or part in {".", ".."} for part in parts): raise DiagnosticEvidenceError("promotion path traversal rejected") candidate = root.joinpath(*parts) _require_contained(candidate, root, allow_missing=False) if candidate.is_symlink() or not candidate.is_file(): raise DiagnosticEvidenceError("promotion artifact must be a regular non-symlink file") try: if candidate.stat().st_nlink != 1: raise DiagnosticEvidenceError("promotion artifact link count is unsafe") except OSError as exc: raise DiagnosticEvidenceError("promotion artifact is unavailable") from exc return candidate def build_diagnostic_promotion_manifest( entries: Sequence[Mapping[str, object]], *, approved_roots: Mapping[str, Path], ) -> tuple[dict[str, object], ...]: """Validate a read-only promotion manifest against packaged policy. The caller supplies only the two already-approved source roots. Policy bytes and every allowlist are loaded from the installed package itself; callers cannot replace them with a permissive mapping. """ try: from checkin_cli.policies import PromotionPolicyError, load_policy policy = load_policy() except (ImportError, PromotionPolicyError, OSError, ValueError) as exc: raise DiagnosticEvidenceError("promotion policy unavailable") from exc roots = _manifest_roots(approved_roots, policy) allowed_kinds = policy.get("allowed_artifact_kinds") forbidden_kinds = policy.get("forbidden_artifact_kinds") suffixes = policy.get("allowed_suffixes") migrations = policy.get("migration_allowlist") config_allowlist = policy.get("config_allowlist") if not all(isinstance(value, list) for value in (allowed_kinds, forbidden_kinds, suffixes, migrations, config_allowlist)): raise DiagnosticEvidenceError("promotion policy allowlists are invalid") result: list[dict[str, object]] = [] seen: set[tuple[str, str]] = set() for entry in entries: if not isinstance(entry, Mapping) or set(entry) != _MANIFEST_KEYS: raise DiagnosticEvidenceError("promotion entry schema is closed") if entry["schema_version"] != _MANIFEST_SCHEMA_VERSION: raise DiagnosticEvidenceError("promotion entry schema version is unsupported") kind = entry["kind"] relative = entry["relative_path"] digest = entry["sha256"] if not isinstance(kind, str) or kind not in allowed_kinds or kind in forbidden_kinds: raise DiagnosticEvidenceError("promotion artifact kind forbidden") if not isinstance(relative, str) or "/" not in relative: raise DiagnosticEvidenceError("promotion path must identify an approved root") root_name, root_relative = relative.split("/", 1) if root_name not in roots or not root_relative: raise DiagnosticEvidenceError("promotion root prefix is invalid") key = (root_name, root_relative) if key in seen: raise DiagnosticEvidenceError("promotion manifest contains a duplicate path") seen.add(key) if not isinstance(digest, str) or _HEX_DIGEST.fullmatch(digest) is None: raise DiagnosticEvidenceError("promotion artifact digest is invalid") parts = root_relative.split("/") if any(part.lower() in _BANNED_PATH_PARTS or part.startswith(".") for part in parts): raise DiagnosticEvidenceError("promotion path names runtime data") allowed_suffixes = {suffix for suffix in suffixes if isinstance(suffix, str)} if Path(root_relative).suffix not in allowed_suffixes: raise DiagnosticEvidenceError("promotion artifact suffix is not allowlisted") if kind == "migration" and root_relative not in migrations: raise DiagnosticEvidenceError("promotion migration is not allowlisted") if kind == "test" and not (root_relative.startswith("tests/") or "/tests/" in root_relative): raise DiagnosticEvidenceError("promotion test path is not under tests") if kind == "documentation" and Path(root_relative).suffix not in {".md", ".html"}: raise DiagnosticEvidenceError("promotion documentation suffix is invalid") if kind == "config_schema" and Path(root_relative).suffix not in {".py", ".pyi"}: raise DiagnosticEvidenceError("promotion config schema must be code-side") # Config pointers are deliberately loaded and type-checked above. A # config_schema artifact is still constrained to code-side files; raw # runtime config values (JSON/YAML/TOML) cannot enter the manifest. if kind == "code" and Path(root_relative).suffix not in {".py", ".pyi", ".toml"}: raise DiagnosticEvidenceError("promotion code suffix is invalid") candidate = _manifest_candidate(roots[root_name], root_relative) try: raw = _read_regular_bytes(candidate, require_private=False) except DiagnosticEvidenceError: raise if not _HEX_DIGEST.fullmatch(_sha256(raw)) or not hmac.compare_digest(_sha256(raw), digest): raise DiagnosticEvidenceError("promotion artifact digest mismatch") result.append( { "kind": kind, "relative_path": relative, "sha256": digest, "schema_version": _MANIFEST_SCHEMA_VERSION, } ) return tuple(result) __all__ = [ "DiagnosticEvidenceError", "DiagnosticEvidenceSource", "DiagnosticSourceDescriptor", "ReadOnlySnapshot", "build_diagnostic_promotion_manifest", "coherent_diagnostic_snapshot", "diagnostic_source", "existing_read_lock", "require_synthetic_provenance", "snapshot_diagnostic_source", "synthetic_replay_fixture", ]