whitehat-security/tests/test_platform_custody_adapter.py

import copy
import json
import sys
from datetime import UTC, datetime
from pathlib import Path

import pytest

from whitehat_security.engagement import AuthorizationError, Engagement
from whitehat_security.platform_custody import (
    CLEANUP_INTERFACE,
    CONTRACT_INTERFACE,
    PROJECTION_INTERFACE,
    PlatformCustodyBroker,
    broker_from_receipt,
    contract_digest,
    digest,
    finalize_run_report,
    interface_artifacts,
    load_schemas,
    resource_names,
    validate_broker_readiness,
    validate_cleanup_receipt,
    validate_projection_contract,
    validate_projection_receipt,
)
from whitehat_security.plane import KillSwitch, admit
from whitehat_security.targets import load_registration


NOW = datetime(2026, 8, 22, 12, 2, tzinfo=UTC)


def contract() -> dict:
    return {
        "interface": CONTRACT_INTERFACE,
        "version": 1,
        "workplan_id": "RAILIANCE-WP-0025",
        "engagement_id": "WH-ENG-FIXTURE-1",
        "status": "approved",
        "engagement_contract_sha256": "1" * 64,
        "target": {
            "id": "audit-core",
            "namespace": "audit-core",
            "deployment": "audit-core",
            "container": "audit-core",
            "sender_external_secret": "audit-core-senders",
            "revision": "a" * 40,
            "image_digest": "sha256:" + "b" * 64,
            "contract_sha256": "c" * 64,
        },
        "runner": {
            "namespace": "whitehat",
            "service_account": "whitehat-runner",
            "secret_name": "whitehat-e2-audit-credentials",
            "mount_root": "/var/run/secrets/whitehat",
            "manifest_sha256": "d" * 64,
        },
        "window": {
            "starts_at": "2026-08-22T12:00:00Z",
            "projection_cutoff": "2026-08-22T12:03:00Z",
            "expires_at": "2026-08-22T12:15:00Z",
        },
        "authority": {
            "remote": "railiance01",
            "registry_path": "platform/workloads/audit-core/senders",
            "registry_field": "senders.json",
            "kv_mount": "platform",
            "kv_prefix": "engagements/WH-ENG-FIXTURE-1/audit-core",
            "eso_service_account": "external-secrets",
            "eso_namespace": "external-secrets",
        },
        "identities": [
            {
                "handle": "token-a",
                "role": "attacker",
                "sender_name": "whitehat-e2-a",
                "tenant": "tenant-a",
                "mount_path": "/var/run/secrets/whitehat/token-a",
                "may_read": True,
                "may_write": True,
            },
            {
                "handle": "token-b",
                "role": "owner",
                "sender_name": "whitehat-e2-b",
                "tenant": "tenant-b",
                "mount_path": "/var/run/secrets/whitehat/token-b",
                "may_read": True,
                "may_write": True,
            },
        ],
    }


def broker_readiness(bound: dict | None = None) -> dict:
    bound = bound or contract()
    return {
        "interface": "railiance.custody-broker-readiness",
        "version": 1,
        "workplan_id": "RAILIANCE-WP-0025",
        "owner": "whitehat-security",
        "reviewer": "whitehat-owner",
        "decision": "approve",
        "created_at": "2026-08-22T12:00:00Z",
        "engagement_id": bound["engagement_id"],
        "target_id": bound["target"]["id"],
        "projection_contract_digest": contract_digest(bound),
        "projection_receipt_interface": PROJECTION_INTERFACE,
        "interface_artifacts": interface_artifacts(),
        "required_roles": ["attacker", "owner"],
        "mount_paths": sorted(item["mount_path"] for item in bound["identities"]),
        "adapter": {
            "repo": "whitehat-security",
            "revision": "a" * 40,
            "path": "src/whitehat_security/platform_custody.py",
            "sha256": "b" * 64,
            "tests_passed": True,
        },
        "cleanup_request_supported": True,
        "secret_values_observed": False,
    }


