tegwick
d2e1377354
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1abf1-7e95-7141-87c9-ee78332bf790
156 lines
8.1 KiB
Python
156 lines
8.1 KiB
Python
"""Synthetic recovery export checks; no node, native backup or credential reads."""
|
|
|
|
import io
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
import zipfile
|
|
from contextlib import redirect_stdout
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools"))
|
|
import vergabe_pilot_export as export
|
|
import vergabe_pilot_export_remote as remote
|
|
from vergabe_pilot_recovery_store import Refused, digest
|
|
|
|
|
|
class ExportTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temporary = tempfile.TemporaryDirectory(prefix="vergabe-export-fixture-")
|
|
self.addCleanup(self.temporary.cleanup)
|
|
self.root = Path(self.temporary.name)
|
|
self.capture = self.root / "capture"
|
|
self.capture.mkdir(mode=0o700)
|
|
for name, content in {"database.dump": b"PGDMP synthetic data",
|
|
"stores.tar.gz": b"synthetic store data"}.items():
|
|
(self.capture / name).write_bytes(content)
|
|
(self.capture / name).chmod(0o600)
|
|
expected = {name: ((self.capture / name).stat().st_size, digest(self.capture / name))
|
|
for name in export.EXPECTED}
|
|
self.patch = patch.object(export, "EXPECTED", expected)
|
|
self.patch.start()
|
|
self.addCleanup(self.patch.stop)
|
|
receipt = {"run_id": export.RUN_ID, "status": "captured_and_resumed", "helm_revision": 4,
|
|
"database": "vergabe_demo_company", "app_resumed": True, "writers_quiesced": True,
|
|
"helper_cleanup_confirmed": True, "restart_fixture_verified": True}
|
|
for name, prefix in [("database.dump", "database"), ("stores.tar.gz", "stores")]:
|
|
receipt[prefix + "_bytes"], receipt[prefix + "_sha256"] = expected[name]
|
|
self.write("receipt.json", receipt)
|
|
self.write("manifest.private.json", {"image": export.IMAGE, "database": "vergabe_demo_company",
|
|
"private_fixture": "synthetic-private-sentinel"})
|
|
|
|
def write(self, name, value):
|
|
path = self.capture / name
|
|
path.write_text(json.dumps(value))
|
|
path.chmod(0o600)
|
|
|
|
def test_changed_capture_refuses_before_crypto(self):
|
|
(self.capture / "database.dump").write_bytes(b"different capture")
|
|
with patch.object(export.subprocess, "run") as crypto, self.assertRaises(Refused):
|
|
export.seal(self.capture, "age")
|
|
crypto.assert_not_called()
|
|
|
|
def test_public_or_linked_source_is_refused(self):
|
|
path = self.capture / "manifest.private.json"
|
|
path.chmod(0o644)
|
|
with self.assertRaises(Refused):
|
|
export.capture_files(self.capture)
|
|
path.unlink()
|
|
path.symlink_to(self.capture / "receipt.json")
|
|
with self.assertRaises(Refused):
|
|
export.capture_files(self.capture)
|
|
|
|
def test_handoff_never_overwrites_a_different_file(self):
|
|
home = self.root / "home"
|
|
home.mkdir(mode=0o700)
|
|
target = home / "packet.age"
|
|
target.write_bytes(b"already there")
|
|
target.chmod(0o600)
|
|
with self.assertRaises(Refused):
|
|
export.handoff(self.capture / "database.dump", home, target.name, os.geteuid(), os.getegid())
|
|
self.assertEqual(target.read_bytes(), b"already there")
|
|
|
|
@unittest.skipUnless(shutil.which("age") and shutil.which("age-keygen"), "requires age fixture tools")
|
|
def test_encrypt_decrypt_transfer_and_retry_preserve_the_whole_packet(self):
|
|
identity = self.root / "synthetic-identity"
|
|
generated = subprocess.run(["age-keygen"], capture_output=True, check=True)
|
|
identity.write_bytes(generated.stdout)
|
|
identity.chmod(0o600)
|
|
public = subprocess.run(["age-keygen", "-y", str(identity)], capture_output=True, check=True).stdout.decode().strip()
|
|
with patch.object(export, "RECIPIENT", public):
|
|
ciphertext, sealed = export.seal(self.capture, shutil.which("age"))
|
|
repeated, again = export.seal(self.capture, shutil.which("age"))
|
|
self.assertEqual(digest(repeated), digest(ciphertext))
|
|
self.assertEqual(again, sealed)
|
|
decrypted = subprocess.run(["age", "-d", "-i", str(identity), str(ciphertext)],
|
|
capture_output=True, check=True)
|
|
with zipfile.ZipFile(io.BytesIO(decrypted.stdout)) as packet:
|
|
self.assertEqual(set(packet.namelist()),
|
|
{"database.dump", "stores.tar.gz", "manifest.private.json", "receipt.json", "packet.json"})
|
|
metadata = json.loads(packet.read("packet.json"))
|
|
for name, row in metadata["files"].items():
|
|
self.assertEqual(packet.read(name), (self.capture / name).read_bytes())
|
|
self.assertEqual(row["sha256"], digest(self.capture / name))
|
|
home = self.root / "handoff"
|
|
home.mkdir(mode=0o700)
|
|
export.handoff(ciphertext, home, export.NAME + ".zip.age", os.geteuid(), os.getegid())
|
|
export.handoff(self.capture / "seal.json", home, export.NAME + ".transfer.json", os.geteuid(), os.getegid())
|
|
result = remote.verify(home)
|
|
self.assertEqual(result["status"], "encrypted_workstation_copy_verified")
|
|
self.assertFalse(result["provider_publication_verified"])
|
|
self.assertFalse(result["decryption_verified"])
|
|
self.assertFalse(result["restore_verified"])
|
|
self.assertNotIn("synthetic-private-sentinel", json.dumps(sealed))
|
|
(home / (export.NAME + ".zip.age")).write_bytes(b"age-encryption.org/v1\ntampered")
|
|
with self.assertRaises(Refused):
|
|
remote.verify(home)
|
|
|
|
def test_changed_source_pin_prevents_remote_execution(self):
|
|
with patch.dict(remote.PINS, {"vergabe_pilot_export": "0" * 64}), self.assertRaises(Refused):
|
|
remote.remote_command()
|
|
|
|
def test_fetch_publishes_only_a_verified_pair_and_reuses_it_on_retry(self):
|
|
handoff = self.root / "remote-home"
|
|
handoff.mkdir(mode=0o700)
|
|
cipher = handoff / (export.NAME + ".zip.age")
|
|
cipher.write_bytes(b"age-encryption.org/v1\nsynthetic-ciphertext")
|
|
cipher.chmod(0o600)
|
|
metadata = {"schema": "railiance.vergabe-pilot-sealed-export.v1", "run_id": export.RUN_ID,
|
|
"recipient": export.RECIPIENT, "ciphertext_sha256": digest(cipher),
|
|
"ciphertext_bytes": cipher.stat().st_size,
|
|
"packet": {"run_id": export.RUN_ID, "image": export.IMAGE,
|
|
"database": "vergabe_demo_company", "files": export.capture_files(self.capture)}}
|
|
transfer = handoff / (export.NAME + ".transfer.json")
|
|
transfer.write_text(json.dumps(metadata))
|
|
transfer.chmod(0o600)
|
|
def execute(args, **kwargs):
|
|
if args[0] == "scp":
|
|
source = args[2].split(":", 1)[1]
|
|
shutil.copy2(handoff / source, Path(args[3]))
|
|
return subprocess.CompletedProcess(args, 0)
|
|
with patch.object(remote, "remote_command", return_value=["fixture-ssh"]), \
|
|
patch.object(remote.Path, "home", return_value=self.root / "workstation"), \
|
|
patch.object(remote, "outside_git", return_value=True), \
|
|
patch.object(remote.sys.stdin, "isatty", return_value=True), \
|
|
patch.object(remote.sys, "argv", ["export", "--fetch"]), \
|
|
patch.object(remote.subprocess, "run", side_effect=execute) as command, \
|
|
redirect_stdout(io.StringIO()):
|
|
self.assertEqual(remote.main(), 0)
|
|
self.assertEqual(remote.main(), 0)
|
|
self.assertEqual(command.call_count, 3)
|
|
path = self.root / "workstation/.local/state/vergabe-pilot-recovery" / export.RUN_ID
|
|
self.assertTrue((path / "workstation-copy.json").exists())
|
|
self.assertEqual({p.suffix for p in path.iterdir()}, {".age", ".json"})
|
|
|
|
def test_nested_backup_destination_inside_a_repository_is_refused(self):
|
|
(self.root / ".git").mkdir()
|
|
self.assertFalse(remote.outside_git(self.root / "private/backup"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|