"""Versioned per-customer authority for the dormant weekly-operations capability.""" from __future__ import annotations import hashlib import json import re from collections.abc import Mapping from dataclasses import dataclass, field from datetime import datetime from typing import ClassVar from pydantic import BaseModel, ConfigDict, Field, StrictInt, StrictStr, ValidationError from .nutrition_weekly_operations_registry_identity import ( WeeklyOperationsRegistryIdentity, parse_weekly_operations_registry_identity, ) from .nutrition_weekly_operations_config import ( JsonValue, WeeklyOperationsAuthorityError, WeeklyOperationsConfig, ) _AUTHORITY_SCHEMA = "nutricoach-weekly-operations-authority-v2" _DIGEST = re.compile(r"[0-9a-f]{64}") _EPOCH = re.compile(r"[a-z0-9][a-z0-9-]{0,63}") class _OwnerInput(BaseModel): model_config: ClassVar[ConfigDict] = ConfigDict(frozen=True, extra="forbid", strict=True) user_id: StrictStr chat_id: StrictStr version: StrictInt class _AuthorityInput(BaseModel): model_config: ClassVar[ConfigDict] = ConfigDict(frozen=True, extra="forbid", strict=True) receipt_schema: StrictStr = Field(alias="schema") candidate_digest: StrictStr config_digest: StrictStr enabled_customer_keys: list[StrictStr] owner: _OwnerInput consent_digest: StrictStr registry_identity: dict[StrictStr, StrictStr | StrictInt] issued_at: StrictStr expires_at: StrictStr feature_epoch: StrictStr @dataclass(frozen=True, slots=True) class WeeklyOperationsOwner: """The versioned owner identity independently bound by a receipt.""" user_id: str chat_id: str version: int def __post_init__(self) -> None: if not self.user_id.strip() or not self.chat_id.strip() or self.version < 1: raise WeeklyOperationsAuthorityError("owner") @dataclass(frozen=True, slots=True) class _ReceiptToken: pass _RECEIPT_TOKEN = _ReceiptToken() @dataclass(frozen=True, slots=True) class WeeklyOperationsAuthorityReceipt: """A sealed authority receipt for an explicitly listed customer set.""" candidate_digest: str config_digest: str enabled_customer_keys: tuple[str, ...] owner: WeeklyOperationsOwner consent_digest: str registry_identity: WeeklyOperationsRegistryIdentity issued_at: datetime expires_at: datetime feature_epoch: str _token: _ReceiptToken = field(repr=False, compare=False) def __post_init__(self) -> None: if self._token is not _RECEIPT_TOKEN: raise WeeklyOperationsAuthorityError("receipt source") _require_digest(self.candidate_digest, "candidate digest") _require_digest(self.config_digest, "config digest") _require_digest(self.consent_digest, "consent digest") if not self.enabled_customer_keys or len(set(self.enabled_customer_keys)) != len(self.enabled_customer_keys): raise WeeklyOperationsAuthorityError("enabled customer keys") if any(not customer_key.strip() for customer_key in self.enabled_customer_keys): raise WeeklyOperationsAuthorityError("enabled customer keys") _require_aware(self.issued_at, "issued at") _require_aware(self.expires_at, "expires at") if self.expires_at <= self.issued_at: raise WeeklyOperationsAuthorityError("issue expiry ordering") if _EPOCH.fullmatch(self.feature_epoch) is None: raise WeeklyOperationsAuthorityError("feature epoch") @property def integrity_digest(self) -> str: """Return unkeyed integrity data; this is not authentication.""" payload = { "schema": _AUTHORITY_SCHEMA, "candidate_digest": self.candidate_digest, "config_digest": self.config_digest, "enabled_customer_keys": self.enabled_customer_keys, "owner": [self.owner.user_id, self.owner.chat_id, self.owner.version], "consent_digest": self.consent_digest, "registry_identity_binding_digest": self.registry_identity.binding_digest, "issued_at": self.issued_at.isoformat(), "expires_at": self.expires_at.isoformat(), "feature_epoch": self.feature_epoch, } encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() return hashlib.sha256(encoded).hexdigest() @dataclass(frozen=True, slots=True) class WeeklyOperationsRuntimeContext: """Current facts that must still agree when an authority is consumed.""" candidate_digest: str customer_key: str owner_user_id: str owner_chat_id: str owner_version: int consent_digest: str consent_granted: bool feature_epoch: str now: datetime def __post_init__(self) -> None: _require_digest(self.candidate_digest, "candidate digest") _require_digest(self.consent_digest, "consent digest") if not self.customer_key.strip(): raise WeeklyOperationsAuthorityError("customer key") if not self.owner_user_id.strip() or not self.owner_chat_id.strip() or self.owner_version < 1: raise WeeklyOperationsAuthorityError("owner") if type(self.consent_granted) is not bool or _EPOCH.fullmatch(self.feature_epoch) is None: raise WeeklyOperationsAuthorityError("runtime state") _require_aware(self.now, "runtime time") def parse_weekly_operations_authority(raw: Mapping[str, JsonValue]) -> WeeklyOperationsAuthorityReceipt: """Parse a complete v1.4 receipt before it can reach the authority gate.""" try: parsed = _AuthorityInput.model_validate(raw) except ValidationError as error: raise WeeklyOperationsAuthorityError("receipt shape") from error if parsed.receipt_schema != _AUTHORITY_SCHEMA: raise WeeklyOperationsAuthorityError("receipt schema") return WeeklyOperationsAuthorityReceipt( candidate_digest=parsed.candidate_digest, config_digest=parsed.config_digest, enabled_customer_keys=tuple(parsed.enabled_customer_keys), owner=WeeklyOperationsOwner( user_id=parsed.owner.user_id, chat_id=parsed.owner.chat_id, version=parsed.owner.version, ), consent_digest=parsed.consent_digest, registry_identity=parse_weekly_operations_registry_identity( parsed.registry_identity ), issued_at=_parse_instant(parsed.issued_at, "issued at"), expires_at=_parse_instant(parsed.expires_at, "expires at"), feature_epoch=parsed.feature_epoch, _token=_RECEIPT_TOKEN, ) def weekly_operations_is_authorized( config: WeeklyOperationsConfig, receipt: WeeklyOperationsAuthorityReceipt, context: WeeklyOperationsRuntimeContext, ) -> bool: """Require compiled, configured, authorized, unexpired, current-consent facts.""" return ( config.is_compiled and config.enabled and context.consent_granted and receipt.candidate_digest == context.candidate_digest and receipt.config_digest == config.digest and context.customer_key in receipt.enabled_customer_keys and receipt.owner.user_id == context.owner_user_id and receipt.owner.chat_id == context.owner_chat_id and receipt.owner.version == context.owner_version and receipt.consent_digest == context.consent_digest and receipt.feature_epoch == config.feature_epoch and receipt.feature_epoch == context.feature_epoch and receipt.issued_at <= context.now < receipt.expires_at ) def _parse_instant(value: str, label: str) -> datetime: try: parsed = datetime.fromisoformat(value) except ValueError as error: raise WeeklyOperationsAuthorityError(label) from error _require_aware(parsed, label) return parsed def _require_aware(value: datetime, label: str) -> None: if value.tzinfo is None or value.utcoffset() is None: raise WeeklyOperationsAuthorityError(label) def _require_digest(value: str, label: str) -> None: if _DIGEST.fullmatch(value) is None: raise WeeklyOperationsAuthorityError(label)