def projection_receipt(bound: dict | None = None) -> dict:
    bound = bound or contract()
    base = {
        "interface": PROJECTION_INTERFACE,
        "version": 1,
        "workplan_id": "RAILIANCE-WP-0025",
        "state": "projected",
        "lease_id": "custody:" + "f" * 32,
        "engagement_id": bound["engagement_id"],
        "target": {
            "id": bound["target"]["id"],
            "revision": bound["target"]["revision"],
            "image_digest": bound["target"]["image_digest"],
        },
        "projection_contract_digest": contract_digest(bound),
        "broker_receipt_digest": digest(broker_readiness(bound)),
        "projected_at": "2026-08-22T12:02:00Z",
        "expires_at": bound["window"]["expires_at"],
        "identities": sorted(
            (
                {
                    "handle": item["handle"],
                    "role": item["role"],
                    "sender_name": item["sender_name"],
                    "mount_path": item["mount_path"],
                }
                for item in bound["identities"]
            ),
            key=lambda item: item["handle"],
        ),
        "resources": {
            "names": resource_names(bound),
            "uids": {
                "store": "uid-store",
                "external_secret": "uid-es",
                "mounted_secret": "uid-secret",
            },
        },
        "cleanup_authority": "railiance-platform",
        "secret_values_observed": False,
    }
    return {**base, "receipt_id": "sha256:" + digest(base)}


def cleanup_receipt(bound: dict | None = None, projection: dict | None = None) -> dict:
    bound = bound or contract()
    projection = projection or projection_receipt(bound)
    return {
        "interface": CLEANUP_INTERFACE,
        "version": 1,
        "workplan_id": "RAILIANCE-WP-0025",
        "state": "cleaned",
        "lease_id": projection["lease_id"],
        "engagement_id": bound["engagement_id"],
        "projection_receipt_id": projection["receipt_id"],
        "cleaned_at": "2026-08-22T12:14:00Z",
        "removed_resources": [resource_names(bound)["store"]],
        "target_ready": True,
        "secret_values_observed": False,
    }


def live_record():
    return {
        "engagement_id": "WH-ENG-FIXTURE-1", "authorization_id": "auth-1",
        "authorizer": "operator", "approved_at": "2026-08-22T11:00:00Z",
        "expires_at": "2026-08-22T12:15:00Z", "target": "https://fixture.invalid",
        "target_id": "audit-core", "target_owner": "audit-core",
        "environment": "build", "source": "runner", "approval_class": "live-e2",
        "plane_namespace": "whitehat", "runner_image_digest": "sha256:abc",
        "routes": ["POST /v1/events"], "fixture_ids": ["a", "b"],
        "credential_lane": "receipt", "credential_role": "runtime",
        "credential_max_ttl_seconds": 900, "techniques": ["e2-differential"],
        "prohibited_techniques": ["saturation"], "rate_limit_per_minute": 10,
        "max_concurrency": 1, "window_start": "2026-08-22T12:00:00Z",
        "window_end": "2026-08-22T12:15:00Z", "operator_contact": "operator",
        "abort_contact": "operator", "posture_claim": "E2",
        "attacker_model": "E2-authenticated-tenant-a",
        "finding_destination": "risk-nexus",
        "target_owner_acknowledged_at": "2026-08-22T11:01:00Z",
    }


def test_four_custody_schemas_are_published():
    schemas = load_schemas()
    assert set(schemas) == {
        "custody-projection-contract.schema.json",
        "custody-broker-readiness.schema.json",
        "custody-projection-receipt.schema.json",
        "custody-cleanup-receipt.schema.json",
    }
    assert "custody-projection-contract" in schemas["custody-projection-contract.schema.json"]["$id"]
    assert "custody-broker-readiness" in schemas["custody-broker-readiness.schema.json"]["$id"]
    assert "custody-projection-receipt" in schemas["custody-projection-receipt.schema.json"]["$id"]
    assert "custody-cleanup-receipt" in schemas["custody-cleanup-receipt.schema.json"]["$id"]


