tegwick
1a35ee8113
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1abf1-7e95-7141-87c9-ee78332bf790
182 lines
10 KiB
Python
182 lines
10 KiB
Python
"""Native restore scope/cleanup/transport checks; synthetic metadata only."""
|
|
|
|
import copy
|
|
import io
|
|
import json
|
|
import os
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
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_restore as restore
|
|
import vergabe_pilot_restore_remote as remote
|
|
from vergabe_pilot_recovery_store import Refused, digest
|
|
|
|
|
|
class RestoreScopeTests(unittest.TestCase):
|
|
def test_restore_has_new_claims_loopback_database_and_no_delivery_credential(self):
|
|
policy, claims, pod = restore.resources("vergabe-restore-0123456789ab", "0123456789ab")
|
|
self.assertEqual(policy["spec"], {"podSelector": {}, "policyTypes": ["Ingress", "Egress"]})
|
|
self.assertEqual({c["metadata"]["name"] for c in claims}, set(restore.CLAIMS.values()))
|
|
spec = pod["spec"]
|
|
self.assertFalse(spec["automountServiceAccountToken"])
|
|
self.assertFalse(spec["enableServiceLinks"])
|
|
self.assertEqual(spec["securityContext"]["fsGroup"], 999)
|
|
self.assertIn("listen_addresses=127.0.0.1", spec["containers"][0]["args"])
|
|
self.assertEqual(spec["containers"][1]["imagePullPolicy"], "Never")
|
|
self.assertNotIn("secretRef", json.dumps(pod))
|
|
self.assertNotIn("vergabe-teilnahme-media", json.dumps(pod))
|
|
self.assertNotIn("vergabe-teilnahme-app-state", json.dumps(pod))
|
|
restore.validate_pod(pod, copy.deepcopy(pod))
|
|
|
|
def test_admission_cannot_inject_a_token_extra_container_or_production_claim(self):
|
|
_, _, expected = restore.resources("vergabe-restore-0123456789ab", "0123456789ab")
|
|
for mode in ("token", "sidecar", "claim", "credential", "command"):
|
|
actual = copy.deepcopy(expected)
|
|
if mode == "token":
|
|
actual["spec"]["automountServiceAccountToken"] = True
|
|
elif mode == "sidecar":
|
|
actual["spec"]["containers"].append({"name": "injected"})
|
|
elif mode == "claim":
|
|
actual["spec"]["volumes"][1]["persistentVolumeClaim"]["claimName"] = "vergabe-teilnahme-media"
|
|
elif mode == "credential":
|
|
actual["spec"]["containers"][1]["envFrom"] = [{"secretRef": {"name": "runtime"}}]
|
|
else:
|
|
actual["spec"]["containers"][1]["command"] = ["unexpected"]
|
|
with self.subTest(mode=mode), self.assertRaises(Refused):
|
|
restore.validate_pod(actual, expected)
|
|
|
|
def test_changed_namespace_uid_cannot_be_deleted(self):
|
|
journal = {"namespace": "vergabe-restore-0123456789ab", "nonce": "0123456789ab", "namespace_uid": "owned"}
|
|
pod = {"metadata": {"name": journal["namespace"], "uid": "replacement", "labels": {restore.LABEL: journal["nonce"]}}}
|
|
with patch.object(restore, "get", return_value={"metadata": {"uid": restore.CLUSTER_UID}}), \
|
|
patch.object(restore, "k", return_value=json.dumps({"items": [pod]}).encode()) as command, \
|
|
self.assertRaises(Refused):
|
|
restore.cleanup(journal)
|
|
self.assertEqual(command.call_count, 1)
|
|
self.assertEqual(command.call_args.args[0], "get")
|
|
|
|
def test_owned_namespace_delete_uses_uid_precondition(self):
|
|
journal = {"namespace": "vergabe-restore-0123456789ab", "nonce": "0123456789ab", "namespace_uid": "owned"}
|
|
namespace = {"metadata": {"name": journal["namespace"], "uid": "owned", "labels": {restore.LABEL: journal["nonce"]}}}
|
|
def request(*args, **kwargs):
|
|
return json.dumps({"items": [namespace]}).encode() if args[0] == "get" else b""
|
|
with patch.object(restore, "get", return_value={"metadata": {"uid": restore.CLUSTER_UID}}), \
|
|
patch.object(restore, "k", side_effect=request) as command:
|
|
self.assertTrue(restore.cleanup(journal))
|
|
delete = next(c for c in command.call_args_list if c.args[0] == "delete")
|
|
self.assertEqual(json.loads(delete.kwargs["payload"])["preconditions"], {"uid": "owned"})
|
|
|
|
def test_failure_is_journaled_and_cleanup_is_attempted(self):
|
|
with tempfile.TemporaryDirectory(prefix="rapps-restore-failure-") as temporary:
|
|
directory = Path(temporary)
|
|
with patch.dict(os.environ, {"SUDO_UID": str(os.geteuid())}), \
|
|
patch.object(restore, "input_point", return_value={}), \
|
|
patch.object(restore, "apply", side_effect=Refused("fixture_admission_refused")), \
|
|
patch.object(restore, "cleanup", return_value=True) as cleanup, \
|
|
patch.object(restore, "production", return_value={}), redirect_stdout(io.StringIO()):
|
|
code = restore.execute("0123456789ab", directory, directory, {})
|
|
self.assertEqual(code, 1)
|
|
cleanup.assert_called_once()
|
|
receipt = json.loads((directory / "journal.json").read_text())["receipt"]
|
|
self.assertEqual(receipt["status"], "failed")
|
|
self.assertFalse(receipt["restore_verified"])
|
|
self.assertEqual(receipt["failure_stage"], "namespace_creation")
|
|
self.assertTrue(receipt["namespace_and_volumes_removed"])
|
|
|
|
def test_altered_network_policy_refuses_before_data_access(self):
|
|
policy, _, _ = restore.resources("vergabe-restore-0123456789ab", "0123456789ab")
|
|
policy["spec"]["egress"] = [{}]
|
|
with patch.object(restore, "k", return_value=json.dumps({"items": [policy]}).encode()), \
|
|
self.assertRaises(Refused):
|
|
restore.isolation_policy("vergabe-restore-0123456789ab")
|
|
|
|
def test_runtime_diagnostics_never_expose_messages_or_unknown_reasons(self):
|
|
pod = {"status": {"phase": "Pending", "containerStatuses": [{"name": "app", "state": {"waiting": {
|
|
"reason": "synthetic-private-value", "message": "synthetic-private-value"}}}]}}
|
|
with patch.object(restore, "get", return_value=pod):
|
|
result = restore.runtime_status("vergabe-restore-0123456789ab")
|
|
self.assertNotIn("synthetic-private-value", json.dumps(result))
|
|
self.assertEqual(result["app_waiting"], "other")
|
|
|
|
def test_replaced_clone_pod_is_not_deleted_for_rollback(self):
|
|
journal = {"namespace": "vergabe-restore-0123456789ab", "nonce": "0123456789ab",
|
|
"namespace_uid": "namespace-owned", "pod_uid": "pod-owned"}
|
|
def read(kind, *args):
|
|
return {"metadata": {"uid": "namespace-owned" if kind == "namespace" else "other-pod",
|
|
"labels": {restore.LABEL: journal["nonce"]}}}
|
|
with patch.object(restore, "get", side_effect=read), patch.object(restore, "k") as command, \
|
|
self.assertRaises(Refused):
|
|
restore.replace_clone_pod(journal, {})
|
|
command.assert_not_called()
|
|
|
|
|
|
class RestoreTransportTests(unittest.TestCase):
|
|
def test_provider_cleanup_and_recovered_hashes_are_required_before_staging(self):
|
|
with tempfile.TemporaryDirectory(prefix="rapps-restore-input-") as temporary:
|
|
kit = Path(temporary)
|
|
recovery = kit / "provider-recovery-fixture"
|
|
point = recovery / "point"
|
|
recovery.mkdir(mode=0o700)
|
|
point.mkdir(mode=0o700)
|
|
files = {}
|
|
for name in restore.FILES[:4]:
|
|
path = point / name
|
|
path.write_bytes(b"synthetic " + name.encode())
|
|
path.chmod(0o600)
|
|
files[name] = {"bytes": path.stat().st_size, "sha256": digest(path)}
|
|
expected = {n: (files[n]["bytes"], files[n]["sha256"]) for n in restore.export.EXPECTED}
|
|
(kit / (restore.export.NAME + ".transfer.json")).write_text(json.dumps({"packet": {
|
|
"run_id": restore.export.RUN_ID, "image": restore.export.IMAGE, "files": files}}))
|
|
(recovery / "fetched.zip.age").write_bytes(b"synthetic encrypted")
|
|
(recovery / "recovery-point.zip").write_bytes(b"synthetic plain")
|
|
body = {"status": "published_read_back_and_decrypted", "run_id": restore.export.RUN_ID,
|
|
"warden_exit_code": 0, "ciphertext_sha256": digest(recovery / "fetched.zip.age"),
|
|
"plaintext_packet_sha256": digest(recovery / "recovery-point.zip"),
|
|
"provider_publication_verified": True, "independent_readback_verified": True,
|
|
"decryption_verified": True, "session_cleanup_confirmed": True, "recovery_directory": str(recovery)}
|
|
receipt = kit / "provider.json"
|
|
receipt.write_text(json.dumps(body))
|
|
receipt.chmod(0o600)
|
|
with patch.object(remote.transfer, "outside_git", return_value=True), \
|
|
patch.object(restore.export, "EXPECTED", expected), \
|
|
patch.object(restore, "CIPHER", body["ciphertext_sha256"]), \
|
|
patch.object(restore, "PLAIN", body["plaintext_packet_sha256"]):
|
|
self.assertEqual(remote.verify_input(receipt), (kit, point, files))
|
|
body["session_cleanup_confirmed"] = False
|
|
receipt.write_text(json.dumps(body))
|
|
with self.assertRaises(Refused):
|
|
remote.verify_input(receipt)
|
|
body["session_cleanup_confirmed"] = True
|
|
receipt.write_text(json.dumps(body))
|
|
(point / "database.dump").write_bytes(b"altered")
|
|
with self.assertRaises(Refused):
|
|
remote.verify_input(receipt)
|
|
|
|
def test_embedded_runner_compiles_without_node_checkout(self):
|
|
command = remote.remote_command("0123456789ab")
|
|
operation = shlex.split(command[-1])
|
|
result = subprocess.run([sys.executable, "-c", operation[5], "--help"], capture_output=True, timeout=10, cwd="/tmp", check=False)
|
|
self.assertEqual(result.returncode, 0)
|
|
self.assertIn(b"--cleanup", result.stdout)
|
|
self.assertEqual(command[:3], ["ssh", "-t", "railiance01"])
|
|
self.assertLess(len(command[-1].encode()), 120000)
|
|
|
|
def test_changed_source_refuses_before_ssh(self):
|
|
with patch.dict(remote.PINS, {"vergabe_pilot_restore": "0" * 64}), self.assertRaises(Refused):
|
|
remote.remote_command("0123456789ab")
|
|
|
|
def test_nonce_cannot_introduce_another_namespace_or_shell_argument(self):
|
|
for value in ("../production", "0123456789ab;other", "", "vergabe-demo-company"):
|
|
with self.subTest(value=value), self.assertRaises(Refused):
|
|
remote.remote_command(value, cleanup=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|