tegwick
f77e2333af
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a11fd9-caf6-7a63-b578-9daaaa134ff2
262 lines
12 KiB
Python
262 lines
12 KiB
Python
"""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"
|