def test_adapter_validates_all_four_document_kinds():
    bound = contract()
    projection = projection_receipt(bound)
    assert validate_projection_contract(bound)["engagement_id"] == "WH-ENG-FIXTURE-1"
    assert validate_projection_receipt(projection, contract=bound)["state"] == "projected"
    assert validate_cleanup_receipt(
        cleanup_receipt(bound, projection), projection=projection, contract=bound
    )["state"] == "cleaned"
    assert validate_broker_readiness(
        broker_readiness(bound), contract=bound, now=NOW
    )["owner"] == "whitehat-security"


def test_adapter_rejects_secret_material():
    tainted = projection_receipt()
    tainted["token"] = "never"
    with pytest.raises(AuthorizationError, match="secret material"):
        validate_projection_receipt(tainted, contract=contract())


def test_receipt_without_contract_is_refused(tmp_path):
    path = tmp_path / "receipt.json"
    path.write_text(json.dumps(projection_receipt()), encoding="utf-8")
    with pytest.raises(AuthorizationError, match="requires a bound contract"):
        broker_from_receipt(path)


def test_receipt_without_broker_is_refused(tmp_path):
    bound = contract()
    receipt = tmp_path / "receipt.json"
    contract_path = tmp_path / "contract.json"
    receipt.write_text(json.dumps(projection_receipt(bound)), encoding="utf-8")
    contract_path.write_text(json.dumps(bound), encoding="utf-8")
    with pytest.raises(AuthorizationError, match="requires a bound broker receipt"):
        broker_from_receipt(receipt, contract_path=contract_path)


def test_noncanonical_receipt_id_is_refused():
    receipt = projection_receipt()
    receipt["receipt_id"] = "sha256:" + "e" * 64
    with pytest.raises(AuthorizationError, match="canonical content digest"):
        validate_projection_receipt(receipt, contract=contract())


def test_stale_contract_digest_is_refused():
    receipt = projection_receipt()
    receipt["projection_contract_digest"] = "1" * 64
    receipt["receipt_id"] = "sha256:" + digest(
        {key: value for key, value in receipt.items() if key != "receipt_id"}
    )
    with pytest.raises(AuthorizationError, match="contract digest"):
        validate_projection_receipt(receipt, contract=contract())


def test_incomplete_resource_uids_are_refused():
    bound = contract()
    receipt = projection_receipt(bound)
    receipt["resources"]["uids"].pop("store")
    receipt["receipt_id"] = "sha256:" + digest(
        {key: value for key, value in receipt.items() if key != "receipt_id"}
    )
    with pytest.raises(AuthorizationError, match="exact Kubernetes UIDs"):
        validate_projection_receipt(receipt, contract=bound)


def test_target_identity_resource_time_and_broker_digest_are_bound():
    bound = contract()
    receipt = projection_receipt(bound)
    mismatched = copy.deepcopy(receipt)
    mismatched["target"]["id"] = "other"
    mismatched["receipt_id"] = "sha256:" + digest(
        {key: value for key, value in mismatched.items() if key != "receipt_id"}
    )
    with pytest.raises(AuthorizationError, match="target"):
        validate_projection_receipt(mismatched, contract=bound)

    mismatched = copy.deepcopy(receipt)
    mismatched["identities"][0]["sender_name"] = "wrong"
    mismatched["receipt_id"] = "sha256:" + digest(
        {key: value for key, value in mismatched.items() if key != "receipt_id"}
    )
    with pytest.raises(AuthorizationError, match="identities"):
        validate_projection_receipt(mismatched, contract=bound)

    mismatched = copy.deepcopy(receipt)
    mismatched["resources"]["names"]["store"] = "custody-wrong"
    mismatched["receipt_id"] = "sha256:" + digest(
        {key: value for key, value in mismatched.items() if key != "receipt_id"}
    )
    with pytest.raises(AuthorizationError, match="resources"):
        validate_projection_receipt(mismatched, contract=bound)

    mismatched = copy.deepcopy(receipt)
    mismatched["expires_at"] = "2026-08-22T12:14:00Z"
    mismatched["receipt_id"] = "sha256:" + digest(
        {key: value for key, value in mismatched.items() if key != "receipt_id"}
    )
    with pytest.raises(AuthorizationError, match="projection/expiry"):
        validate_projection_receipt(mismatched, contract=bound)

    mismatched = copy.deepcopy(receipt)
    mismatched["broker_receipt_digest"] = "0" * 64
    mismatched["receipt_id"] = "sha256:" + digest(
        {key: value for key, value in mismatched.items() if key != "receipt_id"}
    )
    with pytest.raises(AuthorizationError, match="broker digest"):
        PlatformCustodyBroker(
            mismatched, contract=bound, broker=broker_readiness(bound), now=NOW
        )


