#!/usr/bin/env python3 """Permission-sealed preparation/observation for one Task26 logical invite. This tool has no Telegram, provider, service, archive, Git, or JSON-mutation path. The sole profile writers it can call are the shipped candidate store's atomic prepare and expire APIs. Receipts are redacted; only the handoff contains a token. """ from __future__ import annotations import argparse import ctypes import hashlib import importlib import importlib.util import json import os import select import stat import struct import sys import time import zipfile from pathlib import Path from typing import Any from urllib.parse import parse_qs, urlparse CANDIDATE = "2e0894eac92bc396cc4723bf1f18ebc653b95018dd41574df435941c235da925" WHEEL_SHA256 = "af4a9d0a1ffffb6eb7551c1d6dc2b32853ca6d024332a4f8f5702bbf992f141b" PLAN_SHA256 = "7ace03c6dad33d2fc3ef223621cbca68a150fde8429932138e252fb8498ac582" GOLDEN_SHA256 = "1f55b8f967c6564113574d41bf1c15ce977d35c38b5687f1623ba37f235a7bdb" RECOVERY_SHA256 = "f4f2d2347b4b68a080b9c437a813accbef8c4bcc471cbf21d67ffff31e56f89b" ACTOR_ID = "8527916639" OWNER_ID = "8693203710" BOT_USERNAME = "dual_coach_pilot_test_bot" APPROVAL_EVENT_SHA256 = "23845bda574c085e16b6d316e5e84e54b3bc6db3a8803c6b3b93ed7c4b86b049" MODULE_MEMBER = "gateway/platforms/telegram_customer_bootstrap.py" MODULE_SHA256 = "145515d5e110dcebb94fcaa554bcee29544b3a75dcfe058fae944ea3b70042b2" TERMINAL = frozenset({"ACTIVE", "CANCELLED", "EXPIRED", "FAILED"}) IN_CLOSE_WRITE = 0x00000008 IN_MOVED_TO = 0x00000080 IN_CREATE = 0x00000100 O_FLAGS = os.O_CLOEXEC | getattr(os, "O_NOFOLLOW", 0) HERE = Path(__file__).resolve().parent ROOT = HERE.parents[3] WHEEL = ROOT / ".omo/evidence/task26" / ( "task26-repaired-archive-successor-" + CANDIDATE ) / "artifacts/hermes_agent-0.17.0-py3-none-any.whl" PLAN = ROOT / ".omo/plans/dualcoach-production-readiness.md" GOLDEN = ROOT / ".omo/evidence/task26/task26-runbook-rebind-2e0894ea-st_01a0054e/task26-golden-path-2e0894ea.md" RECOVERY = ROOT / ".omo/evidence/task26/task26-runbook-rebind-2e0894ea-st_01a0054e/task26-recovery-runbook-2e0894ea.md" APPROVAL_RECEIPT = HERE / "approval-receipt.json" class HarnessError(RuntimeError): pass def sha_bytes(raw: bytes) -> str: return hashlib.sha256(raw).hexdigest() def sha_file(path: Path) -> str: return sha_bytes(read_regular(path, private=False)) def canonical(value: Any) -> bytes: return (json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + "\n").encode() def stable(info: os.stat_result) -> tuple[int, ...]: return (info.st_dev, info.st_ino, info.st_mode, info.st_nlink, info.st_uid, info.st_size, info.st_mtime_ns, info.st_ctime_ns) def read_regular(path: Path, *, private: bool = True) -> bytes: fd = os.open(path, os.O_RDONLY | O_FLAGS) try: before = os.fstat(fd) if not stat.S_ISREG(before.st_mode) or before.st_nlink != 1 or before.st_uid != os.getuid(): raise HarnessError(f"unsafe regular file: {path}") if private and stat.S_IMODE(before.st_mode) not in {0o400, 0o500, 0o600, 0o700}: raise HarnessError(f"file is not private: {path}") chunks: list[bytes] = [] while chunk := os.read(fd, 1024 * 1024): chunks.append(chunk) if stable(before) != stable(os.fstat(fd)): raise HarnessError(f"file changed during read: {path}") return b"".join(chunks) finally: os.close(fd) def read_json(path: Path, *, private: bool = True) -> dict[str, Any]: try: value = json.loads(read_regular(path, private=private)) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise HarnessError(f"invalid JSON: {path}") from exc if not isinstance(value, dict): raise HarnessError(f"JSON root is not an object: {path}") return value def exclusive_write(path: Path, value: dict[str, Any]) -> None: path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) if path.parent.is_symlink(): raise HarnessError("receipt parent cannot be a symlink") raw = canonical(value) fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | O_FLAGS, 0o600) try: os.fchmod(fd, 0o600) offset = 0 while offset < len(raw): offset += os.write(fd, raw[offset:]) os.fsync(fd) finally: os.close(fd) directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY | O_FLAGS) try: os.fsync(directory) finally: os.close(directory) def verify_seal(path: Path, mode: str) -> dict[str, Any]: seal = read_json(path) expected = { "schema": "task26-invite-harness-permission-seal-v1", "candidate_digest": CANDIDATE, "wheel_sha256": WHEEL_SHA256, "plan_sha256": PLAN_SHA256, "golden_runbook_sha256": GOLDEN_SHA256, "recovery_runbook_sha256": RECOVERY_SHA256, "actor_id": ACTOR_ID, "owner_id": OWNER_ID, "bot_username": BOT_USERNAME, "approval_event_sha256": APPROVAL_EVENT_SHA256, "approval_receipt_sha256": sha_file(APPROVAL_RECEIPT), "harness_sha256": sha_file(Path(__file__).resolve()), } for key, wanted in expected.items(): if seal.get(key) != wanted: raise HarnessError(f"permission seal mismatch: {key}") allowed = seal.get("allowed_modes") if not isinstance(allowed, list) or mode not in allowed: raise HarnessError(f"mode is not permission sealed: {mode}") return seal def loaded_byte_proof() -> dict[str, str]: pins = ((WHEEL, WHEEL_SHA256), (PLAN, PLAN_SHA256), (GOLDEN, GOLDEN_SHA256), (RECOVERY, RECOVERY_SHA256)) for path, wanted in pins: if sha_file(path) != wanted: raise HarnessError(f"bound evidence hash mismatch: {path.name}") with zipfile.ZipFile(WHEEL) as archive: raw = archive.read(MODULE_MEMBER) if sha_bytes(raw) != MODULE_SHA256: raise HarnessError("wheel bootstrap member hash mismatch") spec = importlib.util.find_spec("gateway.platforms.telegram_customer_bootstrap") if spec is None or spec.origin is None or sha_file(Path(spec.origin)) != MODULE_SHA256: raise HarnessError("loaded bootstrap module bytes are not the shipped candidate bytes") return { "wheel_sha256": WHEEL_SHA256, "wheel_module_sha256": MODULE_SHA256, "loaded_module_sha256": MODULE_SHA256, "loaded_byte_match": "PASS", } def bootstrap_path(profile: Path) -> Path: return profile / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json" def validate_ledger_document(path: Path) -> dict[str, Any]: value = read_json(path) if set(value) != {"schema", "sessions", "digest"} or value.get("schema") != "telegram-customer-bootstrap-v1": raise HarnessError("customer bootstrap ledger shape/schema mismatch") sessions = value.get("sessions") if not isinstance(sessions, list): raise HarnessError("customer bootstrap sessions are invalid") payload = {"schema": value["schema"], "sessions": sessions} candidate_bytes = json.dumps( payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False ).encode("utf-8") if value.get("digest") != sha_bytes(candidate_bytes): raise HarnessError("customer bootstrap ledger digest mismatch") ids: list[str] = [] hashes: list[str] = [] for item in sessions: if not isinstance(item, dict): raise HarnessError("customer bootstrap session is invalid") sid = item.get("session_id") sid_hash = item.get("sid_hash") if not isinstance(sid, str) or not isinstance(sid_hash, str): raise HarnessError("customer bootstrap identity is invalid") ids.append(sid) hashes.append(sid_hash) if len(set(ids)) != len(ids) or len(set(hashes)) != len(hashes): raise HarnessError("duplicate bootstrap session/token authority") return value def empty_list(path: Path, field: str) -> None: value = read_json(path) if value.get(field) not in ([], {}): raise HarnessError(f"existing customer authority: {path.relative_to(path.parents[2])}") def preflight(profile: Path, *, customer_key: str | None = None) -> tuple[dict[str, Any], dict[str, Any]]: if profile.is_symlink() or not profile.is_dir() or profile.stat().st_uid != os.getuid(): raise HarnessError("unsafe profile root") ledger_path = bootstrap_path(profile) ledger = validate_ledger_document(ledger_path) sessions = ledger["sessions"] if any(item.get("state") not in TERMINAL for item in sessions): raise HarnessError("nonterminal prior bootstrap session") if any(item.get("role_claims") or item.get("recovery_attempts") for item in sessions): raise HarnessError("open claim/recovery authority in prior bootstrap session") if customer_key is not None and any( isinstance(item.get("customer_draft"), dict) and item["customer_draft"].get("customer_key") == customer_key for item in sessions ): raise HarnessError("customer key is not new") empty_list(profile / "customers/registry.json", "customers") empty_list(profile / "data/onboarding/telegram-room-bootstrap-v1/ledger.json", "sessions") empty_list(profile / "data/owner-actions/customer-service-state.json", "states") empty_list(profile / "data/onboarding/telegram-publication-outbox-v1/ledger.json", "records") empty_list(profile / "data/onboarding/telegram-publication-outbox-v1/emergency.json", "records") onboarding_entries = {item.name for item in (profile / "data/onboarding").iterdir()} expected_entries = { "telegram-customer-bootstrap-v1", "telegram-publication-outbox-v1", "telegram-room-bootstrap-v1", } if onboarding_entries != expected_entries: raise HarnessError("existing or unknown onboarding customer authority") customer_root = profile / "data/customers" if customer_root.exists() and any(customer_root.iterdir()): raise HarnessError("existing data/customers authority") gateway = read_json(profile / "gateway_state.json") platforms = gateway.get("platforms") if not isinstance(platforms, dict) or not isinstance(platforms.get("telegram"), dict): raise HarnessError("gateway state topology is invalid") telegram = platforms["telegram"] if gateway.get("gateway_state") != "stopped" or telegram.get("state") != "disconnected": raise HarnessError("gateway must be stopped and Telegram disconnected") if (profile / "gateway.lock").exists(): raise HarnessError("gateway.lock must be absent") if ACTOR_ID == OWNER_ID: raise HarnessError("private-DM actor cannot be owner/staff") proof = loaded_byte_proof() return ledger, { "status": "PASS", "candidate_digest": CANDIDATE, "actor_id": ACTOR_ID, "owner_id": OWNER_ID, "actor_not_owner_staff": True, "private_dm_required": True, "prior_session_count": len(sessions), "all_prior_sessions_terminal": True, "existing_registry_customer_authorities": 0, "existing_room_customer_authorities": 0, "existing_service_customer_authorities": 0, "existing_onboarding_customer_authorities": 0, "gateway_stopped_telegram_disconnected": True, "open_recovery_authorities": 0, "loaded_byte_proof": proof, "prohibited_operations": [ "network", "Telegram/getUpdates", "drop updates", "cursor edits", "archive restore/prepopulation", "direct JSON mutation", "replay", "forced publication", "service action", "provider action", "Git", ], } class Inotify: def __init__(self, directory: Path) -> None: libc = ctypes.CDLL(None, use_errno=True) self._libc = libc self.fd = libc.inotify_init1(os.O_CLOEXEC) if self.fd < 0: raise OSError(ctypes.get_errno(), "inotify_init1") self.watch = libc.inotify_add_watch( self.fd, os.fsencode(directory), IN_CLOSE_WRITE | IN_MOVED_TO | IN_CREATE ) if self.watch < 0: error = ctypes.get_errno() os.close(self.fd) raise OSError(error, f"inotify_add_watch:{directory}") def close(self) -> None: os.close(self.fd) def await_name(self, name: bytes, timeout: float) -> None: deadline = time.monotonic() + timeout while True: # event-driven only; no sleep or state polling remaining = deadline - time.monotonic() if remaining <= 0: raise HarnessError("bounded filesystem event timeout") readable, _, _ = select.select([self.fd], [], [], remaining) if not readable: raise HarnessError("bounded filesystem event timeout") payload = os.read(self.fd, 65536) offset = 0 while offset + 16 <= len(payload): _watch, _mask, _cookie, length = struct.unpack_from("iIII", payload, offset) observed = payload[offset + 16:offset + 16 + length].split(b"\0", 1)[0] offset += 16 + length if observed == name: return def import_api() -> tuple[Any, Any, Any]: module = importlib.import_module("gateway.platforms.telegram_customer_bootstrap") return ( module.RoomBootstrapStore, module.load_customer_draft, module.room_bootstrap_state_dir, ) def base_receipt(mode: str) -> dict[str, Any]: return { "schema": f"task26-invite-harness-{mode}-receipt-v1", "mode": mode, "candidate_digest": CANDIDATE, "wheel_sha256": WHEEL_SHA256, "plan_sha256": PLAN_SHA256, "golden_runbook_sha256": GOLDEN_SHA256, "recovery_runbook_sha256": RECOVERY_SHA256, "approval_receipt_sha256": sha_file(APPROVAL_RECEIPT), "approval_event_sha256": APPROVAL_EVENT_SHA256, "privacy": "redacted; no raw token, link, Telegram content, or customer answers", } def dry_run(args: argparse.Namespace) -> dict[str, Any]: try: _ledger, result = preflight(args.profile) receipt = {**base_receipt("dry-run"), **result, "mutations": 0, "ready_for_one_invite": True} except HarnessError as exc: receipt = {**base_receipt("dry-run"), "status": "FAIL", "mutations": 0, "ready_for_one_invite": False, "blocker": str(exc)} exclusive_write(args.receipt, receipt) return receipt def verify(args: argparse.Namespace) -> dict[str, Any]: try: ledger, result = preflight(args.profile) receipt = {**base_receipt("verify"), **result, "ledger_sha256": sha_file(bootstrap_path(args.profile)), "ledger_session_count": len(ledger["sessions"]), "mutations": 0, "ready_for_one_invite": True} except HarnessError as exc: receipt = {**base_receipt("verify"), "status": "FAIL", "mutations": 0, "ready_for_one_invite": False, "blocker": str(exc)} exclusive_write(args.receipt, receipt) return receipt def prepare(args: argparse.Namespace) -> dict[str, Any]: if args.handoff.exists() or args.receipt.exists(): raise HarnessError("one-use handoff/receipt path already exists") RoomBootstrapStore, load_customer_draft, room_bootstrap_state_dir = import_api() draft = load_customer_draft(args.draft) before, result = preflight(args.profile, customer_key=draft.customer_key) state_dir = room_bootstrap_state_dir(args.profile) if not state_dir.is_dir(): raise HarnessError("current-layout bootstrap state directory must already exist") watch = Inotify(state_dir) try: # Subscription is complete before the one and only canonical prepare call. store = RoomBootstrapStore(state_dir) prepared = store.prepare_rehearsal_customer_invite( draft, bot_username=BOT_USERNAME, owner_id=OWNER_ID ) watch.await_name(b"ledger.json", 5.0) finally: watch.close() query = parse_qs(urlparse(prepared.customer_link).query, strict_parsing=True) tokens = query.get("start") if not isinstance(tokens, list) or len(tokens) != 1 or not tokens[0].startswith("rc1_"): raise HarnessError("candidate returned an invalid private handoff") token = tokens[0] sid = token[4:] session = prepared.session after = validate_ledger_document(bootstrap_path(args.profile)) if len(after["sessions"]) != len(before["sessions"]) + 1: raise HarnessError("prepare did not create exactly one session") if session.sid_hash != sha_bytes(sid.encode("ascii")): raise HarnessError("candidate SID hash relation mismatch") if sum(item.get("session_id") == session.session_id for item in after["sessions"]) != 1: raise HarnessError("prepared session cardinality mismatch") handoff = { "schema": "task26-private-invite-handoff-v1", "session_id": session.session_id, "start_token": token, "customer_link": prepared.customer_link, "expires_at": prepared.expires_at.isoformat(), "actor_id": ACTOR_ID, "instruction": "Private DM only: actor opens this link and presses Start once.", } # This O_EXCL 0600 artifact is the sole publication boundary and raw-token location. exclusive_write(args.handoff, handoff) receipt = { **base_receipt("prepare"), **result, "status": "PASS_PREPARED", "prepare_invocations": 1, "event_subscription_before_prepare": True, "filesystem_event": "IN_MOVED_TO:ledger.json", "session_id": session.session_id, "session_id_sha256": sha_bytes(session.session_id.encode("ascii")), "sid_hash": session.sid_hash, "start_token_sha256": sha_bytes(token.encode("ascii")), "customer_key_sha256": sha_bytes(draft.customer_key.encode("utf-8")), "state": session.state.value, "generation": session.generation, "expires_at": prepared.expires_at.isoformat(), "ledger_sha256": sha_file(bootstrap_path(args.profile)), "handoff_sha256": sha_file(args.handoff), "handoff_mode": "0600", "raw_token_locations": [str(args.handoff)], "rollback": "prohibited after handoff publication; canonical expiry only", } exclusive_write(args.receipt, receipt) if args.evidence_root is not None: occurrences = [] for root, dirs, files in os.walk(args.evidence_root, followlinks=False): dirs.sort() files.sort() for name in files: path = Path(root) / name if path.is_symlink() or not path.is_file(): continue try: raw = read_regular(path, private=True) except (HarnessError, PermissionError): continue if token.encode("ascii") in raw: occurrences.append(path.resolve()) if occurrences != [args.handoff.resolve()]: raise HarnessError("raw token escaped the single private handoff artifact") return receipt def observe(args: argparse.Namespace) -> dict[str, Any]: ledger, _result = preflight_observe(args.profile, args.session_id, args.sid_hash) state_dir = bootstrap_path(args.profile).parent watch = Inotify(state_dir) ready = {**base_receipt("observe-ready"), "status": "READY", "session_id_sha256": sha_bytes(args.session_id.encode("ascii")), "sid_hash": args.sid_hash, "event_subscription_before_claim": True, "timeout_seconds": args.timeout_seconds} exclusive_write(args.ready, ready) if args.ready_fd is not None: os.write(args.ready_fd, b"1") os.close(args.ready_fd) try: watch.await_name(b"ledger.json", args.timeout_seconds) finally: watch.close() after = validate_ledger_document(bootstrap_path(args.profile)) matches = [item for item in after["sessions"] if item.get("session_id") == args.session_id] if len(matches) != 1: raise HarnessError("observed session identity cardinality mismatch") item = matches[0] claims = item.get("role_claims") if not isinstance(claims, list) or len(claims) != 1: raise HarnessError("accepted Start/claim cardinality is not exactly one") claim = claims[0] if not isinstance(claim, dict) or claim.get("role") != "customer" or claim.get("user_id") != ACTOR_ID: raise HarnessError("claim actor/role mismatch") if claim.get("chat_id") != ACTOR_ID or claim.get("topic_id") != "0" or item.get("owner_id") == ACTOR_ID: raise HarnessError("claim is not an actor-only private DM") if item.get("state") not in {"REGISTERING", "AWAITING_CONSENT", "AWAITING_ACTIVATION", "ACTIVE"}: raise HarnessError("claim did not enter an accepted canonical state") receipt = { **base_receipt("observe"), "status": "PASS_CLAIM_OBSERVED", "event_subscription_before_claim": True, "filesystem_event": "IN_MOVED_TO:ledger.json", "session_id": args.session_id, "session_id_sha256": sha_bytes(args.session_id.encode("ascii")), "sid_hash": args.sid_hash, "actor_id": ACTOR_ID, "private_dm": True, "accepted_start_claim_count": 1, "state": item["state"], "generation": item.get("generation"), "ledger_sha256": sha_file(bootstrap_path(args.profile)), "raw_getUpdates": False, "drop_pending_updates": False, "cursor_edits": False, } exclusive_write(args.receipt, receipt) return receipt def preflight_observe(profile: Path, session_id: str, sid_hash: str) -> tuple[dict[str, Any], dict[str, Any]]: loaded_byte_proof() ledger = validate_ledger_document(bootstrap_path(profile)) matches = [item for item in ledger["sessions"] if item.get("session_id") == session_id] if len(matches) != 1 or matches[0].get("sid_hash") != sid_hash: raise HarnessError("prepared observation binding mismatch") item = matches[0] if item.get("state") != "PREPARED" or item.get("role_claims") or item.get("recovery_attempts"): raise HarnessError("observation requires one clean PREPARED session") if sum(one.get("state") not in TERMINAL for one in ledger["sessions"]) != 1: raise HarnessError("duplicate/nonterminal bootstrap authority") return ledger, item def expire(args: argparse.Namespace) -> dict[str, Any]: RoomBootstrapStore, _load_customer_draft, room_bootstrap_state_dir = import_api() preflight_observe(args.profile, args.session_id, args.sid_hash) store = RoomBootstrapStore(room_bootstrap_state_dir(args.profile)) before = store.get(args.session_id) if time.time() < before.expires_at.timestamp(): raise HarnessError("canonical expiry is unavailable before expires_at") sessions = store.expire_unbound() matches = [item for item in sessions if item.session_id == args.session_id] if len(matches) != 1 or matches[0].state.value != "EXPIRED": raise HarnessError("canonical expiry did not reach EXPIRED") receipt = { **base_receipt("expire"), "status": "PASS_EXPIRED", "expire_invocations": 1, "canonical_api": "RoomBootstrapStore.expire_unbound", "session_id": args.session_id, "session_id_sha256": sha_bytes(args.session_id.encode("ascii")), "sid_hash": args.sid_hash, "state": "EXPIRED", "generation": matches[0].generation, "ledger_sha256": sha_file(bootstrap_path(args.profile)), "token_publication_rollback": False, "cleanup": "terminal receipt complete; archive-first profile cleanup remains separately sealed", } exclusive_write(args.receipt, receipt) return receipt def parser() -> argparse.ArgumentParser: result = argparse.ArgumentParser(description=__doc__) sub = result.add_subparsers(dest="mode", required=True) for name in ("dry-run", "verify", "prepare", "observe", "expire"): command = sub.add_parser(name) command.add_argument("--profile", type=Path, required=True) command.add_argument("--permission", type=Path, required=True) command.add_argument("--receipt", type=Path, required=True) if name == "prepare": command.add_argument("--draft", type=Path, required=True) command.add_argument("--handoff", type=Path, required=True) command.add_argument("--evidence-root", type=Path) if name in {"observe", "expire"}: command.add_argument("--session-id", required=True) command.add_argument("--sid-hash", required=True) if name == "observe": command.add_argument("--ready", type=Path, required=True) command.add_argument("--timeout-seconds", type=float, default=120.0) command.add_argument("--ready-fd", type=int) return result def main() -> int: args = parser().parse_args() try: verify_seal(args.permission, args.mode) handlers = {"dry-run": dry_run, "verify": verify, "prepare": prepare, "observe": observe, "expire": expire} result = handlers[args.mode](args) # stdout is always redacted canonical JSON. sys.stdout.buffer.write(canonical({"mode": args.mode, "status": result["status"], "receipt": str(args.receipt)})) return 2 if result["status"] == "FAIL" else 0 except (HarnessError, OSError, ValueError, KeyError, zipfile.BadZipFile) as exc: sys.stderr.write(f"FAIL: {exc}\n") return 2 if __name__ == "__main__": raise SystemExit(main())