tenant.openagents/omega

1#!/usr/bin/env python3
2"""Observe that an Omega identity survives a downgrade and a rollback.
3
4The installed proof needs recovery evidence, and recovery evidence needs a real
5downgrade/rollback observation. Nothing produced one, so that input was blocked
6and no amount of other evidence could unblock it.
7
8This performs the operation rather than describing it: install the candidate,
9record the identity, downgrade to an older signed artifact, record the identity
10again, roll forward to the candidate, and record it a third time. The claim is
11that the identity reference is byte-identical across all three, which is what
12"an update cannot silently rotate the owner's identity" means in practice.
13
14Only the public identity reference is recorded. A secret key, a recovery
15artifact, or an `nsec`/`ncryptsec` string in the fingerprint is refused, so this
16receipt cannot become a place a secret leaks to.
17
18Usage:
19    observe-omega-rollback-continuity \\
20        --candidate-dmg <rc5.dmg> --older-dmg <rc4.dmg> \\
21        --candidate-evidence <path> --evidence-root <dir> --output <path>
22"""
23
24from __future__ import annotations
25
26import argparse
27import hashlib
28import json
29import shutil
30import subprocess
31import sys
32import time
33from datetime import datetime, timezone
34from pathlib import Path
35from typing import Any
36
37ROLLBACK_CONTINUITY_SCHEMA = "openagents.omega.identity-update-downgrade-rollback.v2"
38INSTALLED_APP = Path("/Applications/Omega.app")
39DATA_ROOT = Path.home() / "Library" / "Application Support" / "Omega RC"
40IDENTITY_MANIFEST = DATA_ROOT / "identity" / "identity.json"
41
42
43class ContinuityError(Exception):
44    pass
45
46
47def sha256_file(path: Path) -> str:
48    digest = hashlib.sha256()
49    with path.open("rb") as handle:
50        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
51            digest.update(chunk)
52    return digest.hexdigest()
53
54
55def canonical_digest(value: Any) -> str:
56    encoded = json.dumps(
57        value, sort_keys=True, separators=(",", ":"), ensure_ascii=False
58    ).encode("utf-8")
59    return hashlib.sha256(encoded).hexdigest()
60
61
62def read_identity_fingerprint() -> str:
63    """The public identity reference, which must not change across an update."""
64    if not IDENTITY_MANIFEST.is_file():
65        raise ContinuityError(
66            f"no identity manifest at {IDENTITY_MANIFEST}; run onboarding first"
67        )
68    manifest = json.loads(IDENTITY_MANIFEST.read_text(encoding="utf-8"))
69    reference = manifest.get("identity", {}).get("identity_ref")
70    if not isinstance(reference, str) or not reference.strip():
71        raise ContinuityError("identity manifest has no usable identity_ref")
72    lowered = reference.lower()
73    # The receipt is a public artifact. A secret-shaped fingerprint would turn
74    # a continuity proof into a leak.
75    if "nsec" in lowered or "ncryptsec" in lowered or len(reference) > 256:
76        raise ContinuityError("identity reference is unsafe to record")
77    return reference
78
79
80def run(command: list[str]) -> None:
81    result = subprocess.run(command, capture_output=True, text=True)
82    if result.returncode != 0:
83        raise ContinuityError(f"{command[0]} failed: {result.stderr.strip()[:200]}")
84
85
86def install_from_dmg(dmg: Path) -> None:
87    """Replace the installed app from a signed disk image."""
88    subprocess.run(
89        ["hdiutil", "detach", "/Volumes/Omega RC", "-quiet"], capture_output=True
90    )
91    run(["hdiutil", "attach", str(dmg), "-nobrowse", "-quiet"])
92    try:
93        source = Path("/Volumes/Omega RC/Omega.app")
94        if not source.is_dir():
95            raise ContinuityError(f"{dmg.name} does not contain Omega.app")
96        if INSTALLED_APP.exists():
97            shutil.rmtree(INSTALLED_APP)
98        subprocess.run(
99            ["cp", "-R", str(source), str(INSTALLED_APP.parent)], check=True
100        )
101    finally:
102        subprocess.run(
103            ["hdiutil", "detach", "/Volumes/Omega RC", "-quiet"], capture_output=True
104        )
105
106
107def installed_binary_sha256() -> str:
108    binary = INSTALLED_APP / "Contents" / "MacOS" / "omega"
109    if not binary.is_file():
110        raise ContinuityError("installed Omega binary is missing")
111    return sha256_file(binary)
112
113
114def write_step(root: Path, name: str, payload: dict[str, Any]) -> dict[str, str]:
115    """Write one observation file and return its reference."""
116    path = root / f"{name}.json"
117    path.write_text(
118        json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8"
119    )
120    return {"path": path.name, "sha256": sha256_file(path)}
121
122
123def observed_at() -> str:
124    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
125
126
127def main() -> int:
128    parser = argparse.ArgumentParser()
129    parser.add_argument("--candidate-dmg", type=Path, required=True)
130    parser.add_argument("--older-dmg", type=Path, required=True)
131    parser.add_argument("--candidate-evidence", type=Path, required=True)
132    parser.add_argument("--evidence-root", type=Path, required=True)
133    parser.add_argument("--output", type=Path, required=True)
134    args = parser.parse_args()
135
136    if sys.platform != "darwin":
137        raise ContinuityError("macos_required")
138
139    evidence = json.loads(args.candidate_evidence.read_text(encoding="utf-8"))
140    candidate_digest = evidence.get("candidate_digest", {}).get("value")
141    if not isinstance(candidate_digest, str) or len(candidate_digest) != 64:
142        raise ContinuityError("candidate evidence has no usable candidate_digest")
143
144    candidate_artifact = sha256_file(args.candidate_dmg)
145    older_artifact = sha256_file(args.older_dmg)
146    if candidate_artifact == older_artifact:
147        raise ContinuityError(
148            "downgrade target must differ from the candidate; a rollback to the "
149            "same artifact proves nothing"
150        )
151
152    root = args.evidence_root
153    root.mkdir(parents=True, exist_ok=True)
154
155    # 1. Candidate installed, identity as it stands.
156    install_from_dmg(args.candidate_dmg)
157    candidate_binary = installed_binary_sha256()
158    before = read_identity_fingerprint()
159    refs: dict[str, dict[str, str]] = {}
160    refs["candidate_before"] = write_step(
161        root,
162        "candidate-before",
163        {
164            "step": "candidate_installed",
165            "artifact_sha256": candidate_artifact,
166            "installed_binary_sha256": candidate_binary,
167            "identity_ref": before,
168            "observed_at": observed_at(),
169        },
170    )
171
172    # 2. Downgrade to the older signed artifact.
173    refs["downgrade_removal"] = write_step(
174        root,
175        "downgrade-removal",
176        {
177            "step": "removed_candidate",
178            "removed_artifact_sha256": candidate_artifact,
179            "observed_at": observed_at(),
180        },
181    )
182    install_from_dmg(args.older_dmg)
183    older_binary = installed_binary_sha256()
184    refs["downgrade_install"] = write_step(
185        root,
186        "downgrade-install",
187        {
188            "step": "installed_older",
189            "artifact_sha256": older_artifact,
190            "installed_binary_sha256": older_binary,
191            "observed_at": observed_at(),
192        },
193    )
194    time.sleep(1)
195    after_downgrade = read_identity_fingerprint()
196    refs["identity_after_downgrade"] = write_step(
197        root,
198        "identity-after-downgrade",
199        {
200            "step": "identity_on_older",
201            "identity_ref": after_downgrade,
202            "matches_before": after_downgrade == before,
203            "observed_at": observed_at(),
204        },
205    )
206
207    # 3. Roll forward to the candidate.
208    refs["rollback_removal"] = write_step(
209        root,
210        "rollback-removal",
211        {
212            "step": "removed_older",
213            "removed_artifact_sha256": older_artifact,
214            "observed_at": observed_at(),
215        },
216    )
217    install_from_dmg(args.candidate_dmg)
218    rolled_binary = installed_binary_sha256()
219    if rolled_binary != candidate_binary:
220        raise ContinuityError(
221            "rollback installed a different binary than the candidate"
222        )
223    refs["rollback_install"] = write_step(
224        root,
225        "rollback-install",
226        {
227            "step": "installed_candidate",
228            "artifact_sha256": candidate_artifact,
229            "installed_binary_sha256": rolled_binary,
230            "observed_at": observed_at(),
231        },
232    )
233    time.sleep(1)
234    after = read_identity_fingerprint()
235    refs["identity_after_rollback"] = write_step(
236        root,
237        "identity-after-rollback",
238        {
239            "step": "identity_on_candidate",
240            "identity_ref": after,
241            "matches_before": after == before,
242            "observed_at": observed_at(),
243        },
244    )
245
246    continuous = before == after_downgrade == after
247    receipt = {
248        "schema": ROLLBACK_CONTINUITY_SCHEMA,
249        "candidate_digest": candidate_digest,
250        "status": "passed" if continuous else "failed",
251        "observed_at": observed_at(),
252        "identity_fingerprint_before": before,
253        "identity_fingerprint_after": after,
254        "sequence": [
255            {
256                "action": "downgrade",
257                "from_artifact_sha256": candidate_artifact,
258                "to_artifact_sha256": older_artifact,
259            },
260            {
261                "action": "update_and_rollback_to_candidate",
262                "from_artifact_sha256": older_artifact,
263                "to_artifact_sha256": candidate_artifact,
264                "installed_binary_sha256": candidate_binary,
265            },
266        ],
267        "evidence_refs": refs,
268    }
269    receipt["evidence_sha256"] = canonical_digest(receipt)
270
271    args.output.parent.mkdir(parents=True, exist_ok=True)
272    args.output.write_text(
273        json.dumps(receipt, indent=2, sort_keys=True) + "\n", encoding="utf-8"
274    )
275    print(
276        json.dumps(
277            {
278                "schema": ROLLBACK_CONTINUITY_SCHEMA,
279                "status": receipt["status"],
280                "identity_stable_across_downgrade_and_rollback": continuous,
281                "output": str(args.output),
282            }
283        )
284    )
285    return 0 if continuous else 1
286
287
288if __name__ == "__main__":
289    try:
290        sys.exit(main())
291    except ContinuityError as error:
292        print(f"error: {error}", file=sys.stderr)
293        sys.exit(2)
294