def test_platform_validator_accepts_the_canonical_fixture():
    scripts = Path.home() / "railiance-platform" / "scripts"
    if str(scripts) not in sys.path:
        sys.path.insert(0, str(scripts))
    import custody_contract

    bound = contract()
    receipt = projection_receipt(bound)
    broker = broker_readiness(bound)
    assert receipt is custody_contract.validate_projection_receipt(receipt, bound)
    assert broker is custody_contract.validate_broker_receipt(broker, bound, now=NOW)


def test_platform_broker_issues_handles_and_supports_cleanup(tmp_path):
    path = tmp_path / "engagement.json"
    path.write_text(json.dumps(live_record()), encoding="utf-8")
    engagement = Engagement.load(path, now=NOW)
    bound = contract()
    receipt = projection_receipt(bound)
    broker = PlatformCustodyBroker(
        receipt, contract=bound, broker=broker_readiness(bound), now=NOW
    )
    assert broker.cleanup_request_supported is True
    lease = admit(
        engagement=engagement,
        registration=load_registration("targets/audit-core-e2.json"),
        broker=broker,
        kill_switch=KillSwitch(tmp_path / "KILL"),
        now=NOW,
        retired=set(),
    )
    assert {handle.role for handle in lease.identities} == {"attacker", "owner"}
    assert lease.lease_id.startswith("custody:")
    with pytest.raises(AuthorizationError, match="custody must revoke"):
        broker.revoke(lease.lease_id)
    cleaned = PlatformCustodyBroker(
        receipt,
        contract=bound,
        cleanup=cleanup_receipt(bound, receipt),
        broker=broker_readiness(bound),
        now=NOW,
    )
    cleaned.revoke(lease.lease_id)


def test_cleanup_receipt_must_match_lease():
    bound = contract()
    projection = projection_receipt(bound)
    with pytest.raises(AuthorizationError, match="lease_id"):
        bad = cleanup_receipt(bound, projection)
        bad["lease_id"] = "custody:" + "0" * 32
        validate_cleanup_receipt(bad, projection=projection, contract=bound)


def test_finalized_report_is_bound_to_cleanup_receipt():
    bound = contract()
    projection = projection_receipt(bound)
    cleanup = cleanup_receipt(bound, projection)
    report = {
        "engagement_id": bound["engagement_id"],
        "target": bound["target"]["id"],
        "target_revision": bound["target"]["image_digest"],
        "credential_revocation": "pending orchestrator cleanup",
        "cleanup": "pending",
    }
    finalized = finalize_run_report(
        report, projection=projection, cleanup=cleanup, contract=bound
    )
    assert finalized["credential_revocation"].startswith("revoked by railiance-platform")
    assert projection["receipt_id"] in finalized["cleanup"]


def test_finalized_report_refuses_wrong_engagement():
    bound = contract()
    projection = projection_receipt(bound)
    cleanup = cleanup_receipt(bound, projection)
    with pytest.raises(AuthorizationError, match="report engagement"):
        finalize_run_report(
            {
                "engagement_id": "WH-ENG-WRONG",
                "target": "audit-core",
                "target_revision": bound["target"]["image_digest"],
            },
            projection=projection,
            cleanup=cleanup,
            contract=bound,
        )