tenant.openagents/omega

1#!/usr/bin/env python3
2
3from __future__ import annotations
4
5import hashlib
6import json
7import re
8import tempfile
9from datetime import datetime
10from pathlib import Path
11from typing import Any
12
13
14MATRIX_SCHEMA = "openagents.omega.identity-proof-matrix.v1"
15TRIPWIRE_SCHEMA = "openagents.omega.installed-secret-tripwires.v1"
16INSTALLED_OBSERVATIONS_SCHEMA = "openagents.omega.identity-installed-observations.v2"
17RECOVERY_EVIDENCE_SCHEMA = "openagents.omega.identity-recovery-evidence.v1"
18ROLLBACK_CONTINUITY_SCHEMA = (
19    "openagents.omega.identity-update-downgrade-rollback.v2"
20)
21MATRIX_CASES = {
22    "disposable-namespace-safety",
23    "create-read-back-restart-sign",
24    "double-create",
25    "concurrent-create-and-process-start",
26    "forged-request-rejection",
27    "stale-request-rejection",
28    "crash-after-secret-write",
29    "crash-after-secret-read-back",
30    "crash-after-manifest-commit",
31    "crash-after-reset-marker",
32    "crash-after-reset-commit",
33    "crash-after-relaunch-acknowledge",
34    "simulated-conflict-custody",
35    "simulated-lost-custody",
36    "simulated-locked-custody",
37    "simulated-symlink-refusal",
38    "simulated-weak-permission-refusal",
39    "simulated-keychain-unavailable",
40    "simulated-corrupt-keychain",
41    "simulated-malformed-event-rejection",
42    "simulated-unadmitted-purpose-rejection",
43    "simulated-conflicting-recovery-selection",
44    "simulated-late-completion-fencing",
45    "simulated-signer-crash-before-completion",
46    "offline-create-and-encrypted-recovery-protection",
47    "wrong-recovery-password-rejection",
48    "corrupt-recovery-artifact-rejection",
49    "encrypted-recovery-and-restart-continuity",
50}
51TRIPWIRE_SURFACES = {
52    "logs",
53    "telemetry",
54    "clipboard",
55    "accessibility",
56    "diagnostics",
57    "crashes",
58}
59MANUAL_JOURNEY_CHECKS = {
60    "identity-first-first-run",
61    "theme-and-agent-setup-baseline",
62    "zed-data-before-after-isolation",
63}
64ACCESSIBILITY_CHECKS = {
65    "keyboard-focus-traversal",
66    "screen-reader-output",
67    "viewport-360-pixels",
68    "larger-ui-font",
69    "light-theme",
70    "dark-theme",
71    "high-contrast",
72    "reduced-motion",
73}
74#: The only checks an owner waiver may cover. A waiver is a record that an
75#: observation *did not happen*, so the set has to stay as small as the thing
76#: that is genuinely unobservable. Every other check is performable on a GPUI
77#: application under ordinary rendering conditions and must be performed.
78WAIVABLE_CHECKS = {"screen-reader-output"}
79ROLLBACK_EVIDENCE_REFS = {
80    "candidate_before",
81    "downgrade_removal",
82    "downgrade_install",
83    "identity_after_downgrade",
84    "rollback_removal",
85    "rollback_install",
86    "identity_after_rollback",
87}
88SHA256 = re.compile(r"^[0-9a-f]{64}$")
89
90
91class IdentityEvidenceError(RuntimeError):
92    pass
93
94
95def reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
96    value: dict[str, Any] = {}
97    for key, item in pairs:
98        if key in value:
99            raise IdentityEvidenceError(f"duplicate JSON key: {key}")
100        value[key] = item
101    return value
102
103
104def load_json(path: Path, label: str) -> dict[str, Any]:
105    try:
106        value = json.loads(
107            path.read_text(encoding="utf-8"), object_pairs_hook=reject_duplicate_keys
108        )
109    except (OSError, UnicodeError, json.JSONDecodeError) as error:
110        raise IdentityEvidenceError(f"cannot read {label}") from error
111    if not isinstance(value, dict):
112        raise IdentityEvidenceError(f"{label} must be a JSON object")
113    return value
114
115
116def sha256_file(path: Path) -> str:
117    digest = hashlib.sha256()
118    with path.open("rb") as source:
119        for chunk in iter(lambda: source.read(1024 * 1024), b""):
120            digest.update(chunk)
121    return digest.hexdigest()
122
123
124def canonical_digest(value: Any) -> str:
125    encoded = json.dumps(
126        value, sort_keys=True, separators=(",", ":"), ensure_ascii=False
127    ).encode("utf-8")
128    return hashlib.sha256(encoded).hexdigest()
129
130
131def require_sha256(value: Any, label: str) -> str:
132    if not isinstance(value, str) or SHA256.fullmatch(value) is None:
133        raise IdentityEvidenceError(f"{label} must be a lowercase SHA-256 digest")
134    return value
135
136
137def validate_identity_matrix(path: Path, candidate_digest: str) -> str:
138    report = load_json(path, "identity proof matrix")
139    if set(report) != {
140        "schema",
141        "status",
142        "candidate",
143        "disposable_keychain",
144        "cases",
145        "evidence_sha256",
146    }:
147        raise IdentityEvidenceError("identity proof matrix keys are not exact")
148    if report.get("schema") != MATRIX_SCHEMA or report.get("status") != "passed":
149        raise IdentityEvidenceError("identity proof matrix is not passed")
150    candidate = report.get("candidate")
151    if not isinstance(candidate, dict) or set(candidate) != {
152        "candidate_digest",
153        "artifact_sha256",
154        "release_record_sha256",
155        "identity_proof_binary_sha256",
156    }:
157        raise IdentityEvidenceError("identity proof matrix candidate binding is not exact")
158    if candidate.get("candidate_digest") != candidate_digest:
159        raise IdentityEvidenceError("identity proof matrix binds a different candidate")
160    for name, value in candidate.items():
161        require_sha256(value, f"identity proof matrix candidate.{name}")
162    locator = report.get("disposable_keychain")
163    if locator != {
164        "service": "com.openagents.omega.identity-proof.v1",
165        "account": "disposable-proof-only",
166        "cleanup": "passed",
167        "production_locator_access": "rejected-by-construction",
168    }:
169        raise IdentityEvidenceError("identity proof matrix locator or cleanup is unsafe")
170    cases = report.get("cases")
171    if not isinstance(cases, list) or len(cases) != len(MATRIX_CASES):
172        raise IdentityEvidenceError("identity proof matrix case inventory is incomplete")
173    observed: set[str] = set()
174    for case in cases:
175        if not isinstance(case, dict) or set(case) != {
176            "case",
177            "status",
178            "evidence_sha256",
179        }:
180            raise IdentityEvidenceError("identity proof matrix case is not exact")
181        name = case.get("case")
182        if name in observed or name not in MATRIX_CASES or case.get("status") != "passed":
183            raise IdentityEvidenceError("identity proof matrix case is invalid")
184        require_sha256(case.get("evidence_sha256"), f"identity proof matrix case {name}")
185        observed.add(name)
186    if observed != MATRIX_CASES:
187        raise IdentityEvidenceError("identity proof matrix case inventory differs")
188    claimed = require_sha256(report.get("evidence_sha256"), "matrix evidence digest")
189    canonical = dict(report)
190    canonical.pop("evidence_sha256")
191    if canonical_digest(canonical) != claimed:
192        raise IdentityEvidenceError("identity proof matrix internal digest differs")
193    return sha256_file(path)
194
195
196def validate_installed_tripwires(path: Path, candidate_digest: str) -> str:
197    report = load_json(path, "installed secret tripwires")
198    if set(report) != {
199        "schema",
200        "candidate_digest",
201        "generated_at",
202        "status",
203        "needle",
204        "surfaces",
205    }:
206        raise IdentityEvidenceError("installed tripwire receipt keys are not exact")
207    if (
208        report.get("schema") != TRIPWIRE_SCHEMA
209        or report.get("candidate_digest") != candidate_digest
210        or report.get("status") != "pass"
211    ):
212        raise IdentityEvidenceError("installed tripwire receipt is not passed or candidate-bound")
213    generated_at = report.get("generated_at")
214    try:
215        parsed_generated_at = datetime.fromisoformat(
216            generated_at.replace("Z", "+00:00")
217        )
218    except (AttributeError, ValueError) as error:
219        raise IdentityEvidenceError("installed tripwire timestamp is invalid") from error
220    if parsed_generated_at.tzinfo is None:
221        raise IdentityEvidenceError("installed tripwire timestamp lacks a timezone")
222    if report.get("needle") != {
223        "source": "protected_file_descriptor",
224        "length": report.get("needle", {}).get("length"),
225        "value_recorded": False,
226    }:
227        raise IdentityEvidenceError("installed tripwire needle handling is unsafe")
228    length = report["needle"]["length"]
229    if not isinstance(length, int) or not 16 <= length <= 4096:
230        raise IdentityEvidenceError("installed tripwire needle length is invalid")
231    surfaces = report.get("surfaces")
232    if not isinstance(surfaces, list) or len(surfaces) != len(TRIPWIRE_SURFACES):
233        raise IdentityEvidenceError("installed tripwire surface inventory is incomplete")
234    observed: set[str] = set()
235    expected_keys = {
236        "name",
237        "path_digest",
238        "status",
239        "files_scanned",
240        "symlinks_skipped",
241        "bytes_scanned",
242        "errors",
243        "match_detected",
244        "evidence_digest",
245    }
246    for surface in surfaces:
247        if not isinstance(surface, dict) or set(surface) != expected_keys:
248            raise IdentityEvidenceError("installed tripwire surface is not exact")
249        name = surface.get("name")
250        if name in observed or name not in TRIPWIRE_SURFACES:
251            raise IdentityEvidenceError("installed tripwire surface name is invalid")
252        if (
253            surface.get("status") not in ("pass", "absent")
254            or surface.get("errors") != 0
255            or surface.get("match_detected") is not False
256        ):
257            raise IdentityEvidenceError("installed tripwire surface did not pass safely")
258        for field in (
259            "files_scanned",
260            "symlinks_skipped",
261            "bytes_scanned",
262            "errors",
263        ):
264            value = surface.get(field)
265            if not isinstance(value, int) or isinstance(value, bool) or value < 0:
266                raise IdentityEvidenceError("installed tripwire surface counts are invalid")
267        require_sha256(surface.get("path_digest"), f"tripwire {name} path digest")
268        require_sha256(surface.get("evidence_digest"), f"tripwire {name} evidence digest")
269        observed.add(name)
270    if observed != TRIPWIRE_SURFACES:
271        raise IdentityEvidenceError("installed tripwire surface inventory differs")
272    return sha256_file(path)
273
274
275def validate_installed_observations(path: Path, candidate_digest: str, evidence_root: Path) -> str:
276    if path.is_symlink() or not path.is_file():
277        raise IdentityEvidenceError(
278            "installed identity observations do not name a regular file"
279        )
280    report = load_json(path, "installed identity observations")
281    if set(report) != {
282        "schema",
283        "candidate_digest",
284        "status",
285        "manual_journey",
286        "accessibility",
287        "waivers",
288        "evidence_sha256",
289    }:
290        raise IdentityEvidenceError("installed identity observation keys are not exact")
291    if (
292        report.get("schema") != INSTALLED_OBSERVATIONS_SCHEMA
293        or report.get("candidate_digest") != candidate_digest
294        or report.get("status") not in {"passed", "passed_with_waivers"}
295    ):
296        raise IdentityEvidenceError(
297            "installed identity observations are not passed or candidate-bound"
298        )
299    waived = validate_observation_group(
300        report.get("manual_journey"),
301        MANUAL_JOURNEY_CHECKS,
302        evidence_root,
303    ) | validate_observation_group(
304        report.get("accessibility"),
305        ACCESSIBILITY_CHECKS,
306        evidence_root,
307    )
308    declared = report.get("waivers")
309    if not isinstance(declared, list) or any(
310        not isinstance(name, str) for name in declared
311    ):
312        raise IdentityEvidenceError("installed observation waiver list is invalid")
313    # A waived check must be declared at the top level, and a declared waiver
314    # must correspond to a waived entry. Without both directions a reader could
315    # see a summary that omits the waiver, or a waiver that hides a pass.
316    if set(declared) != waived or len(declared) != len(set(declared)):
317        raise IdentityEvidenceError(
318            "installed observation waiver list differs from the waived entries"
319        )
320    # The anti-promotion invariant. `waived` never rolls up into a green
321    # status, and a report carrying a waiver can never call itself `passed`.
322    expected_status = "passed_with_waivers" if waived else "passed"
323    if report.get("status") != expected_status:
324        raise IdentityEvidenceError(
325            f"installed observations claim {report.get('status')!r} with "
326            f"{len(waived)} waived check(s); a waiver is not a pass"
327        )
328    claimed = require_sha256(
329        report.get("evidence_sha256"), "installed observation evidence digest"
330    )
331    canonical = dict(report)
332    canonical.pop("evidence_sha256")
333    if canonical_digest(canonical) != claimed:
334        raise IdentityEvidenceError("installed identity observation digest differs")
335    return sha256_file(path)
336
337
338def validate_recovery_evidence(
339    path: Path,
340    candidate_digest: str,
341    candidate_artifact_sha256: str,
342    candidate_binary_sha256: str,
343    evidence_root: Path,
344) -> str:
345    if path.is_symlink() or not path.is_file():
346        raise IdentityEvidenceError("identity recovery evidence is not a regular file")
347    report = load_json(path, "identity recovery evidence")
348    if set(report) != {
349        "schema",
350        "candidate_digest",
351        "status",
352        "identity_matrix",
353        "rollback_continuity",
354        "evidence_sha256",
355    }:
356        raise IdentityEvidenceError("identity recovery evidence keys are not exact")
357    if (
358        report.get("schema") != RECOVERY_EVIDENCE_SCHEMA
359        or report.get("candidate_digest") != candidate_digest
360        or report.get("status") != "passed"
361    ):
362        raise IdentityEvidenceError(
363            "identity recovery evidence is not passed or candidate-bound"
364        )
365    matrix_path, _ = resolve_evidence_reference(
366        report.get("identity_matrix"), evidence_root
367    )
368    rollback_path, _ = resolve_evidence_reference(
369        report.get("rollback_continuity"), evidence_root
370    )
371    if matrix_path == rollback_path:
372        raise IdentityEvidenceError(
373            "recovery matrix and rollback continuity must be distinct references"
374        )
375    validate_identity_matrix(matrix_path, candidate_digest)
376    validate_rollback_continuity(
377        rollback_path,
378        candidate_digest,
379        candidate_artifact_sha256,
380        candidate_binary_sha256,
381        evidence_root,
382    )
383    claimed = require_sha256(
384        report.get("evidence_sha256"), "identity recovery evidence digest"
385    )
386    canonical = dict(report)
387    canonical.pop("evidence_sha256")
388    if canonical_digest(canonical) != claimed:
389        raise IdentityEvidenceError("identity recovery evidence digest differs")
390    return sha256_file(path)
391
392
393def validate_rollback_continuity(
394    path: Path,
395    candidate_digest: str,
396    candidate_artifact_sha256: str,
397    candidate_binary_sha256: str,
398    evidence_root: Path,
399) -> str:
400    report = load_json(path, "identity rollback continuity")
401    if set(report) != {
402        "schema",
403        "candidate_digest",
404        "status",
405        "observed_at",
406        "identity_fingerprint_before",
407        "identity_fingerprint_after",
408        "sequence",
409        "evidence_refs",
410        "evidence_sha256",
411    }:
412        raise IdentityEvidenceError("identity rollback continuity keys are not exact")
413    if (
414        report.get("schema") != ROLLBACK_CONTINUITY_SCHEMA
415        or report.get("candidate_digest") != candidate_digest
416        or report.get("status") != "passed"
417    ):
418        raise IdentityEvidenceError(
419            "identity rollback continuity is not passed or candidate-bound"
420        )
421    require_timestamp(report.get("observed_at"), "identity rollback continuity")
422    before = report.get("identity_fingerprint_before")
423    after = report.get("identity_fingerprint_after")
424    if (
425        not isinstance(before, str)
426        or not before.strip()
427        or len(before) > 256
428        or before != after
429        or "nsec" in before.lower()
430        or "ncryptsec" in before.lower()
431    ):
432        raise IdentityEvidenceError(
433            "identity rollback continuity fingerprint changed or is unsafe"
434        )
435    sequence = report.get("sequence")
436    if not isinstance(sequence, list) or len(sequence) != 2:
437        raise IdentityEvidenceError("identity rollback continuity sequence is incomplete")
438    downgrade, rollback = sequence
439    if not isinstance(downgrade, dict) or set(downgrade) != {
440        "action",
441        "from_artifact_sha256",
442        "to_artifact_sha256",
443    }:
444        raise IdentityEvidenceError("identity downgrade step is not exact")
445    if not isinstance(rollback, dict) or set(rollback) != {
446        "action",
447        "from_artifact_sha256",
448        "to_artifact_sha256",
449        "installed_binary_sha256",
450    }:
451        raise IdentityEvidenceError("identity rollback step is not exact")
452    older_artifact = require_sha256(
453        downgrade.get("to_artifact_sha256"), "downgrade target artifact"
454    )
455    if (
456        downgrade.get("action") != "downgrade"
457        or require_sha256(
458            downgrade.get("from_artifact_sha256"), "downgrade source artifact"
459        )
460        != candidate_artifact_sha256
461        or older_artifact == candidate_artifact_sha256
462        or rollback.get("action") != "update_and_rollback_to_candidate"
463        or require_sha256(
464            rollback.get("from_artifact_sha256"), "rollback source artifact"
465        )
466        != older_artifact
467        or require_sha256(
468            rollback.get("to_artifact_sha256"), "rollback target artifact"
469        )
470        != candidate_artifact_sha256
471        or require_sha256(
472            rollback.get("installed_binary_sha256"), "rollback installed binary"
473        )
474        != candidate_binary_sha256
475    ):
476        raise IdentityEvidenceError(
477            "identity rollback sequence does not bind the candidate and older artifact"
478        )
479    references = report.get("evidence_refs")
480    if not isinstance(references, dict) or set(references) != ROLLBACK_EVIDENCE_REFS:
481        raise IdentityEvidenceError(
482            "identity rollback continuity evidence inventory is not exact"
483        )
484    resolved = [
485        resolve_evidence_reference(reference, evidence_root)
486        for reference in references.values()
487    ]
488    if len({(resolved_path, digest) for resolved_path, digest in resolved}) != len(
489        resolved
490    ):
491        raise IdentityEvidenceError(
492            "identity rollback continuity repeats an evidence reference"
493        )
494    claimed = require_sha256(
495        report.get("evidence_sha256"), "identity rollback continuity digest"
496    )
497    canonical = dict(report)
498    canonical.pop("evidence_sha256")
499    if canonical_digest(canonical) != claimed:
500        raise IdentityEvidenceError("identity rollback continuity digest differs")
501    return sha256_file(path)
502
503
504def validate_observation_group(
505    observations: Any,
506    expected_checks: set[str],
507    evidence_root: Path,
508) -> set[str]:
509    """Validate one observation group and return the checks it waived."""
510    if not isinstance(observations, list) or len(observations) != len(expected_checks):
511        raise IdentityEvidenceError("installed observation check inventory is incomplete")
512    observed: set[str] = set()
513    waived: set[str] = set()
514    for observation in observations:
515        if not isinstance(observation, dict):
516            raise IdentityEvidenceError("installed observation entry is not exact")
517        check = observation.get("check")
518        if check in observed or check not in expected_checks:
519            raise IdentityEvidenceError("installed observation check is invalid")
520        status = observation.get("status")
521        if status not in {"passed", "waived"}:
522            raise IdentityEvidenceError(f"installed observation {check} is not passed")
523        # A waived entry carries a waiver and no facts; a passed entry carries
524        # facts and no waiver. The two shapes are disjoint, so a waived entry
525        # cannot be relabelled `passed` without also inventing the facts the
526        # observation would have produced.
527        if status == "waived":
528            if set(observation) != {
529                "check",
530                "status",
531                "observed_at",
532                "waiver",
533                "evidence_refs",
534            }:
535                raise IdentityEvidenceError("installed observation entry is not exact")
536            validate_observation_waiver(check, observation.get("waiver"))
537            waived.add(check)
538        else:
539            if set(observation) != {
540                "check",
541                "status",
542                "observed_at",
543                "facts",
544                "evidence_refs",
545            }:
546                raise IdentityEvidenceError("installed observation entry is not exact")
547            validate_observation_facts(check, observation.get("facts"))
548        require_timestamp(observation.get("observed_at"), f"installed observation {check}")
549        references = observation.get("evidence_refs")
550        if not isinstance(references, list) or not references:
551            raise IdentityEvidenceError(
552                f"installed observation {check} has no evidence references"
553            )
554        resolved = [resolve_evidence_reference(reference, evidence_root) for reference in references]
555        if len({(path, digest) for path, digest in resolved}) != len(resolved):
556            raise IdentityEvidenceError(
557                f"installed observation {check} repeats an evidence reference"
558            )
559        observed.add(check)
560    if observed != expected_checks:
561        raise IdentityEvidenceError("installed observation check inventory differs")
562    return waived
563
564
565def validate_observation_waiver(check: str, waiver: Any) -> None:
566    """Validate one recorded owner waiver.
567
568    A waiver says an observation was never made. It therefore has to name the
569    owner direction that permits the omission, the date of that direction, and
570    the upstream parity that the direction is conditioned on, so a later reader
571    can re-check the condition rather than trust the word `waived`.
572    """
573    if check not in WAIVABLE_CHECKS:
574        raise IdentityEvidenceError(
575            f"installed observation {check} may not be waived; it is performable"
576        )
577    if not isinstance(waiver, dict) or set(waiver) != {
578        "owner_quote",
579        "owner_direction_date",
580        "basis",
581        "upstream_parity",
582        "issue",
583    }:
584        raise IdentityEvidenceError(f"installed observation {check} waiver is not exact")
585    for field in ("owner_quote", "basis", "issue"):
586        value = waiver.get(field)
587        if not isinstance(value, str) or not value.strip():
588            raise IdentityEvidenceError(
589                f"installed observation {check} waiver has no {field}"
590            )
591    require_timestamp(
592        waiver.get("owner_direction_date"), f"installed observation {check} waiver"
593    )
594    parity = waiver.get("upstream_parity")
595    if not isinstance(parity, dict) or set(parity) != {
596        "product",
597        "version",
598        "method",
599        "observed",
600    }:
601        raise IdentityEvidenceError(
602            f"installed observation {check} waiver has no upstream parity record"
603        )
604    for field in ("product", "version", "method", "observed"):
605        value = parity.get(field)
606        if not isinstance(value, str) or not value.strip():
607            raise IdentityEvidenceError(
608                f"installed observation {check} waiver parity has no {field}"
609            )
610
611
612def validate_observation_facts(check: str, facts: Any) -> None:
613    if not isinstance(facts, dict):
614        raise IdentityEvidenceError(f"installed observation {check} facts are invalid")
615    exact_facts: dict[str, dict[str, Any]] = {
616        "identity-first-first-run": {
617            "identity_presented_before_editor_setup": True,
618            "identity_ready": True,
619        },
620        "theme-and-agent-setup-baseline": {
621            "theme_families": ["Aiur", "Ayu", "Gruvbox"],
622            "agent_setup_visible": True,
623        },
624        "keyboard-focus-traversal": {
625            "all_controls_reachable": True,
626            "reverse_traversal": True,
627            "focus_visible": True,
628            "keyboard_activation": True,
629        },
630        "viewport-360-pixels": {
631            "viewport_width_pixels": 360,
632            "horizontal_overflow": False,
633            "completion_action_visible": True,
634        },
635        "light-theme": {"appearance": "light", "content_legible": True},
636        "dark-theme": {"appearance": "dark", "content_legible": True},
637        "high-contrast": {
638            "system_increase_contrast": True,
639            "content_legible": True,
640            "focus_indicator_visible": True,
641        },
642        "reduced-motion": {
643            "system_reduce_motion": True,
644            "motion_required_for_completion": False,
645        },
646    }
647    if check in exact_facts:
648        if facts != exact_facts[check]:
649            raise IdentityEvidenceError(f"installed observation {check} facts differ")
650        return
651    if check == "zed-data-before-after-isolation":
652        if set(facts) != {"zed_before_sha256", "zed_after_sha256", "unchanged"}:
653            raise IdentityEvidenceError("Zed isolation facts are not exact")
654        before = require_sha256(facts.get("zed_before_sha256"), "Zed before digest")
655        after = require_sha256(facts.get("zed_after_sha256"), "Zed after digest")
656        if before != after or facts.get("unchanged") is not True:
657            raise IdentityEvidenceError("Zed isolation observation changed")
658        return
659    if check == "screen-reader-output":
660        if set(facts) != {
661            "assistive_technology",
662            "identity_status_announced",
663            "controls_named",
664            "secret_value_exposed",
665        }:
666            raise IdentityEvidenceError("screen-reader facts are not exact")
667        technology = facts.get("assistive_technology")
668        if (
669            not isinstance(technology, str)
670            or not technology.strip()
671            or facts.get("identity_status_announced") is not True
672            or facts.get("controls_named") is not True
673            or facts.get("secret_value_exposed") is not False
674        ):
675            raise IdentityEvidenceError("screen-reader observation did not pass safely")
676        return
677    if check == "larger-ui-font":
678        if set(facts) != {
679            "ui_font_size_pixels",
680            "content_clipped",
681            "completion_action_visible",
682        }:
683            raise IdentityEvidenceError("larger UI font facts are not exact")
684        font_size = facts.get("ui_font_size_pixels")
685        if (
686            not isinstance(font_size, int)
687            or isinstance(font_size, bool)
688            or font_size < 18
689            or facts.get("content_clipped") is not False
690            or facts.get("completion_action_visible") is not True
691        ):
692            raise IdentityEvidenceError("larger UI font observation did not pass")
693        return
694    raise IdentityEvidenceError(f"unknown installed observation check {check}")
695
696
697def require_timestamp(value: Any, label: str) -> None:
698    try:
699        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
700    except (AttributeError, ValueError) as error:
701        raise IdentityEvidenceError(f"{label} timestamp is invalid") from error
702    if parsed.tzinfo is None:
703        raise IdentityEvidenceError(f"{label} timestamp lacks a timezone")
704
705
706def resolve_evidence_reference(reference: Any, evidence_root: Path) -> tuple[Path, str]:
707    if not isinstance(reference, dict) or set(reference) != {"path", "sha256"}:
708        raise IdentityEvidenceError("evidence reference must contain exact path and sha256")
709    relative = reference.get("path")
710    expected = require_sha256(reference.get("sha256"), "evidence reference sha256")
711    if (
712        not isinstance(relative, str)
713        or not relative
714        or Path(relative).is_absolute()
715        or ".." in Path(relative).parts
716    ):
717        raise IdentityEvidenceError("evidence reference path is unsafe")
718    if not evidence_root.is_dir() or evidence_root.is_symlink():
719        raise IdentityEvidenceError("evidence root is missing or unsafe")
720    path = evidence_root / relative
721    current = evidence_root
722    for component in Path(relative).parts:
723        current = current / component
724        if current.is_symlink():
725            raise IdentityEvidenceError("evidence reference contains a symbolic link")
726    if not path.is_file():
727        raise IdentityEvidenceError("evidence reference does not name a regular file")
728    if sha256_file(path) != expected:
729        raise IdentityEvidenceError("evidence reference digest differs")
730    return path, expected
731
732
733def self_test() -> None:
734    candidate_digest = "a" * 64
735    candidate = {
736        "candidate_digest": candidate_digest,
737        "artifact_sha256": "b" * 64,
738        "release_record_sha256": "c" * 64,
739        "identity_proof_binary_sha256": "d" * 64,
740    }
741    cases = [
742        {"case": name, "status": "passed", "evidence_sha256": "e" * 64}
743        for name in sorted(MATRIX_CASES)
744    ]
745    matrix = {
746        "schema": MATRIX_SCHEMA,
747        "status": "passed",
748        "candidate": candidate,
749        "disposable_keychain": {
750            "service": "com.openagents.omega.identity-proof.v1",
751            "account": "disposable-proof-only",
752            "cleanup": "passed",
753            "production_locator_access": "rejected-by-construction",
754        },
755        "cases": cases,
756    }
757    matrix["evidence_sha256"] = canonical_digest(matrix)
758    surfaces = [
759        {
760            "name": name,
761            "path_digest": "f" * 64,
762            "status": "pass",
763            "files_scanned": 1,
764            "symlinks_skipped": 0,
765            "bytes_scanned": 1,
766            "errors": 0,
767            "match_detected": False,
768            "evidence_digest": "1" * 64,
769        }
770        for name in sorted(TRIPWIRE_SURFACES)
771    ]
772    tripwires = {
773        "schema": TRIPWIRE_SCHEMA,
774        "candidate_digest": candidate_digest,
775        "generated_at": "2026-07-24T12:00:00+00:00",
776        "status": "pass",
777        "needle": {
778            "source": "protected_file_descriptor",
779            "length": 32,
780            "value_recorded": False,
781        },
782        "surfaces": surfaces,
783    }
784    observation_facts = {
785        "identity-first-first-run": {
786            "identity_presented_before_editor_setup": True,
787            "identity_ready": True,
788        },
789        "theme-and-agent-setup-baseline": {
790            "theme_families": ["Aiur", "Ayu", "Gruvbox"],
791            "agent_setup_visible": True,
792        },
793        "zed-data-before-after-isolation": {
794            "zed_before_sha256": "2" * 64,
795            "zed_after_sha256": "2" * 64,
796            "unchanged": True,
797        },
798        "keyboard-focus-traversal": {
799            "all_controls_reachable": True,
800            "reverse_traversal": True,
801            "focus_visible": True,
802            "keyboard_activation": True,
803        },
804        "screen-reader-output": {
805            "assistive_technology": "self-test reader",
806            "identity_status_announced": True,
807            "controls_named": True,
808            "secret_value_exposed": False,
809        },
810        "viewport-360-pixels": {
811            "viewport_width_pixels": 360,
812            "horizontal_overflow": False,
813            "completion_action_visible": True,
814        },
815        "larger-ui-font": {
816            "ui_font_size_pixels": 18,
817            "content_clipped": False,
818            "completion_action_visible": True,
819        },
820        "light-theme": {"appearance": "light", "content_legible": True},
821        "dark-theme": {"appearance": "dark", "content_legible": True},
822        "high-contrast": {
823            "system_increase_contrast": True,
824            "content_legible": True,
825            "focus_indicator_visible": True,
826        },
827        "reduced-motion": {
828            "system_reduce_motion": True,
829            "motion_required_for_completion": False,
830        },
831    }
832    with tempfile.TemporaryDirectory(prefix="omega-identity-evidence-") as directory:
833        root = Path(directory)
834        matrix_path = root / "matrix.json"
835        tripwire_path = root / "tripwires.json"
836        observation_evidence_path = root / "installed-observation.txt"
837        observation_evidence_path.write_text("installed observation fixture", encoding="utf-8")
838        observation_reference = {
839            "path": observation_evidence_path.name,
840            "sha256": sha256_file(observation_evidence_path),
841        }
842        observations = {
843            "schema": INSTALLED_OBSERVATIONS_SCHEMA,
844            "candidate_digest": candidate_digest,
845            "status": "passed",
846            "manual_journey": [
847                {
848                    "check": check,
849                    "status": "passed",
850                    "observed_at": "2026-07-24T12:00:00+00:00",
851                    "facts": observation_facts[check],
852                    "evidence_refs": [observation_reference],
853                }
854                for check in sorted(MANUAL_JOURNEY_CHECKS)
855            ],
856            "accessibility": [
857                {
858                    "check": check,
859                    "status": "passed",
860                    "observed_at": "2026-07-24T12:00:00+00:00",
861                    "facts": observation_facts[check],
862                    "evidence_refs": [observation_reference],
863                }
864                for check in sorted(ACCESSIBILITY_CHECKS)
865            ],
866            "waivers": [],
867        }
868        observations["evidence_sha256"] = canonical_digest(observations)
869        observations_path = root / "installed-observations.json"
870        matrix_path.write_text(json.dumps(matrix), encoding="utf-8")
871        tripwire_path.write_text(json.dumps(tripwires), encoding="utf-8")
872        observations_path.write_text(json.dumps(observations), encoding="utf-8")
873        matrix_digest = validate_identity_matrix(matrix_path, candidate_digest)
874        tripwire_digest = validate_installed_tripwires(tripwire_path, candidate_digest)
875        observation_digest = validate_installed_observations(
876            observations_path, candidate_digest, root
877        )
878        resolve_evidence_reference(
879            {"path": matrix_path.name, "sha256": matrix_digest}, root
880        )
881        resolve_evidence_reference(
882            {"path": tripwire_path.name, "sha256": tripwire_digest}, root
883        )
884        resolve_evidence_reference(
885            {"path": observations_path.name, "sha256": observation_digest}, root
886        )
887        rollback_refs = {}
888        for name in sorted(ROLLBACK_EVIDENCE_REFS):
889            rollback_evidence_path = root / f"rollback-{name}.txt"
890            rollback_evidence_path.write_text(
891                f"rollback observation fixture {name}", encoding="utf-8"
892            )
893            rollback_refs[name] = {
894                "path": rollback_evidence_path.name,
895                "sha256": sha256_file(rollback_evidence_path),
896            }
897        rollback = {
898            "schema": ROLLBACK_CONTINUITY_SCHEMA,
899            "candidate_digest": candidate_digest,
900            "status": "passed",
901            "observed_at": "2026-07-24T12:00:00+00:00",
902            "identity_fingerprint_before": "B15D 76DE 47C0 0ACE",
903            "identity_fingerprint_after": "B15D 76DE 47C0 0ACE",
904            "sequence": [
905                {
906                    "action": "downgrade",
907                    "from_artifact_sha256": candidate["artifact_sha256"],
908                    "to_artifact_sha256": "3" * 64,
909                },
910                {
911                    "action": "update_and_rollback_to_candidate",
912                    "from_artifact_sha256": "3" * 64,
913                    "to_artifact_sha256": candidate["artifact_sha256"],
914                    "installed_binary_sha256": "4" * 64,
915                },
916            ],
917            "evidence_refs": rollback_refs,
918        }
919        rollback["evidence_sha256"] = canonical_digest(rollback)
920        rollback_path = root / "rollback-continuity.json"
921        rollback_path.write_text(json.dumps(rollback), encoding="utf-8")
922        recovery = {
923            "schema": RECOVERY_EVIDENCE_SCHEMA,
924            "candidate_digest": candidate_digest,
925            "status": "passed",
926            "identity_matrix": {
927                "path": matrix_path.name,
928                "sha256": matrix_digest,
929            },
930            "rollback_continuity": {
931                "path": rollback_path.name,
932                "sha256": sha256_file(rollback_path),
933            },
934        }
935        recovery["evidence_sha256"] = canonical_digest(recovery)
936        recovery_path = root / "recovery-evidence.json"
937        recovery_path.write_text(json.dumps(recovery), encoding="utf-8")
938        validate_recovery_evidence(
939            recovery_path,
940            candidate_digest,
941            candidate["artifact_sha256"],
942            "4" * 64,
943            root,
944        )
945        invalid_rollbacks = []
946        wrong_binary = json.loads(json.dumps(rollback))
947        wrong_binary["sequence"][1]["installed_binary_sha256"] = "5" * 64
948        wrong_binary["evidence_sha256"] = canonical_digest(
949            {
950                key: value
951                for key, value in wrong_binary.items()
952                if key != "evidence_sha256"
953            }
954        )
955        invalid_rollbacks.append(wrong_binary)
956        changed_identity = json.loads(json.dumps(rollback))
957        changed_identity["identity_fingerprint_after"] = "0000 0000 0000 0000"
958        changed_identity["evidence_sha256"] = canonical_digest(
959            {
960                key: value
961                for key, value in changed_identity.items()
962                if key != "evidence_sha256"
963            }
964        )
965        invalid_rollbacks.append(changed_identity)
966        for invalid_rollback in invalid_rollbacks:
967            rollback_path.write_text(json.dumps(invalid_rollback), encoding="utf-8")
968            invalid_recovery = json.loads(json.dumps(recovery))
969            invalid_recovery["rollback_continuity"]["sha256"] = sha256_file(
970                rollback_path
971            )
972            invalid_recovery["evidence_sha256"] = canonical_digest(
973                {
974                    key: value
975                    for key, value in invalid_recovery.items()
976                    if key != "evidence_sha256"
977                }
978            )
979            recovery_path.write_text(json.dumps(invalid_recovery), encoding="utf-8")
980            try:
981                validate_recovery_evidence(
982                    recovery_path,
983                    candidate_digest,
984                    candidate["artifact_sha256"],
985                    "4" * 64,
986                    root,
987                )
988            except IdentityEvidenceError:
989                pass
990            else:
991                raise IdentityEvidenceError(
992                    "invalid rollback continuity was accepted"
993                )
994        rollback_path.write_text(json.dumps(rollback), encoding="utf-8")
995        substituted_recovery = json.loads(json.dumps(recovery))
996        substituted_recovery["rollback_continuity"] = substituted_recovery[
997            "identity_matrix"
998        ]
999        substituted_recovery["evidence_sha256"] = canonical_digest(
1000            {
1001                key: value
1002                for key, value in substituted_recovery.items()
1003                if key != "evidence_sha256"
1004            }
1005        )
1006        recovery_path.write_text(json.dumps(substituted_recovery), encoding="utf-8")
1007        try:
1008            validate_recovery_evidence(
1009                recovery_path,
1010                candidate_digest,
1011                candidate["artifact_sha256"],
1012                "4" * 64,
1013                root,
1014            )
1015        except IdentityEvidenceError:
1016            pass
1017        else:
1018            raise IdentityEvidenceError("recovery evidence substitution was accepted")
1019        matrix["cases"].pop()
1020        matrix_path.write_text(json.dumps(matrix), encoding="utf-8")
1021        try:
1022            validate_identity_matrix(matrix_path, candidate_digest)
1023        except IdentityEvidenceError:
1024            pass
1025        else:
1026            raise IdentityEvidenceError("truncated matrix was accepted")
1027        tripwires["surfaces"][0]["match_detected"] = True
1028        tripwire_path.write_text(json.dumps(tripwires), encoding="utf-8")
1029        try:
1030            validate_installed_tripwires(tripwire_path, candidate_digest)
1031        except IdentityEvidenceError:
1032            pass
1033        else:
1034            raise IdentityEvidenceError("matching tripwire receipt was accepted")
1035        observations["accessibility"][0]["facts"] = {}
1036        observations_path.write_text(json.dumps(observations), encoding="utf-8")
1037        try:
1038            validate_installed_observations(observations_path, candidate_digest, root)
1039        except IdentityEvidenceError:
1040            pass
1041        else:
1042            raise IdentityEvidenceError("invalid accessibility facts were accepted")
1043        valid_observations = json.loads(
1044            json.dumps(observations)
1045        )
1046        valid_observations["accessibility"][0]["facts"] = observation_facts[
1047            valid_observations["accessibility"][0]["check"]
1048        ]
1049        invalid_variants = []
1050        missing = json.loads(json.dumps(valid_observations))
1051        missing["accessibility"].pop()
1052        missing["evidence_sha256"] = canonical_digest(
1053            {key: value for key, value in missing.items() if key != "evidence_sha256"}
1054        )
1055        invalid_variants.append(missing)
1056        duplicate = json.loads(json.dumps(valid_observations))
1057        duplicate["accessibility"][1]["check"] = duplicate["accessibility"][0]["check"]
1058        duplicate["evidence_sha256"] = canonical_digest(
1059            {key: value for key, value in duplicate.items() if key != "evidence_sha256"}
1060        )
1061        invalid_variants.append(duplicate)
1062        stale = json.loads(json.dumps(valid_observations))
1063        stale["candidate_digest"] = "9" * 64
1064        stale["evidence_sha256"] = canonical_digest(
1065            {key: value for key, value in stale.items() if key != "evidence_sha256"}
1066        )
1067        invalid_variants.append(stale)
1068        forged = json.loads(json.dumps(valid_observations))
1069        forged["manual_journey"][0]["evidence_refs"][0]["sha256"] = "8" * 64
1070        forged["evidence_sha256"] = canonical_digest(
1071            {key: value for key, value in forged.items() if key != "evidence_sha256"}
1072        )
1073        invalid_variants.append(forged)
1074        for invalid in invalid_variants:
1075            observations_path.write_text(json.dumps(invalid), encoding="utf-8")
1076            try:
1077                validate_installed_observations(observations_path, candidate_digest, root)
1078            except IdentityEvidenceError:
1079                pass
1080            else:
1081                raise IdentityEvidenceError("invalid installed observations were accepted")
1082    print("Omega identity evidence receipt self-test passed")
1083
1084
1085if __name__ == "__main__":
1086    self_test()
1087