tegwick
08906da7d0
Some checks failed
Use-case regression gate / usecase-regressions (push) Has been cancelled
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3
288 lines
11 KiB
Python
288 lines
11 KiB
Python
"""Historical evidence metadata sealing against disposable files only."""
|
|
|
|
import fcntl
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import subprocess
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).parents[1]
|
|
spec = importlib.util.spec_from_file_location("backup_custody", ROOT / "scripts/backup_custody.py")
|
|
custody = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(custody)
|
|
|
|
|
|
@pytest.fixture
|
|
def evidence(tmp_path, monkeypatch):
|
|
state = tmp_path / "state"
|
|
state.mkdir(mode=0o700)
|
|
directory = state / "releases"
|
|
directory.mkdir(mode=0o755)
|
|
(directory / "fixture.json").write_bytes(b'{"retained":"synthetic evidence"}\n')
|
|
(directory / "fixture.json").chmod(0o600)
|
|
monkeypatch.setattr(custody, "STATE", state)
|
|
# Native entry point always selects root. Exercise real chmod/fd/hash IO as
|
|
# the test user, and verify the root chown syscall separately below.
|
|
monkeypatch.setattr(
|
|
custody,
|
|
"desired",
|
|
lambda r: dict(
|
|
r,
|
|
uid=os.getuid(),
|
|
gid=os.getgid(),
|
|
mode="0o700" if r["kind"] == "directory" else "0o600",
|
|
),
|
|
)
|
|
rows = [custody.fingerprint("releases"), custody.fingerprint("releases/fixture.json")]
|
|
plan = {"targets": rows, "proposal_digest": "sha256:" + "a" * 64}
|
|
return plan, tmp_path / "attempt", state
|
|
|
|
|
|
def test_real_metadata_seal_preserves_bytes_paths_and_explicit_rollback(evidence):
|
|
plan, directory, state = evidence
|
|
raw = (state / "releases/fixture.json").read_bytes()
|
|
result = custody.transact(plan, directory)
|
|
assert result["status"] == "completed_evidence_private_root_custody_verified"
|
|
assert result["verified_entries"] == 2 and result["bytes_modified"] is False
|
|
assert (state / "releases").stat().st_mode & 0o777 == 0o700
|
|
assert (state / "releases/fixture.json").read_bytes() == raw
|
|
assert json.loads((directory / "receipt.json").read_text()) == result
|
|
result = custody.transact(plan, directory, rollback=True)
|
|
assert result["status"] == "original_metadata_restored_verified"
|
|
assert result["prior_receipt"]["status"] == "completed_evidence_private_root_custody_verified"
|
|
assert [custody.fingerprint(r["relative_path"]) for r in plan["targets"]] == plan["targets"]
|
|
|
|
|
|
def test_changed_body_refuses_before_any_metadata_write(evidence):
|
|
plan, directory, state = evidence
|
|
(state / "releases/fixture.json").write_bytes(b"changed")
|
|
with pytest.raises(custody.Refused, match="evidence_drift"):
|
|
custody.transact(plan, directory)
|
|
assert not directory.exists()
|
|
assert (state / "releases").stat().st_mode & 0o777 == 0o755
|
|
|
|
|
|
def test_replaced_inode_or_child_listing_refuses(evidence):
|
|
plan, directory, state = evidence
|
|
(state / "releases/unreviewed").write_bytes(b"new")
|
|
with pytest.raises(custody.Refused):
|
|
custody.transact(plan, directory)
|
|
assert not directory.exists()
|
|
|
|
|
|
def test_partial_seal_has_durable_preimages_and_can_restore(evidence, monkeypatch):
|
|
plan, directory, _ = evidence
|
|
original = custody.change
|
|
|
|
def interrupted(row, before, after):
|
|
original(row, before, after)
|
|
raise SystemExit(9)
|
|
|
|
monkeypatch.setattr(custody, "change", interrupted)
|
|
with pytest.raises(SystemExit):
|
|
custody.transact(plan, directory)
|
|
receipt = json.loads((directory / "receipt.json").read_text())
|
|
assert receipt["status"] == "seal_started" and len(receipt["changes"]) == 1
|
|
assert receipt["changes"][0]["before"] == plan["targets"][0]
|
|
monkeypatch.setattr(custody, "change", original)
|
|
result = custody.transact(plan, directory, rollback=True)
|
|
assert result["status"] == "original_metadata_restored_verified"
|
|
|
|
|
|
def test_postflight_failure_is_not_success_and_retains_journal(evidence):
|
|
plan, directory, _ = evidence
|
|
|
|
def fail():
|
|
raise RuntimeError("private provider text")
|
|
|
|
with pytest.raises(RuntimeError):
|
|
custody.transact(plan, directory, postflight=fail)
|
|
raw = (directory / "receipt.json").read_text()
|
|
assert json.loads(raw)["status"] == "requires_inspection"
|
|
assert "private provider text" not in raw
|
|
|
|
|
|
def test_interfering_rollback_and_duplicate_attempt_refuse(evidence):
|
|
plan, directory, state = evidence
|
|
custody.transact(plan, directory)
|
|
original = (directory / "receipt.json").read_bytes()
|
|
with pytest.raises(custody.Refused):
|
|
custody.transact(plan, directory)
|
|
assert (directory / "receipt.json").read_bytes() == original
|
|
(state / "releases/fixture.json").write_bytes(b"interference")
|
|
with pytest.raises(custody.Refused):
|
|
custody.transact(plan, directory, rollback=True)
|
|
assert (directory / "receipt.json").read_bytes() == original
|
|
|
|
|
|
def test_symlink_ancestor_and_final_file_refuse(evidence):
|
|
_, _, state = evidence
|
|
(state / "alias").symlink_to(state / "releases")
|
|
with pytest.raises(OSError):
|
|
custody.fingerprint("alias/fixture.json")
|
|
(state / "releases/link").symlink_to(state / "releases/fixture.json")
|
|
with pytest.raises(OSError):
|
|
custody.fingerprint("releases/link")
|
|
with pytest.raises(custody.Refused):
|
|
custody.fingerprint("../escape")
|
|
|
|
|
|
def test_hardlink_and_public_state_refuse(evidence):
|
|
_, _, state = evidence
|
|
os.link(state / "releases/fixture.json", state / "another")
|
|
with pytest.raises(custody.Refused):
|
|
custody.fingerprint("releases/fixture.json")
|
|
state.chmod(0o755)
|
|
with pytest.raises(custody.Refused):
|
|
custody.fingerprint("releases")
|
|
|
|
|
|
def test_root_ownership_change_uses_fd_and_retains_regular_bytes(evidence, monkeypatch):
|
|
plan, _, state = evidence
|
|
before = plan["targets"][1]
|
|
after = dict(before, uid=0, gid=0)
|
|
checks = iter([before, after])
|
|
calls = []
|
|
monkeypatch.setattr(custody, "fingerprint", lambda _: next(checks))
|
|
|
|
def chown(fd, uid, gid):
|
|
assert os.fstat(fd).st_ino == before["inode"]
|
|
calls.append((uid, gid))
|
|
|
|
monkeypatch.setattr(custody.os, "fchown", chown)
|
|
custody.change(before, before, after)
|
|
assert calls == [(0, 0)]
|
|
assert (state / "releases/fixture.json").read_bytes() == b'{"retained":"synthetic evidence"}\n'
|
|
|
|
|
|
def test_nonterminal_launcher_refuses_before_ssh(monkeypatch):
|
|
spec = importlib.util.spec_from_file_location(
|
|
"seal_launcher", ROOT / "scripts/seal-backup-evidence.py"
|
|
)
|
|
launcher = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(launcher)
|
|
monkeypatch.setattr(launcher, "prepare", lambda: ({"proposal_digest": "fixture"}, ""))
|
|
monkeypatch.setattr(launcher.sys, "stdin", SimpleNamespace(isatty=lambda: False))
|
|
monkeypatch.setattr(launcher.subprocess, "run", lambda *_args, **_kwargs: pytest.fail("no SSH"))
|
|
with pytest.raises(SystemExit, match="Founder terminal"):
|
|
launcher.main(["--execute", "--expect-digest", "fixture"])
|
|
|
|
|
|
def test_digest_or_path_scope_refuses_before_native_calls(monkeypatch):
|
|
fake_os = SimpleNamespace(**vars(os))
|
|
fake_os.geteuid = lambda: 0
|
|
monkeypatch.setattr(custody, "os", fake_os)
|
|
monkeypatch.setattr(custody, "native_fence", lambda _: pytest.fail("no native call"))
|
|
with pytest.raises(custody.Refused, match="proposal_digest_differs"):
|
|
custody.run({"proposal_digest": "sha256:" + "a" * 64}, "seal")
|
|
plan = {"targets": []}
|
|
plan["proposal_digest"] = "sha256:" + custody.sha(custody.canonical(plan))
|
|
with pytest.raises(custody.Refused, match="exact_eight_paths_required"):
|
|
custody.run(plan, "seal")
|
|
|
|
|
|
@pytest.mark.parametrize("fault", [None, "receipt", "pid", "helper", "mount", "cluster", "timer"])
|
|
def test_native_fence_requires_terminal_unchanged_receipt_absent_process_and_helper(
|
|
tmp_path, monkeypatch, fault
|
|
):
|
|
pid = os.getpid() if fault == "pid" else 999999999
|
|
raw = json.dumps({"pid": pid, "phase": "completed"}).encode()
|
|
path = tmp_path / "receipt.json"
|
|
path.write_bytes(raw)
|
|
path.chmod(0o600)
|
|
|
|
@contextmanager
|
|
def opened(_):
|
|
with path.open("rb") as stream:
|
|
yield stream.fileno()
|
|
|
|
monkeypatch.setattr(custody, "opened", opened)
|
|
fake_os = SimpleNamespace(**vars(os))
|
|
|
|
def stat_as_native(fd):
|
|
result = os.fstat(fd)
|
|
return SimpleNamespace(
|
|
st_mode=result.st_mode, st_nlink=result.st_nlink, st_uid=0, st_size=result.st_size
|
|
)
|
|
|
|
fake_os.fstat = stat_as_native
|
|
monkeypatch.setattr(custody, "os", fake_os)
|
|
pods = []
|
|
if fault in ("helper", "mount"):
|
|
pods = [
|
|
{
|
|
"metadata": {
|
|
"name": "csoc-date-" + custody.DATE[:20] if fault == "helper" else "other",
|
|
"uid": "fixture",
|
|
},
|
|
"spec": {
|
|
"volumes": [{"hostPath": {"path": str(custody.STATE)}}]
|
|
if fault == "mount"
|
|
else []
|
|
},
|
|
}
|
|
]
|
|
|
|
def command(args, **_):
|
|
if args[0] == "systemctl":
|
|
output = (
|
|
b"ActiveState=active\nUnitFileState=enabled"
|
|
if fault == "timer"
|
|
else (b"ActiveState=inactive\nUnitFileState=disabled")
|
|
)
|
|
return subprocess.CompletedProcess(args, 0, output, b"")
|
|
value = (
|
|
{"items": pods}
|
|
if "pods" in args
|
|
else {"metadata": {"uid": "wrong" if fault == "cluster" else custody.CLUSTER}}
|
|
)
|
|
return subprocess.CompletedProcess(args, 0, json.dumps(value).encode(), b"")
|
|
|
|
monkeypatch.setattr(custody.subprocess, "run", command)
|
|
plan = {
|
|
"receipts": [
|
|
{
|
|
"relative_path": "date-operations/fixture/receipt.json",
|
|
"sha256": "wrong" if fault == "receipt" else custody.sha(raw),
|
|
"pid": pid,
|
|
"terminal_status": "completed",
|
|
}
|
|
]
|
|
}
|
|
if fault:
|
|
with pytest.raises(custody.Refused):
|
|
custody.native_fence(plan)
|
|
else:
|
|
custody.native_fence(plan)
|
|
|
|
|
|
def test_all_four_real_operation_locks_are_held_and_released(tmp_path, monkeypatch):
|
|
state = tmp_path / "state"
|
|
state.mkdir(mode=0o700)
|
|
monkeypatch.setattr(custody, "STATE", state)
|
|
names = (
|
|
"date-operations/supervisor.lock",
|
|
"backup.lock",
|
|
"snapshot.lock",
|
|
custody.PRIVATE + "/execute.lock",
|
|
)
|
|
for name in names:
|
|
p = state / name
|
|
p.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
|
|
p.write_bytes(b"retained")
|
|
p.chmod(0o600)
|
|
before = [custody.fingerprint(name) for name in names]
|
|
with custody.locked():
|
|
for name in names:
|
|
with (state / name).open("rb") as stream:
|
|
with pytest.raises(BlockingIOError):
|
|
fcntl.flock(stream, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
for name in names:
|
|
with (state / name).open("rb") as stream:
|
|
fcntl.flock(stream, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
assert [custody.fingerprint(name) for name in names] == before
|