tenant.openagents/omega

1#!/usr/bin/env python3
2
3from __future__ import annotations
4
5import argparse
6import concurrent.futures
7import hashlib
8import json
9import os
10import plistlib
11import re
12import secrets
13import subprocess
14import sys
15import tempfile
16import threading
17from pathlib import Path
18from typing import Any
19
20
21SCHEMA = "openagents.omega.identity-proof-matrix.v1"
22PROTOCOL = "openagents.omega.identity-proof.v1"
23SERVICE = "com.openagents.omega.identity-proof.v1"
24ACCOUNT = "disposable-proof-only"
25ALLOWED_BUNDLE_IDENTIFIERS = {"com.openagents.omega", "com.openagents.omega.rc"}
26SHA256 = re.compile(r"^[0-9a-f]{64}$")
27
28
29class ProofError(Exception):
30    pass
31
32
33def digest_bytes(value: bytes) -> str:
34    return hashlib.sha256(value).hexdigest()
35
36
37def digest_file(path: Path) -> str:
38    digest = hashlib.sha256()
39    with path.open("rb") as source:
40        for chunk in iter(lambda: source.read(1024 * 1024), b""):
41            digest.update(chunk)
42    return digest.hexdigest()
43
44
45def canonical_digest(value: Any) -> str:
46    return digest_bytes(
47        json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()
48    )
49
50
51def load_object(path: Path, label: str) -> dict[str, Any]:
52    try:
53        value = json.loads(path.read_text(encoding="utf-8"))
54    except (OSError, UnicodeError, json.JSONDecodeError) as error:
55        raise ProofError(f"{label}_invalid") from error
56    if not isinstance(value, dict):
57        raise ProofError(f"{label}_invalid")
58    return value
59
60
61def require_digest(value: Any, label: str) -> str:
62    if not isinstance(value, str) or SHA256.fullmatch(value) is None:
63        raise ProofError(f"{label}_invalid")
64    return value
65
66
67def require_object(value: Any, label: str) -> dict[str, Any]:
68    if not isinstance(value, dict):
69        raise ProofError(f"{label}_invalid")
70    return value
71
72
73def validate_bundle_identifier(info: dict[str, Any], release: dict[str, Any]) -> None:
74    installed_identifier = info.get("CFBundleIdentifier")
75    release_identifier = release.get("bundle_identifier")
76    if (
77        installed_identifier not in ALLOWED_BUNDLE_IDENTIFIERS
78        or release_identifier != installed_identifier
79    ):
80        raise ProofError("installed_bundle_identifier_invalid")
81
82
83def validate_candidate(
84    app: Path, release_path: Path, artifact: Path, evidence_path: Path
85) -> tuple[Path, dict[str, str]]:
86    if sys.platform != "darwin":
87        raise ProofError("macos_required")
88    if app != Path("/Applications/Omega.app") or app.is_symlink() or not app.is_dir():
89        raise ProofError("installed_app_not_exact")
90    for path, label in (
91        (release_path, "release_record"),
92        (artifact, "artifact"),
93        (evidence_path, "candidate_evidence"),
94    ):
95        if not path.is_absolute() or path.is_symlink() or not path.is_file():
96            raise ProofError(f"{label}_not_regular")
97
98    release = load_object(release_path, "release_record")
99    evidence = load_object(evidence_path, "candidate_evidence")
100    if evidence.get("schema") != "openagents.omega.identity-candidate-evidence.v1":
101        raise ProofError("candidate_evidence_schema_invalid")
102    candidate = evidence.get("candidate")
103    if not isinstance(candidate, dict):
104        raise ProofError("candidate_missing")
105    candidate_digest = require_digest(
106        require_object(evidence.get("candidate_digest"), "candidate_digest").get("value"),
107        "candidate_digest",
108    )
109    if canonical_digest(candidate) != candidate_digest:
110        raise ProofError("candidate_digest_mismatch")
111
112    release_digest = digest_file(release_path)
113    artifact_digest = digest_file(artifact)
114    release_artifact_digest = require_digest(
115        require_object(release.get("digests"), "release_digests").get("package_sha256"),
116        "package_digest",
117    )
118    if release_artifact_digest != artifact_digest:
119        raise ProofError("artifact_digest_mismatch")
120    candidate_artifact = require_object(candidate.get("artifact"), "candidate_artifact")
121    if (
122        candidate_artifact.get("sha256") != artifact_digest
123        or candidate_artifact.get("release_record_sha256") != release_digest
124        or candidate_artifact.get("name") != release.get("artifact_name")
125        or artifact.name != release.get("artifact_name")
126    ):
127        raise ProofError("candidate_artifact_binding_mismatch")
128
129    info_path = app / "Contents/Info.plist"
130    driver = app / "Contents/MacOS/omega-identity-proof"
131    if driver.is_symlink() or not driver.is_file() or info_path.is_symlink():
132        raise ProofError("installed_driver_not_regular")
133    try:
134        info = plistlib.loads(info_path.read_bytes())
135    except (OSError, plistlib.InvalidFileException) as error:
136        raise ProofError("installed_info_invalid") from error
137    validate_bundle_identifier(info, release)
138
139    components = require_object(release.get("components"), "release_components")
140    component = require_object(
141        components.get("identity_proof_driver"), "identity_proof_component"
142    )
143    expected_driver_digest = require_digest(
144        component.get("binary_sha256"), "identity_proof_binary_digest"
145    )
146    if component != {
147        "path": "Contents/MacOS/omega-identity-proof",
148        "binary_sha256": expected_driver_digest,
149        "protocol": PROTOCOL,
150        "keyring_service": SERVICE,
151        "keyring_account": ACCOUNT,
152    }:
153        raise ProofError("identity_proof_component_invalid")
154    driver_digest = digest_file(driver)
155    packaged = require_object(
156        candidate_artifact.get("packaged_app"), "candidate_packaged_app"
157    )
158    if (
159        driver_digest != expected_driver_digest
160        or packaged.get("identity_proof_binary_sha256") != driver_digest
161    ):
162        raise ProofError("installed_driver_digest_mismatch")
163    signature = subprocess.run(
164        ["codesign", "--verify", "--strict", "--verbose=2", str(driver)],
165        stdout=subprocess.DEVNULL,
166        stderr=subprocess.DEVNULL,
167    )
168    if signature.returncode != 0:
169        raise ProofError("installed_driver_signature_invalid")
170    return driver, {
171        "candidate_digest": candidate_digest,
172        "artifact_sha256": artifact_digest,
173        "release_record_sha256": release_digest,
174        "identity_proof_binary_sha256": driver_digest,
175    }
176
177
178class Driver:
179    def __init__(self, binary: Path):
180        self.binary = binary
181
182    def invoke(
183        self,
184        root: Path,
185        command: list[str],
186        expected_crash: bool = False,
187        offline: bool = False,
188        password: bytearray | None = None,
189    ) -> dict[str, Any]:
190        arguments = [str(self.binary), "--root", str(root), *command]
191        read_descriptor: int | None = None
192        write_descriptor: int | None = None
193        try:
194            if password is not None:
195                read_descriptor, write_descriptor = os.pipe()
196                os.write(write_descriptor, password)
197                os.close(write_descriptor)
198                write_descriptor = None
199                arguments.extend(["--password-fd", str(read_descriptor)])
200            if offline:
201                sandbox = Path("/usr/bin/sandbox-exec")
202                if not sandbox.is_file():
203                    raise ProofError("deny_network_sandbox_unavailable")
204                arguments = [
205                    str(sandbox),
206                    "-p",
207                    "(version 1) (allow default) (deny network*)",
208                    *arguments,
209                ]
210            process = subprocess.run(
211                arguments,
212                capture_output=True,
213                text=True,
214                timeout=60,
215                pass_fds=() if read_descriptor is None else (read_descriptor,),
216            )
217        finally:
218            if write_descriptor is not None:
219                os.close(write_descriptor)
220            if read_descriptor is not None:
221                os.close(read_descriptor)
222        if expected_crash:
223            if process.returncode not in (-9, 137):
224                raise ProofError("expected_crash_not_observed")
225            return {"crash": "observed"}
226        if process.returncode != 0:
227            raise ProofError("driver_command_rejected")
228        try:
229            outcome = json.loads(process.stdout)
230        except json.JSONDecodeError as error:
231            raise ProofError("driver_output_invalid") from error
232        if not isinstance(outcome, dict) or outcome.get("protocol") != PROTOCOL:
233            raise ProofError("driver_protocol_invalid")
234        return outcome
235
236    def rejected(self, root: Path, command: list[str]) -> str:
237        process = subprocess.run(
238            [str(self.binary), "--root", str(root), *command],
239            capture_output=True,
240            text=True,
241            timeout=30,
242        )
243        if process.returncode == 0:
244            raise ProofError("expected_rejection_not_observed")
245        return "rejected"
246
247
248class FakeDriver:
249    def __init__(self):
250        self.lock = threading.Lock()
251        self.states: dict[str, dict[str, Any]] = {}
252        self.keychain_access = False
253
254    def invoke(
255        self,
256        root: Path,
257        command: list[str],
258        expected_crash: bool = False,
259        offline: bool = False,
260        password: bytearray | None = None,
261    ) -> dict[str, Any]:
262        with self.lock:
263            key = str(root)
264            name = command[0]
265            if name == "initialize":
266                self.states[key] = {
267                    "state": "absent",
268                    "identity": None,
269                    "receipt": None,
270                    "recovery_identity": None,
271                    "password_digest": None,
272                }
273                return self.outcome(
274                    "initialized",
275                    {
276                        "root": key,
277                        "keyring_service": SERVICE,
278                        "keyring_account": ACCOUNT,
279                    },
280                )
281            state = self.states[key]
282            if name == "safety":
283                return self.outcome(
284                    "safety-checked", {"production_locator_access": "rejected-by-construction"}
285                )
286            if name == "create":
287                receipt = command[command.index("--receipt") + 1]
288                if state["identity"] is None:
289                    state.update(
290                        state="ready",
291                        identity=f"omega-nostr-self-test-{len(self.states):02d}",
292                        receipt=receipt,
293                    )
294                elif state["receipt"] != receipt:
295                    raise ProofError("fake_distinct_create")
296                if expected_crash:
297                    return {"crash": "observed"}
298                return self.custody("create-complete", state)
299            if name == "resume-create":
300                return self.custody("create-resumed", state)
301            if name in ("inspect", "process-start"):
302                return self.custody("inspection" if name == "inspect" else "process-start", state)
303            if name == "sign":
304                return self.outcome("signed", {"event_id": "d" * 64, "signature": "e" * 128})
305            if name == "probe-forged":
306                return self.outcome("forged-request-rejected", {"typed_error": "rejected"})
307            if name == "probe-stale":
308                return self.outcome("stale-request-rejected", {"typed_error": "rejected"})
309            if name == "simulate-safe":
310                scenario = command[command.index("--scenario") + 1]
311                return self.outcome(
312                    "safe-scenario-simulated",
313                    {
314                        "scenario": scenario,
315                        "mode": "deterministic-no-keychain-simulation",
316                        "expected_outcome": "typed-rejection",
317                        "production_locator_access": "rejected-by-construction",
318                    },
319                )
320            if name == "protect-recovery":
321                if not offline or password is None:
322                    raise ProofError("fake_recovery_boundary_missing")
323                state["recovery_identity"] = state["identity"]
324                state["password_digest"] = digest_bytes(password)
325                return self.custody("recovery-protected", state)
326            if name == "probe-wrong-recovery":
327                if not offline or password is None:
328                    raise ProofError("fake_recovery_boundary_missing")
329                if digest_bytes(password) == state["password_digest"]:
330                    raise ProofError("fake_wrong_password_not_wrong")
331                return self.outcome(
332                    "wrong-recovery-password-rejected",
333                    {
334                        "scenario": "wrong-password",
335                        "expected_outcome": "recovery-decryption-rejected",
336                        "production_locator_access": "rejected-by-construction",
337                    },
338                )
339            if name == "probe-corrupt-recovery":
340                if not offline or password is None:
341                    raise ProofError("fake_recovery_boundary_missing")
342                return self.outcome(
343                    "corrupt-recovery-artifact-rejected",
344                    {
345                        "scenario": "corrupt-artifact",
346                        "expected_outcome": "invalid-recovery-artifact-rejected",
347                        "production_locator_access": "rejected-by-construction",
348                    },
349                )
350            if name == "recover":
351                if not offline or password is None:
352                    raise ProofError("fake_recovery_boundary_missing")
353                if digest_bytes(password) != state["password_digest"]:
354                    raise ProofError("fake_recovery_password_invalid")
355                state.update(
356                    state="ready",
357                    identity=state["recovery_identity"],
358                    receipt=command[command.index("--receipt") + 1],
359                )
360                return self.custody("recovery-complete", state)
361            if name == "reset":
362                state["reset_pending"] = True
363                if expected_crash:
364                    return {"crash": "observed"}
365                return self.custody("reset-marked", state, displayed_state="reset-pending")
366            if name == "resume-reset":
367                state.update(state="absent", identity=None, receipt=None, reset_pending=False)
368                if expected_crash:
369                    return {"crash": "observed"}
370                return self.custody("reset-resumed", state)
371            raise ProofError("fake_command_unknown")
372
373    def rejected(self, root: Path, command: list[str]) -> str:
374        with self.lock:
375            state = self.states[str(root)]
376            if command[0] == "create":
377                receipt = command[command.index("--receipt") + 1]
378                if state["receipt"] != receipt:
379                    return "rejected"
380            raise ProofError("fake_expected_rejection_missing")
381
382    @staticmethod
383    def outcome(name: str, facts: Any) -> dict[str, Any]:
384        return {"protocol": PROTOCOL, "outcome": name, "facts": facts}
385
386    def custody(
387        self, name: str, state: dict[str, Any], displayed_state: str | None = None
388    ) -> dict[str, Any]:
389        identity = state["identity"]
390        facts = {
391            "state": displayed_state or state["state"],
392            "identity": None if identity is None else {"identity_ref": identity},
393        }
394        return self.outcome(name, facts)
395
396
397def find_identity(value: Any) -> str:
398    if isinstance(value, dict):
399        direct = value.get("identity_ref")
400        if isinstance(direct, str) and direct:
401            return direct
402        for child in value.values():
403            try:
404                return find_identity(child)
405            except ProofError:
406                pass
407    elif isinstance(value, list):
408        for child in value:
409            try:
410                return find_identity(child)
411            except ProofError:
412                pass
413    raise ProofError("public_identity_missing")
414
415
416def find_state(value: Any) -> str:
417    if isinstance(value, dict):
418        direct = value.get("state")
419        if isinstance(direct, str):
420            return direct
421        for child in value.values():
422            try:
423                return find_state(child)
424            except ProofError:
425                pass
426    elif isinstance(value, list):
427        for child in value:
428            try:
429                return find_state(child)
430            except ProofError:
431                pass
432    raise ProofError("custody_state_missing")
433
434
435class Matrix:
436    def __init__(self, driver: Driver, base: Path):
437        self.driver = driver
438        self.base = base
439        self.counter = 0
440        self.roots: list[Path] = []
441        self.active: tuple[Path, str] | None = None
442        self.results: list[dict[str, str]] = []
443
444    def root(self) -> Path:
445        self.counter += 1
446        root = self.base / f"omega-identity-proof-case-{self.counter:02d}"
447        initialized = self.driver.invoke(root, ["initialize", "--confirm-disposable"])
448        facts = initialized.get("facts", {})
449        if (
450            facts.get("keyring_service") != SERVICE
451            or facts.get("keyring_account") != ACCOUNT
452        ):
453            raise ProofError("disposable_locator_mismatch")
454        self.roots.append(root)
455        return root
456
457    def record(self, name: str, facts: Any) -> None:
458        self.results.append(
459            {"case": name, "status": "passed", "evidence_sha256": canonical_digest(facts)}
460        )
461
462    def create(
463        self,
464        root: Path,
465        receipt: str,
466        crash: str | None = None,
467        offline: bool = False,
468    ) -> str:
469        command = ["create", "--receipt", receipt]
470        if crash:
471            command.extend(["--crash-after", crash])
472            self.driver.invoke(root, command, expected_crash=True, offline=offline)
473            created = self.driver.invoke(root, ["resume-create"])
474        else:
475            created = self.driver.invoke(root, command, offline=offline)
476        identity = find_identity(created)
477        self.active = (root, identity)
478        return identity
479
480    def cleanup(self) -> None:
481        if self.active is None:
482            return
483        root, identity = self.active
484        try:
485            reset = self.driver.invoke(
486                root,
487                ["reset", "--identity-ref", identity, "--receipt", "matrix-cleanup"],
488            )
489            if reset.get("outcome") != "reset-marked":
490                raise ProofError("cleanup_reset_not_marked")
491            self.driver.invoke(root, ["resume-reset"])
492            started = self.driver.invoke(root, ["process-start"])
493            if started.get("outcome") != "process-start":
494                raise ProofError("cleanup_relaunch_not_acknowledged")
495        except Exception as error:
496            raise ProofError("disposable_keychain_cleanup_failed") from error
497        finally:
498            self.active = None
499
500    def verify_all_clean(self) -> None:
501        cleanup_error = False
502        for root in self.roots:
503            try:
504                inspected = self.driver.invoke(root, ["inspect"])
505                try:
506                    identity = find_identity(inspected)
507                except ProofError:
508                    identity = None
509                if identity is not None:
510                    self.active = (root, identity)
511                    self.cleanup()
512                final = self.driver.invoke(root, ["process-start"])
513                if find_state(final) != "absent":
514                    cleanup_error = True
515            except ProofError:
516                cleanup_error = True
517        if cleanup_error:
518            raise ProofError("disposable_keychain_cleanup_failed")
519
520    def run(self) -> list[dict[str, str]]:
521        try:
522            root = self.root()
523            safety = self.driver.invoke(root, ["safety"])
524            if safety.get("facts", {}).get("production_locator_access") != "rejected-by-construction":
525                raise ProofError("production_locator_not_rejected")
526            self.record("disposable-namespace-safety", safety)
527
528            recovery_password = bytearray(secrets.token_urlsafe(48).encode("ascii"))
529            wrong_password = bytearray(secrets.token_urlsafe(48).encode("ascii"))
530            try:
531                root = self.root()
532                recovery_identity = self.create(
533                    root,
534                    "matrix-offline-recovery-create",
535                    offline=True,
536                )
537                protected = self.driver.invoke(
538                    root,
539                    ["protect-recovery", "--identity-ref", recovery_identity],
540                    offline=True,
541                    password=recovery_password,
542                )
543                wrong = self.driver.invoke(
544                    root,
545                    ["probe-wrong-recovery"],
546                    offline=True,
547                    password=wrong_password,
548                )
549                corrupt = self.driver.invoke(
550                    root,
551                    ["probe-corrupt-recovery"],
552                    offline=True,
553                    password=recovery_password,
554                )
555                if (
556                    wrong.get("outcome") != "wrong-recovery-password-rejected"
557                    or wrong.get("facts", {}).get("scenario") != "wrong-password"
558                    or corrupt.get("outcome") != "corrupt-recovery-artifact-rejected"
559                    or corrupt.get("facts", {}).get("scenario") != "corrupt-artifact"
560                ):
561                    raise ProofError("recovery_rejection_result_invalid")
562                self.record(
563                    "offline-create-and-encrypted-recovery-protection",
564                    {"network_policy": "deny-network", "outcome": protected},
565                )
566                self.record("wrong-recovery-password-rejection", wrong)
567                self.record("corrupt-recovery-artifact-rejection", corrupt)
568                self.cleanup()
569                recovered = self.driver.invoke(
570                    root,
571                    ["recover", "--receipt", "matrix-offline-recovery-adopt"],
572                    offline=True,
573                    password=recovery_password,
574                )
575                if find_identity(recovered) != recovery_identity:
576                    raise ProofError("recovery_changed_identity")
577                self.active = (root, recovery_identity)
578                restarted = self.driver.invoke(root, ["process-start"])
579                if find_identity(restarted) != recovery_identity:
580                    raise ProofError("recovery_restart_changed_identity")
581                self.record(
582                    "encrypted-recovery-and-restart-continuity",
583                    {
584                        "network_policy": "deny-network",
585                        "recovery": recovered,
586                        "restart": restarted,
587                    },
588                )
589                self.cleanup()
590            finally:
591                recovery_password[:] = b"\0" * len(recovery_password)
592                wrong_password[:] = b"\0" * len(wrong_password)
593
594            safe_scenarios = (
595                "conflict-custody",
596                "lost-custody",
597                "locked-custody",
598                "symlink-refusal",
599                "weak-permission-refusal",
600                "keychain-unavailable",
601                "corrupt-keychain",
602                "malformed-event-rejection",
603                "unadmitted-purpose-rejection",
604                "conflicting-recovery-selection",
605                "late-completion-fencing",
606                "signer-crash-before-completion",
607            )
608            for scenario in safe_scenarios:
609                result = self.driver.invoke(
610                    root, ["simulate-safe", "--scenario", scenario]
611                )
612                facts = result.get("facts", {})
613                if (
614                    result.get("outcome") != "safe-scenario-simulated"
615                    or facts.get("scenario") != scenario
616                    or facts.get("mode") != "deterministic-no-keychain-simulation"
617                    or facts.get("production_locator_access")
618                    != "rejected-by-construction"
619                ):
620                    raise ProofError("safe_scenario_result_invalid")
621                self.record(f"simulated-{scenario}", result)
622
623            identity = self.create(root, "matrix-create")
624            inspected = self.driver.invoke(root, ["inspect"])
625            restarted = self.driver.invoke(root, ["process-start"])
626            signed = self.driver.invoke(
627                root,
628                ["sign", "--identity-ref", identity, "--request", "matrix-sign"],
629            )
630            self.record("create-read-back-restart-sign", [inspected, restarted, signed])
631            repeated = self.driver.invoke(root, ["create", "--receipt", "matrix-create"])
632            if find_identity(repeated) != identity:
633                raise ProofError("double_create_changed_identity")
634            distinct = self.driver.rejected(
635                root, ["create", "--receipt", "matrix-distinct-create"]
636            )
637            self.record("double-create", [repeated, distinct])
638            forged = self.driver.invoke(root, ["probe-forged", "--request", "matrix-forged"])
639            self.record("forged-request-rejection", forged)
640            self.cleanup()
641
642            root = self.root()
643            with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
644                futures = [
645                    executor.submit(
646                        self.driver.invoke,
647                        root,
648                        ["create", "--receipt", "matrix-concurrent-create"],
649                    )
650                    for _ in range(2)
651                ]
652                concurrent_results = [future.result() for future in futures]
653            identities = {find_identity(result) for result in concurrent_results}
654            if len(identities) != 1:
655                raise ProofError("concurrent_create_changed_identity")
656            current_identity = identities.pop()
657            self.active = (root, current_identity)
658            with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
659                starts = [
660                    executor.submit(self.driver.invoke, root, ["process-start"])
661                    for _ in range(2)
662                ]
663                start_results = [future.result() for future in starts]
664            self.record("concurrent-create-and-process-start", [concurrent_results, start_results])
665            stale_identity = current_identity
666            self.cleanup()
667
668            root = self.root()
669            current_identity = self.create(root, "matrix-current-after-reset")
670            stale = self.driver.invoke(
671                root,
672                [
673                    "probe-stale",
674                    "--stale-identity-ref",
675                    stale_identity,
676                    "--request",
677                    "matrix-stale",
678                ],
679            )
680            self.record("stale-request-rejection", stale)
681            self.cleanup()
682
683            for crash in ("secret-write", "secret-read-back", "manifest-commit"):
684                root = self.root()
685                self.create(root, f"matrix-crash-{crash}", crash)
686                self.record(f"crash-after-{crash}", {"crash": "observed", "resume": "passed"})
687                self.cleanup()
688
689            root = self.root()
690            identity = self.create(root, "matrix-reset-marker-create")
691            self.driver.invoke(
692                root,
693                [
694                    "reset",
695                    "--identity-ref",
696                    identity,
697                    "--receipt",
698                    "matrix-reset-marker",
699                    "--crash-after",
700                    "reset-marker",
701                ],
702                expected_crash=True,
703            )
704            self.driver.invoke(root, ["resume-reset"])
705            self.driver.invoke(root, ["process-start"])
706            self.active = None
707            self.record("crash-after-reset-marker", {"crash": "observed", "resume": "passed"})
708
709            root = self.root()
710            identity = self.create(root, "matrix-reset-commit-create")
711            self.driver.invoke(
712                root,
713                ["reset", "--identity-ref", identity, "--receipt", "matrix-reset-commit"],
714            )
715            self.driver.invoke(
716                root, ["resume-reset", "--crash-after", "reset-commit"], expected_crash=True
717            )
718            self.driver.invoke(root, ["process-start"])
719            self.active = None
720            self.record("crash-after-reset-commit", {"crash": "observed", "resume": "passed"})
721
722            root = self.root()
723            identity = self.create(root, "matrix-relaunch-create")
724            self.driver.invoke(
725                root,
726                ["reset", "--identity-ref", identity, "--receipt", "matrix-relaunch"],
727            )
728            self.driver.invoke(root, ["resume-reset"])
729            self.driver.invoke(
730                root,
731                ["process-start", "--crash-after", "relaunch-acknowledge"],
732                expected_crash=True,
733            )
734            self.driver.invoke(root, ["process-start"])
735            self.active = None
736            self.record(
737                "crash-after-relaunch-acknowledge",
738                {"crash": "observed", "resume": "passed"},
739            )
740            return self.results
741        finally:
742            self.cleanup()
743            self.verify_all_clean()
744
745
746def run_matrix(driver: Path, bindings: dict[str, str], output: Path) -> None:
747    if not output.is_absolute() or output.is_symlink():
748        raise ProofError("output_path_invalid")
749    output.parent.mkdir(parents=True, exist_ok=True)
750    with tempfile.TemporaryDirectory(prefix="omega-proof-matrix-") as temporary:
751        base = Path(temporary).resolve()
752        results = Matrix(Driver(driver), base).run()
753    record = {
754        "schema": SCHEMA,
755        "status": "passed",
756        "candidate": bindings,
757        "disposable_keychain": {
758            "service": SERVICE,
759            "account": ACCOUNT,
760            "cleanup": "passed",
761            "production_locator_access": "rejected-by-construction",
762        },
763        "cases": results,
764    }
765    record["evidence_sha256"] = canonical_digest(record)
766    output.write_text(json.dumps(record, indent=2, sort_keys=True) + "\n", encoding="utf-8")
767    print(json.dumps({"schema": SCHEMA, "status": "passed", "evidence_sha256": record["evidence_sha256"]}))
768
769
770def self_test() -> None:
771    candidate = {"artifact": {"sha256": "a" * 64}, "omega": {"commit": "b" * 40}}
772    digest = canonical_digest(candidate)
773    if canonical_digest(json.loads(json.dumps(candidate))) != digest:
774        raise ProofError("canonical_digest_not_deterministic")
775    public_record = {
776        "schema": SCHEMA,
777        "candidate": {"candidate_digest": digest},
778        "disposable_keychain": {"service": SERVICE, "account": ACCOUNT},
779        "cases": [{"case": "fixture", "status": "passed", "evidence_sha256": "c" * 64}],
780    }
781    encoded = json.dumps(public_record, sort_keys=True)
782    forbidden = ("nsec", "private_key", "/Users/", "/tmp/", "identity.json")
783    if any(value in encoded for value in forbidden):
784        raise ProofError("public_evidence_leak")
785    if require_digest(digest, "candidate_digest") != digest:
786        raise ProofError("digest_validation_failed")
787    validate_bundle_identifier(
788        {"CFBundleIdentifier": "com.openagents.omega.rc"},
789        {"bundle_identifier": "com.openagents.omega.rc"},
790    )
791    for installed_identifier, release_identifier in (
792        ("com.openagents.omega.rc", "com.openagents.omega"),
793        ("com.example.omega", "com.example.omega"),
794    ):
795        try:
796            validate_bundle_identifier(
797                {"CFBundleIdentifier": installed_identifier},
798                {"bundle_identifier": release_identifier},
799            )
800        except ProofError:
801            pass
802        else:
803            raise ProofError("bundle_identifier_validation_failed")
804    try:
805        require_digest("A" * 64, "candidate_digest")
806    except ProofError:
807        pass
808    else:
809        raise ProofError("uppercase_digest_accepted")
810    with tempfile.TemporaryDirectory(prefix="omega-proof-matrix-self-test-") as temporary:
811        fake = FakeDriver()
812        cases = Matrix(fake, Path(temporary).resolve()).run()
813    expected_cases = {
814        "disposable-namespace-safety",
815        "create-read-back-restart-sign",
816        "double-create",
817        "concurrent-create-and-process-start",
818        "forged-request-rejection",
819        "stale-request-rejection",
820        "crash-after-secret-write",
821        "crash-after-secret-read-back",
822        "crash-after-manifest-commit",
823        "crash-after-reset-marker",
824        "crash-after-reset-commit",
825        "crash-after-relaunch-acknowledge",
826        "simulated-conflict-custody",
827        "simulated-lost-custody",
828        "simulated-locked-custody",
829        "simulated-symlink-refusal",
830        "simulated-weak-permission-refusal",
831        "simulated-keychain-unavailable",
832        "simulated-corrupt-keychain",
833        "simulated-malformed-event-rejection",
834        "simulated-unadmitted-purpose-rejection",
835        "simulated-conflicting-recovery-selection",
836        "simulated-late-completion-fencing",
837        "simulated-signer-crash-before-completion",
838        "offline-create-and-encrypted-recovery-protection",
839        "wrong-recovery-password-rejection",
840        "corrupt-recovery-artifact-rejection",
841        "encrypted-recovery-and-restart-continuity",
842    }
843    if {case["case"] for case in cases} != expected_cases:
844        raise ProofError("matrix_case_coverage_incomplete")
845    if fake.keychain_access:
846        raise ProofError("self_test_touched_keychain")
847    print(json.dumps({"schema": SCHEMA, "status": "self-test-passed", "keychain_access": "none"}))
848
849
850def parse_args() -> argparse.Namespace:
851    parser = argparse.ArgumentParser(
852        description="Run the installed signed Omega disposable identity proof matrix"
853    )
854    parser.add_argument("--self-test", action="store_true")
855    parser.add_argument("--release-record", type=Path)
856    parser.add_argument("--artifact", type=Path)
857    parser.add_argument("--candidate-evidence", type=Path)
858    parser.add_argument("--app", type=Path, default=Path("/Applications/Omega.app"))
859    parser.add_argument(
860        "--output",
861        type=Path,
862        default=Path.cwd() / "target/omega-identity-proof/matrix-evidence.json",
863    )
864    return parser.parse_args()
865
866
867def main() -> int:
868    args = parse_args()
869    try:
870        if args.self_test:
871            self_test()
872            return 0
873        if args.release_record is None or args.artifact is None or args.candidate_evidence is None:
874            raise ProofError("candidate_inputs_required")
875        app = args.app
876        release = args.release_record
877        artifact = args.artifact
878        evidence = args.candidate_evidence
879        output = args.output
880        driver, bindings = validate_candidate(app, release, artifact, evidence)
881        run_matrix(driver, bindings, output)
882        return 0
883    except ProofError as error:
884        print(json.dumps({"schema": SCHEMA, "status": "failed", "reason": str(error)}))
885        return 1
886    except Exception:
887        print(json.dumps({"schema": SCHEMA, "status": "failed", "reason": "unexpected_error"}))
888        return 1
889
890
891if __name__ == "__main__":
892    sys.exit(main())
893