tegwick
a2cb673e0a
Some checks failed
Use-case regression gate / usecase-regressions (push) Failing after 6m5s
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a11538-05ea-7b63-b3c7-a2df5f38898d
113 lines
4.6 KiB
Python
113 lines
4.6 KiB
Python
"""Only an authenticated pair may leave the attended decryption boundary."""
|
|
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import subprocess
|
|
import tarfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture
|
|
def recovery():
|
|
spec = importlib.util.spec_from_file_location(
|
|
"recover_pair", Path(__file__).resolve().parents[1] / "scripts/recover-paired-backup.py"
|
|
)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def packet(path, bad=None):
|
|
contents = {"database.dump": b"PGDMPsynthetic", "content.tar.gz": b"synthetic assets"}
|
|
receipt = {
|
|
"status": "captured",
|
|
"writers_quiesced": True,
|
|
"app_resumed": True,
|
|
"database": "coulomb_social_db",
|
|
"started_at": "2026-10-07T20:00:00Z",
|
|
"image": "synthetic@sha256:fixture",
|
|
}
|
|
for name, key in [("database.dump", "database"), ("content.tar.gz", "content")]:
|
|
receipt[key + "_bytes"] = len(contents[name])
|
|
receipt[key + "_sha256"] = hashlib.sha256(contents[name]).hexdigest()
|
|
contents["receipt.json"] = json.dumps(receipt).encode()
|
|
with tarfile.open(path, "w") as bundle:
|
|
for name, data in contents.items():
|
|
entry = tarfile.TarInfo(name)
|
|
entry.size = len(data)
|
|
if bad == "link" and name == "content.tar.gz":
|
|
entry.type, entry.linkname, entry.size = tarfile.SYMTYPE, "/tmp/private", 0
|
|
bundle.addfile(entry, io.BytesIO(data))
|
|
if bad in {"duplicate", "traversal", "foreign"}:
|
|
entry = tarfile.TarInfo(
|
|
{"duplicate": "database.dump", "traversal": "../escaped", "foreign": "extra.env"}[
|
|
bad
|
|
]
|
|
)
|
|
entry.size = 1
|
|
bundle.addfile(entry, io.BytesIO(b"x"))
|
|
path.chmod(0o600)
|
|
return receipt
|
|
|
|
|
|
@pytest.mark.parametrize("bad", ["duplicate", "traversal", "foreign", "link"])
|
|
def test_rejects_entire_unsafe_packet_before_extraction(recovery, tmp_path, bad):
|
|
source = tmp_path / "packet.tar"
|
|
packet(source, bad)
|
|
destination = tmp_path / "out"
|
|
destination.mkdir(mode=0o700)
|
|
with pytest.raises(recovery.Refused, match="invalid_packet_members"):
|
|
recovery.unpack_packet(source, destination)
|
|
assert list(destination.iterdir()) == []
|
|
|
|
|
|
def test_real_age_roundtrip_checks_both_stores_and_capture_binding(recovery, tmp_path):
|
|
tmp_path.chmod(0o700)
|
|
source = tmp_path / "packet.tar"
|
|
inner = packet(source)
|
|
identity = tmp_path / "synthetic-identity"
|
|
generated = subprocess.run(["age-keygen", "-o", str(identity)], capture_output=True)
|
|
assert generated.returncode == 0
|
|
public = subprocess.run(["age-keygen", "-y", str(identity)], capture_output=True, check=True)
|
|
encrypted = tmp_path / "packet.tar.age"
|
|
subprocess.run(
|
|
["age", "-r", public.stdout.decode().strip(), "-o", str(encrypted), str(source)],
|
|
capture_output=True,
|
|
check=True,
|
|
)
|
|
encrypted.chmod(0o600)
|
|
transfer = {
|
|
"status": "offsite_fetched_verified",
|
|
"application": "coulomb-social",
|
|
"remote_retrieval_verified": True,
|
|
"version_id": "synthetic-version",
|
|
"destination": recovery.PREFIX + "synthetic.tar.age",
|
|
"captured_at": inner["started_at"],
|
|
"image": inner["image"],
|
|
"ciphertext_sha256": recovery.digest(encrypted),
|
|
"ciphertext_bytes": encrypted.stat().st_size,
|
|
}
|
|
receipt_path = tmp_path / "transfer.json"
|
|
receipt_path.write_text(json.dumps(transfer))
|
|
receipt_path.chmod(0o600)
|
|
destination = tmp_path / "recovered"
|
|
result = recovery.recover(encrypted, receipt_path, identity, destination)
|
|
assert result["status"] == "decrypted_pair_verified"
|
|
assert not result["application_restore_verified"]
|
|
assert (destination / "database.dump").read_bytes() == b"PGDMPsynthetic"
|
|
assert (destination / "content.tar.gz").read_bytes() == b"synthetic assets"
|
|
assert destination.stat().st_mode & 0o777 == 0o700
|
|
assert all(path.stat().st_mode & 0o777 == 0o600 for path in destination.iterdir())
|
|
assert not (destination / "packet.tar").exists()
|
|
transfer["captured_at"] = "different capture"
|
|
receipt_path.write_text(json.dumps(transfer))
|
|
with pytest.raises(recovery.Refused, match="capture_binding_mismatch"):
|
|
recovery.recover(encrypted, receipt_path, identity, tmp_path / "wrong-capture")
|
|
encrypted.write_bytes(b"tampered")
|
|
with pytest.raises(recovery.Refused, match="verified_transfer_required"):
|
|
recovery.recover(encrypted, receipt_path, identity, tmp_path / "corrupt")
|
|
assert not (tmp_path / "corrupt").exists()
|