DataOps-platform/tests/security/test_trusted_delivery_controls.py

"""Security contracts for P3-WP06 trusted-delivery control profiles and gates."""

from __future__ import annotations

from datetime import UTC, datetime, timedelta
from pathlib import Path

import pytest

from app.core.system.trusted_delivery_controls import (
    ControlProfileCatalog,
    EvidenceLedger,
    SupplyChainReleaseGate,
    TrustedCapabilityRegistry,
    TrustedDeliveryControlError,
)


def _digest(seed: str) -> str:
    return (seed * 64)[:64]


def _future() -> str:
    return (datetime.now(UTC) + timedelta(minutes=10)).isoformat()


def test_control_profiles_are_versioned_evidence_mappings_not_compliance_claims():
    catalog = ControlProfileCatalog()

    profiles = [catalog.resolve(name, "1.0") for name in (
        "china_general", "finance", "medical", "government"
    )]

    assert [profile["profile_id"] for profile in profiles] == [
        "china_general", "finance", "medical", "government"
    ]
    assert all(profile["conformance_state"] == "ENGINEERING_EVIDENCE_ONLY" for profile in profiles)
    assert all(profile["external_required"] is True for profile in profiles)
    assert all(set(profile["external_dependencies"]) == {"security_legal", "identity", "network"} for profile in profiles)
    assert all(profile["evidence_mapping"] for profile in profiles)
    with pytest.raises(TrustedDeliveryControlError, match="unknown control profile"):
        catalog.resolve("medical", "2.0")


def test_capability_registry_defaults_disabled_and_only_issues_bound_signed_envelopes():
    registry = TrustedCapabilityRegistry()
    with pytest.raises(TrustedDeliveryControlError, match="disabled"):
        registry.issue_envelope(
            "kms", "kms-key-wrap", {"request_digest": _digest("a"), "expires_at": _future()}
        )

    fake = TrustedCapabilityRegistry.for_tests({
        "kms": {"version": "v1", "config_digest": _digest("b")},
    })
    envelope = fake.issue_envelope(
        "kms", "kms-key-wrap", {"request_digest": _digest("a"), "expires_at": _future()}
    )

    assert set(envelope) == {"capability", "version", "config_digest", "request_digest", "expires_at", "receipt", "signature"}
    assert fake.verify_envelope(envelope) is True
    assert "credential" not in str(envelope).lower()
    with pytest.raises(TrustedDeliveryControlError, match="replay"):
        fake.issue_envelope(
            "kms", "kms-key-wrap", {"request_digest": _digest("a"), "expires_at": _future()}
        )
    with pytest.raises(TrustedDeliveryControlError, match="unsupported"):
        fake.issue_envelope(
            "kms", "kms-key-wrap", {"request_digest": _digest("c"), "expires_at": _future(), "url": "https://evil.example"}
        )


def test_capability_envelope_rejects_expired_wrong_bound_or_tampered_receipt():
    fake = TrustedCapabilityRegistry.for_tests({
        "dlp": {"version": "v2", "config_digest": _digest("d")},
    })
    with pytest.raises(TrustedDeliveryControlError, match="expired"):
        fake.issue_envelope(
            "dlp", "dlp-export-check", {
                "request_digest": _digest("e"),
                "expires_at": (datetime.now(UTC) - timedelta(seconds=1)).isoformat(),
            },
        )
    envelope = fake.issue_envelope(
        "dlp", "dlp-export-check", {"request_digest": _digest("f"), "expires_at": _future()}
    )
    envelope["request_digest"] = _digest("0")
    with pytest.raises(TrustedDeliveryControlError, match="signature"):
        fake.verify_envelope(envelope)


