from __future__ import annotations import base64 from contextlib import redirect_stderr, redirect_stdout from datetime import UTC, datetime, timedelta import hashlib import io import json from pathlib import Path import subprocess import sys import tempfile import unittest from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey META_ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(META_ROOT / "tools" / "deployment")) from govoplan_deploy.backup_evidence import verify_backup_evidence # noqa: E402 from govoplan_deploy.bundle import ( # noqa: E402 atomic_write, bundle_paths, canonical_json, read_env, ) from govoplan_deploy.cli import main as deploy_main # noqa: E402 from govoplan_deploy.distribution import ( # noqa: E402 DistributionError, canonical_signed_payload, canonical_json as canonical_distribution_json, ) from govoplan_deploy.model import default_spec, parse_spec # noqa: E402 from govoplan_deploy.planning import ( # noqa: E402 release_change_requires_backup, verify_stored_backup_evidence, ) class BackupEvidenceTests(unittest.TestCase): def setUp(self) -> None: self.now = datetime(2026, 8, 3, 12, tzinfo=UTC) self.private = Ed25519PrivateKey.generate() public = ( self.private.public_key() .public_bytes( serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo, ) .decode("ascii") ) self.keyring = { "schema_version": "1", "purpose": "govoplan-backup-evidence", "keys": [ { "key_id": "backup-controller-1", "algorithm": "ed25519", "status": "active", "public_key_pem": public, "not_before": (self.now - timedelta(days=1)).isoformat(), "expires_at": (self.now + timedelta(days=365)).isoformat(), } ], } self.release = { "channel": "stable", "version": "1.2.3", "manifest_sha256": "a" * 64, "composition_sha256": "b" * 64, "api_image": "registry.example/api@sha256:" + "c" * 64, "web_image": "registry.example/web@sha256:" + "d" * 64, } def test_verifies_coordinated_restore_drill_and_release_binding(self) -> None: summary = verify_backup_evidence( self._evidence(), self.keyring, installation_id="govoplan-test", profile="self-hosted", release=self.release, now=self.now, ) self.assertEqual("recovery-1", summary["recovery_point_id"]) self.assertEqual("restore-1", summary["restore_drill_id"]) self.assertEqual("backup-controller-1", summary["signature_key_id"]) def test_tampering_staleness_and_partial_restore_fail_closed(self) -> None: tampered = self._evidence() tampered["components"]["objects"]["object_count"] = 999 with self.assertRaisesRegex(DistributionError, "signature verification"): verify_backup_evidence( tampered, self.keyring, installation_id="govoplan-test", profile="self-hosted", release=self.release, now=self.now, ) stale = self._evidence(captured=self.now - timedelta(days=2)) with self.assertRaisesRegex(DistributionError, "stale"): verify_backup_evidence( stale, self.keyring, installation_id="govoplan-test", profile="self-hosted", release=self.release, now=self.now, ) partial = self._evidence() partial["restore_drill"]["objects_verified"] = False partial["signatures"] = [self._signature(partial)] with self.assertRaisesRegex(DistributionError, "objects_verified"): verify_backup_evidence( partial, self.keyring, installation_id="govoplan-test", profile="self-hosted", release=self.release, now=self.now, ) def test_wrong_release_key_purpose_and_component_skew_fail_closed(self) -> None: wrong_release = dict(self.release) wrong_release["version"] = "1.2.4" with self.assertRaisesRegex(DistributionError, "release field"): verify_backup_evidence( self._evidence(), self.keyring, installation_id="govoplan-test", profile="self-hosted", release=wrong_release, now=self.now, ) wrong_keyring = dict(self.keyring) wrong_keyring["purpose"] = "govoplan-runtime-distribution" with self.assertRaisesRegex(DistributionError, "wrong purpose"): verify_backup_evidence( self._evidence(), wrong_keyring, installation_id="govoplan-test", profile="self-hosted", release=self.release, now=self.now, ) skewed = self._evidence() skewed["components"]["database"]["captured_at"] = ( self.now - timedelta(hours=1) ).isoformat() skewed["signatures"] = [self._signature(skewed)] with self.assertRaisesRegex(DistributionError, "one recovery point"): verify_backup_evidence( skewed, self.keyring, installation_id="govoplan-test", profile="self-hosted", release=self.release, now=self.now, ) false_rto = self._evidence() false_rto["restore_drill"]["measured_rto_seconds"] = 1 false_rto["signatures"] = [self._signature(false_rto)] with self.assertRaisesRegex(DistributionError, "RTO"): verify_backup_evidence( false_rto, self.keyring, installation_id="govoplan-test", profile="self-hosted", release=self.release, now=self.now, ) def test_provider_signing_tool_emits_canonical_verified_evidence(self) -> None: self.now = datetime.now(UTC) self.keyring["keys"][0]["not_before"] = ( self.now - timedelta(days=1) ).isoformat() self.keyring["keys"][0]["expires_at"] = ( self.now + timedelta(days=365) ).isoformat() evidence = self._evidence() evidence["signatures"] = [] with tempfile.TemporaryDirectory(prefix="govoplan-backup-signer-") as value: root = Path(value) source = root / "unsigned.json" output = root / "signed.json" keyring = root / "keyring.json" private_key = root / "private.pem" atomic_write(source, canonical_json(evidence), mode=0o600) atomic_write(keyring, canonical_json(self.keyring), mode=0o600) atomic_write( private_key, self.private.private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption(), ), mode=0o600, ) result = subprocess.run( [ sys.executable, str(META_ROOT / "tools/deployment/sign-backup-evidence.py"), "--input", str(source), "--output", str(output), "--trusted-keyring", str(keyring), "--signing-key", f"backup-controller-1={private_key}", ], cwd=META_ROOT, check=False, capture_output=True, text=True, ) self.assertEqual(0, result.returncode, result.stderr) encoded = output.read_bytes() signed = json.loads(encoded) self.assertEqual(canonical_distribution_json(signed), encoded) summary = verify_backup_evidence( signed, self.keyring, installation_id="govoplan-test", profile="self-hosted", release=self.release, ) self.assertEqual("backup-controller-1", summary["signature_key_id"]) def test_cli_adoption_gates_the_next_release_against_previous_receipt(self) -> None: self.now = datetime.now(UTC) self.keyring["keys"][0]["not_before"] = ( self.now - timedelta(days=1) ).isoformat() self.keyring["keys"][0]["expires_at"] = ( self.now + timedelta(days=365) ).isoformat() evidence = self._evidence() encoded_evidence = canonical_distribution_json(evidence) encoded_keyring = canonical_distribution_json(self.keyring) with tempfile.TemporaryDirectory(prefix="govoplan-backup-evidence-") as value: paths = bundle_paths(Path(value)) paths.root.chmod(0o700) raw = default_spec( installation_id="govoplan-test", profile="self-hosted", public_url="https://govoplan.example.test", ingress_mode="existing-proxy", trusted_proxy_cidrs=("127.0.0.1/32",), ).to_dict() raw["release"] = { **raw["release"], **self.release, } current = parse_spec(raw) atomic_write(paths.spec, canonical_json(current.to_dict()), mode=0o600) source_evidence = paths.root / "source-backup.json" source_keyring = paths.root / "source-keyring.json" atomic_write(source_evidence, encoded_evidence, mode=0o600) atomic_write(source_keyring, encoded_keyring, mode=0o600) output = io.StringIO() with redirect_stdout(output), redirect_stderr(output): result = deploy_main( [ "verify-backup", "--directory", str(paths.root), "--evidence", str(source_evidence), "--evidence-sha256", hashlib.sha256(encoded_evidence).hexdigest(), "--trusted-keyring", str(source_keyring), "--adopt", ] ) self.assertEqual(0, result, output.getvalue()) runtime_environment = read_env(paths.env) self.assertEqual( "verified", runtime_environment["GOVOPLAN_BACKUP_EVIDENCE_STATE"], ) self.assertEqual( "recovery-1", runtime_environment["GOVOPLAN_BACKUP_RECOVERY_POINT_ID"], ) self.assertNotIn("snapshot:postgres", str(runtime_environment)) self.assertNotIn("urn:kms", str(runtime_environment)) receipt = { "installation_id": current.installation_id, "profile": current.profile, "release": dict(self.release), } atomic_write(paths.receipt, canonical_json(receipt), mode=0o600) target_raw = current.to_dict() target_raw["release"]["version"] = "1.2.4" target_raw["release"]["manifest_sha256"] = "9" * 64 target = parse_spec(target_raw) self.assertTrue(release_change_requires_backup(target, receipt)) summary = verify_stored_backup_evidence( target, paths, receipt=receipt, ) self.assertEqual("recovery-1", summary["recovery_point_id"]) def _evidence(self, *, captured: datetime | None = None) -> dict[str, object]: captured = captured or self.now - timedelta(hours=2) started = captured + timedelta(minutes=15) completed = captured + timedelta(minutes=30) issued = completed + timedelta(minutes=10) artifact_time = captured.isoformat() payload: dict[str, object] = { "schema_version": "1", "evidence_id": "backup-1", "installation_id": "govoplan-test", "deployment_subject": { "profile": "self-hosted", "topology": "compose", "subject_ref": "urn:govoplan:installation:govoplan-test", }, "release": dict(self.release), "recovery_point": { "id": "recovery-1", "captured_at": captured.isoformat(), "consistency": "application-quiesced", "rpo_seconds": 300, "write_fence": { "mode": "application-quiesce", "token_sha256": "e" * 64, "established_at": captured.isoformat(), }, }, "components": { "database": { "provider": "postgres", "artifact_ref": "snapshot:postgres:backup-1", "artifact_sha256": "1" * 64, "snapshot_id": "pg-snapshot-1", "lsn": "0/16B6C50", "protected": True, "encryption_key_ref": "urn:kms:key:database-backup", "captured_at": artifact_time, }, "objects": { "provider": "s3", "artifact_ref": "s3://backup/govoplan-test/recovery-1", "manifest_sha256": "2" * 64, "version_id": "object-snapshot-1", "object_count": 4, "total_bytes": 1024, "protected": True, "encryption_key_ref": "urn:kms:key:object-backup", "captured_at": artifact_time, }, "configuration": { "artifact_ref": "backup:configuration:recovery-1", "sha256": "3" * 64, "protected": True, "encryption_key_ref": "urn:kms:key:configuration-backup", "captured_at": artifact_time, }, "key_custody": { "provider": "kms", "keyset_ref": "urn:kms:keyset:govoplan-test", "keyset_version": "version-4", "recoverable": True, "captured_at": artifact_time, }, }, "restore_drill": { "drill_id": "restore-1", "recovery_point_id": "recovery-1", "started_at": started.isoformat(), "completed_at": completed.isoformat(), "isolated_target_ref": "urn:govoplan:restore-target:restore-1", "release_manifest_sha256": self.release["manifest_sha256"], "migration_heads_sha256": "4" * 64, "representative_object_manifest_sha256": "2" * 64, "database_verified": True, "objects_verified": True, "configuration_verified": True, "key_custody_verified": True, "semantic_checks": [ { "id": "institutional-journey", "status": "passed", "evidence_ref": "evidence:journey:institutional-1", } ], "measured_rpo_seconds": 120, "measured_rto_seconds": 900, "evidence_ref": "evidence:restore:restore-1", }, "issued_at": issued.isoformat(), "expires_at": (self.now + timedelta(days=7)).isoformat(), "revoked": False, "signatures": [], } payload["signatures"] = [self._signature(payload)] return payload def _signature(self, payload: dict[str, object]) -> dict[str, str]: return { "key_id": "backup-controller-1", "algorithm": "ed25519", "value": base64.b64encode( self.private.sign(canonical_signed_payload(payload)) ).decode("ascii"), } if __name__ == "__main__": unittest.main()