#!/usr/bin/env python3 """Event-first observer for the exact Task26 replacement consent card.""" from __future__ import annotations import argparse import ctypes import hashlib import json import os import selectors import stat import struct import sys import time from dataclasses import dataclass from pathlib import Path from typing import Final, cast CANDIDATE: Final = "2e0894eac92bc396cc4723bf1f18ebc653b95018dd41574df435941c235da925" WHEEL: Final = "af4a9d0a1ffffb6eb7551c1d6dc2b32853ca6d024332a4f8f5702bbf992f141b" PLAN: Final = "d816dbbd6a27d5826b251d279a9831de096ef7cd79b5fd131ceff7253b7369be" GOLDEN_V6: Final = "8416cd7c83cab5540af2e1e01dbcc5b1bdfd49a59d8ef21ea49512c28d749218" RECOVERY_V6: Final = "a2e15845764e7041957ae4518096fe73f87ec35b57021980cdd4df3ea18ac00d" SESSION_ID: Final = "cb_PCczfFXoI4GjvCxLBs1oRA" SID_HASH: Final = "53b4f95b5b4db9a611128976a7b951bbe22d9619e05b2ff07fadf5b02f9a5380" CUSTOMER_KEY: Final = "task26_live_2e_r2_20260815_8527916639" ACTOR_ID: Final = "8527916639" CHAT_ID: Final = "8527916639" TOPIC_ID: Final = "0" CLAIM_MESSAGE_ID: Final = "158" CARD_MESSAGE_ID: Final = "159" OWNER_ID: Final = "8693203710" CONSENT_RECEIPT_SHA256: Final = "5e3067fe3c865224fc73ff5a9a65058c87960f05d9af53bddccaf6f32dc83939" LEDGER_RELATIVE: Final = Path("data/onboarding/telegram-customer-bootstrap-v1/ledger.json") REGISTRY_RELATIVE: Final = Path("customers/registry.json") SCHEMA: Final = "task26-consent-stage-observer-receipt-v1" RUNTIME_HASHES: Final = { "gateway/platforms/telegram.py": "b41060dea28eb3bbb83217068f5218d5df2c879dba00e73c9ca4e577a6049dad", "gateway/platforms/telegram_customer_bootstrap.py": "145515d5e110dcebb94fcaa554bcee29544b3a75dcfe058fae944ea3b70042b2", "gateway/platforms/nutrition_coaching.py": "07f868adf9fe0565e6eac89fd76b09601adea315e2a122a09ee748614d912c2f", } VENV_SITE: Final = Path("/home/cube/projects/richard/hermes-agent/.venv/lib/python3.12/site-packages") PROFILE_MODULE_HASHES: Final = { "workspace/checkin_cli/checkin_cli/customer_admin.py": "fe4a8725be379e47b1b2fab3029147705d7def887d94af747e5057c93322c40f", "workspace/checkin_cli/checkin_cli/nutrition_onboarding.py": "d1f848d1fa8bc71e018f45c83e8e19e93671dfa83157d3feff15b8f7ec0f1d8a", } IN_MOVED_TO = 0x00000080 IN_Q_OVERFLOW = 0x00004000 IN_IGNORED = 0x00008000 EVENT = struct.Struct("iIII") PRIVATE_FILE = 0o600 PRIVATE_DIR = 0o700 O_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0) class ObserverError(RuntimeError): """Fail-closed consent observer violation.""" class ObserverTimeout(ObserverError): """Bounded event wait expired.""" @dataclass(frozen=True) class Baseline: registry: dict[str, object] ledger: dict[str, object] session: dict[str, object] registry_sha256: str ledger_sha256: str expected_consent_receipt_sha256: str def canonical_bytes(value: object) -> bytes: return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode() def digest_json(value: object) -> str: return hashlib.sha256(canonical_bytes(value)).hexdigest() def sha256_file(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def _private_regular(path: Path, label: str) -> None: if path.is_symlink(): raise ObserverError(f"{label} is a symlink") info = path.stat() if not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() or info.st_nlink != 1: raise ObserverError(f"{label} is not one owner-held regular file") if stat.S_IMODE(info.st_mode) != PRIVATE_FILE: raise ObserverError(f"{label} is not private 0600") def _private_dir(path: Path, label: str) -> None: if path.is_symlink(): raise ObserverError(f"{label} is a symlink") info = path.stat() if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid(): raise ObserverError(f"{label} is not an owner-held directory") if stat.S_IMODE(info.st_mode) != PRIVATE_DIR: raise ObserverError(f"{label} is not private 0700") def _load_json(path: Path, label: str) -> tuple[dict[str, object], str]: _private_regular(path, label) fd = os.open(path, os.O_RDONLY | os.O_CLOEXEC | O_NOFOLLOW) try: info = os.fstat(fd) raw = os.read(fd, info.st_size + 1) if len(raw) != info.st_size or info.st_size > 16 * 1024 * 1024: raise ObserverError(f"{label} changed during stable read") finally: os.close(fd) try: value = json.loads(raw) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise ObserverError(f"{label} is invalid JSON") from exc if not isinstance(value, dict): raise ObserverError(f"{label} is not a JSON object") return value, hashlib.sha256(raw).hexdigest() def _session(ledger: dict[str, object]) -> dict[str, object]: if set(ledger) != {"schema", "sessions", "digest"} or ledger.get("schema") != "telegram-customer-bootstrap-v1": raise ObserverError("bootstrap ledger schema drift") rows = ledger.get("sessions") if not isinstance(rows, list) or ledger.get("digest") != digest_json({"schema": ledger["schema"], "sessions": rows}): raise ObserverError("bootstrap ledger digest drift") matches = [row for row in rows if isinstance(row, dict) and row.get("session_id") == SESSION_ID] if len(matches) != 1: raise ObserverError("exact bootstrap session is unavailable") return matches[0] def _customer(registry: dict[str, object]) -> dict[str, object]: customers = registry.get("customers") if not isinstance(customers, list): raise ObserverError("registry schema drift") matches = [row for row in customers if isinstance(row, dict) and row.get("customer_key") == CUSTOMER_KEY] if len(matches) != 1: raise ObserverError("exact registry customer is unavailable") return matches[0] def _expected_receipt_digest() -> str: return digest_json({ "chat_id": None, "topic_id": None, "customer_user_id": int(ACTOR_ID), "consent_card_message_id": int(CARD_MESSAGE_ID), "bootstrap_generation": 5, "notice_version": "privacy-v1", }) def validate_runtime(profile: Path) -> None: for relative, expected in RUNTIME_HASHES.items(): if sha256_file(VENV_SITE / relative) != expected: raise ObserverError(f"loaded runtime hash drift: {relative}") for relative, expected in PROFILE_MODULE_HASHES.items(): if sha256_file(profile / relative) != expected: raise ObserverError(f"profile authority hash drift: {relative}") def arm_baseline(profile: Path, *, enforce_runtime: bool = True) -> Baseline: if not profile.is_absolute(): profile = profile.resolve() _private_dir(profile, "profile root") _private_dir((profile / REGISTRY_RELATIVE).parent, "registry parent") _private_dir((profile / LEDGER_RELATIVE).parent, "bootstrap authority parent") if enforce_runtime: validate_runtime(profile) registry, registry_hash = _load_json(profile / REGISTRY_RELATIVE, "customer registry") ledger, ledger_hash = _load_json(profile / LEDGER_RELATIVE, "bootstrap ledger") row = _session(ledger) draft = row.get("customer_draft") if not isinstance(draft, dict): raise ObserverError("current customer draft authority drift") claim = [{"role": "customer", "user_id": ACTOR_ID, "chat_id": CHAT_ID, "topic_id": TOPIC_ID, "message_id": CLAIM_MESSAGE_ID}] if ( row.get("sid_hash") != SID_HASH or draft.get("customer_key") != CUSTOMER_KEY or draft.get("customer_user_id") != ACTOR_ID or row.get("state") != "AWAITING_CONSENT" or row.get("generation") != 4 or row.get("owner_id") != OWNER_ID or row.get("role_claims") != claim or row.get("consent_publication_attempt") != 1 or row.get("consent_card_message_id") != CARD_MESSAGE_ID or row.get("recovery_attempts") != [] or row.get("failure_code") is not None ): raise ObserverError("current consent card/session/claim authority drift") customer = _customer(registry) if ( registry.get("owner") != {"user_id": OWNER_ID, "chat_id": OWNER_ID, "topic_id": "0"} or customer.get("enabled") is not False or customer.get("telegram") != {"user_id": ACTOR_ID, "chat_id": CHAT_ID, "topic_id": TOPIC_ID} or customer.get("ai_processing_consent") != {"granted": False, "recorded_on": None, "notice_version": None} ): raise ObserverError("current disabled registry consent/route authority drift") expected = _expected_receipt_digest() if expected != CONSENT_RECEIPT_SHA256: raise ObserverError("sealed exact callback receipt digest drift") return Baseline(registry, ledger, row, registry_hash, ledger_hash, expected) def validate_registry_transition(before: dict[str, object], after: dict[str, object]) -> str: old_customer, new_customer = _customer(before), _customer(after) expected = cast(dict[str, object], json.loads(json.dumps(old_customer))) consent = new_customer.get("ai_processing_consent") if not isinstance(consent, dict): raise ObserverError("registry consent commit is invalid") expected["ai_processing_consent"] = { "granted": True, "recorded_on": consent.get("recorded_on"), "notice_version": "privacy-v1", } recorded = consent.get("recorded_on") if isinstance(consent, dict) else None if not isinstance(recorded, str) or len(recorded) != 10: raise ObserverError("registry consent commit date is invalid") expected_registry = json.loads(json.dumps(before)) index = next(i for i, row in enumerate(expected_registry["customers"]) if row.get("customer_key") == CUSTOMER_KEY) expected_registry["customers"][index] = expected if after != expected_registry: raise ObserverError("registry changed beyond exact privacy-v1 consent grant") return recorded def validate_ledger_transition(before: dict[str, object], after: dict[str, object]) -> dict[str, object]: old, new = _session(before), _session(after) expected = json.loads(json.dumps(old)) expected.update({ "state": "AWAITING_ACTIVATION", "generation": 5, "consent_card_message_id": None, "updated_at": new.get("updated_at"), }) if not isinstance(new.get("updated_at"), str) or new == old: raise ObserverError("bootstrap consent commit timestamp is invalid") if new != expected: raise ObserverError("bootstrap transition is not exact generation 4 to 5 consent commit") expected_ledger = json.loads(json.dumps(before)) rows = expected_ledger["sessions"] index = next(i for i, row in enumerate(rows) if row.get("session_id") == SESSION_ID) rows[index] = expected expected_ledger["digest"] = digest_json({"schema": expected_ledger["schema"], "sessions": rows}) if after != expected_ledger: raise ObserverError("bootstrap ledger changed beyond exact consent session") return new def _inotify_init() -> int: libc = ctypes.CDLL(None, use_errno=True) fd = libc.inotify_init1(os.O_CLOEXEC | os.O_NONBLOCK) if fd < 0: raise ObserverError(f"inotify_init1 failed: errno {ctypes.get_errno()}") return int(fd) def _add_watch(fd: int, path: Path) -> int: libc = ctypes.CDLL(None, use_errno=True) watch = libc.inotify_add_watch(fd, os.fsencode(path), IN_MOVED_TO) if watch < 0: raise ObserverError(f"inotify_add_watch failed: errno {ctypes.get_errno()}") return int(watch) def _events(raw: bytes) -> list[tuple[int, int, str]]: result = [] offset = 0 while offset + EVENT.size <= len(raw): watch, mask, _cookie, length = EVENT.unpack_from(raw, offset) offset += EVENT.size name = raw[offset:offset + length].split(b"\0", 1)[0].decode("utf-8", "strict") offset += length if mask & IN_Q_OVERFLOW: raise ObserverError("inotify queue overflow") if mask & IN_IGNORED: raise ObserverError("authority watch invalidated") result.append((watch, mask, name)) return result class ConsentWatcher: def __init__(self, profile: Path, *, enforce_runtime: bool) -> None: self.profile = profile.resolve() self.enforce_runtime = enforce_runtime self.fd = -1 self.registry_watch = -1 self.ledger_watch = -1 self.baseline: Baseline | None = None def arm(self) -> Baseline: self.fd = _inotify_init() self.registry_watch = _add_watch(self.fd, (self.profile / REGISTRY_RELATIVE).parent) self.ledger_watch = _add_watch(self.fd, (self.profile / LEDGER_RELATIVE).parent) self.baseline = arm_baseline(self.profile, enforce_runtime=self.enforce_runtime) try: queued = os.read(self.fd, 65536) except BlockingIOError: queued = b"" for watch, _mask, name in _events(queued): if (watch, name) in {(self.registry_watch, REGISTRY_RELATIVE.name), (self.ledger_watch, LEDGER_RELATIVE.name)}: raise ObserverError("authority changed across subscription baseline") return self.baseline def wait(self, timeout: float) -> tuple[str, dict[str, object], str, str]: if self.baseline is None: raise ObserverError("observer is not armed") deadline = time.monotonic() + timeout registry_seen = False recorded_on = "" registry_after_hash = "" selector = selectors.DefaultSelector() selector.register(self.fd, selectors.EVENT_READ) try: while True: remaining = deadline - time.monotonic() if remaining <= 0: raise ObserverTimeout("bounded monotonic deadline expired") ready = selector.select(remaining) if not ready: raise ObserverTimeout("bounded monotonic deadline expired") for watch, _mask, name in _events(os.read(self.fd, 65536)): if watch == self.registry_watch and name == REGISTRY_RELATIVE.name: after, registry_after_hash = _load_json(self.profile / REGISTRY_RELATIVE, "customer registry") if after == self.baseline.registry and registry_seen: continue recorded_on = validate_registry_transition(self.baseline.registry, after) registry_seen = True elif watch == self.ledger_watch and name == LEDGER_RELATIVE.name: if not registry_seen: raise ObserverError("bootstrap consent commit preceded registry consent commit") after, ledger_after_hash = _load_json(self.profile / LEDGER_RELATIVE, "bootstrap ledger") row = validate_ledger_transition(self.baseline.ledger, after) return recorded_on, row, registry_after_hash, ledger_after_hash finally: selector.close() def close(self) -> None: if self.fd >= 0: os.close(self.fd) self.fd = -1 def bindings() -> dict[str, object]: return {"candidate": CANDIDATE, "wheel_sha256": WHEEL, "plan_sha256": PLAN, "runbook_sha256": {"golden_v6": GOLDEN_V6, "recovery_v6": RECOVERY_V6}} def callback_authority() -> dict[str, str]: return {"actor_id": ACTOR_ID, "chat_id": CHAT_ID, "topic_id": TOPIC_ID, "message_id": CARD_MESSAGE_ID, "claim_message_id": CLAIM_MESSAGE_ID, "session_id": SESSION_ID, "sid_hash": SID_HASH, "customer_key": CUSTOMER_KEY, "action": "canonical_customer_consent_accept", "notice_version": "privacy-v1"} def _secure_write(path: Path, value: dict[str, object]) -> None: path = path.resolve() _private_dir(path.parent, "receipt parent") flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC | O_NOFOLLOW fd = os.open(path, flags, PRIVATE_FILE) try: raw = canonical_bytes(value) + b"\n" os.write(fd, raw) os.fsync(fd) if stat.S_IMODE(os.fstat(fd).st_mode) != PRIVATE_FILE: raise ObserverError("receipt mode drift") finally: os.close(fd) def _ready_wire(fd: int | None, value: dict[str, object]) -> None: if fd is not None: os.write(fd, canonical_bytes(value)) os.close(fd) def ready_receipt(baseline: Baseline) -> dict[str, object]: return {"schema": SCHEMA, "status": "READY_BEFORE_CONSENT", "event_subscription": "inotify-established-before-human-callback", "safe_window_seconds": 1800, "bindings": bindings(), "callback_authority": callback_authority(), "expected_consent_receipt_sha256": baseline.expected_consent_receipt_sha256, "baseline": {"registry_sha256": baseline.registry_sha256, "bootstrap_ledger_sha256": baseline.ledger_sha256, "bootstrap_state": "AWAITING_CONSENT", "bootstrap_generation": 4}, "privacy": "no-raw-update-no-token-no-network"} def tree_digest(root: Path) -> str: values = [] for path in sorted(item for item in root.rglob("*") if item.is_file()): values.append((str(path.relative_to(root)), sha256_file(path))) return digest_json(values) def run(args: argparse.Namespace) -> dict[str, object]: watcher = ConsentWatcher(args.profile, enforce_runtime=not args.skip_runtime_hashes) try: baseline = watcher.arm() ready = ready_receipt(baseline) if args.mode == "arm-only": _secure_write(args.receipt, ready) return ready _secure_write(args.ready, ready) _ready_wire(args.ready_fd, ready) print(canonical_bytes(ready).decode(), flush=True) recorded_on, row, registry_after, ledger_after = watcher.wait(args.timeout) receipt: dict[str, object] = {"schema": SCHEMA, "status": "PASS_CONSENT_COMMITTED", "event_subscription": "inotify-before-human-callback", "bindings": bindings(), "callback_authority": callback_authority(), "commit": {"registry_consent": "granted/privacy-v1", "recorded_on": recorded_on, "bootstrap_transition": "AWAITING_CONSENT:g4->AWAITING_ACTIVATION:g5", "consent_card_retired": row.get("consent_card_message_id") is None, "registry_sha256_after": registry_after, "bootstrap_ledger_sha256_after": ledger_after, "consent_receipt_sha256": baseline.expected_consent_receipt_sha256}, "privacy": "redacted-no-raw-update-no-token-no-network"} _secure_write(args.receipt, receipt) return receipt finally: watcher.close() def parser() -> argparse.ArgumentParser: result = argparse.ArgumentParser(description=__doc__) sub = result.add_subparsers(dest="mode", required=True) arm = sub.add_parser("arm-only") arm.add_argument("--profile", type=Path, required=True) arm.add_argument("--receipt", type=Path, required=True) arm.add_argument("--skip-runtime-hashes", action="store_true", help=argparse.SUPPRESS) observe = sub.add_parser("observe") observe.add_argument("--profile", type=Path, required=True) observe.add_argument("--ready", type=Path, required=True) observe.add_argument("--receipt", type=Path, required=True) observe.add_argument("--timeout", type=float, required=True) observe.add_argument("--ready-fd", type=int) observe.add_argument("--skip-runtime-hashes", action="store_true", help=argparse.SUPPRESS) return result def main(argv: list[str] | None = None) -> int: args = parser().parse_args(argv) try: if args.mode == "observe" and (args.timeout <= 0 or args.timeout > 1800): raise ObserverError("timeout must be in (0, 1800]") receipt = run(args) except ObserverTimeout as exc: print(f"FAIL: {exc}", file=sys.stderr) return 3 except (ObserverError, OSError, ValueError, KeyError) as exc: print(f"FAIL: {exc}", file=sys.stderr) return 2 print(canonical_bytes(receipt).decode()) return 0 if __name__ == "__main__": raise SystemExit(main())