#!/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 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_V5_SHA256 = "3c6a0f24124142fc1b03c350a357d5530aa1a561c4b332e2e6c29a0de74c87e2" RECOVERY_V5_SHA256 = "e2fbff811e5b0caf4551010c1a47d640bba17febffb951f0525b6ff4dccd71d3" RUNBOOK_V5_RECEIPT_SHA256 = "2fb0019a8d9e46cc197f72685a89fc54ca5fb98a52840f41ade5d51f76d86736" RESET_MANIFEST_SHA256 = "7ef0eb8f4b27ab311a11909c25167df9493348d996319d009eec36a0ebf6f146" RESET_EVIDENCE_DIGEST = "d2879b128573a9d75a55958d2fa3e1a24be1636f0a7a20affe8f2ef9b94dfd33" 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] 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_V5 = RUNBOOK_ROOT / "task26-golden-path-2e0894ea-v5.md" RECOVERY_V5 = RUNBOOK_ROOT / "task26-recovery-runbook-2e0894ea-v5.md" RUNBOOK_V5_RECEIPT = RUNBOOK_ROOT / "runbook-binding-receipt-v5.json" APPROVAL_RECEIPT = HERE / "approval-receipt.json" SCHEMA = HERE / "schema-v2.json" SCHEMA_SHA256 = "6a4249d263db04bea1432ad2393150ed77d3243a1365ed8dbe1eed5c0261df9e" OPERATIONS = HERE / "OPERATIONS-v2.md" OPERATIONS_SHA256 = "82099d8a3658a494c3cf13558e43fe7846e7e826cfcdec6c10515f05a697752d" 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_V5, GOLDEN_V5_SHA256), (RECOVERY_V5, RECOVERY_V5_SHA256), (RUNBOOK_V5_RECEIPT, RUNBOOK_V5_RECEIPT_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-v2", "candidate_digest": CANDIDATE, "wheel_sha256": WHEEL_SHA256, "plan_sha256": PLAN_SHA256, "golden_runbook_v5_sha256": GOLDEN_V5_SHA256, "recovery_runbook_v5_sha256": RECOVERY_V5_SHA256, "runbook_binding_v5_sha256": RUNBOOK_V5_RECEIPT_SHA256, "reset_manifest_sha256": RESET_MANIFEST_SHA256, "reset_evidence_digest": RESET_EVIDENCE_DIGEST, "actor_id": ACTOR_ID, "owner_id": OWNER_ID, "approval_event_sha256": APPROVAL_EVENT_SHA256, "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": 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 absent_baseline(profile: Path) -> dict[str, Any]: require_directory(profile) require_directory(profile / "data") binding = reset_binding(profile) onboarding = profile / "data/onboarding" if onboarding.exists() or onboarding.is_symlink(): raise HarnessError("absent-ledger baseline requires onboarding parent absent") present = [relative for relative in CLEARED_SIBLINGS if (profile / relative).exists() or (profile / relative).is_symlink()] if present: raise HarnessError("unexpected post-reset sibling authority: " + present[0]) proof = loaded_byte_proof() return { "status": "PASS", "baseline": "RESET_BOUND_LEDGER_ABSENT", "ledger_present": False, "prior_session_count": 0, "all_prior_sessions_terminal": True, "sibling_authorities_clean": True, "nearest_stable_parent": "data", "actor_not_owner_staff": ACTOR_ID != OWNER_ID, "loaded_byte_proof": proof, **binding, } 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 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 prior claim/recovery authority") 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") return ledger, {"status": "PASS", "baseline": "AUTHENTICATED_LEDGER_PRESENT", "ledger_present": True, "prior_session_count": len(sessions), "all_prior_sessions_terminal": True, "loaded_byte_proof": loaded_byte_proof()} def preflight(profile: Path, customer_key: str | None = None) -> tuple[dict[str, Any], dict[str, Any]]: path = bootstrap_path(profile) if path.exists() or path.is_symlink(): return existing_baseline(profile, customer_key) result = absent_baseline(profile) return {"schema": "telegram-customer-bootstrap-v1", "sessions": [], "digest": None}, result 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-v2-{mode}-receipt-v1", "mode": mode, "candidate_digest": CANDIDATE, "wheel_sha256": WHEEL_SHA256, "plan_sha256": PLAN_SHA256, "golden_runbook_v5_sha256": GOLDEN_V5_SHA256, "recovery_runbook_v5_sha256": RECOVERY_V5_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_one_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_one_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() draft = load_draft(args.draft) before, result = preflight(args.profile, draft.customer_key) absent = not result["ledger_present"] watch_root = args.profile / "data" if absent else bootstrap_path(args.profile).parent event_name = b"onboarding" if absent else b"ledger.json" watch = Inotify(watch_root) try: if absent and ((args.profile / "data/onboarding").exists() or (args.profile / "data/onboarding").is_symlink()): raise HarnessError("onboarding authority race appeared after subscription") 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"]) != len(before["sessions"]) + 1: raise HarnessError("prepare did not establish exactly one new session") 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: raise HarnessError("new session is not exactly one generation-1 PREPARED authority") 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-v2", "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": "data" if absent else "telegram-customer-bootstrap-v1", "event_subscription_before_initializer": True, "onboarding_create_event_count": event_count if absent else 0, "ledger_write_event_count": event_count if not absent else 0, "candidate_initializer_invocations": 1, "prepare_invocations": 1, "final_ledger_publications_expected_from_candidate": 2 if absent else 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 observe(args: argparse.Namespace) -> dict[str, Any]: loaded_byte_proof() ledger = validate_ledger(bootstrap_path(args.profile)) matches = [item for item in ledger["sessions"] if item.get("session_id") == args.session_id] if len(matches) != 1 or matches[0].get("sid_hash") != args.sid_hash or matches[0].get("state") != "PREPARED": raise HarnessError("observe requires exact clean PREPARED binding") 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 = next(one for one in after["sessions"] if one.get("session_id") == args.session_id) 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)) matches = [item for item in ledger["sessions"] if item.get("session_id") == args.session_id] if len(matches) != 1 or matches[0].get("sid_hash") != args.sid_hash or matches[0].get("state") != "PREPARED": raise HarnessError("expiry binding mismatch") if time.time() < __import__("datetime").datetime.fromisoformat(matches[0]["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())