#!/usr/bin/env python3 """Task26 invite harness v2: reset-bound absent-ledger initialization and observation.""" 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 subprocess 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 = "d816dbbd6a27d5826b251d279a9831de096ef7cd79b5fd131ceff7253b7369be" RESET_PLAN_SHA256 = "7ace03c6dad33d2fc3ef223621cbca68a150fde8429932138e252fb8498ac582" GOLDEN_V6_SHA256 = "8416cd7c83cab5540af2e1e01dbcc5b1bdfd49a59d8ef21ea49512c28d749218" RECOVERY_V6_SHA256 = "a2e15845764e7041957ae4518096fe73f87ec35b57021980cdd4df3ea18ac00d" RUNBOOK_V6_RECEIPT_SHA256 = "6fa1e309e0e0f663ca82131a393fcbaa189338ec4c29a1140d3003bece66e18c" RESET_MANIFEST_SHA256 = "7ef0eb8f4b27ab311a11909c25167df9493348d996319d009eec36a0ebf6f146" RESET_EVIDENCE_DIGEST = "d2879b128573a9d75a55958d2fa3e1a24be1636f0a7a20affe8f2ef9b94dfd33" BASELINE_READINESS_SHA256 = "16497e3bb42dd97c66caabc4abd57aa692f32951f0c0d012d754d9988b9cde6a" PRIOR_EXPIRY_SHA256 = "d906f1f79b85c897981d328e06a9bd9503a6b9cf0ca4f58852ee5ad7926cdf2d" PRIOR_SESSION_ID = "cb_v4olwxbpSQatMtVLR4QLmw" PRIOR_SID_HASH = "63bdc993abc7b4f0025e69b10246faf8b5f4da3b7c040f02987a77bef3fa7c59" REPLACEMENT_KEY = "task26_live_2e_r2_20260815_8527916639" REPLACEMENT_DRAFT_SHA256 = "ee05174e7287a87f9c5e40dae1034cf8b3228edc6b9516df8e6349667baffc7c" REPLACEMENT_AUTHORIZATION_SHA256 = "4a3551a9bbccf9bba66ff6b6cfd9500e8e30db65b6244032af4972ea475511ef" ACTOR_ID = "8527916639" OWNER_ID = "8693203710" BOT_USERNAME = "dual_coach_pilot_test_bot" 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] EVIDENCE = ROOT / ".omo/evidence/task26" WHEEL = EVIDENCE / ("task26-repaired-archive-successor-" + CANDIDATE) / "artifacts/hermes_agent-0.17.0-py3-none-any.whl" PLAN = ROOT / ".omo/plans/dualcoach-production-readiness.md" RUNBOOK_ROOT = EVIDENCE / "task26-runbook-rebind-2e0894ea-st_01a0054e" GOLDEN_V6 = RUNBOOK_ROOT / "task26-golden-path-2e0894ea-v6.md" RECOVERY_V6 = RUNBOOK_ROOT / "task26-recovery-runbook-2e0894ea-v6.md" RUNBOOK_V6_RECEIPT = RUNBOOK_ROOT / "runbook-binding-receipt-v6.json" APPROVAL_RECEIPT = HERE / "replacement-approval-receipt.redacted.json" SCHEMA = HERE / "schema-v3.json" SCHEMA_SHA256 = "5d11d2189049c5f3b438b107281a092322753cd2831a30bbb9e12b2df6160229" OPERATIONS = HERE / "OPERATIONS-v3.md" OPERATIONS_SHA256 = "c04b6abccaba4982cb8392c3d6d9585fb011aac6237b549bea3e7e1f84c576fc" REPLACEMENT_DRAFT = HERE / "task26-live-2e-r2-customer-draft.private.json" BASELINE_READINESS = EVIDENCE / "reset-controller-st_01a0054d/reset-baseline-completion-v1/readiness-receipt.json" PRIOR_EXPIRY = EVIDENCE / "task26-invite-harness-st_01a0056a-v2/task26-live-2e-expiry.redacted.json" RESET_RELATIVE = Path("data/profile-reset-archives/task26-live-reset-2e0894ea") class HarnessError(RuntimeError): pass def sha_bytes(raw: bytes) -> str: return hashlib.sha256(raw).hexdigest() 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 sha_file(path: Path) -> str: return sha_bytes(read_regular(path, private=False)) 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 require_directory(path: Path, mode: int = 0o700) -> None: info = path.lstat() if not stat.S_ISDIR(info.st_mode) or stat.S_ISLNK(info.st_mode) or info.st_uid != os.getuid(): raise HarnessError(f"unsafe directory: {path}") if stat.S_IMODE(info.st_mode) != mode: raise HarnessError(f"wrong directory mode: {path}") 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") fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | O_FLAGS, 0o600) try: raw = canonical(value) os.fchmod(fd, 0o600) offset = 0 while offset < len(raw): offset += os.write(fd, raw[offset:]) os.fsync(fd) finally: os.close(fd) def wheel_module_bytes() -> bytes: if sha_file(WHEEL) != WHEEL_SHA256: raise HarnessError("wheel hash mismatch") with zipfile.ZipFile(WHEEL) as archive: raw = archive.read(MODULE_MEMBER) if sha_bytes(raw) != MODULE_SHA256: raise HarnessError("wheel bootstrap member hash mismatch") return raw def loaded_byte_proof() -> dict[str, str]: pins = ( (PLAN, PLAN_SHA256), (GOLDEN_V6, GOLDEN_V6_SHA256), (RECOVERY_V6, RECOVERY_V6_SHA256), (RUNBOOK_V6_RECEIPT, RUNBOOK_V6_RECEIPT_SHA256), (BASELINE_READINESS, BASELINE_READINESS_SHA256), (PRIOR_EXPIRY, PRIOR_EXPIRY_SHA256), (REPLACEMENT_DRAFT, REPLACEMENT_DRAFT_SHA256), ) for path, wanted in pins: if sha_file(path) != wanted: raise HarnessError(f"bound evidence hash mismatch: {path.name}") raw = wheel_module_bytes() source = raw.decode("utf-8") initializer_fragments = ( "self._prepare_paths()", "self.state_dir.mkdir(mode=0o700, parents=True, exist_ok=True)", "self._write_unlocked(())", "os.replace(temporary_name, self.ledger_path)", "os.fsync(directory)", ) if any(fragment not in source for fragment in initializer_fragments): raise HarnessError("candidate absent-ledger initializer proof 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 bytes differ from candidate wheel") return { "candidate_absent_ledger_initializer": "PASS", "initializer_source_sha256": MODULE_SHA256, "atomic_replace_and_directory_fsync": "PASS", "loaded_byte_match": "PASS", } def verify_seal(path: Path, mode: str) -> dict[str, Any]: seal = read_json(path) if sha_file(SCHEMA) != SCHEMA_SHA256: raise HarnessError("harness schema hash mismatch") if sha_file(OPERATIONS) != OPERATIONS_SHA256: raise HarnessError("operations hash mismatch") expected = { "schema": "task26-invite-harness-permission-seal-v3", "candidate_digest": CANDIDATE, "wheel_sha256": WHEEL_SHA256, "plan_sha256": PLAN_SHA256, "golden_runbook_v6_sha256": GOLDEN_V6_SHA256, "recovery_runbook_v6_sha256": RECOVERY_V6_SHA256, "runbook_binding_v6_sha256": RUNBOOK_V6_RECEIPT_SHA256, "reset_manifest_sha256": RESET_MANIFEST_SHA256, "reset_evidence_digest": RESET_EVIDENCE_DIGEST, "baseline_readiness_sha256": BASELINE_READINESS_SHA256, "prior_expiry_receipt_sha256": PRIOR_EXPIRY_SHA256, "prior_session_id": PRIOR_SESSION_ID, "prior_sid_hash": PRIOR_SID_HASH, "replacement_customer_key": REPLACEMENT_KEY, "replacement_draft_sha256": REPLACEMENT_DRAFT_SHA256, "replacement_authorization_event_sha256": REPLACEMENT_AUTHORIZATION_SHA256, "replacement_maximum": 1, "third_invite_authorized": False, "actor_id": ACTOR_ID, "owner_id": OWNER_ID, "approval_receipt_sha256": sha_file(APPROVAL_RECEIPT), "schema_sha256": SCHEMA_SHA256, "operations_sha256": OPERATIONS_SHA256, "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}") if mode not in seal.get("allowed_modes", []): raise HarnessError(f"mode is not permission sealed: {mode}") return seal def reset_binding(profile: Path) -> dict[str, Any]: root = profile / RESET_RELATIVE manifest_path = root / "manifest.json" receipt_path = root / "receipt.json" if sha_file(manifest_path) != RESET_MANIFEST_SHA256: raise HarnessError("reset manifest hash mismatch") manifest = read_json(manifest_path) receipt = read_json(receipt_path) expected = { "candidate_digest": CANDIDATE, "wheel_sha256": WHEEL_SHA256, "plan_sha256": RESET_PLAN_SHA256, "evidence_digest": RESET_EVIDENCE_DIGEST, "restore_or_prepopulate": False, } if any(manifest.get(key) != value for key, value in expected.items()): raise HarnessError("reset manifest binding mismatch") if ( receipt.get("status") != "PASS" or receipt.get("manifest_sha256") != RESET_MANIFEST_SHA256 or receipt.get("evidence_digest") != RESET_EVIDENCE_DIGEST or receipt.get("post_reset_empty_baseline") is not True ): raise HarnessError("reset execution receipt binding mismatch") return {"reset_manifest_sha256": RESET_MANIFEST_SHA256, "reset_evidence_digest": RESET_EVIDENCE_DIGEST, "reset_receipt_sha256": sha_file(receipt_path)} def bootstrap_path(profile: Path) -> Path: return profile / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json" def validate_ledger(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("unexpected customer-bootstrap schema") sessions = value.get("sessions") if not isinstance(sessions, list): raise HarnessError("customer-bootstrap sessions are invalid") payload = json.dumps({"schema": value["schema"], "sessions": sessions}, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode() if value.get("digest") != sha_bytes(payload): raise HarnessError("customer-bootstrap ledger digest mismatch") ids = [item.get("session_id") for item in sessions if isinstance(item, dict)] hashes = [item.get("sid_hash") for item in sessions if isinstance(item, dict)] if len(ids) != len(sessions) or len(set(ids)) != len(ids) or len(set(hashes)) != len(hashes): raise HarnessError("duplicate or malformed customer-bootstrap identity") return value CLEARED_SIBLINGS = ( "customers", "gateway.lock", "gateway_state.json", "sessions", "state.db", ".clean_shutdown", "data/customers", "data/owner-actions", "data/activation-completion-notices.jsonl", "data/customer-activation-audit.jsonl", "data/customer-activation-journal.json", "data/customer-activation-receipt.json", "data/nutrition-onboarding-projection-journal.jsonl", "data/scheduled-deliveries.jsonl", "data/scheduled-deliveries-fence.json", "data/customer-schedule-claims", "data/recovery-audits", "data/activation-readiness", ) def replacement_evidence() -> dict[str, Any]: readiness = read_json(BASELINE_READINESS) expiry = read_json(PRIOR_EXPIRY) approval = read_json(APPROVAL_RECEIPT) if readiness.get("status") != "PASS" or readiness.get("customer_rows") != 0: raise HarnessError("baseline completion readiness is invalid") if ( expiry.get("status") != "PASS_EXPIRED" or expiry.get("session_id") != PRIOR_SESSION_ID or expiry.get("sid_hash") != PRIOR_SID_HASH or expiry.get("state") != "EXPIRED" ): raise HarnessError("prior expiry receipt is invalid") if ( approval.get("authorization_event_sha256") != REPLACEMENT_AUTHORIZATION_SHA256 or approval.get("replacement_count") != 1 or approval.get("third_invite_authorized") is not False or approval.get("prior_session_id") != PRIOR_SESSION_ID ): raise HarnessError("replacement authorization receipt is invalid") return { "baseline_readiness_sha256": BASELINE_READINESS_SHA256, "prior_expiry_receipt_sha256": PRIOR_EXPIRY_SHA256, "replacement_authorization_event_sha256": REPLACEMENT_AUTHORIZATION_SHA256, } def service_inactive() -> None: result = subprocess.run( ["systemctl", "--user", "show", "hermes-gateway-dualcoachtest.service", "-p", "ActiveState", "-p", "SubState", "-p", "MainPID"], text=True, capture_output=True, check=False, ) values = dict(line.split("=", 1) for line in result.stdout.splitlines() if "=" in line) if result.returncode != 0 or values != {"MainPID": "0", "ActiveState": "inactive", "SubState": "dead"}: raise HarnessError("gateway service must be inactive/dead MainPID0") def existing_baseline(profile: Path, customer_key: str | None = None) -> tuple[dict[str, Any], dict[str, Any]]: require_directory(profile) require_directory(profile / "data") path = bootstrap_path(profile) ledger = validate_ledger(path) state = path.parent require_directory(state) if {item.name for item in state.iterdir()} != {"ledger.json", "ledger.lock"}: raise HarnessError("unexpected customer-bootstrap authority file") for item in state.iterdir(): if item.is_symlink() or not item.is_file() or item.stat().st_uid != os.getuid() or item.stat().st_mode & 0o777 != 0o600: raise HarnessError("unsafe customer-bootstrap authority mode/type") sessions = ledger["sessions"] if len(sessions) != 1: raise HarnessError("replacement requires exactly one prior session") prior = sessions[0] if ( prior.get("session_id") != PRIOR_SESSION_ID or prior.get("sid_hash") != PRIOR_SID_HASH or prior.get("state") != "EXPIRED" or prior.get("generation") != 2 or prior.get("role_claims") != [] or prior.get("recovery_attempts") != [] ): raise HarnessError("prior session is not the exact expired unclaimed lineage") prior_draft = prior.get("customer_draft") if not isinstance(prior_draft, dict) or prior_draft.get("customer_user_id") is not None: raise HarnessError("prior expired lineage has customer identity authority") registry = read_json(profile / "customers/registry.json") if registry.get("customers") != [] or registry.get("registry_mode") != "ordinary_v1": raise HarnessError("canonical registry is not empty") onboarding = profile / "data/onboarding" if {item.name for item in onboarding.iterdir()} != {"telegram-customer-bootstrap-v1"}: raise HarnessError("unexpected onboarding authority") owner_actions = profile / "data/owner-actions" if {item.name for item in owner_actions.iterdir()} != {"draft-deliveries.json.lock"}: raise HarnessError("unexpected owner-action authority") gateway = read_json(profile / "gateway_state.json") platforms = gateway.get("platforms") telegram = platforms.get("telegram") if isinstance(platforms, dict) else None if gateway.get("gateway_state") != "stopped" or not isinstance(telegram, dict) or telegram.get("state") != "disconnected": raise HarnessError("gateway state is not stopped/disconnected") lock = profile / "gateway.lock" if lock.is_symlink() or not lock.is_file() or lock.stat().st_mode & 0o777 != 0o600: raise HarnessError("gateway lock baseline is unsafe") service_inactive() if customer_key is not None and customer_key != REPLACEMENT_KEY: raise HarnessError("only the sealed replacement customer key is allowed") if customer_key == prior_draft.get("customer_key"): raise HarnessError("replacement customer key reuses prior lineage") proof = loaded_byte_proof() evidence = replacement_evidence() return ledger, { "status": "PASS", "baseline": "ONE_EXPIRED_REPLACEMENT_READY", "ledger_present": True, "prior_session_count": 1, "prior_state": "EXPIRED", "prepared_session_count": 0, "accepted_claim_count": 0, "registry_customer_count": 0, "service": "inactive/dead MainPID0", "replacement_maximum": 1, "third_invite_authorized": False, "loaded_byte_proof": proof, **evidence, **reset_binding(profile), } def preflight(profile: Path, customer_key: str | None = None) -> tuple[dict[str, Any], dict[str, Any]]: path = bootstrap_path(profile) if not path.exists() or path.is_symlink(): raise HarnessError("replacement requires exact prior expired ledger") return existing_baseline(profile, customer_key) class Inotify: def __init__(self, directory: Path) -> None: libc = ctypes.CDLL(None, use_errno=True) self.fd = libc.inotify_init1(os.O_CLOEXEC) if self.fd < 0: raise OSError(ctypes.get_errno(), "inotify_init1") watch = libc.inotify_add_watch(self.fd, os.fsencode(directory), IN_CREATE | IN_MOVED_TO | IN_CLOSE_WRITE) if 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_count(self, name: bytes, timeout: float) -> int: deadline = time.monotonic() + timeout count = 0 while True: remaining = deadline - time.monotonic() if remaining <= 0 and count == 0: raise HarnessError("bounded filesystem event timeout") readable, _, _ = select.select([self.fd], [], [], max(0.0, remaining) if count == 0 else 0.0) if not readable: if count: return count 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: count += 1 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-v3-{mode}-receipt-v1", "mode": mode, "candidate_digest": CANDIDATE, "wheel_sha256": WHEEL_SHA256, "plan_sha256": PLAN_SHA256, "golden_runbook_v6_sha256": GOLDEN_V6_SHA256, "recovery_runbook_v6_sha256": RECOVERY_V6_SHA256, "reset_manifest_sha256": RESET_MANIFEST_SHA256, "reset_evidence_digest": RESET_EVIDENCE_DIGEST, "privacy": "redacted; no raw invite token or customer content", } def read_mode(args: argparse.Namespace, mode: str) -> dict[str, Any]: try: ledger, result = preflight(args.profile) receipt = {**base_receipt(mode), **result, "status": "PASS", "mutations": 0, "ready_for_exactly_one_replacement_invite": True, "ledger_session_count": len(ledger["sessions"])} if result["ledger_present"]: receipt["ledger_sha256"] = sha_file(bootstrap_path(args.profile)) except (HarnessError, OSError) as exc: receipt = {**base_receipt(mode), "status": "FAIL", "blocker": str(exc), "mutations": 0, "ready_for_exactly_one_replacement_invite": False} 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 output already exists") Store, load_draft, state_dir_for = import_api() read_regular(args.draft) if args.draft.stat().st_mode & 0o777 != 0o600 or sha_file(args.draft) != REPLACEMENT_DRAFT_SHA256: raise HarnessError("only the sealed replacement draft is accepted") draft = load_draft(args.draft) if draft.customer_key != REPLACEMENT_KEY or draft.customer_user_id is not None: raise HarnessError("replacement draft authority mismatch") before, result = preflight(args.profile, draft.customer_key) watch_root = bootstrap_path(args.profile).parent event_name = b"ledger.json" watch = Inotify(watch_root) try: store = Store(state_dir_for(args.profile)) prepared = store.prepare_rehearsal_customer_invite(draft, bot_username=BOT_USERNAME, owner_id=OWNER_ID) event_count = watch.await_name_count(event_name, 5.0) finally: watch.close() if event_count != 1: raise HarnessError("expected exactly one watched authority creation/write event") ledger_path = bootstrap_path(args.profile) after = validate_ledger(ledger_path) state = ledger_path.parent # Candidate Path.mkdir(parents=True, mode=0700) applies 0700 to the leaf; # its intermediate parent follows the process umask (0775 on this runtime). # The enclosing profile/data roots are 0700, so this does not widen access. require_directory(args.profile / "data/onboarding", mode=0o775) require_directory(state) if {item.name for item in state.iterdir()} != {"ledger.json", "ledger.lock"}: raise HarnessError("unexpected candidate initializer output") for item in state.iterdir(): if item.is_symlink() or not item.is_file() or item.stat().st_mode & 0o777 != 0o600: raise HarnessError("candidate initializer output mode/type mismatch") if len(after["sessions"]) != 2 or after["sessions"][0] != before["sessions"][0]: raise HarnessError("prepare did not preserve one prior and add one replacement") new = [item for item in after["sessions"] if item.get("session_id") == prepared.session.session_id] if ( len(new) != 1 or new[0].get("state") != "PREPARED" or new[0].get("generation") != 1 or prepared.session.session_id == PRIOR_SESSION_ID or prepared.session.sid_hash == PRIOR_SID_HASH ): raise HarnessError("new session is not a distinct generation-1 PREPARED replacement") tokens = parse_qs(urlparse(prepared.customer_link).query, strict_parsing=True).get("start") if not isinstance(tokens, list) or len(tokens) != 1 or not tokens[0].startswith("rc1_"): raise HarnessError("candidate private handoff is invalid") token = tokens[0] if prepared.session.sid_hash != sha_bytes(token[4:].encode("ascii")): raise HarnessError("candidate SID hash mismatch") handoff = {"schema": "task26-private-invite-handoff-v3", "session_id": prepared.session.session_id, "start_token": token, "customer_link": prepared.customer_link, "expires_at": prepared.expires_at.isoformat(), "actor_id": ACTOR_ID} exclusive_write(args.handoff, handoff) receipt = { **base_receipt("prepare"), "status": "PASS_PREPARED", "baseline_before": result["baseline"], "watch_root": "telegram-customer-bootstrap-v1", "event_subscription_before_initializer": True, "event_subscription_before_prepare": True, "onboarding_create_event_count": 0, "ledger_write_event_count": event_count, "candidate_initializer_invocations": 1, "prepare_invocations": 1, "replacement_prepare_invocations": 1, "final_ledger_publications_expected_from_candidate": 1, "atomic_replace_and_directory_fsync_source_proof": True, "session_id": prepared.session.session_id, "session_id_sha256": sha_bytes(prepared.session.session_id.encode()), "sid_hash": prepared.session.sid_hash, "start_token_sha256": sha_bytes(token.encode()), "state": "PREPARED", "generation": 1, "ledger_sha256": sha_file(ledger_path), "handoff_sha256": sha_file(args.handoff), "handoff_mode": "0600", "raw_token_locations": [str(args.handoff)], } exclusive_write(args.receipt, receipt) if args.evidence_root: locations = [] for path in sorted(item for item in args.evidence_root.rglob("*") if item.is_file() and not item.is_symlink()): try: if token.encode() in read_regular(path): locations.append(path.resolve()) except (HarnessError, PermissionError): continue if locations != [args.handoff.resolve()]: raise HarnessError("raw token escaped the one private handoff") return receipt def replacement_session(ledger: dict[str, Any], session_id: str, sid_hash: str) -> dict[str, Any]: sessions = ledger["sessions"] if len(sessions) != 2: raise HarnessError("replacement lifecycle requires exactly two sessions") prior, replacement = sessions if ( prior.get("session_id") != PRIOR_SESSION_ID or prior.get("sid_hash") != PRIOR_SID_HASH or prior.get("state") != "EXPIRED" or prior.get("role_claims") != [] or prior.get("recovery_attempts") != [] ): raise HarnessError("prior expired lineage changed") draft = replacement.get("customer_draft") if ( replacement.get("session_id") != session_id or replacement.get("sid_hash") != sid_hash or session_id == PRIOR_SESSION_ID or sid_hash == PRIOR_SID_HASH or not isinstance(draft, dict) or draft.get("customer_key") != REPLACEMENT_KEY or draft.get("customer_user_id") is not None ): raise HarnessError("replacement session binding mismatch") return replacement def observe(args: argparse.Namespace) -> dict[str, Any]: loaded_byte_proof() ledger = validate_ledger(bootstrap_path(args.profile)) replacement = replacement_session(ledger, args.session_id, args.sid_hash) if replacement.get("state") != "PREPARED" or replacement.get("role_claims") != []: raise HarnessError("observe requires exact clean PREPARED replacement") watch = Inotify(bootstrap_path(args.profile).parent) ready = {**base_receipt("observe-ready"), "status": "READY", "event_subscription_before_claim": True, "session_id_sha256": sha_bytes(args.session_id.encode()), "sid_hash": args.sid_hash} 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: count = watch.await_name_count(b"ledger.json", args.timeout_seconds) finally: watch.close() if count != 1: raise HarnessError("duplicate claim ledger writes observed at boundary") after = validate_ledger(bootstrap_path(args.profile)) item = replacement_session(after, args.session_id, args.sid_hash) claims = item.get("role_claims") if not isinstance(claims, list) or len(claims) != 1: raise HarnessError("accepted claim cardinality is not one") claim = claims[0] if claim.get("user_id") != ACTOR_ID or claim.get("chat_id") != ACTOR_ID or claim.get("topic_id") != "0": raise HarnessError("accepted claim is not exact actor private DM") receipt = {**base_receipt("observe"), "status": "PASS_CLAIM_OBSERVED", "accepted_start_claim_count": 1, "actor_id": ACTOR_ID, "private_dm": True, "session_id": args.session_id, "sid_hash": args.sid_hash, "state": item.get("state"), "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 expire(args: argparse.Namespace) -> dict[str, Any]: Store, _load, state_dir_for = import_api() ledger = validate_ledger(bootstrap_path(args.profile)) replacement = replacement_session(ledger, args.session_id, args.sid_hash) if replacement.get("state") != "PREPARED" or replacement.get("role_claims") != []: raise HarnessError("expiry binding mismatch") if time.time() < __import__("datetime").datetime.fromisoformat(replacement["expires_at"]).timestamp(): raise HarnessError("canonical expiry unavailable before expires_at") sessions = Store(state_dir_for(args.profile)).expire_unbound() item = next(one for one in sessions if one.session_id == args.session_id) if item.state.value != "EXPIRED": raise HarnessError("canonical expiry failed") receipt = {**base_receipt("expire"), "status": "PASS_EXPIRED", "canonical_api": "RoomBootstrapStore.expire_unbound", "expire_invocations": 1, "session_id": args.session_id, "sid_hash": args.sid_hash, "state": "EXPIRED", "ledger_sha256": sha_file(bootstrap_path(args.profile))} 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) if args.mode in {"dry-run", "verify"}: result = read_mode(args, args.mode) else: result = {"prepare": prepare, "observe": observe, "expire": expire}[args.mode](args) 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, StopIteration, zipfile.BadZipFile) as exc: sys.stderr.write(f"FAIL: {exc}\n") return 2 if __name__ == "__main__": raise SystemExit(main())