railiance-platform/tests/test_openbao_isolated_restore.py

"""RPF-WP-0064-T02/T03 unit tests without Docker: admission, verifier, launcher.

Adversarial receipt fixtures are derived from one valid synthetic receipt built
here; nothing in this module touches Docker, OpenBao or custody material.
"""
from __future__ import annotations

import copy
import hashlib
import json
import os
import shutil
import subprocess
import sys
import tempfile
import unittest
from datetime import UTC, datetime, timedelta
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
import openbao_restore_rehearsal as R  # noqa: E402

A = R.adapter
V = R.load("verify_restore_receipt", "scripts/openbao-isolated-restore/verify-restore-receipt.py")
FAKE_IMAGE = "quay.io/openbao/openbao@sha256:" + hashlib.sha256(b"image").hexdigest()


def digest(seed: str) -> str:
    return "sha256:" + hashlib.sha256(seed.encode()).hexdigest()


class Base(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.work = Path(tempfile.mkdtemp(prefix="rpf64-unit-"))
        cls.root, cls.commit = R.make_source_root(cls.work)

    @classmethod
    def tearDownClass(cls):
        shutil.rmtree(cls.work, ignore_errors=True)

    def admission(self, **override):
        now = datetime.now(UTC).replace(microsecond=0)
        adm = json.loads((R.ROOT / "docs/openbao-restore-admission.example.json").read_text())
        adm.update({"attempt_id": "RPF-WP-0064-T04-unit-01", "material": "synthetic",
                    "not_before": A.iso(now - timedelta(minutes=5)), "not_after": A.iso(now + timedelta(hours=1)),
                    "image": FAKE_IMAGE, "adapter_commit": self.commit, "source_sha256": A.source_digests(self.root)})
        adm["snapshot"].update({"ciphertext_sha256": digest("c"), "ciphertext_bytes": 10,
                                "plaintext_sha256": digest("p"), "source_receipt_sha256": digest("r"),
                                "age_recipient": R.Rehearsal.age_pair()[1],
                                "source_cluster_id": "9b5aa97f-933c-bd94-095a-5b5d7f766da5",
                                "source_applied_index": 50})
        adm.update(override)
        return adm

    def receipt(self):
        rc = json.loads((R.ROOT / "docs/openbao-restore-receipt-v2.example.json").read_text())
        adm = self.admission()
        now = datetime.now(UTC).replace(microsecond=0)
        rc.update({"attempt_id": adm["attempt_id"], "material": "synthetic", "admission_sha256": digest("adm"),
                   "signer_principal": "founder@rehearsal", "anchor_protected": False,
                   "signer_fingerprint": "SHA256:" + "AbCd" * 10 + "AbC", "adapter_commit": self.commit,
                   "source_sha256": adm["source_sha256"], "started_at": A.iso(now - timedelta(seconds=40)),
                   "finished_at": A.iso(now), "not_after": adm["not_after"], "elapsed_seconds": 40,
                   "image": FAKE_IMAGE, "image_id": digest("id")})
        rc["snapshot"].update({k: adm["snapshot"][k] for k in rc["snapshot"] if k in adm["snapshot"]})
        rc["restored"].update({"cluster_id": adm["snapshot"]["source_cluster_id"], "applied_index": 50})
        for row in rc["objects"]:
            row["observed_sha256"] = digest(row["object"])
        return rc

    def verify(self, rc, **kw):
        return V.verify(copy.deepcopy(rc), source_root=self.root, **kw)


class AdmissionTests(Base):
    def test_valid_synthetic_admission(self):
        A.validate_admission(self.admission())

    def test_example_admission_refused_as_placeholder(self):
        with self.assertRaises(A.Refused):
            A.validate_admission(json.loads((R.ROOT / "docs/openbao-restore-admission.example.json").read_text()))

    def test_refusals(self):
        cases = [
            {"phase": "B"}, {"phase_b_admitted": True}, {"image": "quay.io/openbao/openbao:2.5.4"},
            {"image": "quay.io/openbao/openbao@sha256:" + "0" * 64}, {"image": "docker.io/x/openbao@" + digest("i")},
            {"max_duration_seconds": 2701}, {"material": "original", "max_duration_seconds": 600},
            {"namespace": "secrets-engine-recovery-attempt"}, {"readback_credential": "operator-oidc"},
            {"unseal": {"key_threshold": 1, "key_shares": 1}}, {"extra": True}, {"adapter_commit": "HEAD"},
        ]
        for override in cases:
            with self.subTest(override=override), self.assertRaises(A.Refused):
                A.validate_admission(self.admission(**override))
        adm = self.admission()
        adm["source_sha256"].pop(next(iter(adm["source_sha256"])))
        with self.assertRaises(A.Refused):
            A.validate_admission(adm)

    def test_window(self):
        adm = self.admission()
        A.check_window(adm)
        with self.assertRaises(A.Refused):
            A.check_window(adm, A.parse_iso(adm["not_after"]))

    def test_source_closure_detects_dirty_or_changed(self):
        adm = self.admission()
        A.check_source_closure(adm, self.root)
        with self.assertRaises(A.Refused):
            A.check_source_closure(dict(adm, adapter_commit="f" * 40), self.root)
        target = self.root / "openbao/auth/secrets-engine-jwt-role.json"
        original = target.read_bytes()
        try:
            target.write_bytes(original + b" ")
            with self.assertRaises(A.Refused):
                A.check_source_closure(adm, self.root)
        finally:
            target.write_bytes(original)

    def signed(self, adm, namespace=A.NAMESPACE):
        anchor = self.work / "anchor"
        anchor.mkdir(mode=0o700, exist_ok=True)
        key = anchor / "k"
        if not key.exists():
            subprocess.run(["ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-C", "f", "-f", str(key)], check=True)
            pub = (anchor / "k.pub").read_text().strip()
            (anchor / "allowed_signers").write_text(f'founder@rehearsal namespaces="{A.NAMESPACE}" {pub}\n')
        raw = json.dumps(adm).encode()
        path = Path(tempfile.mkstemp(dir=self.work)[1])
        path.write_bytes(raw)
        subprocess.run(["ssh-keygen", "-q", "-Y", "sign", "-f", str(key), "-n", namespace, str(path)], check=True)
        sig = Path(str(path) + ".sig")
        os.chmod(sig, 0o600)
        return path, sig, A.sha256_bytes(raw), anchor / "allowed_signers"

    def test_signature_namespace_and_bytes_bound(self):
        path, sig, sha, signers = self.signed(self.admission())
        _adm, _raw, signer, protected = A.load_admission(path, sig, sha, signers)
        self.assertEqual(signer["principal"], "founder@rehearsal")
        self.assertFalse(protected)  # runtime-owned anchor: rehearsal only
        with self.assertRaises(A.Refused):
            A.load_admission(path, sig, digest("other"), signers)
        path2, sig2, sha2, _ = self.signed(self.admission(), namespace="secrets-engine-recovery-attempt")
        with self.assertRaises(A.Refused):
            A.load_admission(path2, sig2, sha2, signers)

    def test_original_material_needs_protected_anchor_and_t06_pin(self):
        path, sig, sha, signers = self.signed(self.admission(material="original"))
        with self.assertRaises(A.Refused):
            A.load_admission(path, sig, sha, signers)
        self.assertEqual(A.PINNED_IMAGE_DIGEST,
                         "sha256:436eaf9778cad75507ff70ea26ace30dcbe15606e619ac3823495663d7f7c115")

    def test_normalise_and_project(self):
        source = {"token_ttl": "5m", "token_policies": ["b", "a"], "_note": "x"}
        ok, projected = A.project(source, {"token_ttl": 300, "token_policies": ["a", "b"], "extra": 1}, ("_note",))
        self.assertTrue(ok)
        self.assertEqual(projected, {"token_policies": ["a", "b"], "token_ttl": 300})
        self.assertFalse(A.project(source, {"token_ttl": 600, "token_policies": ["a", "b"]}, ("_note",))[0])

    def test_reads_never_include_kv_data(self):
        self.assertFalse(any("/data/" in p for p in A.ALLOWED_READS))
        self.assertIn(A.KV_METADATA_PATH, A.ALLOWED_READS)


class VerifierTests(Base):
    def test_valid_receipts(self):
        self.assertEqual(self.verify(self.receipt()), "restored")
        rc = self.receipt()
        rc["objects"][-1].update({"present": False, "source_match": False, "observed_sha256": None})
        rc.update(result="refused", reason="object_missing")
        self.assertEqual(self.verify(rc), "refused")
        rc = self.receipt()
        rc.update(result="aborted", reason="unseal_refused", objects=[])
        self.assertEqual(self.verify(rc), "aborted")

    def test_example_keys_match_verifier_schema(self):
        rc = json.loads((R.ROOT / "docs/openbao-restore-receipt-v2.example.json").read_text())
        self.assertEqual(set(rc), V.TOP)
        self.assertEqual(set(rc["teardown"]), V.TEARDOWN)
        self.assertEqual(tuple(o["object"] for o in rc["objects"]), V.REQUIRED_OBJECTS)
        self.assertEqual(tuple(o["object"] for o in A.OBJECTS), V.REQUIRED_OBJECTS)
        self.assertEqual(set(A.CLAIMS_NOT_MADE), V.CLAIMS)
        self.assertEqual(A.ABORT_REASONS, V.ABORT)
        with self.assertRaises(V.Reject):
            self.verify(rc)  # placeholders

    def reject(self, mutate, code=None, **kw):
        rc = self.receipt()
        mutate(rc)
        with self.assertRaises(V.Reject) as ctx:
            self.verify(rc, **kw)
        if code:
            self.assertEqual(str(ctx.exception), code)

    def test_adversarial_fixtures(self):
        share = "Kq3/" + "x9Z+" * 10
        cases = {
            "unknown_field": (lambda r: r.update(notes="ok"), "schema_fields"),
            "missing_field": (lambda r: r.pop("teardown"), "schema_fields"),
            "missing_teardown": (lambda r: r["teardown"].update(container_removed=False), "teardown_missing"),
            "reaper_on_success": (lambda r: r["teardown"].update(performed_by="reaper"), "teardown_missing"),
            "unrevoked": (lambda r: r["teardown"].update(readback_credential_revoked=False), "teardown_missing"),
            "data_field": (lambda r: r["objects"][0].update(data={"x": 1}), "object_fields"),
            "kv_data_path": (lambda r: r["objects"][-1].update(
                path="platform/data/workloads/keycape/secrets-engine-openbao"), "kv_data_path"),
            "kv_data_read": (lambda r: r.update(kv_data_read=True), "secret_policy"),
            "credential_shaped": (lambda r: r.update(signer_principal="hvs.CAESIJ" + "a" * 30), None),
            "share_shaped": (lambda r: r["objects"][0].update(path=share), "secret_shaped_value"),
            "hex_key_shaped": (lambda r: r["objects"][0].update(path="f00d" * 16), "secret_shaped_value"),
            "age_identity": (lambda r: r["objects"][0].update(path="AGE-SECRET-KEY-1QQQ"), "secret_shaped_value"),
            "pem": (lambda r: r["objects"][0].update(path="-----BEGIN PRIVATE"), "secret_shaped_value"),
            "placeholder_digest": (lambda r: r["snapshot"].update(ciphertext_sha256="sha256:" + "1" * 64),
                                   "snapshot_digest"),
            "placeholder_text": (lambda r: r["objects"][0].update(path="replace-with-path"), "placeholder"),
            "untyped_digest": (lambda r: r.update(admission_sha256=hashlib.sha256(b"x").hexdigest()),
                               "admission_digest"),
            "claims_missing": (lambda r: r["claims_not_made"].remove("consumer_continuity"), "claims_not_made"),
            "deadline_overrun": (lambda r: r.update(elapsed_seconds=4000, started_at=A.iso(
                A.parse_iso(r["finished_at"]) - timedelta(seconds=4000))), "deadline_overrun"),
            "original_short_deadline": (lambda r: r.update(material="original", anchor_protected=True,
                                                           deadline_seconds=600), "deadline"),
            "original_unprotected_anchor": (lambda r: r.update(material="original"), "anchor_unprotected"),
            "restored_with_mismatch": (lambda r: r["objects"][3].update(source_match=False), "result_inconsistent"),
            "restored_network": (lambda r: r["network"].update(interfaces=["eth0", "lo"]), "network_not_contained"),
            "cluster_mismatch": (lambda r: r["restored"].update(cluster_id_match=False), "restore_inconsistent"),
            "objects_incomplete": (lambda r: r["objects"].pop(), "objects_incomplete"),
            "bad_reason": (lambda r: r.update(result="aborted", reason="because"), "result_inconsistent"),
            "phase_b": (lambda r: r.update(phase="B"), "phase"),
            "source_hash": (lambda r: r["source_sha256"].update(
                {"openbao/auth/secrets-engine-jwt-role.json": digest("zz")}), "source_hash_mismatch"),
            "source_commit": (lambda r: r.update(adapter_commit="a" * 40), "source_commit_mismatch"),
        }
        for name, (mutate, code) in cases.items():
            with self.subTest(name):
                self.reject(mutate, code)

    def test_not_after_window(self):
        self.reject(lambda r: None, "verification_window_closed",
                    not_after=datetime(2026, 1, 1, tzinfo=UTC))

    def test_duplicate_keys_rejected_by_cli(self):
        path = self.work / "dup.json"
        path.write_text('{"schema": "a", "schema": "b"}')
        done = subprocess.run([sys.executable, str(R.VERIFIER), "--receipt", str(path), "--source-root",
                               str(self.root)], capture_output=True)
        self.assertEqual(done.returncode, 1)
        self.assertIn(b"duplicate_key", done.stdout)

    def test_cli_pass_prints_digest_only(self):
        path = self.work / "good.json"
        path.write_text(json.dumps(self.receipt()))
        done = subprocess.run([sys.executable, str(R.VERIFIER), "--receipt", str(path), "--source-root",
                               str(self.root), "--expect-result", "restored"], capture_output=True)
        self.assertEqual(done.returncode, 0, done.stdout)
        self.assertRegex(done.stdout.decode(), r"^PASS: restored receipt sha256:[a-f0-9]{64}\n$")


class LauncherTests(Base):
    launcher = R.ROOT / "scripts/openbao-isolated-restore/run-isolated-restore.sh"

    def run_launcher(self, *args):
        return subprocess.run(["bash", str(self.launcher), *args], capture_output=True, stdin=subprocess.DEVNULL)

    def test_refusals_before_any_prompt(self):
        f = self.work / "f"
        f.write_text("x")
        self.assertIn(b"Usage", self.run_launcher().stdout)
        self.assertIn(b"exact sha256", self.run_launcher(str(f), str(f), "abc", str(f), str(self.work)).stdout)
        out = self.run_launcher(str(f), str(f), digest("x"), str(f), str(self.work)).stdout
        self.assertIn(b"interactive terminal", out)
        self.assertNotIn(b"hidden", out)

    def test_preflight_refuses_unsigned_admission_without_prompt(self):
        adm = self.work / "adm.json"
        raw = json.dumps(self.admission(material="original")).encode()
        adm.write_bytes(raw)
        os.chmod(adm, 0o600)
        done = subprocess.run([sys.executable, str(R.ROOT / "scripts/openbao-isolated-restore/preflight.py"),
                               "--admission", str(adm), "--admission-sig", str(adm), "--admission-sha256",
                               A.sha256_bytes(raw), "--ciphertext", str(adm), "--receipt", str(self.work / "r.json")],
                              capture_output=True, stdin=subprocess.DEVNULL)
        self.assertEqual(done.returncode, 1)
        self.assertIn(b"admission_refused", done.stdout)
        self.assertNotIn(b"hidden", done.stdout + done.stderr)


if __name__ == "__main__":
    unittest.main()