"""Security and delivery-chain contracts for P3-WP07.""" from __future__ import annotations import hashlib import importlib.util import json import os import re import subprocess import sys from datetime import UTC, datetime, timedelta from pathlib import Path import jsonschema import pytest from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey FAKE_PRIVATE_KEY = bytes.fromhex( "4f3b2775a4df921336c3882f071d9edce4a6e946b928345f7bf68b55d4eaa3d8" ) SECOND_PRIVATE_KEY = bytes.fromhex("1f" * 32) ROOT = Path(__file__).resolve().parents[1] CLI = ROOT / "deployment/enterprise/enterprise_delivery.py" CHART = ROOT / "deployment/helm/dataops-platform" VALUES = CHART / "values.enterprise.example.json" SCHEMA = CHART / "values.schema.json" SPEC = importlib.util.spec_from_file_location("wp07_delivery", CLI) assert SPEC and SPEC.loader DELIVERY = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(DELIVERY) def _sha(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def _run(*args: str, expect: int = 0) -> dict[str, object]: completed = subprocess.run( [sys.executable, str(CLI), *args], cwd=ROOT, capture_output=True, text=True ) assert completed.returncode == expect, completed.stderr return json.loads(completed.stdout) def _evidence(root: Path, artifact: bytes = b"candidate") -> dict[str, Path]: root.mkdir() artifact_path = root / "artifact.tar.gz" artifact_path.write_bytes(artifact) sbom = root / "sbom.json" sbom_value = { "bomFormat": "CycloneDX", "specVersion": "1.5", "serialNumber": "urn:uuid:11111111-1111-4111-8111-111111111111", "version": 1, "metadata": { "properties": [ {"name": "dataops:artifact-sha256", "value": _sha(artifact_path)} ] }, "components": [], } sbom.write_text(json.dumps(sbom_value)) now = datetime.now(UTC) scan = root / "scan.json" license_file = root / "license.json" private = Ed25519PrivateKey.from_private_bytes(FAKE_PRIVATE_KEY) payload = {"artifact_sha256": _sha(artifact_path), "sbom_sha256": _sha(sbom)} signed = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode() signature = root / "signature.json" signature.write_text( json.dumps( { "algorithm": "Ed25519", "artifact_sha256": payload["artifact_sha256"], "sbom_sha256": payload["sbom_sha256"], "provider": "fake_local_only", "key_id": "wp07-local-test", "issued_at": (now - timedelta(minutes=1)).isoformat(), "expires_at": (now + timedelta(hours=1)).isoformat(), "signature": private.sign(signed).hex(), } ) ) signature_sha = _sha(signature) scan.write_text( json.dumps( { "artifact_sha256": _sha(artifact_path), "sbom_sha256": _sha(sbom), "status": "clear", "provider": "fake_local_only", "key_id": "wp07-local-test", "issued_at": (now - timedelta(minutes=1)).isoformat(), "expires_at": (now + timedelta(hours=1)).isoformat(), "signature_sha256": signature_sha, } ) ) license_file.write_text( json.dumps( { "artifact_sha256": _sha(artifact_path), "sbom_sha256": _sha(sbom), "status": "clear", "provider": "fake_local_only", "key_id": "wp07-local-test", "issued_at": (now - timedelta(minutes=1)).isoformat(), "expires_at": (now + timedelta(hours=1)).isoformat(), "signature_sha256": signature_sha, } ) ) return { "artifact": artifact_path, "sbom": sbom, "scan": scan, "license": license_file, "signature": signature, } def _contract_manifest( path: Path, artifact: Path, sbom: Path, *, openapi: Path | None = None ) -> dict[str, object]: component_identity = hashlib.sha256(b"[]").hexdigest() return { "artifact_sha256": _sha(artifact), "sbom_sha256": _sha(sbom), "component_identity_sha256": component_identity, "openapi": str(openapi or ROOT / "docs/architecture/OPENAPI.yaml"), "config": str(SCHEMA), "chart": str(CHART / "Chart.yaml"), "values": str(VALUES), "migration_head": "20260811_495", "migration_chain": ["20260811_495"], } def _release_record(artifact: Path, sbom: Path) -> dict[str, str]: sbom_value = json.loads(sbom.read_text()) identity = [ { "bom-ref": component.get("bom-ref", ""), "name": component.get("name", ""), "version": component.get("version", ""), "purl": component.get("purl", ""), } for component in sbom_value["components"] ] identity.sort(key=lambda item: json.dumps(item, sort_keys=True, separators=(",", ":"))) return { "artifact_sha256": _sha(artifact), "sbom_sha256": _sha(sbom), "sbom_serial": sbom_value["serialNumber"], "component_identity_sha256": hashlib.sha256( json.dumps(identity, sort_keys=True, separators=(",", ":")).encode() ).hexdigest(), } def _contract_sha(manifest: dict[str, object]) -> str: openapi = Path(str(manifest["openapi"])) config = Path(str(manifest["config"])) chart = Path(str(manifest["chart"])) values = Path(str(manifest["values"])) value = { "openapi_sha256": _sha(openapi), "config_schema_sha256": _sha(config), "chart_sha256": _sha(chart), "values_sha256": _sha(values), "migration_head": manifest["migration_head"], "migration_chain": manifest["migration_chain"], "openapi_operations": sorted( re.findall(r"^ (/[^:\s]+):", openapi.read_text(), re.MULTILINE) ), "config_required": sorted(json.loads(config.read_text()).get("required", [])), } return hashlib.sha256( json.dumps(value, sort_keys=True, separators=(",", ":")).encode() ).hexdigest() def _registry_signer(private_key: bytes, trust_root: str) -> dict[str, str]: public = Ed25519PrivateKey.from_private_bytes(private_key).public_key().public_bytes( serialization.Encoding.Raw, serialization.PublicFormat.Raw ) return { "algorithm": "Ed25519", "trust_root": trust_root, "public_key": public.hex(), "public_key_sha256": hashlib.sha256(public).hexdigest(), } def _temporary_registry(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: value = json.loads((ROOT / "deployment/enterprise/trust_store.json").read_text()) value["providers"]["external"] = { "enabled": True, "approved_signers": { "key-a": _registry_signer(FAKE_PRIVATE_KEY, "test-external-root-a"), "key-b": _registry_signer(SECOND_PRIVATE_KEY, "test-external-root-b"), }, } value["providers"]["fake_local_only"]["approved_signers"]["key-b"] = _registry_signer( SECOND_PRIVATE_KEY, "test-local-root-b" ) path = tmp_path / "trust-store.json" path.write_text(json.dumps(value)) monkeypatch.setattr(DELIVERY, "TRUST_STORE", path) return path def _candidate( state: Path, evidence: dict[str, Path], environment: str ) -> dict[str, object]: return _run( "candidate", "--state-dir", str(state), "--environment", environment, "--version", "20260812.2", "--artifact", str(evidence["artifact"]), "--sbom", str(evidence["sbom"]), "--scan", str(evidence["scan"]), "--license", str(evidence["license"]), "--signature", str(evidence["signature"]), "--approval-ref", "CHG-1201", "--test-only", ) def test_chart_has_real_service_hooks_and_strict_network_policy(): schema = json.loads(SCHEMA.read_text()) jsonschema.validate(json.loads(VALUES.read_text()), schema) service = (CHART / "templates/service.yaml").read_text() controls = (CHART / "templates/controls.yaml").read_text() workloads = (CHART / "templates/workloads.yaml").read_text() assert ( "kind: Service" in service and "selector:" in service and "targetPort:" in service ) assert "helm.sh/hook: pre-install,pre-upgrade" in controls assert "dataops-hook-bootstrap" in controls assert 'helm.sh/hook-weight: "-40"' in controls assert 'helm.sh/hook-weight: "-39"' in controls assert 'helm.sh/hook-weight: "-30"' in controls assert 'helm.sh/hook-weight: "-20"' in controls assert 'helm.sh/hook-weight: "-10"' in controls assert "dataops-role-init" in controls and "dataops-db-migrate" in controls assert "kube-dns" in controls and "dataops-postgres" in controls assert "dataops-neo4j" in controls and "dataops-minio" in controls assert "dataops-hook-egress" in controls and "dataops-backend-ingress" in controls assert "dataops-ingress-controller" in controls assert "namespaceSelector:" in controls and "ports:" in controls assert "egress: [{}]" not in controls and "- {}" not in controls assert "depends_on" not in workloads assert "before-hook-creation" in controls runtime_service_account = workloads.split("---", 1)[0] assert "namespace: {{ .Values.namespace }}" in runtime_service_account def test_candidate_recomputes_evidence_and_rejects_tamper_expiry_and_symlink( tmp_path: Path, ): state = tmp_path / "state" evidence = _evidence(tmp_path / "evidence") candidate = _candidate(state, evidence, "development") assert candidate["status"] == "candidate_created" persisted = state / "candidates" / f"{candidate['candidate_digest']}.json" os.link(persisted, state / "linked-state.json") assert ( _run( "candidate", "--state-dir", str(state), "--environment", "development", "--version", "20260812.2", "--artifact", str(evidence["artifact"]), "--sbom", str(evidence["sbom"]), "--scan", str(evidence["scan"]), "--license", str(evidence["license"]), "--signature", str(evidence["signature"]), "--approval-ref", "CHG-1201", expect=2, )["status"] == "rejected" ) evidence["artifact"].write_bytes(b"tampered") rejected = _run( "candidate", "--state-dir", str(state), "--environment", "development", "--version", "20260812.2", "--artifact", str(evidence["artifact"]), "--sbom", str(evidence["sbom"]), "--scan", str(evidence["scan"]), "--license", str(evidence["license"]), "--signature", str(evidence["signature"]), "--approval-ref", "CHG-1201", expect=2, ) assert rejected["status"] == "rejected" expired = _evidence(tmp_path / "expired") scan_value = json.loads(expired["scan"].read_text()) scan_value["expires_at"] = "2000-01-01T00:00:00+00:00" expired["scan"].write_text(json.dumps(scan_value)) assert _run( "candidate", "--state-dir", str(state), "--environment", "test", "--version", "expired", "--artifact", str(expired["artifact"]), "--sbom", str(expired["sbom"]), "--scan", str(expired["scan"]), "--license", str(expired["license"]), "--signature", str(expired["signature"]), "--approval-ref", "CHG-expired", expect=2, )["status"] == "rejected" assert any(term in rejected["error"] for term in ("scan", "signature", "SBOM")) state_link = tmp_path / "state-link" state_link.symlink_to(state, target_is_directory=True) rejected = _run( "candidate", "--state-dir", str(state_link), "--environment", "development", "--version", "bad", "--artifact", str(evidence["artifact"]), "--sbom", str(evidence["sbom"]), "--scan", str(evidence["scan"]), "--license", str(evidence["license"]), "--signature", str(evidence["signature"]), "--approval-ref", "CHG-1202", expect=2, ) assert rejected["status"] == "rejected" unsafe = tmp_path / "unsafe-state" unsafe.mkdir(mode=0o700) outside = tmp_path / "outside" outside.mkdir(mode=0o700) (unsafe / "candidates").symlink_to(outside, target_is_directory=True) clean = _evidence(tmp_path / "clean") assert ( _run( "candidate", "--state-dir", str(unsafe), "--environment", "development", "--version", "unsafe", "--artifact", str(clean["artifact"]), "--sbom", str(clean["sbom"]), "--scan", str(clean["scan"]), "--license", str(clean["license"]), "--signature", str(clean["signature"]), "--approval-ref", "CHG-unsafe", expect=2, )["status"] == "rejected" ) assert not list(outside.iterdir()) promote_state = tmp_path / "promotion-state" promote_evidence = _evidence(tmp_path / "promotion-evidence") promote_candidate = _candidate(promote_state, promote_evidence, "development") promote_outside = tmp_path / "promotion-outside" promote_outside.mkdir(mode=0o700) (promote_state / "current").symlink_to(promote_outside, target_is_directory=True) assert _run( "promote", "--state-dir", str(promote_state), "--environment", "development", "--candidate-digest", promote_candidate["candidate_digest"], "--expected-base-digest", "none", expect=2, )["status"] == "rejected" assert not list(promote_outside.iterdir()) def test_candidate_rejects_self_signed_external_provider(tmp_path: Path): state = tmp_path / "state" evidence = _evidence(tmp_path / "external") signature = json.loads(evidence["signature"].read_text()) signature["provider"] = "external" signature["key_id"] = "caller-supplied-root" evidence["signature"].write_text(json.dumps(signature)) signature_sha = _sha(evidence["signature"]) for evidence_name in ("scan", "license"): value = json.loads(evidence[evidence_name].read_text()) value["provider"] = "external" value["key_id"] = "caller-supplied-root" value["signature_sha256"] = signature_sha evidence[evidence_name].write_text(json.dumps(value)) assert _run( "candidate", "--state-dir", str(state), "--environment", "production", "--version", "external", "--artifact", str(evidence["artifact"]), "--sbom", str(evidence["sbom"]), "--scan", str(evidence["scan"]), "--license", str(evidence["license"]), "--signature", str(evidence["signature"]), "--approval-ref", "CHG-external", expect=2, )["status"] == "rejected" def test_candidate_uses_configured_key_and_binds_scan_to_same_sbom(tmp_path: Path): """The evidence payload cannot introduce its own trust root or SBOM binding.""" state = tmp_path / "state" evidence = _evidence(tmp_path / "registry-key") signature = json.loads(evidence["signature"].read_text()) assert "public_key" not in signature signature_sha = _sha(evidence["signature"]) for evidence_name in ("scan", "license"): value = json.loads(evidence[evidence_name].read_text()) value["signature_sha256"] = signature_sha value["sbom_sha256"] = _sha(evidence["sbom"]) evidence[evidence_name].write_text(json.dumps(value)) assert _candidate(state, evidence, "development")["status"] == "candidate_created" payload_root = _evidence(tmp_path / "payload-root") signature = json.loads(payload_root["signature"].read_text()) signature["public_key"] = "00" * 32 payload_root["signature"].write_text(json.dumps(signature)) signature_sha = _sha(payload_root["signature"]) for evidence_name in ("scan", "license"): value = json.loads(payload_root[evidence_name].read_text()) value["signature_sha256"] = signature_sha payload_root[evidence_name].write_text(json.dumps(value)) assert _run( "candidate", "--state-dir", str(tmp_path / "payload-root-state"), "--environment", "test", "--version", "payload-root", "--artifact", str(payload_root["artifact"]), "--sbom", str(payload_root["sbom"]), "--scan", str(payload_root["scan"]), "--license", str(payload_root["license"]), "--signature", str(payload_root["signature"]), "--approval-ref", "CHG-root", "--test-only", expect=2, )["status"] == "rejected" mismatched = _evidence(tmp_path / "mismatched-sbom") scan = json.loads(mismatched["scan"].read_text()) scan["sbom_sha256"] = "0" * 64 mismatched["scan"].write_text(json.dumps(scan)) assert _run( "candidate", "--state-dir", str(tmp_path / "mismatch-state"), "--environment", "test", "--version", "mismatch", "--artifact", str(mismatched["artifact"]), "--sbom", str(mismatched["sbom"]), "--scan", str(mismatched["scan"]), "--license", str(mismatched["license"]), "--signature", str(mismatched["signature"]), "--approval-ref", "CHG-sbom", "--test-only", expect=2, )["status"] == "rejected" def test_evidence_reader_rejects_symbolic_parent_directory(tmp_path: Path): actual = tmp_path / "actual" actual.mkdir() evidence = actual / "evidence.json" evidence.write_text('{"safe": true}') linked = tmp_path / "linked" linked.symlink_to(actual, target_is_directory=True) with pytest.raises(DELIVERY.DeliveryError, match="symbolic"): DELIVERY.safe_bytes(linked / "evidence.json") def test_compatibility_requires_three_bound_contract_manifests(tmp_path: Path): artifact = tmp_path / "artifact.tar.gz" artifact.write_bytes(b"valid artifact") evidence = _evidence(tmp_path / "evidence", b"valid artifact") assert _run( "compatibility", "--output", str(tmp_path / "matrix.json"), "--base-version", "base", "--candidate-version", "candidate", "--base-artifact", str(artifact), "--base-sbom", str(evidence["sbom"]), "--candidate-artifact", str(artifact), "--candidate-sbom", str(evidence["sbom"]), "--rollback-artifact", str(artifact), "--rollback-sbom", str(evidence["sbom"]), expect=2, )["status"] == "rejected" def test_compatibility_rejects_caller_authored_unsigned_approval(tmp_path: Path): base_artifact = tmp_path / "base.tar.gz" candidate_artifact = tmp_path / "candidate.tar.gz" base_artifact.write_bytes(b"base") candidate_artifact.write_bytes(b"unrelated candidate") base_evidence = _evidence(tmp_path / "base-evidence", b"base") candidate_evidence = _evidence(tmp_path / "candidate-evidence", b"unrelated candidate") manifests: dict[str, Path] = {} sources = { "base": (base_artifact, base_evidence["sbom"]), "candidate": (candidate_artifact, candidate_evidence["sbom"]), "rollback": (base_artifact, base_evidence["sbom"]), } contract_values: dict[str, dict[str, object]] = {} for name, (artifact, sbom) in sources.items(): path = tmp_path / f"{name}-contract.json" value = _contract_manifest(path, artifact, sbom) path.write_text(json.dumps(value)) manifests[name] = path contract_values[name] = value approval = { "base": _release_record(base_artifact, base_evidence["sbom"]), "candidate": _release_record(candidate_artifact, candidate_evidence["sbom"]), "base_contract_sha256": _contract_sha(contract_values["base"]), "candidate_contract_sha256": _contract_sha(contract_values["candidate"]), "approval_ref": "CALLER-SELF-APPROVED", "provider": "external", "key_id": "caller-created", "signature": "00" * 64, } approval_path = tmp_path / "approval.json" approval_path.write_text(json.dumps(approval)) assert _run( "compatibility", "--output", str(tmp_path / "matrix.json"), "--base-version", "base", "--candidate-version", "candidate", "--base-artifact", str(base_artifact), "--base-sbom", str(base_evidence["sbom"]), "--candidate-artifact", str(candidate_artifact), "--candidate-sbom", str(candidate_evidence["sbom"]), "--rollback-artifact", str(base_artifact), "--rollback-sbom", str(base_evidence["sbom"]), "--base-contract", str(manifests["base"]), "--candidate-contract", str(manifests["candidate"]), "--rollback-contract", str(manifests["rollback"]), "--compatibility-approval", str(approval_path), expect=2, )["status"] == "rejected" def test_compatibility_rejects_expired_registry_signed_approval( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ): _temporary_registry(tmp_path, monkeypatch) base_artifact, candidate_artifact = tmp_path / "base.tar.gz", tmp_path / "candidate.tar.gz" base_artifact.write_bytes(b"base") candidate_artifact.write_bytes(b"candidate") base_evidence = _evidence(tmp_path / "base-evidence", b"base") candidate_evidence = _evidence(tmp_path / "candidate-evidence", b"candidate") manifests, values = {}, {} for name, artifact, sbom in ( ("base", base_artifact, base_evidence["sbom"]), ("candidate", candidate_artifact, candidate_evidence["sbom"]), ("rollback", base_artifact, base_evidence["sbom"]), ): path = tmp_path / f"{name}.json" value = _contract_manifest(path, artifact, sbom) path.write_text(json.dumps(value)) manifests[name], values[name] = path, value legacy_payload = { "base": _release_record(base_artifact, base_evidence["sbom"]), "candidate": _release_record(candidate_artifact, candidate_evidence["sbom"]), "base_contract_sha256": _contract_sha(values["base"]), "candidate_contract_sha256": _contract_sha(values["candidate"]), "approval_ref": "APPROVAL-EXPIRED", } approval = { **legacy_payload, "provider": "external", "key_id": "key-a", "algorithm": "Ed25519", "issued_at": "2000-01-01T00:00:00Z", "expires_at": "2000-01-01T00:05:00Z", "signature": Ed25519PrivateKey.from_private_bytes(FAKE_PRIVATE_KEY).sign( DELIVERY.canonical(legacy_payload) ).hex(), } approval_path = tmp_path / "approval.json" approval_path.write_text(json.dumps(approval)) args = DELIVERY.parser().parse_args([ "compatibility", "--output", str(tmp_path / "matrix.json"), "--base-version", "base", "--candidate-version", "candidate", "--base-artifact", str(base_artifact), "--base-sbom", str(base_evidence["sbom"]), "--candidate-artifact", str(candidate_artifact), "--candidate-sbom", str(candidate_evidence["sbom"]), "--rollback-artifact", str(base_artifact), "--rollback-sbom", str(base_evidence["sbom"]), "--base-contract", str(manifests["base"]), "--candidate-contract", str(manifests["candidate"]), "--rollback-contract", str(manifests["rollback"]), "--compatibility-approval", str(approval_path), ]) with pytest.raises(DELIVERY.DeliveryError, match="approval evidence expired"): DELIVERY.command_compatibility(args) now = datetime.now(UTC) approval["issued_at"] = now.isoformat(timespec="seconds").replace("+00:00", "Z") approval["expires_at"] = (now + timedelta(minutes=30)).isoformat( timespec="seconds" ).replace("+00:00", "Z") signed = { **legacy_payload, "issued_at": approval["issued_at"], "expires_at": approval["expires_at"], } approval["signature"] = Ed25519PrivateKey.from_private_bytes(FAKE_PRIVATE_KEY).sign( DELIVERY.canonical(signed) ).hex() approval_path.write_text(json.dumps(approval)) assert DELIVERY.command_compatibility(args)["status"] == "compatible" def test_hook_policy_lifecycle_contract_is_documented(): plan = (ROOT / "docs/phase3/P3_WP07_ENTERPRISE_DELIVERY_IMPLEMENTATION_PLAN.md").read_text() assert "hook NetworkPolicy" in plan assert "pre-install/pre-upgrade" in plan assert "before-hook-creation" in plan assert "role-init/db-migrate" in plan def test_evidence_path_replacement_between_lstat_and_open_is_rejected( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ): evidence = tmp_path / "evidence.json" evidence.write_text('{"old": true}') replacement = tmp_path / "replacement.json" replacement.write_text('{"new": true}') original_open = DELIVERY.os.open def replace_before_open(path: object, *args: object, **kwargs: object) -> int: if path == evidence.name and kwargs.get("dir_fd") is not None: replacement.replace(evidence) return original_open(path, *args, **kwargs) monkeypatch.setattr(DELIVERY.os, "open", replace_before_open) with pytest.raises(DELIVERY.DeliveryError, match="changed while reading"): DELIVERY.safe_bytes(evidence) def test_promotion_only_allows_adjacent_environment_and_cas_rollback(tmp_path: Path): state, evidence = tmp_path / "state", _evidence(tmp_path / "evidence") development = _candidate(state, evidence, "development") _run( "promote", "--state-dir", str(state), "--environment", "development", "--candidate-digest", development["candidate_digest"], "--expected-base-digest", "none", ) direct = _candidate(state, evidence, "production") assert ( _run( "promote", "--state-dir", str(state), "--environment", "production", "--candidate-digest", direct["candidate_digest"], "--expected-base-digest", "none", expect=2, )["status"] == "rejected" ) test = _candidate(state, evidence, "test") _run( "promote", "--state-dir", str(state), "--environment", "test", "--candidate-digest", test["candidate_digest"], "--expected-base-digest", development["candidate_digest"], ) mismatched = _candidate(state, _evidence(tmp_path / "different", b"different"), "staging") assert ( _run( "promote", "--state-dir", str(state), "--environment", "staging", "--candidate-digest", mismatched["candidate_digest"], "--expected-base-digest", test["candidate_digest"], expect=2, )["status"] == "rejected" ) staging = _candidate(state, evidence, "staging") _run( "promote", "--state-dir", str(state), "--environment", "staging", "--candidate-digest", staging["candidate_digest"], "--expected-base-digest", test["candidate_digest"], ) production = _candidate(state, evidence, "production") _run( "promote", "--state-dir", str(state), "--environment", "production", "--candidate-digest", production["candidate_digest"], "--expected-base-digest", staging["candidate_digest"], ) rolled = _run( "rollback", "--state-dir", str(state), "--environment", "production", "--target-digest", staging["candidate_digest"], "--approval-ref", "CHG-1203", ) assert rolled["status"] == "rollback_recorded" def test_backup_restore_recomputes_signed_checksums_and_requires_encryption_for_production( tmp_path: Path, ): source = tmp_path / "backup-source" source.mkdir() source.chmod(0o700) data = source / "postgres.sql" data.write_text("safe backup") manifest = _run( "backup-plan", "--backup-dir", str(source), "--release-artifact", str(data), "--config-artifact", str(data), "--key-reference", "kms://external/backup", "--encryption-provider", "fake", "--test-only", ) result = _run( "restore-plan", "--backup-manifest", manifest["manifest_path"], "--target", "recovery-1201", "--confirm-target", "recovery-1201", "--environment", "test", "--test-only", ) assert result["status"] == "restore_to_new_target_planned" data.write_text("tampered") assert ( _run( "restore-plan", "--backup-manifest", manifest["manifest_path"], "--target", "recovery-1202", "--confirm-target", "recovery-1202", "--environment", "test", "--test-only", expect=2, )["status"] == "rejected" ) assert ( _run( "backup-plan", "--backup-dir", str(tmp_path / "disabled"), "--release-artifact", str(data), "--config-artifact", str(data), "--key-reference", "kms://external/backup", "--encryption-provider", "disabled", "--environment", "production", expect=2, )["status"] == "rejected" ) def test_production_external_backup_fails_closed_without_approved_kms_and_signer( tmp_path: Path, ): source = tmp_path / "external-backup" source.mkdir(mode=0o700) data = source / "postgres.sql" data.write_text("backup") assert _run( "backup-plan", "--backup-dir", str(source), "--release-artifact", str(data), "--config-artifact", str(data), "--key-reference", "kms://enterprise/production-backup", "--encryption-provider", "external", "--environment", "production", expect=2, )["status"] == "rejected" def test_restore_rejects_reused_new_target(tmp_path: Path): source = tmp_path / "replay-backup" source.mkdir(mode=0o700) data = source / "postgres.sql" data.write_text("backup") manifest = _run( "backup-plan", "--backup-dir", str(source), "--release-artifact", str(data), "--config-artifact", str(data), "--key-reference", "kms://external/backup", "--encryption-provider", "fake", "--test-only", ) args = ( "restore-plan", "--backup-manifest", manifest["manifest_path"], "--target", "recovery-once", "--confirm-target", "recovery-once", "--environment", "test", "--test-only", ) assert _run(*args)["status"] == "restore_to_new_target_planned" assert _run(*args, expect=2)["status"] == "rejected" def test_restore_rejects_manifest_supplied_public_key(tmp_path: Path): source = tmp_path / "self-signed-backup" source.mkdir(mode=0o700) data = source / "postgres.sql" data.write_text("backup") manifest = _run( "backup-plan", "--backup-dir", str(source), "--release-artifact", str(data), "--config-artifact", str(data), "--key-reference", "kms://external/backup", "--encryption-provider", "fake", "--test-only", ) manifest_path = Path(str(manifest["manifest_path"])) value = json.loads(manifest_path.read_text()) value["integrity_signature"]["public_key"] = "00" * 32 manifest_path.write_text(json.dumps(value)) assert _run( "restore-plan", "--backup-manifest", str(manifest_path), "--target", "self-signed-target", "--confirm-target", "self-signed-target", "--environment", "test", "--test-only", expect=2, )["status"] == "rejected" def test_restore_rejects_registry_key_b_when_manifest_names_key_a( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ): _temporary_registry(tmp_path, monkeypatch) source = tmp_path / "key-swap-backup" source.mkdir(mode=0o700) data = source / "postgres.sql" data.write_text("backup") backup_args = DELIVERY.parser().parse_args([ "backup-plan", "--backup-dir", str(source), "--release-artifact", str(data), "--config-artifact", str(data), "--key-reference", "kms://external/backup", "--encryption-provider", "fake", "--test-only", ]) manifest = DELIVERY.command_backup(backup_args) manifest_path = Path(str(manifest["manifest_path"])) value = json.loads(manifest_path.read_text()) payload = { "manifest_sha256": value["manifest_sha256"], "encryption_provider": value["encryption_provider"], "key_reference": value["key_reference"], } value["integrity_signature"]["key_id"] = "key-b" value["integrity_signature"]["signature"] = Ed25519PrivateKey.from_private_bytes( SECOND_PRIVATE_KEY ).sign(DELIVERY.canonical(payload)).hex() manifest_path.write_text(json.dumps(value)) restore_args = DELIVERY.parser().parse_args([ "restore-plan", "--backup-manifest", str(manifest_path), "--target", "key-swap-target", "--confirm-target", "key-swap-target", "--environment", "test", "--test-only", ]) with pytest.raises(DELIVERY.DeliveryError, match="signer identity mismatch"): DELIVERY.command_restore(restore_args) def test_compatibility_recomputes_live_release_inputs(tmp_path: Path): artifact = tmp_path / "release.tar.gz" artifact.write_bytes(b"release") evidence = _evidence(tmp_path / "compatibility-evidence", b"release") contracts = {} for name in ("base", "candidate", "rollback"): path = tmp_path / f"{name}-contract.json" path.write_text(json.dumps(_contract_manifest(path, artifact, evidence["sbom"]))) contracts[name] = path report = _run( "compatibility", "--output", str(tmp_path / "matrix.json"), "--base-version", "20260811", "--candidate-version", "20260812", "--base-artifact", str(artifact), "--base-sbom", str(evidence["sbom"]), "--candidate-artifact", str(artifact), "--candidate-sbom", str(evidence["sbom"]), "--rollback-artifact", str(artifact), "--rollback-sbom", str(evidence["sbom"]), "--base-contract", str(contracts["base"]), "--candidate-contract", str(contracts["candidate"]), "--rollback-contract", str(contracts["rollback"]), ) assert report["status"] == "compatible" matrix = json.loads((tmp_path / "matrix.json").read_text()) assert matrix["migration_head"] == "20260811_495" assert matrix["openapi_sha256"] == _sha(ROOT / "docs/architecture/OPENAPI.yaml") assert matrix["chart_sha256"] assert matrix["rollback"]["artifact_sha256"] == _sha(artifact) assert matrix["comparison"]["base_to_candidate"] == "verified" incompatible = tmp_path / "incompatible.tar.gz" incompatible.write_bytes(b"totally unrelated") incompatible_evidence = _evidence(tmp_path / "incompatible-evidence", b"totally unrelated") changed_contract = tmp_path / "changed-contract.json" changed_contract.write_text( json.dumps(_contract_manifest(changed_contract, incompatible, incompatible_evidence["sbom"])) ) assert _run( "compatibility", "--output", str(tmp_path / "rejected.json"), "--base-version", "20260811", "--candidate-version", "20260812", "--base-artifact", str(artifact), "--base-sbom", str(evidence["sbom"]), "--candidate-artifact", str(incompatible), "--candidate-sbom", str(incompatible_evidence["sbom"]), "--rollback-artifact", str(artifact), "--rollback-sbom", str(evidence["sbom"]), "--base-contract", str(contracts["base"]), "--candidate-contract", str(changed_contract), "--rollback-contract", str(contracts["rollback"]), expect=2, )["status"] == "rejected" base_openapi = tmp_path / "base-openapi.yaml" candidate_openapi = tmp_path / "candidate-openapi.yaml" base_openapi.write_text("openapi: 3.0.0\npaths:\n /kept:\n get: {}\n") candidate_openapi.write_text("openapi: 3.0.0\npaths:\n /new:\n get: {}\n") base_contract = _contract_manifest( tmp_path / "base-contract.json", artifact, evidence["sbom"], openapi=base_openapi ) for name, value in { "base": base_contract, "candidate": {**base_contract, "openapi": str(candidate_openapi)}, "rollback": base_contract, }.items(): (tmp_path / f"{name}-contract.json").write_text(json.dumps(value)) assert _run( "compatibility", "--output", str(tmp_path / "api-rejected.json"), "--base-version", "20260811", "--candidate-version", "20260812", "--base-artifact", str(artifact), "--base-sbom", str(evidence["sbom"]), "--candidate-artifact", str(artifact), "--candidate-sbom", str(evidence["sbom"]), "--rollback-artifact", str(artifact), "--rollback-sbom", str(evidence["sbom"]), "--base-contract", str(tmp_path / "base-contract.json"), "--candidate-contract", str(tmp_path / "candidate-contract.json"), "--rollback-contract", str(tmp_path / "rollback-contract.json"), expect=2, )["status"] == "rejected"