DataOps-platform/tests/test_phase3_wp07_enterprise_delivery_contract.py

"""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"