tenant.openagents/omega

1#!/usr/bin/env python3
2"""Generate candidate-bound Omega identity evidence without self-attestation."""
3
4from __future__ import annotations
5
6import argparse
7import hashlib
8import json
9import plistlib
10import re
11import shutil
12import subprocess
13import sys
14import tempfile
15from contextlib import contextmanager
16from datetime import datetime
17from pathlib import Path
18from typing import Any, Iterator
19
20from omega_identity_evidence import (
21    IdentityEvidenceError,
22    resolve_evidence_reference,
23    validate_identity_matrix,
24    validate_installed_observations,
25    validate_installed_tripwires,
26    validate_recovery_evidence,
27)
28
29
30ROOT = Path(__file__).resolve().parent.parent
31CONTRACT_PATH = Path("crates/omega_identity/src/contract.rs")
32VECTOR_PATH = Path("crates/omega_identity/fixtures/omega_nostr_identity_v1.json")
33SCHEMA_PATH = Path(
34    "crates/omega_identity/fixtures/identity_candidate_evidence_schema_v1.json"
35)
36VERIFIER_PATH = Path("script/verify-omega-identity")
37PINNED_SIGNING_IDENTITY = (
38    "Developer ID Application: OpenAgents, Inc. (HQWSG26L43)"
39)
40PINNED_TEAM_ID = "HQWSG26L43"
41ASSURANCE_SPEC_ADMISSION = {
42    "assurance_spec_id": "assurance.omega.identity.first.onboarding",
43    "assurance_revision": 2,
44    "admitted_document_sha256": "84dfc9daa7b3f6f92b1895b78854835ce58530c2b81f220f6ca3b62c799f72b1",
45    "admitted_proposal_sha256": "86a0b414be71df319c1fbf705eae5b7c1cc7d8405caa67478b11ab84baa0f6cd",
46    "admission_receipt_ref": "authority.decision.2d660f7602ee99ac6c4a7a2fe906ba0f",
47    "admission_receipt_sha256": "4336ef7443cba679f2aac0176761b8a8e3ed7e2b3e93cefc25c66948e80f2f96",
48    "product_spec_sha256": "fdbf66ee5c9c89a357be14d19aa21197313ab39b2c95bdc330743bda37a91c12",
49}
50EXPECTED_PACKAGES = {
51    "atomic-write-file": {
52        "version": "0.3.0",
53        "checksum": "84790c55b5704b0d35130bf16a4ce22a8e70eb0ea773522557524d9a4852663d",
54    },
55    "keyring": {
56        "version": "3.6.3",
57        "checksum": "eebcc3aff044e5944a8fbaf69eb277d11986064cba30c468730e8b9909fb551c",
58    },
59    "nostr": {
60        "version": "0.44.4",
61        "checksum": "98cf5d15d70d1f8f4059e5f79923ac15891eb691d2843d01191e0585fb064d70",
62    },
63}
64GATE_REQUIREMENTS = {
65    "assurance_spec_admission": [
66        "admitted AssuranceSpec identifier and digest",
67        "candidate-bound result for every admitted scenario",
68    ],
69    "custody_scenarios": [
70        "pristine install",
71        "existing identity",
72        "conflict custody",
73        "lost custody",
74        "locked custody",
75        "symlink and weak-permission custody refusal",
76        "keychain unavailable",
77        "corrupt keychain data",
78    ],
79    "recovery_scenarios": [
80        "offline identity creation",
81        "offline encrypted recovery",
82        "corrupt recovery artifact rejection",
83        "wrong recovery password rejection",
84        "restart and rollback identity continuity",
85    ],
86    "forged_request_rejection": [
87        "wrong-identity signing request rejection",
88        "unadmitted purpose and malformed event rejection",
89        "owner-selection proof for conflicting recovery candidates",
90    ],
91    "stale_task_fencing": [
92        "secret-write, read-back, manifest-commit, reset, and relaunch crash points",
93        "double Create and concurrent application starts",
94        "late background completion cannot replace newer durable state",
95        "signer crash before completion",
96    ],
97    "installed_secret_tripwires": [
98        "installed-candidate logs scan",
99        "installed-candidate telemetry scan",
100        "installed-candidate clipboard scan",
101        "installed-candidate UI/accessibility tree scan",
102        "installed-candidate diagnostics scan",
103        "installed-candidate crash-output scan",
104    ],
105    "manual_journey": [
106        "identity-first first run",
107        "Theme and Agent Setup accepted-baseline comparison",
108        "Zed data before/after digest comparison",
109    ],
110    "owner_observation": [
111        "named owner observation bound to candidate_digest",
112    ],
113    "independent_verification": [
114        "named independent verifier record bound to candidate_digest",
115    ],
116    "accessibility": [
117        "keyboard and focus traversal",
118        "screen-reader output",
119        "360-pixel width and larger UI font",
120        "light, dark, high-contrast, and reduced-motion results",
121    ],
122    "install_lifecycle": [
123        "pristine install and side-by-side Zed result",
124        "restart, update, downgrade, and rollback result",
125        "uninstall and reinstall result",
126        "identity continuity and Zed-isolation result",
127    ],
128}
129STRUCTURED_RECEIPT_GATES = {
130    "custody_scenarios": "identity_matrix",
131    "recovery_scenarios": "recovery_evidence",
132    "forged_request_rejection": "identity_matrix",
133    "stale_task_fencing": "identity_matrix",
134    "installed_secret_tripwires": "installed_tripwires",
135    "manual_journey": "installed_observations",
136    "accessibility": "installed_observations",
137}
138
139
140class EvidenceError(RuntimeError):
141    pass
142
143
144def sha256_file(path: Path) -> str:
145    digest = hashlib.sha256()
146    with path.open("rb") as handle:
147        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
148            digest.update(chunk)
149    return digest.hexdigest()
150
151
152def canonical_digest(value: Any) -> str:
153    canonical = json.dumps(
154        value, ensure_ascii=True, separators=(",", ":"), sort_keys=True
155    ).encode("utf-8")
156    return hashlib.sha256(canonical).hexdigest()
157
158
159def git(*arguments: str) -> str:
160    result = subprocess.run(
161        ["git", *arguments],
162        cwd=ROOT,
163        check=True,
164        capture_output=True,
165        text=True,
166    )
167    return result.stdout.strip()
168
169
170def require_match(pattern: str, text: str, label: str) -> str:
171    match = re.search(pattern, text, flags=re.MULTILINE)
172    if match is None:
173        raise EvidenceError(f"could not read {label}")
174    return match.group(1)
175
176
177def require_sha256(value: Any, label: str) -> str:
178    if not isinstance(value, str) or re.fullmatch(r"[0-9a-f]{64}", value) is None:
179        raise EvidenceError(f"{label} must be a lowercase SHA-256 digest")
180    return value
181
182
183def cargo_lock_packages(lock_text: str) -> dict[str, dict[str, str]]:
184    packages: dict[str, dict[str, str]] = {}
185    for block in lock_text.split("[[package]]")[1:]:
186        name_match = re.search(r'^name = "([^"]+)"$', block, flags=re.MULTILINE)
187        version_match = re.search(r'^version = "([^"]+)"$', block, flags=re.MULTILINE)
188        checksum_match = re.search(
189            r'^checksum = "([0-9a-f]{64})"$', block, flags=re.MULTILINE
190        )
191        if name_match is None or version_match is None or checksum_match is None:
192            continue
193        name = name_match.group(1)
194        if name in EXPECTED_PACKAGES:
195            if name in packages:
196                raise EvidenceError(
197                    f"Cargo.lock snapshot contains more than one checksummed {name}"
198                )
199            packages[name] = {
200                "version": version_match.group(1),
201                "checksum": checksum_match.group(1),
202            }
203    return packages
204
205
206def run_identity_verifier(verifier_path: Path) -> None:
207    result = subprocess.run(
208        [str(verifier_path)],
209        cwd=ROOT,
210        capture_output=True,
211        text=True,
212    )
213    if result.returncode != 0:
214        detail = result.stderr.strip() or result.stdout.strip()
215        raise EvidenceError(f"identity verifier failed: {detail}")
216
217
218def verify_codesign(
219    path: Path, release_record: dict[str, Any], artifact_kind: str
220) -> dict[str, Any]:
221    if shutil.which("codesign") is None:
222        raise EvidenceError("codesign is required to validate the macOS candidate")
223    verification = subprocess.run(
224        ["codesign", "--verify", "--deep", "--strict", str(path)],
225        capture_output=True,
226        text=True,
227    )
228    if verification.returncode != 0:
229        detail = verification.stderr.strip() or verification.stdout.strip()
230        raise EvidenceError(f"{artifact_kind} codesign verification failed: {detail}")
231    details = subprocess.run(
232        ["codesign", "-dv", "--verbose=4", str(path)],
233        capture_output=True,
234        text=True,
235    )
236    if details.returncode != 0:
237        detail = details.stderr.strip() or details.stdout.strip()
238        raise EvidenceError(f"{artifact_kind} codesign metadata failed: {detail}")
239    metadata = "\n".join((details.stdout, details.stderr))
240    authorities = re.findall(r"(?m)^Authority=(.+)$", metadata)
241    team_match = re.search(r"(?m)^TeamIdentifier=(.+)$", metadata)
242    team_id = None if team_match is None else team_match.group(1)
243    if PINNED_SIGNING_IDENTITY not in authorities:
244        raise EvidenceError(
245            f"{artifact_kind} signing Authority does not match release record"
246        )
247    if team_id != PINNED_TEAM_ID:
248        raise EvidenceError(
249            f"{artifact_kind} TeamIdentifier does not match release record"
250        )
251    return {
252        "verified": True,
253        "authority": PINNED_SIGNING_IDENTITY,
254        "team_id": team_id,
255    }
256
257
258def file_contains_literal(path: Path, literal: bytes) -> bool:
259    overlap = max(len(literal) - 1, 0)
260    previous = b""
261    with path.open("rb") as handle:
262        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
263            combined = previous + chunk
264            if literal in combined:
265                return True
266            previous = combined[-overlap:] if overlap else b""
267    return False
268
269
270def scan_packaged_app(
271    application: Path, forbidden_literals: list[str]
272) -> dict[str, Any]:
273    regular_files = sorted(
274        path
275        for path in application.rglob("*")
276        if path.is_file() and not path.is_symlink()
277    )
278    matches: dict[str, list[str]] = {}
279    for literal in forbidden_literals:
280        encoded = literal.encode("utf-8")
281        matching_paths = [
282            path.relative_to(application).as_posix()
283            for path in regular_files
284            if file_contains_literal(path, encoded)
285        ]
286        if matching_paths:
287            matches[literal] = matching_paths
288    if matches:
289        summary = ", ".join(
290            f"{literal}: {paths}" for literal, paths in sorted(matches.items())
291        )
292        raise EvidenceError(
293            f"fresh mounted-app identity boundary scan found forbidden literals: {summary}"
294        )
295    return {
296        "status": "passed",
297        "forbidden_literals": forbidden_literals,
298        "matches": {},
299        "scope": "all regular, non-symlink files under mounted Omega.app",
300    }
301
302
303@contextmanager
304def mounted_dmg(artifact_path: Path) -> Iterator[Path]:
305    if shutil.which("hdiutil") is None:
306        raise EvidenceError("hdiutil is required to validate the macOS candidate DMG")
307    with tempfile.TemporaryDirectory(prefix="omega-identity-dmg-") as directory:
308        mountpoint = Path(directory)
309        attach = subprocess.run(
310            [
311                "hdiutil",
312                "attach",
313                "-readonly",
314                "-nobrowse",
315                "-mountpoint",
316                str(mountpoint),
317                str(artifact_path),
318            ],
319            capture_output=True,
320            text=True,
321        )
322        if attach.returncode != 0:
323            detail = attach.stderr.strip() or attach.stdout.strip()
324            raise EvidenceError(f"artifact is not a mountable read-only DMG: {detail}")
325        try:
326            yield mountpoint
327        finally:
328            detach = subprocess.run(
329                ["hdiutil", "detach", str(mountpoint)],
330                capture_output=True,
331                text=True,
332            )
333            if detach.returncode != 0:
334                forced_detach = subprocess.run(
335                    ["hdiutil", "detach", "-force", str(mountpoint)],
336                    capture_output=True,
337                    text=True,
338                )
339                if forced_detach.returncode != 0:
340                    detail = forced_detach.stderr.strip() or forced_detach.stdout.strip()
341                    raise EvidenceError(f"could not detach validated DMG: {detail}")
342
343
344def inspect_packaged_app(
345    artifact_path: Path,
346    release_record: dict[str, Any],
347    forbidden_literals: list[str],
348    *,
349    verify_signatures: bool,
350) -> dict[str, Any]:
351    digests = release_record.get("digests", {})
352    expected_omega_digest = require_sha256(
353        digests.get("omega_binary_sha256"), "digests.omega_binary_sha256"
354    )
355    expected_cli_digest = require_sha256(
356        digests.get("cli_binary_sha256"), "digests.cli_binary_sha256"
357    )
358    identity_proof_component = release_record.get("components", {}).get(
359        "identity_proof_driver", {}
360    )
361    if identity_proof_component.get("path") != "Contents/MacOS/omega-identity-proof":
362        raise EvidenceError("identity proof driver path is invalid")
363    if identity_proof_component.get("protocol") != "openagents.omega.identity-proof.v1":
364        raise EvidenceError("identity proof driver protocol is invalid")
365    if identity_proof_component.get("keyring_service") != "com.openagents.omega.identity-proof.v1":
366        raise EvidenceError("identity proof driver Keychain service is invalid")
367    if identity_proof_component.get("keyring_account") != "disposable-proof-only":
368        raise EvidenceError("identity proof driver Keychain account is invalid")
369    expected_identity_proof_digest = require_sha256(
370        identity_proof_component.get("binary_sha256"),
371        "components.identity_proof_driver.binary_sha256",
372    )
373    expected_product = release_record.get("product")
374    expected_bundle_identifier = release_record.get("bundle_identifier")
375    expected_bundle_version = release_record.get(
376        "bundle_version", release_record.get("version")
377    )
378    if not all(
379        isinstance(value, str) and value
380        for value in (
381            expected_product,
382            expected_bundle_identifier,
383            expected_bundle_version,
384        )
385    ):
386        raise EvidenceError(
387            "release record product, bundle_identifier, and bundle version are required"
388        )
389
390    dmg_signing = (
391        verify_codesign(artifact_path, release_record, "DMG")
392        if verify_signatures
393        else {
394            "verified": False,
395            "authority": None,
396            "team_id": None,
397            "internal_self_test_bypass": True,
398        }
399    )
400    with mounted_dmg(artifact_path) as mountpoint:
401        applications = sorted(
402            path
403            for path in mountpoint.rglob("Omega.app")
404            if path.is_dir() and not path.is_symlink()
405        )
406        if len(applications) != 1:
407            raise EvidenceError(
408                f"DMG must contain exactly one Omega.app, found {len(applications)}"
409            )
410        application = applications[0]
411        application_signing = (
412            verify_codesign(application, release_record, "Omega.app")
413            if verify_signatures
414            else {
415                "verified": False,
416                "authority": None,
417                "team_id": None,
418                "internal_self_test_bypass": True,
419            }
420        )
421        info_path = application / "Contents/Info.plist"
422        omega_path = application / "Contents/MacOS/omega"
423        cli_path = application / "Contents/MacOS/cli"
424        identity_proof_path = application / "Contents/MacOS/omega-identity-proof"
425        for required_path in (info_path, omega_path, cli_path, identity_proof_path):
426            if not required_path.is_file() or required_path.is_symlink():
427                raise EvidenceError(
428                    f"packaged app is missing a regular {required_path.relative_to(application)}"
429                )
430        try:
431            info = plistlib.loads(info_path.read_bytes())
432        except (OSError, plistlib.InvalidFileException) as error:
433            raise EvidenceError(f"packaged Info.plist is invalid: {error}") from error
434
435        if info.get("CFBundleName") != expected_product:
436            raise EvidenceError(
437                f"Info.plist CFBundleName={info.get('CFBundleName')!r}, "
438                f"expected {expected_product!r}"
439            )
440        if info.get("CFBundleDisplayName") not in (None, expected_product):
441            raise EvidenceError(
442                "Info.plist CFBundleDisplayName contains an unexpected product name"
443            )
444        if info.get("CFBundleIdentifier") != expected_bundle_identifier:
445            raise EvidenceError("Info.plist CFBundleIdentifier differs from release record")
446        if info.get("CFBundleShortVersionString") != expected_bundle_version:
447            raise EvidenceError(
448                "Info.plist CFBundleShortVersionString differs from release record "
449                "bundle version"
450            )
451        bundle_build_version = info.get("CFBundleVersion")
452        if not isinstance(bundle_build_version, str) or not bundle_build_version:
453            raise EvidenceError("Info.plist CFBundleVersion must be a non-empty build version")
454
455        omega_digest = sha256_file(omega_path)
456        cli_digest = sha256_file(cli_path)
457        identity_proof_digest = sha256_file(identity_proof_path)
458        if omega_digest != expected_omega_digest:
459            raise EvidenceError("embedded omega binary digest differs from release record")
460        if cli_digest != expected_cli_digest:
461            raise EvidenceError("embedded cli binary digest differs from release record")
462        if identity_proof_digest != expected_identity_proof_digest:
463            raise EvidenceError("embedded identity proof driver digest differs from release record")
464        identity_proof_signing = (
465            verify_codesign(identity_proof_path, release_record, "omega-identity-proof")
466            if verify_signatures
467            else {
468                "verified": False,
469                "authority": None,
470                "team_id": None,
471                "internal_self_test_bypass": True,
472            }
473        )
474        source_commit = release_record.get("source", {}).get("commit")
475        if not isinstance(source_commit, str) or len(source_commit) != 40:
476            raise EvidenceError("release record source commit is invalid")
477        if not file_contains_literal(omega_path, source_commit.encode("ascii")):
478            raise EvidenceError(
479                "embedded omega binary does not contain the release source commit marker"
480            )
481        observed_scan = scan_packaged_app(application, forbidden_literals)
482        return {
483            "application_name": application.name,
484            "bundle_identifier": info["CFBundleIdentifier"],
485            "bundle_version": expected_bundle_version,
486            "bundle_build_version": bundle_build_version,
487            "omega_binary_sha256": omega_digest,
488            "cli_binary_sha256": cli_digest,
489            "identity_proof_binary_sha256": identity_proof_digest,
490            "source_commit_marker": source_commit,
491            "signing": {
492                "dmg": dmg_signing,
493                "application": application_signing,
494                "identity_proof_driver": identity_proof_signing,
495            },
496            "fresh_boundary_scan": observed_scan,
497        }
498
499
500def pending_gate(required_evidence: list[str]) -> dict[str, Any]:
501    return {
502        "status": "pending",
503        "attestor": None,
504        "observed_at": None,
505        "evidence": {},
506        "required_evidence": required_evidence,
507    }
508
509
510def valid_observed_at(value: Any) -> bool:
511    if not isinstance(value, str) or not value:
512        return False
513    try:
514        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
515    except ValueError:
516        return False
517    return parsed.tzinfo is not None
518
519
520def ingest_attestations(
521    attestations_dir: Path | None,
522    candidate_digest: str,
523    *,
524    evidence_root: Path | None = None,
525    required_receipts: dict[str, str] | None = None,
526) -> tuple[dict[str, dict[str, Any]], dict[str, str]]:
527    gates = {
528        name: pending_gate(requirements)
529        for name, requirements in GATE_REQUIREMENTS.items()
530    }
531    errors: dict[str, str] = {}
532    if attestations_dir is None:
533        return gates, errors
534    if not attestations_dir.is_dir():
535        errors["_attestations_dir"] = f"not a directory: {attestations_dir}"
536        return gates, errors
537
538    expected_paths = {attestations_dir / f"{name}.json" for name in gates}
539    unexpected = sorted(
540        path.name
541        for path in attestations_dir.glob("*.json")
542        if path not in expected_paths
543    )
544    if unexpected:
545        errors["_unexpected_files"] = ", ".join(unexpected)
546
547    for gate_name, gate in gates.items():
548        path = attestations_dir / f"{gate_name}.json"
549        if not path.is_file():
550            continue
551        try:
552            attestation = json.loads(path.read_text(encoding="utf-8"))
553        except (OSError, json.JSONDecodeError) as error:
554            errors[gate_name] = f"invalid JSON: {error}"
555            continue
556        if not isinstance(attestation, dict):
557            errors[gate_name] = "attestation must be a JSON object"
558            continue
559        if set(attestation) != {
560            "candidate_digest",
561            "attestor",
562            "observed_at",
563            "evidence",
564        }:
565            errors[gate_name] = "attestation has missing or unknown fields"
566            continue
567        if attestation["candidate_digest"] != candidate_digest:
568            errors[gate_name] = "candidate_digest does not match this candidate"
569            continue
570        if (
571            not isinstance(attestation["attestor"], str)
572            or not attestation["attestor"].strip()
573        ):
574            errors[gate_name] = "attestor must be a non-empty string"
575            continue
576        if not valid_observed_at(attestation["observed_at"]):
577            errors[gate_name] = "observed_at must be an ISO-8601 timestamp with timezone"
578            continue
579        evidence = attestation["evidence"]
580        if not isinstance(evidence, dict) or set(evidence) != set(gate["required_evidence"]):
581            errors[gate_name] = (
582                "evidence keys must exactly match required_evidence"
583            )
584            continue
585        receipt_name = STRUCTURED_RECEIPT_GATES.get(gate_name)
586        if receipt_name is not None and required_receipts is not None:
587            if evidence_root is None:
588                errors[gate_name] = "structured evidence requires --evidence-root"
589                continue
590            try:
591                resolved = [
592                    resolve_evidence_reference(reference, evidence_root)
593                    for reference in evidence.values()
594                ]
595            except IdentityEvidenceError as error:
596                errors[gate_name] = str(error)
597                continue
598            expected_digest = required_receipts.get(receipt_name)
599            if expected_digest is None or any(
600                digest != expected_digest for _path, digest in resolved
601            ):
602                errors[gate_name] = (
603                    f"every evidence reference must bind the validated {receipt_name} receipt"
604                )
605                continue
606        elif any(
607            not (
608                isinstance(reference, str)
609                and reference.strip()
610                or receipt_name is not None
611                and isinstance(reference, dict)
612            )
613            for reference in evidence.values()
614        ):
615            errors[gate_name] = "evidence values must be non-empty references or results"
616            continue
617        gate.update(
618            {
619                "status": "passed",
620                "attestor": attestation["attestor"].strip(),
621                "observed_at": attestation["observed_at"],
622                "evidence": evidence,
623            }
624        )
625    owner = gates["owner_observation"]
626    independent = gates["independent_verification"]
627    if (
628        owner["status"] == "passed"
629        and independent["status"] == "passed"
630        and owner["attestor"] == independent["attestor"]
631    ):
632        errors["independent_verification"] = (
633            "independent verifier must differ from the owner observer"
634        )
635        gates["independent_verification"] = pending_gate(
636            GATE_REQUIREMENTS["independent_verification"]
637        )
638    return gates, errors
639
640
641def validate_generated_record(
642    record: dict[str, Any], *, allow_internal_self_test_bypass: bool = False
643) -> None:
644    if record.get("schema") != "openagents.omega.identity-candidate-evidence.v1":
645        raise EvidenceError("unexpected evidence schema")
646    gates = record.get("gates")
647    if not isinstance(gates, dict) or set(gates) != set(GATE_REQUIREMENTS):
648        raise EvidenceError("generated evidence has the wrong required gate inventory")
649    admitted = all(gate.get("status") == "passed" for gate in gates.values()) and not record.get(
650        "attestation_errors"
651    )
652    if record.get("candidate_admitted") is not admitted:
653        raise EvidenceError("candidate_admitted disagrees with required gates")
654    expected_status = "admitted" if admitted else "pending_required_gates"
655    if record.get("status") != expected_status:
656        raise EvidenceError("candidate status disagrees with required gates")
657    digest = record.get("candidate_digest", {}).get("value")
658    if digest != canonical_digest(record["candidate"]):
659        raise EvidenceError("candidate digest does not match canonical candidate binding")
660    if not allow_internal_self_test_bypass:
661        signing = (
662            record.get("candidate", {})
663            .get("artifact", {})
664            .get("packaged_app", {})
665            .get("signing", {})
666        )
667        expected = {
668            "verified": True,
669            "authority": PINNED_SIGNING_IDENTITY,
670            "team_id": PINNED_TEAM_ID,
671        }
672        if (
673            signing.get("dmg") != expected
674            or signing.get("application") != expected
675            or signing.get("identity_proof_driver") != expected
676        ):
677            raise EvidenceError(
678                "public candidate evidence requires exact pinned verified signing facts"
679            )
680
681
682def build_record(
683    release_record_path: Path,
684    artifact_path: Path,
685    cargo_lock_snapshot: Path,
686    *,
687    allow_dirty: bool,
688    attestations_dir: Path | None,
689    identity_matrix_path: Path | None = None,
690    installed_tripwires_path: Path | None = None,
691    installed_observations_path: Path | None = None,
692    recovery_evidence_path: Path | None = None,
693    evidence_root: Path | None = None,
694    verify_signatures: bool = True,
695) -> dict[str, Any]:
696    contract_path = ROOT / CONTRACT_PATH
697    vector_path = ROOT / VECTOR_PATH
698    schema_path = ROOT / SCHEMA_PATH
699    verifier_path = ROOT / VERIFIER_PATH
700    for required_path in (
701        release_record_path,
702        artifact_path,
703        cargo_lock_snapshot,
704        contract_path,
705        vector_path,
706        schema_path,
707        verifier_path,
708    ):
709        if not required_path.is_file():
710            raise EvidenceError(f"required file is missing: {required_path}")
711
712    commit = git("rev-parse", "HEAD")
713    dirty = bool(git("status", "--porcelain"))
714    if dirty and not allow_dirty:
715        raise EvidenceError(
716            "working tree is dirty; candidate evidence must bind a clean Omega commit"
717        )
718
719    release_record = json.loads(release_record_path.read_text(encoding="utf-8"))
720    if release_record.get("product") != "Omega":
721        raise EvidenceError("release record product must be Omega")
722    if release_record.get("signing_identity") != PINNED_SIGNING_IDENTITY:
723        raise EvidenceError(
724            "release record signing_identity differs from pinned OpenAgents identity"
725        )
726    if release_record.get("team_id") != PINNED_TEAM_ID:
727        raise EvidenceError("release record team_id differs from pinned OpenAgents team")
728    release_commit = release_record.get("source", {}).get("commit")
729    if release_commit != commit:
730        raise EvidenceError(
731            f"release record commit {release_commit!r} does not match checkout {commit}"
732        )
733
734    artifact_digest = sha256_file(artifact_path)
735    digests = release_record.get("digests", {})
736    if digests.get("package_sha256") != artifact_digest:
737        raise EvidenceError(
738            "release record package digest does not match the explicit artifact"
739        )
740    if release_record.get("artifact_name") != artifact_path.name:
741        raise EvidenceError("release artifact name does not match the explicit artifact")
742
743    snapshot_digest = sha256_file(cargo_lock_snapshot)
744    build_lock = release_record.get("build_inputs", {}).get(
745        "cargo_lock_snapshot", {}
746    )
747    if build_lock.get("sha256") != snapshot_digest:
748        raise EvidenceError(
749            "release build Cargo.lock digest does not match the explicit snapshot"
750        )
751    if build_lock.get("name") != cargo_lock_snapshot.name:
752        raise EvidenceError(
753            "release build Cargo.lock name does not match the explicit snapshot"
754        )
755    actual_packages = cargo_lock_packages(
756        cargo_lock_snapshot.read_text(encoding="utf-8")
757    )
758    if actual_packages != EXPECTED_PACKAGES:
759        raise EvidenceError(
760            "build Cargo.lock snapshot differs from frozen identity packages/checksums"
761        )
762
763    verification = release_record.get("identity_boundaries", {})
764    source_verifier = verification.get("source_verifier", {})
765    packaged_scan = verification.get("packaged_scan", {})
766    verifier_digest = sha256_file(verifier_path)
767    if source_verifier.get("path") != VERIFIER_PATH.as_posix():
768        raise EvidenceError("release record source verifier path is invalid")
769    if source_verifier.get("sha256") != verifier_digest:
770        raise EvidenceError("release record verifier script digest does not match checkout")
771    if source_verifier.get("passed") is not True:
772        raise EvidenceError("identity_boundaries.source_verifier.passed must be true")
773    if packaged_scan.get("passed") is not True:
774        raise EvidenceError("identity_boundaries.packaged_scan.passed must be true")
775    forbidden_literals = packaged_scan.get("forbidden_literals")
776    if forbidden_literals != ["BUZZ_PRIVATE_KEY", "identity.key", "get_nsec"]:
777        raise EvidenceError(
778            "identity_boundaries.packaged_scan.forbidden_literals is invalid"
779        )
780    run_identity_verifier(verifier_path)
781
782    contract_text = contract_path.read_text(encoding="utf-8")
783    contract_version = int(
784        require_match(
785            r"^pub const IDENTITY_CONTRACT_VERSION: u32 = ([0-9]+);$",
786            contract_text,
787            "identity contract version",
788        )
789    )
790    buzz_commit = require_match(
791        r'^pub const BUZZ_SOURCE_COMMIT: &str = "([0-9a-f]{40})";$',
792        contract_text,
793        "reviewed Buzz source commit",
794    )
795    vector = json.loads(vector_path.read_text(encoding="utf-8"))
796    vector_version = int(
797        require_match(r"_v([0-9]+)\.json$", vector_path.name, "public vector version")
798    )
799    if vector.get("source_commit") != buzz_commit:
800        raise EvidenceError("public vector and identity contract bind different Buzz commits")
801    if vector.get("manifest", {}).get("contract_version") != contract_version:
802        raise EvidenceError("public vector and identity contract versions differ")
803
804    packaged_app = inspect_packaged_app(
805        artifact_path,
806        release_record,
807        forbidden_literals,
808        verify_signatures=verify_signatures,
809    )
810    candidate = {
811        "omega": {
812            "repository": "https://github.com/OpenAgentsInc/omega",
813            "commit": commit,
814            "dirty": dirty,
815        },
816        "identity_contract": {
817            "version": contract_version,
818            "source_path": CONTRACT_PATH.as_posix(),
819            "source_sha256": sha256_file(contract_path),
820        },
821        "public_vector": {
822            "version": vector_version,
823            "path": VECTOR_PATH.as_posix(),
824            "sha256": sha256_file(vector_path),
825        },
826        "reviewed_source": {
827            "repository": "https://github.com/OpenAgentsInc/buzz",
828            "commit": buzz_commit,
829        },
830        "assurance_spec_admission": ASSURANCE_SPEC_ADMISSION,
831        "packages": [
832            {
833                "name": name,
834                "version": package["version"],
835                "cargo_checksum": package["checksum"],
836            }
837            for name, package in sorted(actual_packages.items())
838        ],
839        "build_cargo_lock": {
840            "name": cargo_lock_snapshot.name,
841            "sha256": snapshot_digest,
842        },
843        "artifact": {
844            "name": release_record["artifact_name"],
845            "version": release_record.get("version"),
846            "sha256": artifact_digest,
847            "release_record_sha256": sha256_file(release_record_path),
848            "packaged_app": packaged_app,
849        },
850        "automated_tripwires": {
851            "verifier_path": VERIFIER_PATH.as_posix(),
852            "verifier_script_sha256": verifier_digest,
853            "source_verification": "passed",
854            "packaged_boundary_scan": "passed",
855            "packaged_forbidden_literals": forbidden_literals,
856            "fresh_mounted_app_scan": packaged_app["fresh_boundary_scan"],
857            "scope": (
858                "Static and packaged boundary tripwires; these are not exhaustive "
859                "installed-candidate proof."
860            ),
861        },
862    }
863    candidate_digest = canonical_digest(candidate)
864    required_receipts: dict[str, str] | None = None
865    machine_evidence_errors: dict[str, str] = {}
866    if verify_signatures:
867        required_receipts = {}
868        if identity_matrix_path is None:
869            machine_evidence_errors["_identity_matrix"] = "--identity-matrix is required"
870        else:
871            try:
872                required_receipts["identity_matrix"] = validate_identity_matrix(
873                    identity_matrix_path, candidate_digest
874                )
875            except IdentityEvidenceError as error:
876                machine_evidence_errors["_identity_matrix"] = str(error)
877        if installed_tripwires_path is None:
878            machine_evidence_errors["_installed_tripwires"] = (
879                "--installed-tripwires is required"
880            )
881        else:
882            try:
883                required_receipts["installed_tripwires"] = validate_installed_tripwires(
884                    installed_tripwires_path, candidate_digest
885                )
886            except IdentityEvidenceError as error:
887                machine_evidence_errors["_installed_tripwires"] = str(error)
888        if installed_observations_path is None:
889            machine_evidence_errors["_installed_observations"] = (
890                "--installed-observations is required"
891            )
892        elif evidence_root is None:
893            machine_evidence_errors["_installed_observations"] = (
894                "--installed-observations requires --evidence-root"
895            )
896        else:
897            try:
898                required_receipts["installed_observations"] = (
899                    validate_installed_observations(
900                        installed_observations_path, candidate_digest, evidence_root
901                    )
902                )
903            except IdentityEvidenceError as error:
904                machine_evidence_errors["_installed_observations"] = str(error)
905        if recovery_evidence_path is None:
906            machine_evidence_errors["_recovery_evidence"] = (
907                "--recovery-evidence is required"
908            )
909        elif evidence_root is None:
910            machine_evidence_errors["_recovery_evidence"] = (
911                "--recovery-evidence requires --evidence-root"
912            )
913        else:
914            try:
915                required_receipts["recovery_evidence"] = validate_recovery_evidence(
916                    recovery_evidence_path,
917                    candidate_digest,
918                    candidate["artifact"]["sha256"],
919                    candidate["artifact"]["packaged_app"]["omega_binary_sha256"],
920                    evidence_root,
921                )
922            except IdentityEvidenceError as error:
923                machine_evidence_errors["_recovery_evidence"] = str(error)
924    gates, attestation_errors = ingest_attestations(
925        attestations_dir,
926        candidate_digest,
927        evidence_root=evidence_root,
928        required_receipts=required_receipts,
929    )
930    attestation_errors.update(machine_evidence_errors)
931    if dirty:
932        attestation_errors["_dirty_source"] = (
933            "development-only dirty source binding cannot be admitted"
934        )
935    if not verify_signatures:
936        attestation_errors["_signature_bypass"] = (
937            "internal self-test signature bypass cannot be admitted"
938        )
939    admitted = all(gate["status"] == "passed" for gate in gates.values()) and not (
940        attestation_errors
941    )
942    record = {
943        "schema": "openagents.omega.identity-candidate-evidence.v1",
944        "schema_version": 1,
945        "status": "admitted" if admitted else "pending_required_gates",
946        "candidate_admitted": admitted,
947        "candidate": candidate,
948        "candidate_digest": {
949            "algorithm": "sha256",
950            "canonicalization": "RFC 8259 JSON; UTF-8; sorted keys; compact separators",
951            "value": candidate_digest,
952        },
953        "gates": gates,
954        "attestation_errors": attestation_errors,
955    }
956    validate_generated_record(
957        record, allow_internal_self_test_bypass=not verify_signatures
958    )
959    return record
960
961
962def write_record(path: Path, record: dict[str, Any]) -> None:
963    path.parent.mkdir(parents=True, exist_ok=True)
964    path.write_text(
965        json.dumps(record, indent=2, sort_keys=True) + "\n",
966        encoding="utf-8",
967    )
968
969
970def create_self_test_dmg(
971    directory: Path, version: str, source_commit: str
972) -> tuple[Path, str, str, str]:
973    source = directory / "dmg-source"
974    executable_directory = source / "Omega.app/Contents/MacOS"
975    executable_directory.mkdir(parents=True)
976    omega_path = executable_directory / "omega"
977    cli_path = executable_directory / "cli"
978    identity_proof_path = executable_directory / "omega-identity-proof"
979    omega_path.write_bytes(
980        f"omega evidence self-test executable\nZED_COMMIT_SHA={source_commit}\n".encode(
981            "ascii"
982        )
983    )
984    cli_path.write_bytes(b"omega cli evidence self-test executable\n")
985    identity_proof_path.write_bytes(
986        b"omega identity proof evidence self-test executable\n"
987    )
988    omega_path.chmod(0o755)
989    cli_path.chmod(0o755)
990    identity_proof_path.chmod(0o755)
991    info = {
992        "CFBundleName": "Omega",
993        "CFBundleDisplayName": "Omega",
994        "CFBundleIdentifier": "com.openagents.omega.rc",
995        "CFBundleShortVersionString": version,
996        "CFBundleVersion": version,
997        "CFBundleExecutable": "omega",
998        "CFBundlePackageType": "APPL",
999    }
1000    info_path = source / "Omega.app/Contents/Info.plist"
1001    info_path.write_bytes(plistlib.dumps(info))
1002    artifact_path = directory / "Omega-evidence-self-test.dmg"
1003    result = subprocess.run(
1004        [
1005            "hdiutil",
1006            "create",
1007            "-quiet",
1008            "-format",
1009            "UDRO",
1010            "-volname",
1011            "Omega Evidence Self Test",
1012            "-srcfolder",
1013            str(source),
1014            str(artifact_path),
1015        ],
1016        capture_output=True,
1017        text=True,
1018    )
1019    if result.returncode != 0:
1020        detail = result.stderr.strip() or result.stdout.strip()
1021        raise EvidenceError(f"could not create self-test DMG: {detail}")
1022    return (
1023        artifact_path,
1024        sha256_file(omega_path),
1025        sha256_file(cli_path),
1026        sha256_file(identity_proof_path),
1027    )
1028
1029
1030def self_test() -> None:
1031    if shutil.which("hdiutil") is None:
1032        raise EvidenceError("self-test requires hdiutil")
1033    with tempfile.TemporaryDirectory(prefix="omega-identity-evidence-") as directory:
1034        temp = Path(directory)
1035        version = "evidence-self-test"
1036        source_commit = git("rev-parse", "HEAD")
1037        artifact_path, omega_digest, cli_digest, identity_proof_digest = create_self_test_dmg(
1038            temp, version, source_commit
1039        )
1040        lock_snapshot = temp / "Cargo.lock.snapshot"
1041        lock_snapshot.write_bytes((ROOT / "Cargo.lock").read_bytes())
1042        release_record = json.loads(
1043            (ROOT / "crates/app_identity/fixtures/release_record_v1.json").read_text(
1044                encoding="utf-8"
1045            )
1046        )
1047        release_record["artifact_name"] = artifact_path.name
1048        release_record["version"] = version
1049        release_record["bundle_version"] = version
1050        release_record["signing_identity"] = PINNED_SIGNING_IDENTITY
1051        release_record["team_id"] = PINNED_TEAM_ID
1052        release_record["source"]["commit"] = source_commit
1053        release_record["digests"].update(
1054            {
1055                "package_sha256": sha256_file(artifact_path),
1056                "omega_binary_sha256": omega_digest,
1057                "cli_binary_sha256": cli_digest,
1058            }
1059        )
1060        release_record.setdefault("components", {})["identity_proof_driver"] = {
1061            "path": "Contents/MacOS/omega-identity-proof",
1062            "binary_sha256": identity_proof_digest,
1063            "protocol": "openagents.omega.identity-proof.v1",
1064            "keyring_service": "com.openagents.omega.identity-proof.v1",
1065            "keyring_account": "disposable-proof-only",
1066        }
1067        release_record["build_inputs"] = {
1068            "cargo_lock_snapshot": {
1069                "name": lock_snapshot.name,
1070                "sha256": sha256_file(lock_snapshot),
1071            }
1072        }
1073        release_record["identity_boundaries"] = {
1074            "source_verifier": {
1075                "path": VERIFIER_PATH.as_posix(),
1076                "sha256": sha256_file(ROOT / VERIFIER_PATH),
1077                "passed": True,
1078            },
1079            "packaged_scan": {
1080                "passed": True,
1081                "forbidden_literals": [
1082                    "BUZZ_PRIVATE_KEY",
1083                    "identity.key",
1084                    "get_nsec",
1085                ],
1086            },
1087        }
1088        release_record_path = temp / "release-record.json"
1089        release_record_path.write_text(
1090            json.dumps(release_record, indent=2) + "\n", encoding="utf-8"
1091        )
1092
1093        wrong_signer_record = json.loads(json.dumps(release_record))
1094        wrong_signer_record["signing_identity"] = "Developer ID Application: Not OpenAgents"
1095        wrong_signer_path = temp / "wrong-signer-release-record.json"
1096        wrong_signer_path.write_text(
1097            json.dumps(wrong_signer_record) + "\n", encoding="utf-8"
1098        )
1099        try:
1100            build_record(
1101                wrong_signer_path,
1102                artifact_path,
1103                lock_snapshot,
1104                allow_dirty=True,
1105                attestations_dir=None,
1106            )
1107        except EvidenceError as error:
1108            if "differs from pinned OpenAgents identity" not in str(error):
1109                raise
1110        else:
1111            raise EvidenceError("release record with an unpinned signer was accepted")
1112
1113        wrong_team_record = json.loads(json.dumps(release_record))
1114        wrong_team_record["team_id"] = "NOTOPENAGENTS"
1115        wrong_team_path = temp / "wrong-team-release-record.json"
1116        wrong_team_path.write_text(
1117            json.dumps(wrong_team_record) + "\n", encoding="utf-8"
1118        )
1119        try:
1120            build_record(
1121                wrong_team_path,
1122                artifact_path,
1123                lock_snapshot,
1124                allow_dirty=True,
1125                attestations_dir=None,
1126            )
1127        except EvidenceError as error:
1128            if "differs from pinned OpenAgents team" not in str(error):
1129                raise
1130        else:
1131            raise EvidenceError("release record with an unpinned team was accepted")
1132
1133        try:
1134            build_record(
1135                release_record_path,
1136                artifact_path,
1137                lock_snapshot,
1138                allow_dirty=True,
1139                attestations_dir=None,
1140            )
1141        except EvidenceError as error:
1142            if "DMG codesign verification failed" not in str(error):
1143                raise
1144        else:
1145            raise EvidenceError(
1146                "default production path accepted an unsigned self-test DMG"
1147            )
1148
1149        pending = build_record(
1150            release_record_path,
1151            artifact_path,
1152            lock_snapshot,
1153            allow_dirty=True,
1154            attestations_dir=None,
1155            verify_signatures=False,
1156        )
1157        if pending["candidate_admitted"] or pending["status"] != "pending_required_gates":
1158            raise EvidenceError("unattested self-test candidate was admitted")
1159        second = build_record(
1160            release_record_path,
1161            artifact_path,
1162            lock_snapshot,
1163            allow_dirty=True,
1164            attestations_dir=None,
1165            verify_signatures=False,
1166        )
1167        if pending != second:
1168            raise EvidenceError("same inputs did not produce identical evidence")
1169
1170        attestations = temp / "attestations"
1171        attestations.mkdir()
1172        observed_at = "2026-07-24T12:00:00Z"
1173        for gate_name in GATE_REQUIREMENTS:
1174            (attestations / f"{gate_name}.json").write_text(
1175                json.dumps(
1176                    {
1177                        "candidate_digest": pending["candidate_digest"]["value"],
1178                        "attestor": (
1179                            "self-test:independent"
1180                            if gate_name == "independent_verification"
1181                            else f"self-test:owner:{gate_name}"
1182                        ),
1183                        "observed_at": observed_at,
1184                        "evidence": {
1185                            requirement: (
1186                                f"self-test evidence for {gate_name}: {requirement}"
1187                            )
1188                            for requirement in GATE_REQUIREMENTS[gate_name]
1189                        },
1190                    }
1191                )
1192                + "\n",
1193                encoding="utf-8",
1194            )
1195        coverage_path = attestations / "custody_scenarios.json"
1196        incomplete_coverage = json.loads(coverage_path.read_text(encoding="utf-8"))
1197        incomplete_coverage["evidence"].pop(
1198            GATE_REQUIREMENTS["custody_scenarios"][-1]
1199        )
1200        coverage_path.write_text(
1201            json.dumps(incomplete_coverage) + "\n", encoding="utf-8"
1202        )
1203        coverage_gates, coverage_errors = ingest_attestations(
1204            attestations, pending["candidate_digest"]["value"]
1205        )
1206        if (
1207            coverage_gates["custody_scenarios"]["status"] != "pending"
1208            or "custody_scenarios" not in coverage_errors
1209        ):
1210            raise EvidenceError("incomplete requirement coverage was accepted")
1211        incomplete_coverage["evidence"] = {
1212            requirement: (
1213                f"self-test evidence for custody_scenarios: {requirement}"
1214            )
1215            for requirement in GATE_REQUIREMENTS["custody_scenarios"]
1216        }
1217        coverage_path.write_text(
1218            json.dumps(incomplete_coverage) + "\n", encoding="utf-8"
1219        )
1220        receipt_path = temp / "machine-receipt.json"
1221        receipt_path.write_text('{"status":"passed"}\n', encoding="utf-8")
1222        receipt_digest = sha256_file(receipt_path)
1223        for gate_name in STRUCTURED_RECEIPT_GATES:
1224            attestation_path = attestations / f"{gate_name}.json"
1225            attestation = json.loads(attestation_path.read_text(encoding="utf-8"))
1226            attestation["evidence"] = {
1227                requirement: {
1228                    "path": receipt_path.name,
1229                    "sha256": receipt_digest,
1230                }
1231                for requirement in GATE_REQUIREMENTS[gate_name]
1232            }
1233            attestation_path.write_text(json.dumps(attestation) + "\n", encoding="utf-8")
1234        structured_gates, structured_errors = ingest_attestations(
1235            attestations,
1236            pending["candidate_digest"]["value"],
1237            evidence_root=temp,
1238            required_receipts={
1239                "identity_matrix": receipt_digest,
1240                "installed_tripwires": receipt_digest,
1241                "installed_observations": receipt_digest,
1242                "recovery_evidence": receipt_digest,
1243            },
1244        )
1245        if structured_errors or any(
1246            structured_gates[name]["status"] != "passed"
1247            for name in STRUCTURED_RECEIPT_GATES
1248        ):
1249            raise EvidenceError("valid structured machine evidence was rejected")
1250        receipt_path.write_text('{"status":"tampered"}\n', encoding="utf-8")
1251        _, tampered_errors = ingest_attestations(
1252            attestations,
1253            pending["candidate_digest"]["value"],
1254            evidence_root=temp,
1255            required_receipts={
1256                "identity_matrix": receipt_digest,
1257                "installed_tripwires": receipt_digest,
1258                "installed_observations": receipt_digest,
1259                "recovery_evidence": receipt_digest,
1260            },
1261        )
1262        if not all(name in tampered_errors for name in STRUCTURED_RECEIPT_GATES):
1263            raise EvidenceError("tampered structured machine evidence was accepted")
1264        receipt_path.write_text('{"status":"passed"}\n', encoding="utf-8")
1265        fully_attested = build_record(
1266            release_record_path,
1267            artifact_path,
1268            lock_snapshot,
1269            allow_dirty=True,
1270            attestations_dir=attestations,
1271            verify_signatures=False,
1272        )
1273        if any(
1274            gate["status"] != "passed"
1275            for gate in fully_attested["gates"].values()
1276        ):
1277            raise EvidenceError("valid self-test attestations did not pass every gate")
1278        if fully_attested["candidate_admitted"]:
1279            raise EvidenceError("signature-bypassed self-test candidate was admitted")
1280        if "_signature_bypass" not in fully_attested["attestation_errors"]:
1281            raise EvidenceError("signature-bypassed self-test omitted its admission blocker")
1282        representable_admission = json.loads(json.dumps(fully_attested))
1283        representable_admission["attestation_errors"].pop("_signature_bypass")
1284        representable_admission["attestation_errors"].pop("_dirty_source", None)
1285        representable_admission["candidate"]["omega"]["dirty"] = False
1286        representable_admission["candidate"]["artifact"]["packaged_app"]["signing"] = {
1287            "dmg": {
1288                "verified": True,
1289                "authority": PINNED_SIGNING_IDENTITY,
1290                "team_id": PINNED_TEAM_ID,
1291            },
1292            "application": {
1293                "verified": True,
1294                "authority": PINNED_SIGNING_IDENTITY,
1295                "team_id": PINNED_TEAM_ID,
1296            },
1297            "identity_proof_driver": {
1298                "verified": True,
1299                "authority": PINNED_SIGNING_IDENTITY,
1300                "team_id": PINNED_TEAM_ID,
1301            },
1302        }
1303        representable_admission["candidate_digest"]["value"] = canonical_digest(
1304            representable_admission["candidate"]
1305        )
1306        representable_admission["candidate_admitted"] = True
1307        representable_admission["status"] = "admitted"
1308        validate_generated_record(representable_admission)
1309
1310        fake_artifact = temp / "arbitrary-fake.dmg"
1311        fake_artifact.write_bytes(b"not a disk image\n")
1312        fake_record = json.loads(json.dumps(release_record))
1313        fake_record["artifact_name"] = fake_artifact.name
1314        fake_record["digests"]["package_sha256"] = sha256_file(fake_artifact)
1315        fake_record_path = temp / "fake-release-record.json"
1316        fake_record_path.write_text(json.dumps(fake_record) + "\n", encoding="utf-8")
1317        try:
1318            build_record(
1319                fake_record_path,
1320                fake_artifact,
1321                lock_snapshot,
1322                allow_dirty=True,
1323                attestations_dir=None,
1324                verify_signatures=False,
1325            )
1326        except EvidenceError as error:
1327            if "not a mountable read-only DMG" not in str(error):
1328                raise
1329        else:
1330            raise EvidenceError("arbitrary fake .dmg passed package validation")
1331    print("Omega identity candidate evidence harness: self-test passed")
1332
1333
1334def parse_arguments() -> argparse.Namespace:
1335    parser = argparse.ArgumentParser(
1336        description=(
1337            "Bind an Omega identity-first evidence record to an explicit RC package "
1338            "and build Cargo.lock snapshot."
1339        )
1340    )
1341    parser.add_argument("--release-record", type=Path)
1342    parser.add_argument("--artifact", type=Path)
1343    parser.add_argument("--cargo-lock-snapshot", type=Path)
1344    parser.add_argument("--attestations-dir", type=Path)
1345    parser.add_argument("--identity-matrix", type=Path)
1346    parser.add_argument("--installed-tripwires", type=Path)
1347    parser.add_argument("--installed-observations", type=Path)
1348    parser.add_argument("--recovery-evidence", type=Path)
1349    parser.add_argument("--evidence-root", type=Path)
1350    parser.add_argument(
1351        "--output",
1352        type=Path,
1353        default=ROOT / "target/omega-identity-evidence/candidate-evidence.json",
1354    )
1355    parser.add_argument(
1356        "--allow-dirty",
1357        action="store_true",
1358        help="development only: record dirty=true instead of refusing the checkout",
1359    )
1360    parser.add_argument("--self-test", action="store_true")
1361    return parser.parse_args()
1362
1363
1364def main() -> int:
1365    arguments = parse_arguments()
1366    try:
1367        if arguments.self_test:
1368            if any(
1369                value is not None
1370                for value in (
1371                    arguments.release_record,
1372                    arguments.artifact,
1373                    arguments.cargo_lock_snapshot,
1374                    arguments.attestations_dir,
1375                    arguments.identity_matrix,
1376                    arguments.installed_tripwires,
1377                    arguments.installed_observations,
1378                    arguments.recovery_evidence,
1379                    arguments.evidence_root,
1380                )
1381            ):
1382                raise EvidenceError("--self-test does not accept candidate inputs")
1383            self_test()
1384            return 0
1385        if (
1386            arguments.release_record is None
1387            or arguments.artifact is None
1388            or arguments.cargo_lock_snapshot is None
1389        ):
1390            raise EvidenceError(
1391                "--release-record, --artifact, and --cargo-lock-snapshot are required"
1392            )
1393        record = build_record(
1394            arguments.release_record.resolve(),
1395            arguments.artifact.resolve(),
1396            arguments.cargo_lock_snapshot.resolve(),
1397            allow_dirty=arguments.allow_dirty,
1398            attestations_dir=(
1399                None
1400                if arguments.attestations_dir is None
1401                else arguments.attestations_dir.resolve()
1402            ),
1403            identity_matrix_path=(
1404                None if arguments.identity_matrix is None else arguments.identity_matrix.resolve()
1405            ),
1406            installed_tripwires_path=(
1407                None
1408                if arguments.installed_tripwires is None
1409                else arguments.installed_tripwires.resolve()
1410            ),
1411            installed_observations_path=(
1412                None
1413                if arguments.installed_observations is None
1414                else arguments.installed_observations.absolute()
1415            ),
1416            recovery_evidence_path=(
1417                None
1418                if arguments.recovery_evidence is None
1419                else arguments.recovery_evidence.absolute()
1420            ),
1421            evidence_root=(
1422                None if arguments.evidence_root is None else arguments.evidence_root.absolute()
1423            ),
1424        )
1425        write_record(arguments.output.resolve(), record)
1426        print(f"wrote candidate evidence: {arguments.output.resolve()}")
1427        if record["candidate_admitted"]:
1428            print("candidate admitted: every required gate has a valid attestation")
1429            return 0
1430        print(
1431            "candidate is not admitted: one or more required attestations are pending "
1432            "or invalid",
1433            file=sys.stderr,
1434        )
1435        return 3
1436    except (EvidenceError, json.JSONDecodeError, OSError, subprocess.CalledProcessError) as error:
1437        print(f"error: {error}", file=sys.stderr)
1438        return 2
1439
1440
1441if __name__ == "__main__":
1442    raise SystemExit(main())
1443