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, 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)