"""Synthetic exact-custody and offline-image guards; no operator/provider access.""" import base64 import hashlib import importlib.util import json import stat import subprocess from pathlib import Path import pytest def module(name, monkeypatch): scripts = Path(__file__).resolve().parents[1] / "scripts" monkeypatch.syspath_prepend(str(scripts)) spec = importlib.util.spec_from_file_location(name.replace("-", "_"), scripts / (name + ".py")) result = importlib.util.module_from_spec(spec) spec.loader.exec_module(result) return result @pytest.mark.parametrize("failure", [None, "write", "readback"]) def test_caller_writer_uses_cas0_and_revokes_on_every_outcome(monkeypatch, failure): caller = module("admit-scoped-caller", monkeypatch) value = "synthetic-caller-credential" monkeypatch.setattr(caller.secrets, "token_urlsafe", lambda *args: value) calls = [] def transport(args, payload=None, token=None, missing=False): calls.append((args, payload, token)) if args[0] == "token": return json.dumps({"auth": {"client_token": "synthetic-writer", "policies": [caller.WRITER], "lease_duration": 300}}).encode() if args == ["token", "revoke", "-self"]: return b"{}" if "platform/metadata/workloads/email-connect/callers/user-engine" in args: return None if args[0] == "write": assert payload == {"options": {"cas": 0}, "data": {"TOKEN": value}} if failure == "write": raise caller.Refused("synthetic_failure") return b'{"data":{"version":1}}' if failure == "readback": raise caller.Refused("synthetic_failure") return json.dumps({"data": {"data": {"TOKEN": value}, "metadata": {"version": 1}}}).encode() if failure: with pytest.raises(caller.Refused): caller.provision(transport) else: receipt = caller.provision(transport) assert receipt["kv_version"] == 1 and receipt["readback_equal"] assert value not in json.dumps(receipt) assert calls[-1][0] == ["token", "revoke", "-self"] assert calls[-1][2] == "synthetic-writer" def test_caller_writer_rejects_inherited_authority_before_any_secret_access(monkeypatch): caller = module("admit-scoped-caller", monkeypatch) seen = [] def transport(args, *rest, **kwargs): seen.append(args) if args[0] == "token": return json.dumps({"auth": {"client_token": "synthetic-writer", "policies": [caller.WRITER], "identity_policies": ["platform-admin"], "lease_duration": 300}}).encode() return b"{}" with pytest.raises(caller.Refused, match="bounded_writer_required"): caller.provision(transport) assert len(seen) == 2 and seen[-1] == ["token", "revoke", "-self"] def test_unreviewed_plan_refuses_before_decision_network_call(monkeypatch): caller = module("admit-scoped-caller", monkeypatch) monkeypatch.setattr(caller, "urlopen", lambda *a, **kw: pytest.fail("network forbidden")) with pytest.raises(caller.Refused, match="reviewed_plan_hash_required"): caller.approved_plan({}, "unreviewed", None, "http://synthetic.invalid") def test_caller_self_revocation_reaches_actual_bao_cli_endpoint(monkeypatch): import shutil from http.server import BaseHTTPRequestHandler, HTTPServer from threading import Thread if not shutil.which("bao"): pytest.skip("OpenBao CLI integration requires installed bao") caller = module("admit-scoped-caller", monkeypatch) seen = [] class Handler(BaseHTTPRequestHandler): def do_POST(self): seen.append((self.path, self.headers.get("X-Vault-Token"))) self.send_response(204) self.end_headers() def do_PUT(self): self.do_POST() def log_message(self, *args): pass server = HTTPServer(("127.0.0.1", 0), Handler) worker = Thread(target=server.serve_forever, daemon=True) worker.start() monkeypatch.setenv("BAO_ADDR", f"http://127.0.0.1:{server.server_port}") monkeypatch.setenv("VAULT_ADDR", f"http://127.0.0.1:{server.server_port}") monkeypatch.delenv("BAO_NAMESPACE", raising=False) monkeypatch.delenv("VAULT_NAMESPACE", raising=False) try: caller.bao(["token", "revoke", "-self"], token="synthetic-only-writer") finally: server.shutdown() server.server_close() worker.join(timeout=2) assert seen == [("/v1/auth/token/revoke-self", "synthetic-only-writer")] def test_caller_failure_receipt_allows_only_static_diagnostic_codes(monkeypatch): caller = module("admit-scoped-caller", monkeypatch) assert caller.failure_receipt(caller.Refused("caller_readback_differs"))["failure_code"] == "caller_readback_differs" for error in [caller.Refused("synthetic-sensitive-value"), ValueError("synthetic-sensitive-value")]: receipt = caller.failure_receipt(error) assert receipt["failure_code"] == "contained_operation_failed" assert "synthetic-sensitive-value" not in json.dumps(receipt) def test_recovery_kit_public_module_is_readable_by_rehearsal_uid(monkeypatch, tmp_path): kit = module("prepare-ledger-recovery-kit", monkeypatch) cipher = tmp_path / "ledger.tar.age" cipher.write_bytes(b"synthetic-encrypted-ledger") cipher.chmod(0o600) hashed = hashlib.sha256(cipher.read_bytes()).hexdigest() evidence = tmp_path / "evidence.json" evidence.write_text(json.dumps({ "publication": {"ciphertext_sha256": hashed, "ciphertext_bytes": cipher.stat().st_size, "version_id": "synthetic-version"}, "application_recovery": {"application_restore_verified": True, "application": "email-connect", "ciphertext_sha256": hashed, "version_id": "synthetic-version", "recovery_image": "synthetic-image"}, })) def docker(args, **kwargs): if args[2] == "inspect": return subprocess.CompletedProcess(args, 0, json.dumps([{"Id": "sha256:" + "a" * 64}]).encode()) Path(args[args.index("--output") + 1]).write_bytes(b"synthetic-image-archive") return subprocess.CompletedProcess(args, 0, b"") monkeypatch.setattr(kit.subprocess, "run", docker) destination = tmp_path / "kit" kit.prepare(evidence, cipher, destination) assert stat.S_IMODE(destination.stat().st_mode) == 0o700 assert stat.S_IMODE((destination / "ledger_archive.py").stat().st_mode) == 0o644 for name in ["ledger.tar.age", "receiver-image.tar", "transfer-receipt.json", "kit.tar.gz"]: assert stat.S_IMODE((destination / name).stat().st_mode) == 0o600 @pytest.mark.parametrize("protected,ephemeral,corrupt", [ (True, None, False), (False, None, False), (True, {}, False), (True, None, True), ]) def test_scaleway_key_escrow_requires_exact_protected_nonexpiring_readback(monkeypatch, protected, ephemeral, corrupt): escrow = module("scaleway-recovery-custody", monkeypatch) identity = b"synthetic-only-private-identity" secret_id = "00000000-0000-0000-0000-000000000001" project_id = "00000000-0000-0000-0000-000000000002" monkeypatch.setattr(escrow, "recipient", lambda value: escrow.RECIPIENT) seen = [] def transport(secret, suffix="", payload=None): seen.append((secret, suffix, payload)) if suffix == "": return {"name": escrow.NAME, "project_id": project_id, "region": escrow.REGION, "type": "opaque", "protected": protected, "ephemeral_policy": ephemeral} if "?" in suffix: return {"total_count": 0, "versions": []} if payload is not None: assert base64.b64decode(payload["data"]) == identity assert payload["disable_previous"] is False return {"revision": 1} return {"secret_id": secret_id, "revision": 1, "data": base64.b64encode(b"synthetic-corrupt" if corrupt else identity).decode()} if protected and ephemeral is None and not corrupt: result = escrow.escrow(secret_id, project_id, identity, transport) assert result["readback_equal"] and not result["worker_access_denial_verified"] assert identity.decode() not in json.dumps(result) else: with pytest.raises(escrow.Refused): escrow.escrow(secret_id, project_id, identity, transport) if not protected or ephemeral is not None: assert len(seen) == 1 def test_offline_recovery_image_is_bound_to_export_and_capture_image(monkeypatch, tmp_path): rehearsal = module("rehearse-ledger", monkeypatch) image = tmp_path / "receiver-image.tar" image.write_bytes(b"synthetic-image-export") image.chmod(0o600) manifest = tmp_path / "manifest.json" manifest.write_text(json.dumps({"schema": "email-connect.ledger-recovery-kit.v1", "image": rehearsal.IMAGE, "image_id": "sha256:" + "a" * 64, "files": {"receiver-image.tar": {"sha256": hashlib.sha256(image.read_bytes()).hexdigest(), "bytes": image.stat().st_size}}})) manifest.chmod(0o600) assert rehearsal.recovery_image(manifest) == "sha256:" + "a" * 64 image.write_bytes(b"synthetic-changed-export") with pytest.raises(rehearsal.Refused, match="offline_image_archive_changed"): rehearsal.recovery_image(manifest) @pytest.mark.parametrize("lifecycle,corrupt", [(False, False), (True, False), (False, True)]) def test_kit_requires_no_expiry_and_separate_version_hash_readback(monkeypatch, tmp_path, lifecycle, corrupt): import io publisher = module("publish-ledger-recovery-kit", monkeypatch) kit = tmp_path / "kit.tar.gz" content = b"synthetic-kit-with-no-private-identity" kit.write_bytes(content) expected = {"status": "recovery_kit_prepared", "private_identity_included": False, "kit_sha256": hashlib.sha256(content).hexdigest(), "kit_bytes": len(content)} monkeypatch.setattr(publisher, "private_directory", lambda path: None) calls = [] class Owner: def get_bucket_versioning(self, **kwargs): return {"Status": "Enabled"} def get_bucket_lifecycle_configuration(self, **kwargs): return {"Rules": [{"Expiration": {"Days": 30}}]} if lifecycle else {"Rules": []} def put_object(self, **kwargs): calls.append(kwargs) assert kwargs["IfNoneMatch"] == "*" return {"VersionId": "synthetic-kit-version"} class Reader: def get_object(self, **kwargs): assert kwargs["VersionId"] == "synthetic-kit-version" return {"VersionId": "synthetic-kit-version", "ContentLength": len(content), "Body": io.BytesIO(b"synthetic-corrupt" if corrupt else content)} if lifecycle or corrupt: with pytest.raises(publisher.Refused): publisher.publish(Owner(), Reader(), kit, expected, tmp_path / "readback") if lifecycle: assert calls == [] else: result = publisher.publish(Owner(), Reader(), kit, expected, tmp_path / "readback") assert result["separate_reader_readback_verified"] assert not result["reader_application_binding_verified"] assert not result["private_identity_included"] def test_low_disk_headroom_refuses_before_capture(monkeypatch, tmp_path): from types import SimpleNamespace backup = module("backup-ledger", monkeypatch) monkeypatch.setattr(backup.shutil, "disk_usage", lambda *args: SimpleNamespace(free=1023)) with pytest.raises(backup.Refused, match="backup_disk_headroom_required"): backup.capacity(tmp_path / "not-created", minimum_free=1024) assert not (tmp_path / "not-created").exists() def test_retention_budget_refuses_without_deleting_ciphertext(monkeypatch, tmp_path): from types import SimpleNamespace backup = module("backup-ledger", monkeypatch) state = tmp_path / "state" state.mkdir(mode=0o700) archive = state / "synthetic.age" archive.write_bytes(b"synthetic-protected-ciphertext") monkeypatch.setattr(backup.shutil, "disk_usage", lambda *args: SimpleNamespace(free=10**10)) with pytest.raises(backup.Refused, match="backup_local_retention_review_required"): backup.capacity(state, budget=512 * 1024**2) assert archive.read_bytes() == b"synthetic-protected-ciphertext"