"""Independent validation of the gateway's strict staff-membership proof.""" from __future__ import annotations import hashlib import json import stat from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Mapping SCHEMA = "telegram-staff-membership-evidence-v1" MAX_AGE = timedelta(minutes=5) _HEX_FIELDS = ( "candidate_digest", "candidate_core_digest", "candidate_inventory_digest", "hermes_wheel_sha256", "profile_wheel_sha256", "deployment_receipt_sha256", "registry_sha256", "config_sha256", "staff_chat_inventory_sha256", "activation_journal_sha256", ) class StaffMembershipEvidenceError(ValueError): pass def _canonical(value: object) -> bytes: return json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":")).encode() def _digest(value: object) -> str: return hashlib.sha256(_canonical(value)).hexdigest() def _file_digest(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def _private_json(path: Path, label: str) -> dict[str, object]: try: metadata = path.lstat() value = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: raise StaffMembershipEvidenceError(f"{label} is unavailable") from exc if (path.is_symlink() or not stat.S_ISREG(metadata.st_mode) or stat.S_IMODE(metadata.st_mode) != 0o600 or metadata.st_nlink != 1 or not isinstance(value, dict)): raise StaffMembershipEvidenceError(f"{label} is invalid") return value def validate_staff_membership_evidence( path: Path, *, profile_root: Path, registry_path: Path, customer_id: str, customer_user_id: str, now: datetime | None = None, ) -> dict[str, object]: """Validate all current bindings without mutating profile state.""" evidence_path = Path(path) if not evidence_path.is_absolute(): evidence_path = profile_root / evidence_path try: resolved = evidence_path.resolve(strict=True) except (OSError, RuntimeError) as exc: raise StaffMembershipEvidenceError("staff membership evidence is unavailable") from exc if not resolved.is_relative_to(profile_root.resolve()): raise StaffMembershipEvidenceError("staff membership evidence escapes the profile") payload = _private_json(resolved, "staff membership evidence") if payload.get("schema") != SCHEMA or payload.get("phase") != "pre_activation": raise StaffMembershipEvidenceError("staff membership evidence is invalid") body = {key: value for key, value in payload.items() if key != "evidence_sha256"} if payload.get("evidence_sha256") != _digest(body): raise StaffMembershipEvidenceError("staff membership evidence digest is invalid") if payload.get("customer_id") != customer_id or str(payload.get("customer_user_id")) != customer_user_id: raise StaffMembershipEvidenceError("staff membership customer binding is stale") if payload.get("registry_sha256") != _file_digest(registry_path): raise StaffMembershipEvidenceError("staff membership registry binding is stale") config_path = profile_root / "config.yaml" if payload.get("config_sha256") != _file_digest(config_path): raise StaffMembershipEvidenceError("staff membership config binding is stale") for field in _HEX_FIELDS: value = payload.get(field) if (not isinstance(value, str) or len(value) != 64 or any(character not in "0123456789abcdef" for character in value)): raise StaffMembershipEvidenceError(f"staff membership {field} is missing or stale") deployment_value = payload.get("deployment_receipt_path") if not isinstance(deployment_value, str): raise StaffMembershipEvidenceError("staff membership deployment receipt is missing") deployment_path = Path(deployment_value) if not deployment_path.is_absolute(): deployment_path = profile_root / deployment_path deployment = _private_json(deployment_path, "staff membership deployment receipt") if _file_digest(deployment_path) != payload["deployment_receipt_sha256"]: raise StaffMembershipEvidenceError("staff membership deployment receipt is stale") for field in ("candidate_digest", "candidate_core_digest", "candidate_inventory_digest", "hermes_wheel_sha256", "profile_wheel_sha256"): if payload[field] != deployment.get(field): raise StaffMembershipEvidenceError(f"staff membership {field} deployment binding is stale") inventory = payload.get("staff_chat_inventory") if not isinstance(inventory, list) or _digest(inventory) != payload.get("staff_chat_inventory_sha256"): raise StaffMembershipEvidenceError("staff membership inventory is invalid") group_ids: set[object] = set() private_ids: set[object] = set() for row in inventory: if not isinstance(row, Mapping): raise StaffMembershipEvidenceError("staff membership inventory is invalid") if row.get("kind") == "membership_chat": group_ids.add(row.get("chat_id")) elif (row.get("kind") == "private_staff_dm" and row.get("chat_id") == row.get("staff_user_id") and int(str(row.get("chat_id"))) > 0): private_ids.add(row.get("chat_id")) else: raise StaffMembershipEvidenceError("staff membership inventory is invalid") membership = payload.get("membership_results") private = payload.get("private_dm_results") if (not isinstance(membership, list) or {row.get("chat_id") for row in membership if isinstance(row, Mapping)} != group_ids or any(not isinstance(row, Mapping) or row.get("status") not in {"left", "kicked"} for row in membership) or not isinstance(private, list) or {row.get("chat_id") for row in private if isinstance(row, Mapping)} != private_ids or any(not isinstance(row, Mapping) or row.get("identity_separated") is not True or str(row.get("chat_id")) == customer_user_id for row in private)): raise StaffMembershipEvidenceError("customer absence evidence is incomplete") epoch = payload.get("subscription_epoch_id") session = payload.get("bootstrap_session_id") generation = payload.get("bootstrap_generation") if (not isinstance(epoch, str) or not epoch or not isinstance(session, str) or not session or type(generation) is not int or generation < 0): raise StaffMembershipEvidenceError("staff membership lifecycle binding is invalid") try: armed = datetime.fromisoformat(str(payload["subscription_armed_at_utc"])) started = datetime.fromisoformat(str(payload["request_started_at_utc"])) observed = datetime.fromisoformat(str(payload["observed_at_utc"])) except (KeyError, ValueError) as exc: raise StaffMembershipEvidenceError("staff membership timestamps are invalid") from exc current = now or datetime.now(UTC) if (any(value.tzinfo is None for value in (armed, started, observed)) or not armed <= started <= observed <= current or current - observed > MAX_AGE): raise StaffMembershipEvidenceError("staff membership evidence is stale") return { "staff_membership_evidence_sha256": _file_digest(resolved), "staff_chat_inventory_sha256": payload["staff_chat_inventory_sha256"], "membership_subscription_epoch_id": epoch, "staff_membership_evidence_path": str(resolved), } __all__ = ["StaffMembershipEvidenceError", "validate_staff_membership_evidence"]