def test_capability_registry_rejects_credentials_raw_config_and_unknown_scan_payloads():
    assert TrustedCapabilityRegistry().dependency_status() == {
        "dlp": "TBD_EXTERNAL", "iam": "TBD_EXTERNAL", "kms": "TBD_EXTERNAL",
        "siem": "TBD_EXTERNAL", "target": "TBD_EXTERNAL",
    }
    with pytest.raises(TrustedDeliveryControlError, match="unsupported"):
        TrustedCapabilityRegistry.for_tests({
            "siem": {"version": "v1", "config_digest": _digest("a"), "url": "https://evil.example"},
        })
    with pytest.raises(TrustedDeliveryControlError, match="unsafe"):
        TrustedCapabilityRegistry.for_tests({
            "siem": {"version": "secret-v1", "config_digest": _digest("a")},
        })
    with pytest.raises(TrustedDeliveryControlError, match="unsupported"):
        SupplyChainReleaseGate().evaluate({
            "artifact_digest": _digest("a"), "sbom_digest": _digest("b"),
            "license_policy_digest": _digest("c"), "vulnerability_scan_digest": _digest("d"),
            "approval_id": "release-approval-1", "approval_version": "v1",
            "scan_decision": "passed", "evidence_expires_at": _future(),
            "raw_scan": "password=must-not-persist",
        })


def test_control_contract_deployment_copy_is_exact():
    root = Path(__file__).resolve().parents[2]
    source = root / "app/core/system/trusted_delivery_controls.py"
    deployed = root / "deployment/app/core/system/trusted_delivery_controls.py"
    assert deployed.read_bytes() == source.read_bytes()


def test_supply_chain_gate_fails_closed_and_returns_digest_only_approval_summary():
    gate = SupplyChainReleaseGate(now_factory=lambda: datetime(2026, 8, 11, tzinfo=UTC))
    with pytest.raises(TrustedDeliveryControlError, match="missing"):
        gate.evaluate({"artifact_digest": _digest("a")})
    with pytest.raises(TrustedDeliveryControlError, match="rejected"):
        gate.evaluate({
            "artifact_digest": _digest("a"), "sbom_digest": _digest("b"),
            "license_policy_digest": _digest("c"), "vulnerability_scan_digest": _digest("d"),
            "approval_id": "release-approval-1", "approval_version": "v1",
            "scan_decision": "failed", "evidence_expires_at": "2026-08-12T00:00:00+00:00",
        })
    summary = gate.evaluate({
        "artifact_digest": _digest("a"), "sbom_digest": _digest("b"),
        "license_policy_digest": _digest("c"), "vulnerability_scan_digest": _digest("d"),
        "approval_id": "release-approval-1", "approval_version": "v1",
        "scan_decision": "passed", "evidence_expires_at": "2026-08-12T00:00:00+00:00",
    })
    assert summary["decision_code"] == "RELEASE_GATE_APPROVED"
    assert set(summary) == {"decision_code", "artifact_digest", "sbom_digest", "license_policy_digest", "vulnerability_scan_digest", "approval_id", "approval_version", "evidence_expires_at"}
    assert "scan" not in str(summary).lower() or "digest" in str(summary).lower()


def test_evidence_ledger_exports_only_immutable_references_and_blocks_hold_or_single_party_destruction():
    ledger = EvidenceLedger()
    ledger.append({
        "evidence_ref": "trusted-evidence:delivery-001", "evidence_digest": _digest("a"),
        "decision_code": "DELIVERY_ALLOWED",
    })
    assert ledger.export_summary() == [{
        "evidence_ref": "trusted-evidence:delivery-001", "evidence_digest": _digest("a"),
        "decision_code": "DELIVERY_ALLOWED",
    }]
    with pytest.raises(TrustedDeliveryControlError, match="active legal hold"):
        ledger.approve_destruction({"approval_refs": ["case-1", "case-2"], "active_hold": True})
    with pytest.raises(TrustedDeliveryControlError, match="two independent"):
        ledger.approve_destruction({"approval_refs": ["case-1", "case-1"], "active_hold": False})
    result = ledger.approve_destruction({"approval_refs": ["case-1", "case-2"], "active_hold": False})
    assert result["decision_code"] == "DESTRUCTION_DUAL_APPROVED"
    with pytest.raises(TrustedDeliveryControlError, match="immutable"):
        ledger.tamper({"evidence_ref": "trusted-evidence:delivery-001", "replacement": "raw secret"})
    assert ledger.audit_events()[-1]["decision_code"] == "TAMPER_REJECTED"