tenant.openagents/omega

1#!/usr/bin/env python3
2"""Search the surfaces an installed Omega writes for a marked secret.
3
4The installed proof requires a tripwire receipt and nothing produced one, so
5that input was permanently blocked. This produces it. The method is a
6marked-secret search: the disposable canary the candidate was given during the
7journey is searched for on every surface a secret could plausibly escape onto,
8and the receipt records which surfaces were read and that none of them carried
9it. The needle's value is never written to the receipt or to any log — recording
10the secret you are proving was not leaked would defeat the exercise — so only
11its length is kept.
12
13Three properties keep this receipt from being a formality.
14
15**The needle is not this script's to invent.** It is read from a descriptor the
16caller opened on the canary file, and a run without `--needle-fd` refuses rather
17than minting a fresh secret no other process has ever seen. An earlier version
18generated its own needle, wrote it into a pipe, closed both ends of that pipe in
19the same function, and then searched the disk for it. Nothing could ever have
20written that value anywhere, so `pass` was guaranteed by construction whatever
21the product did. A receipt that cannot fail is worse than no receipt.
22
23**The scan is proved able to fire on this run.** Before any surface is read the
24needle is planted in a private file and the same scanner is pointed at it. If
25that file does not come back as a match the run blocks, so a scanner that has
26been broken — by an edit, by an unreadable temporary directory, by a needle that
27arrived empty — cannot report that nothing was found.
28
29**Every surface is read where Omega actually writes, through the interface that
30surface actually has.** `crates/paths/src/paths.rs` puts `logs_dir()` and the
31telemetry log under `~/Library/Logs/<storage slug>` and `crashes_dir()` under
32`~/Library/Logs/DiagnosticReports`; none of those is under the application
33support root an earlier version scanned, so four of the six surfaces were being
34searched for in directories the product never creates. The clipboard is a
35pasteboard and the accessibility surface is a tree; neither is ever a file in a
36data directory, so each is captured through its own API into a private file that
37is deleted when the run ends.
38
39A surface that exists and does not carry the needle records `pass`. A surface
40whose location is right and whose file the product has not written records
41`absent`. A surface that could not be observed at all records `blocked`, which
42the receipt validator refuses — "nothing was found there" and "nobody looked"
43must not read the same, and only one of them may support a release.
44
45Usage:
46    collect-omega-installed-tripwires \\
47        --app /Applications/Omega.app \\
48        --candidate-evidence <path> \\
49        --output <path> \\
50        --needle-fd 3 3<"$OMEGA_PRIVATE_PROOF_DIR/disposable-canary"
51"""
52
53from __future__ import annotations
54
55import argparse
56import hashlib
57import json
58import os
59import platform
60import shutil
61import stat
62import subprocess
63import sys
64import tempfile
65from collections.abc import Callable
66from datetime import datetime, timezone
67from pathlib import Path
68from typing import Any
69
70TRIPWIRE_SCHEMA = "openagents.omega.installed-secret-tripwires.v1"
71
72#: Bounds the receipt validator enforces on the needle, mirrored here so a
73#: canary that could not support a receipt is refused before anything is
74#: scanned rather than after.
75MINIMUM_NEEDLE_BYTES = 16
76MAXIMUM_NEEDLE_BYTES = 4096
77
78#: `AppChannel::storage_slug` for the RC channel, and the display name
79#: `app_name()` uses for the data root. The two differ, and using either where
80#: the other belongs is what put four surfaces in directories that do not
81#: exist.
82STORAGE_SLUG = "omega-rc"
83DISPLAY_NAME = "Omega RC"
84
85#: Refuse to read anything enormous; a scan that hangs is a scan nobody runs.
86MAX_FILE_BYTES = 64 * 1024 * 1024
87
88#: Surfaces whose file the product writes only under a condition the journey
89#: may not have met. A missing telemetry log or an empty crash directory is a
90#: fact about the run, not a gap in the search, and records `absent`. Every
91#: other surface must be observed or blocked.
92OPTIONAL_SURFACES = frozenset({"telemetry", "crashes"})
93
94
95class TripwireError(Exception):
96    pass
97
98
99def sha256_hex(data: bytes) -> str:
100    return hashlib.sha256(data).hexdigest()
101
102
103def data_root(override: Path | None = None) -> Path:
104    """`paths::data_dir()` on macOS: the application support root."""
105    return override or Path.home() / "Library" / "Application Support" / DISPLAY_NAME
106
107
108def logs_root(override: Path | None = None) -> Path:
109    """`paths::logs_dir()` on macOS.
110
111    Not the data root. `paths.rs` sends macOS logs to `~/Library/Logs/<slug>`,
112    and the live log file and the telemetry log are both there.
113    """
114    return override or Path.home() / "Library" / "Logs" / STORAGE_SLUG
115
116
117def crashes_root(override: Path | None = None) -> Path:
118    """`paths::crashes_dir()` on macOS."""
119    return override or Path.home() / "Library" / "Logs" / "DiagnosticReports"
120
121
122def telemetry_file(logs: Path) -> Path:
123    """`client::telemetry` writes `telemetry.log` inside the logs directory."""
124    return logs / "telemetry.log"
125
126
127def filesystem_surfaces(
128    data: Path, logs: Path, crashes: Path
129) -> dict[str, dict[str, Any]]:
130    """Where each file-backed surface actually lives.
131
132    `anchor` is the directory that must exist for the search to have happened
133    in the right place at all. An absent anchor is not an empty surface: it
134    means this run looked somewhere the product does not write, which is the
135    defect this table was rewritten to remove, so it blocks.
136    """
137    return {
138        "logs": {
139            "anchor": logs,
140            "roots": [logs],
141            # The telemetry log lives in the same directory and is its own
142            # surface below. Excluding it here keeps the two counts disjoint so
143            # neither borrows the other's coverage.
144            "exclude": frozenset({telemetry_file(logs)}),
145        },
146        "telemetry": {
147            "anchor": logs,
148            "roots": [telemetry_file(logs)],
149            "exclude": frozenset(),
150        },
151        "diagnostics": {
152            "anchor": data,
153            "roots": [data / "db", data / "hang_traces"],
154            "exclude": frozenset(),
155        },
156        "crashes": {
157            "anchor": crashes.parent,
158            "roots": [crashes],
159            "exclude": frozenset(),
160        },
161    }
162
163
164class ScanCounts:
165    """Exactly what a scan inspected, and what it found."""
166
167    def __init__(self) -> None:
168        self.files_scanned = 0
169        self.symlinks_skipped = 0
170        self.bytes_scanned = 0
171        self.errors = 0
172        self.match_detected = False
173        self.present = False
174
175
176def scan_paths(
177    roots: list[Path],
178    needle: bytes,
179    exclude: frozenset[Path] = frozenset(),
180) -> ScanCounts:
181    """Scan a set of files and directories for the needle, counting the reads.
182
183    Symlinks are skipped rather than followed. Following them would let a link
184    out of the surface inflate the byte count with files this candidate never
185    wrote, which would make the receipt claim more coverage than it earned.
186    """
187    counts = ScanCounts()
188
189    def read_file(path: Path) -> None:
190        try:
191            file_stat = path.lstat()
192            if not stat.S_ISREG(file_stat.st_mode):
193                return
194            if file_stat.st_size > MAX_FILE_BYTES:
195                counts.errors += 1
196                return
197            payload = path.read_bytes()
198        except OSError:
199            counts.errors += 1
200            return
201        counts.files_scanned += 1
202        counts.bytes_scanned += len(payload)
203        if needle in payload:
204            counts.match_detected = True
205
206    for root in roots:
207        if root.is_symlink():
208            counts.symlinks_skipped += 1
209            continue
210        if not root.exists():
211            continue
212        counts.present = True
213        if root.is_file():
214            if root not in exclude:
215                read_file(root)
216            continue
217        for current, directories, files in os.walk(root, followlinks=False):
218            current_path = Path(current)
219            for directory in list(directories):
220                if (current_path / directory).is_symlink():
221                    counts.symlinks_skipped += 1
222                    directories.remove(directory)
223            for file_name in files:
224                path = current_path / file_name
225                if path in exclude:
226                    continue
227                if path.is_symlink():
228                    counts.symlinks_skipped += 1
229                    continue
230                read_file(path)
231
232    return counts
233
234
235def surface_record(
236    name: str, path_label: str, status: str, counts: ScanCounts | None = None
237) -> dict[str, Any]:
238    """One surface's line in the receipt, in the exact shape the validator reads."""
239    counts = counts or ScanCounts()
240    body = {
241        "name": name,
242        "status": status,
243        "files_scanned": counts.files_scanned,
244        "symlinks_skipped": counts.symlinks_skipped,
245        "bytes_scanned": counts.bytes_scanned,
246        "errors": counts.errors,
247        "match_detected": counts.match_detected,
248    }
249    return {
250        **body,
251        "path_digest": sha256_hex(path_label.encode("utf-8")),
252        "evidence_digest": sha256_hex(json.dumps(body, sort_keys=True).encode("utf-8")),
253    }
254
255
256def resolve_status(name: str, counts: ScanCounts, anchor_present: bool) -> str:
257    """Turn a scan into one of the four things that can honestly be said.
258
259    `blocked` is the answer whenever the search could not be made where it had
260    to be made. It is deliberately a status the receipt validator refuses.
261    """
262    if not anchor_present:
263        return "blocked"
264    if counts.errors:
265        return "error"
266    if counts.match_detected:
267        return "match"
268    if not counts.present:
269        return "absent" if name in OPTIONAL_SURFACES else "blocked"
270    return "pass"
271
272
273def scan_filesystem_surface(
274    name: str, plan: dict[str, Any], needle: bytes
275) -> tuple[dict[str, Any], str]:
276    anchor: Path = plan["anchor"]
277    roots: list[Path] = plan["roots"]
278    counts = scan_paths(roots, needle, plan["exclude"])
279    anchor_present = anchor.is_dir()
280    status = resolve_status(name, counts, anchor_present)
281    label = "\n".join(sorted(str(root) for root in roots))
282    detail = f"{status} at {label}"
283    if not anchor_present:
284        detail = (
285            f"blocked: {anchor} does not exist, so this run did not look where "
286            "the product writes"
287        )
288    return surface_record(name, label, status, counts), detail
289
290
291def appkit_pasteboard_bytes() -> bytes | None:
292    """Every flavour of every item on the general pasteboard, through PyObjC.
293
294    `None` means this interface was not available or did not answer, which
295    falls through to `pbpaste`. It never falls through to a pass.
296    """
297    try:
298        from AppKit import NSPasteboard  # type: ignore[import-not-found]
299    except ImportError:
300        return None
301    try:
302        pasteboard = NSPasteboard.generalPasteboard()
303        payload = bytearray()
304        for item in pasteboard.pasteboardItems() or []:
305            for kind in item.types() or []:
306                data = item.dataForType_(kind)
307                if data is not None:
308                    payload += bytes(data)
309    except Exception:  # noqa: BLE001 - any failed read is a read that did not happen
310        return None
311    return bytes(payload)
312
313
314def pasteboard_bytes() -> tuple[bytes | None, str]:
315    """Read the general pasteboard through its own interface.
316
317    PyObjC reads every flavour of every item when it is available. Where it is
318    not, `pbpaste` reads the text flavours, which is a real read of the same
319    pasteboard and is recorded as exactly that much coverage. A pasteboard is
320    never a file in a data directory, so there is no third option that reads
321    one.
322    """
323    through_appkit = appkit_pasteboard_bytes()
324    if through_appkit is not None:
325        return through_appkit, "NSPasteboard general pasteboard, every type of every item"
326
327    if shutil.which("pbpaste") is None:
328        return None, ""
329    payload = bytearray()
330    read_any = False
331    for flavour in ("txt", "rtf"):
332        try:
333            result = subprocess.run(
334                ["pbpaste", "-Prefer", flavour],
335                capture_output=True,
336                timeout=30,
337                check=False,
338            )
339        except (OSError, subprocess.SubprocessError):
340            continue
341        if result.returncode == 0:
342            read_any = True
343            payload += result.stdout
344    if not read_any:
345        return None, ""
346    return bytes(payload), "pbpaste general pasteboard, text and rich-text flavours"
347
348
349def pasteboard_inventory() -> int | None:
350    """How many bytes the pasteboard itself says it is holding.
351
352    `clipboard info` answers with a flavour and a size for everything on the
353    general pasteboard. It is the corroboration for an empty read: a read that
354    came back with nothing while the pasteboard says it holds something means
355    the flavours were not all reachable, not that the surface is clean.
356    """
357    try:
358        result = subprocess.run(
359            ["osascript", "-e", "clipboard info"],
360            capture_output=True,
361            text=True,
362            timeout=30,
363            check=False,
364        )
365    except (OSError, subprocess.SubprocessError):
366        return None
367    if result.returncode != 0:
368        return None
369    total = 0
370    for token in result.stdout.split(","):
371        token = token.strip()
372        if token.isdigit():
373            total += int(token)
374    return total
375
376
377def pasteboard_capture_status(
378    payload: bytes | None, declared: int | None
379) -> tuple[bool, str]:
380    """Whether a pasteboard read may be recorded as an observation of the surface.
381
382    An empty read is only an observation when the pasteboard agrees it is
383    empty. Otherwise this run saw less than the surface holds, and a partial
384    look at a surface is not a clean one.
385    """
386    if payload is None:
387        return False, "the general pasteboard could not be read through any interface"
388    if payload:
389        return True, ""
390    if declared is None:
391        return False, (
392            "the pasteboard read returned nothing and the pasteboard's own "
393            "inventory could not be read to confirm it is empty"
394        )
395    if declared > 0:
396        return False, (
397            f"the pasteboard holds {declared} bytes in flavours this read "
398            "cannot see, so it was not fully scanned"
399        )
400    return True, ""
401
402
403def capture_pasteboard(directory: Path) -> tuple[Path | None, str, str]:
404    if platform.system() != "Darwin":
405        return None, "", "the pasteboard is a macOS interface and this host is not macOS"
406    payload, method = pasteboard_bytes()
407    readable, reason = pasteboard_capture_status(
408        payload, pasteboard_inventory() if not payload else None
409    )
410    if not readable or payload is None:
411        return None, "", reason
412    destination = directory / "pasteboard.bin"
413    handle = os.open(destination, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
414    with os.fdopen(handle, "wb") as capture:
415        capture.write(payload)
416    return destination, method, ""
417
418
419def candidate_pid(app: Path | None, explicit: int | None) -> int | None:
420    """The running candidate this run reads the accessibility tree of."""
421    if explicit is not None:
422        return explicit
423    if app is None:
424        return None
425    executables = str(app / "Contents" / "MacOS")
426    try:
427        result = subprocess.run(
428            ["pgrep", "-f", executables],
429            capture_output=True,
430            text=True,
431            timeout=30,
432            check=False,
433        )
434    except (OSError, subprocess.SubprocessError):
435        return None
436    for line in result.stdout.split():
437        try:
438            return int(line)
439        except ValueError:
440            continue
441    return None
442
443
444def accessibility_tree(pid: int) -> tuple[str | None, str]:
445    """Every label and value the process publishes to assistive technology.
446
447    This is the surface a screen reader consumes, read through the same
448    `AXUIElement` interface it uses. It is never a file in a data directory,
449    and a run that cannot reach it says so rather than recording an empty
450    directory as a clean surface.
451    """
452    script = f"""
453    tell application "System Events"
454        if not (exists (first process whose unix id is {pid})) then return "NOPROC"
455        tell (first process whose unix id is {pid})
456            set out to ""
457            try
458                repeat with w in windows
459                    repeat with e in (entire contents of w)
460                        try
461                            set d to description of e
462                            if d is not missing value then set out to out & d & linefeed
463                        end try
464                        try
465                            set v to value of e
466                            if v is not missing value then set out to out & (v as text) & linefeed
467                        end try
468                    end repeat
469                end repeat
470            end try
471            return out
472        end tell
473    end tell
474    """
475    try:
476        result = subprocess.run(
477            ["osascript", "-e", script],
478            capture_output=True,
479            text=True,
480            timeout=120,
481            check=False,
482        )
483    except (OSError, subprocess.SubprocessError):
484        return None, "the accessibility tree could not be read; osascript did not answer"
485    if result.returncode != 0:
486        return None, (
487            "the accessibility tree could not be read. Grant this terminal "
488            "Accessibility in System Settings > Privacy & Security > Accessibility"
489        )
490    if result.stdout.strip() == "NOPROC":
491        return None, "the candidate is not running, so it publishes no accessibility tree"
492    return result.stdout, ""
493
494
495def capture_accessibility(directory: Path, pid: int | None) -> tuple[Path | None, str, str]:
496    if platform.system() != "Darwin":
497        return None, "", "the accessibility tree is a macOS interface and this host is not macOS"
498    if pid is None:
499        return None, "", "no running candidate was found to read an accessibility tree from"
500    tree, reason = accessibility_tree(pid)
501    if tree is None:
502        return None, "", reason
503    destination = directory / "accessibility.txt"
504    handle = os.open(destination, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
505    with os.fdopen(handle, "wb") as capture:
506        capture.write(tree.encode("utf-8", errors="surrogateescape"))
507    return destination, f"AXUIElement tree of pid {pid} through System Events", ""
508
509
510def scan_captured_surface(
511    name: str,
512    label: str,
513    capture: Path | None,
514    reason: str,
515    needle: bytes,
516) -> tuple[dict[str, Any], str]:
517    """A surface that had to be captured through an API before it could be read."""
518    if capture is None:
519        return surface_record(name, label, "blocked"), f"blocked: {reason}"
520    counts = scan_paths([capture], needle)
521    status = resolve_status(name, counts, anchor_present=True)
522    return surface_record(name, label, status, counts), f"{status}: {reason}"
523
524
525def read_needle(file_descriptor: int) -> bytes:
526    """Read the canary from the descriptor the caller opened on it.
527
528    The descriptor is the whole point: the secret never becomes an argument an
529    unrelated process can read out of the process table, and this script never
530    invents one. A needle nothing has ever seen cannot be leaked, so searching
531    for one proves nothing at all.
532    """
533    chunks: list[bytes] = []
534    remaining = MAXIMUM_NEEDLE_BYTES + 2
535    while remaining > 0:
536        try:
537            chunk = os.read(file_descriptor, remaining)
538        except OSError as error:
539            raise TripwireError(f"--needle-fd {file_descriptor} could not be read") from error
540        if not chunk:
541            break
542        chunks.append(chunk)
543        remaining -= len(chunk)
544    needle = b"".join(chunks).rstrip(b"\r\n")
545    if not MINIMUM_NEEDLE_BYTES <= len(needle) <= MAXIMUM_NEEDLE_BYTES:
546        raise TripwireError(
547            f"the canary must carry {MINIMUM_NEEDLE_BYTES}..{MAXIMUM_NEEDLE_BYTES} "
548            f"bytes; this descriptor carried {len(needle)}"
549        )
550    return needle
551
552
553def verify_scanner_can_fire(
554    needle: bytes,
555    directory: Path,
556    scanner: Callable[[list[Path], bytes], ScanCounts] | None = None,
557) -> bool:
558    """Plant this run's needle and require the scanner to find it.
559
560    Without this, every way the search can be silently disabled — a needle that
561    arrived empty, an edit that broke the comparison, a walk that reads nothing
562    — reports the same `pass` as a clean product.
563    """
564    control = directory / "scanner-control"
565    control.mkdir(mode=0o700)
566    planted = control / "planted"
567    handle = os.open(planted, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
568    with os.fdopen(handle, "wb") as target:
569        target.write(b"prefix" + needle + b"suffix")
570    counts = (scanner or scan_paths)([control], needle)
571    shutil.rmtree(control, ignore_errors=True)
572    return counts.match_detected and counts.files_scanned == 1
573
574
575def self_test() -> None:
576    """Every property this receipt rests on has to be able to fail here."""
577    with tempfile.TemporaryDirectory() as directory:
578        workspace = Path(directory)
579        needle = b"canary-" + b"0123456789abcdef" * 2
580
581        # The scanner finds a planted needle, and distinguishes clean from
582        # missing. A scanner that cannot do this makes every run pass.
583        root = workspace / "logs"
584        root.mkdir()
585        (root / "clean.log").write_bytes(b"nothing to see")
586        clean = scan_paths([root], needle)
587        if clean.match_detected or not clean.present or clean.files_scanned != 1:
588            raise TripwireError("scanner reported a clean surface incorrectly")
589        (root / "leaked.log").write_bytes(b"secret=" + needle)
590        dirty = scan_paths([root], needle)
591        if not dirty.match_detected:
592            raise TripwireError("scanner failed to detect a planted needle")
593        missing = scan_paths([workspace / "missing"], needle)
594        if missing.present or missing.files_scanned != 0:
595            raise TripwireError("scanner did not distinguish absent from clean")
596
597        # An excluded path is not counted by the surface that excludes it.
598        excluded = scan_paths([root], needle, frozenset({root / "leaked.log"}))
599        if excluded.match_detected or excluded.files_scanned != 1:
600            raise TripwireError("scanner did not honour the exclusion")
601
602        # The live-fire control has to catch a scanner that cannot match.
603        if not verify_scanner_can_fire(needle, workspace):
604            raise TripwireError("the scanner control failed on a working scanner")
605        blind = verify_scanner_can_fire(
606            needle, workspace, scanner=lambda roots, needle: ScanCounts()
607        )
608        if blind:
609            raise TripwireError("the scanner control passed a scanner that never matches")
610
611        # An empty pasteboard read is an observation only when the pasteboard
612        # agrees it is empty.
613        for payload, declared, expected in (
614            (b"something", None, True),
615            (None, None, False),
616            (b"", None, False),
617            (b"", 0, True),
618            (b"", 12, False),
619        ):
620            readable, _ = pasteboard_capture_status(payload, declared)
621            if readable is not expected:
622                raise TripwireError(
623                    f"a pasteboard read of {payload!r} against {declared} was "
624                    f"judged {readable}"
625                )
626
627        # A needle outside the receipt's bounds is refused before anything is
628        # scanned, and a short one is refused too.
629        short_path = workspace / "short-canary"
630        short_path.write_bytes(b"tooshort")
631        with open(short_path, "rb") as short:
632            try:
633                read_needle(short.fileno())
634            except TripwireError:
635                pass
636            else:
637                raise TripwireError("a canary below the minimum length was accepted")
638        long_path = workspace / "canary"
639        long_path.write_bytes(needle + b"\n")
640        with open(long_path, "rb") as canary:
641            if read_needle(canary.fileno()) != needle:
642                raise TripwireError("the canary read back differently than it was written")
643
644        # The surfaces resolve where `paths.rs` writes, not under the
645        # application support root. This is the regression that let four of six
646        # surfaces record `absent` while the product logged 191 KB elsewhere.
647        logs = logs_root()
648        if logs != Path.home() / "Library" / "Logs" / STORAGE_SLUG:
649            raise TripwireError("the logs surface does not resolve to paths::logs_dir()")
650        if "Application Support" in str(logs):
651            raise TripwireError("the logs surface resolves under the data root again")
652        if crashes_root() != Path.home() / "Library" / "Logs" / "DiagnosticReports":
653            raise TripwireError("the crash surface does not resolve to paths::crashes_dir()")
654        plan = filesystem_surfaces(data_root(), logs, crashes_root())
655        if plan["telemetry"]["roots"] != [logs / "telemetry.log"]:
656            raise TripwireError("the telemetry surface does not resolve to the telemetry log")
657        if set(plan) | {"clipboard", "accessibility"} != {
658            "logs",
659            "telemetry",
660            "diagnostics",
661            "crashes",
662            "clipboard",
663            "accessibility",
664        }:
665            raise TripwireError("the surface inventory is not the six the validator requires")
666
667        # A required surface whose anchor does not exist blocks; it never
668        # reports the absence as a clean scan.
669        empty = ScanCounts()
670        if resolve_status("logs", empty, anchor_present=False) != "blocked":
671            raise TripwireError("a missing logs anchor did not block")
672        if resolve_status("logs", empty, anchor_present=True) != "blocked":
673            raise TripwireError("a missing required surface did not block")
674        if resolve_status("crashes", empty, anchor_present=True) != "absent":
675            raise TripwireError("an optional absent surface did not record absent")
676        scanned = ScanCounts()
677        scanned.present = True
678        if resolve_status("logs", scanned, anchor_present=True) != "pass":
679            raise TripwireError("a scanned clean surface did not pass")
680        scanned.match_detected = True
681        if resolve_status("logs", scanned, anchor_present=True) != "match":
682            raise TripwireError("a matched surface did not report the match")
683
684    print(json.dumps({"schema": TRIPWIRE_SCHEMA, "status": "self-test-passed"}))
685
686
687def main() -> int:
688    parser = argparse.ArgumentParser()
689    parser.add_argument(
690        "--app",
691        type=Path,
692        default=Path("/Applications/Omega.app"),
693        help="the installed candidate whose accessibility tree is read",
694    )
695    parser.add_argument("--candidate-evidence", type=Path)
696    parser.add_argument(
697        "--needle-fd",
698        type=int,
699        default=None,
700        help=(
701            "descriptor already open on the disposable canary the candidate was "
702            "given. Required: this script does not invent a secret nothing has seen."
703        ),
704    )
705    parser.add_argument("--candidate-pid", type=int, default=None)
706    parser.add_argument("--data-root", type=Path, default=None)
707    parser.add_argument("--logs-root", type=Path, default=None)
708    parser.add_argument("--crashes-root", type=Path, default=None)
709    parser.add_argument("--output", type=Path)
710    parser.add_argument("--self-test", action="store_true")
711    args = parser.parse_args()
712
713    if args.self_test:
714        self_test()
715        return 0
716
717    if args.candidate_evidence is None or args.output is None:
718        raise TripwireError("--candidate-evidence and --output are required")
719    if args.needle_fd is None:
720        raise TripwireError(
721            "--needle-fd is required. Pass the descriptor of the disposable "
722            "canary the candidate was actually given, for example "
723            "`--needle-fd 3 3<\"$OMEGA_PRIVATE_PROOF_DIR/disposable-canary\"`. "
724            "A needle this script generates itself has never been seen by any "
725            "other process, so searching for it can only ever pass."
726        )
727
728    evidence = json.loads(args.candidate_evidence.read_text(encoding="utf-8"))
729    # The candidate evidence records a structured digest, not a bare string;
730    # the validator compares against its `value`.
731    candidate_digest = evidence.get("candidate_digest", {}).get("value")
732    if not isinstance(candidate_digest, str) or len(candidate_digest) != 64:
733        raise TripwireError(
734            "candidate evidence has no usable candidate_digest to bind to"
735        )
736
737    needle = read_needle(args.needle_fd)
738
739    surfaces: list[dict[str, Any]] = []
740    details: dict[str, str] = {}
741    workspace = Path(tempfile.mkdtemp(prefix="omega-tripwires-"))
742    os.chmod(workspace, 0o700)
743    try:
744        if not verify_scanner_can_fire(needle, workspace):
745            raise TripwireError(
746                "the scanner did not detect a needle planted in front of it, so "
747                "nothing this run could report about the product would mean "
748                "anything"
749            )
750
751        plan = filesystem_surfaces(
752            data_root(args.data_root),
753            logs_root(args.logs_root),
754            crashes_root(args.crashes_root),
755        )
756        for name in sorted(plan):
757            record, detail = scan_filesystem_surface(name, plan[name], needle)
758            surfaces.append(record)
759            details[name] = detail
760
761        capture, method, reason = capture_pasteboard(workspace)
762        record, detail = scan_captured_surface(
763            "clipboard", "NSPasteboard:general", capture, method or reason, needle
764        )
765        surfaces.append(record)
766        details["clipboard"] = detail
767
768        capture, method, reason = capture_accessibility(
769            workspace, candidate_pid(args.app, args.candidate_pid)
770        )
771        record, detail = scan_captured_surface(
772            "accessibility",
773            f"AXUIElement:{args.app}",
774            capture,
775            method or reason,
776            needle,
777        )
778        surfaces.append(record)
779        details["accessibility"] = detail
780    finally:
781        shutil.rmtree(workspace, ignore_errors=True)
782
783    leaked = [surface["name"] for surface in surfaces if surface["match_detected"]]
784    blocked = [
785        surface["name"] for surface in surfaces if surface["status"] in ("blocked", "error")
786    ]
787    if leaked:
788        status = "fail"
789    elif blocked:
790        status = "blocked"
791    else:
792        status = "pass"
793
794    receipt = {
795        "schema": TRIPWIRE_SCHEMA,
796        "candidate_digest": candidate_digest,
797        "generated_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
798        "status": status,
799        # The value is deliberately absent. Recording the secret you are proving
800        # was not leaked would defeat the exercise.
801        "needle": {
802            "source": "protected_file_descriptor",
803            "length": len(needle),
804            "value_recorded": False,
805        },
806        "surfaces": surfaces,
807    }
808
809    args.output.parent.mkdir(parents=True, exist_ok=True)
810    args.output.write_text(
811        json.dumps(receipt, indent=2, sort_keys=True) + "\n", encoding="utf-8"
812    )
813
814    print(
815        json.dumps(
816            {
817                "schema": TRIPWIRE_SCHEMA,
818                "status": status,
819                "surfaces_scanned": len(surfaces),
820                "leaked_surfaces": leaked,
821                "blocked_surfaces": blocked,
822                # How each surface was read, so a reader can see the coverage
823                # the receipt's counts came from. The receipt itself carries no
824                # free-text field the validator would have to trust.
825                "surface_detail": details,
826                "output": str(args.output),
827            },
828            indent=2,
829            sort_keys=True,
830        )
831    )
832    if leaked:
833        return 1
834    return 3 if blocked else 0
835
836
837if __name__ == "__main__":
838    try:
839        sys.exit(main())
840    except TripwireError as error:
841        print(f"error: {error}", file=sys.stderr)
842        sys.exit(2)
843