"""Candidate-bound authority for owner nonclinical risk acceptance.""" from __future__ import annotations import hashlib import hmac import json import os import stat from collections.abc import Mapping from dataclasses import dataclass from pathlib import Path from typing import Final _SCHEMA: Final = "owner-risk-acceptance-authority-v1" _CAPABILITY: Final = "owner_risk_acceptance" @dataclass(frozen=True, slots=True) class OwnerRiskAcceptanceCapability: candidate_epoch: str owner_user_id: int owner_chat_id: int owner_topic_id: int authority_seal_sha256: str def load_owner_risk_acceptance( extra: Mapping[str, object], *, candidate_digest: str, owner_user_id: int, owner_chat_id: int, owner_topic_id: int, ) -> OwnerRiskAcceptanceCapability | None: room = extra.get("room_bootstrap") if not isinstance(room, Mapping): return None room_values: dict[str, object] = {} for key, item in room.items(): if not isinstance(key, str): raise ValueError("owner risk acceptance mapping is invalid") room_values[key] = item raw = room_values.get("owner_risk_acceptance") if raw is None: return None if not isinstance(raw, Mapping): raise ValueError("owner risk acceptance config is invalid") config_values: dict[str, object] = {} for key, item in raw.items(): if not isinstance(key, str): raise ValueError("owner risk acceptance mapping is invalid") config_values[key] = item enabled = config_values.get("enabled") if enabled is not True: if set(config_values) != {"enabled"}: raise ValueError("disabled owner risk acceptance is invalid") return None if set(config_values) != { "enabled", "receipt_path", "receipt_sha256", }: raise ValueError("owner risk acceptance config is invalid") expected_file_digest = _digest( config_values.get("receipt_sha256"), "receipt SHA", ) receipt_path = _authorized_receipt_path( config_values.get("receipt_path"), ) receipt_bytes = receipt_path.read_bytes() if not hmac.compare_digest( hashlib.sha256(receipt_bytes).hexdigest(), expected_file_digest, ): raise ValueError("owner risk acceptance receipt SHA mismatch") try: decoded: object = json.loads(receipt_bytes) except json.JSONDecodeError as exc: raise ValueError("owner risk acceptance receipt is invalid") from exc if not isinstance(decoded, Mapping): raise ValueError("owner risk acceptance receipt is invalid") document: dict[str, object] = {} for key, item in decoded.items(): if not isinstance(key, str): raise ValueError("owner risk acceptance mapping is invalid") document[key] = item if ( set(document) != { "authorization_seal_sha256", "capability", "candidate_digest", "owner", "schema", "status", } or document.get("schema") != _SCHEMA or document.get("capability") != _CAPABILITY or document.get("status") != "AUTHORIZED" ): raise ValueError("owner risk acceptance receipt is invalid") sealed_candidate = _digest( document.get("candidate_digest"), "candidate digest", ) _ = _digest(candidate_digest, "active candidate digest") if not hmac.compare_digest(sealed_candidate, candidate_digest): raise ValueError("owner risk acceptance candidate is invalid") owner = document.get("owner") if not isinstance(owner, Mapping): raise ValueError("owner risk acceptance owner is invalid") owner_values: dict[str, object] = {} for key, item in owner.items(): if not isinstance(key, str): raise ValueError("owner risk acceptance mapping is invalid") owner_values[key] = item if ( set(owner_values) != {"chat_id", "topic_id", "user_id"} or any( type(owner_values.get(name)) is not int for name in ("chat_id", "topic_id", "user_id") ) or ( owner_values.get("user_id"), owner_values.get("chat_id"), owner_values.get("topic_id"), ) != (owner_user_id, owner_chat_id, owner_topic_id) ): raise ValueError("owner risk acceptance owner is invalid") expected_seal = _digest( document.get("authorization_seal_sha256"), "authorization seal", ) core = { key: value for key, value in document.items() if key != "authorization_seal_sha256" } actual_seal = hashlib.sha256(_canonical(core)).hexdigest() if not hmac.compare_digest(actual_seal, expected_seal): raise ValueError("owner risk acceptance seal is invalid") return OwnerRiskAcceptanceCapability( candidate_epoch=sealed_candidate, owner_user_id=owner_user_id, owner_chat_id=owner_chat_id, owner_topic_id=owner_topic_id, authority_seal_sha256=expected_seal, ) def _authorized_receipt_path(value: object) -> Path: if not isinstance(value, str) or not value: raise ValueError("owner risk acceptance receipt path is invalid") path = Path(value) root = _credential_root() if not path.is_absolute(): if path.name != value: raise ValueError("owner risk acceptance receipt path is invalid") path = root / path if path.is_symlink(): raise ValueError("owner risk acceptance receipt path is invalid") resolved = path.resolve(strict=True) if resolved.parent != root: raise ValueError("owner risk acceptance receipt root is invalid") metadata = resolved.stat() if ( not stat.S_ISREG(metadata.st_mode) or metadata.st_mode & 0o077 or metadata.st_uid != os.getuid() ): raise ValueError("owner risk acceptance receipt mode is invalid") return resolved def _credential_root() -> Path: pin_value = os.environ.get("TASK26_AUTHORITY_PIN") if pin_value: return Path(pin_value).resolve(strict=True).parent credential_directory = os.environ.get("CREDENTIALS_DIRECTORY") if credential_directory: root = Path(credential_directory) if root.is_symlink(): raise ValueError("systemd credential directory is invalid") return root.resolve(strict=True) raise ValueError("Task26 authority credentials are unavailable") def _digest(value: object, label: str) -> str: if ( not isinstance(value, str) or len(value) != 64 or set(value) - set("0123456789abcdef") ): raise ValueError(f"owner risk acceptance {label} is invalid") return value def _canonical(value: object) -> bytes: return json.dumps( value, sort_keys=True, separators=(",", ":"), ).encode() __all__ = [ "OwnerRiskAcceptanceCapability", "load_owner_risk_acceptance", ]