from __future__ import annotations import copy import importlib.util import json import sys import unittest from datetime import UTC, datetime from pathlib import Path ROOT = Path(__file__).resolve().parents[1] SCRIPTS = ROOT / "scripts" if str(SCRIPTS) not in sys.path: sys.path.insert(0, str(SCRIPTS)) import custody_contract as module NOW = datetime(2026, 8, 22, 22, 1, tzinfo=UTC) def projection_contract() -> dict: return { "interface": module.CONTRACT_INTERFACE, "version": 1, "workplan_id": module.WORKPLAN_ID, "engagement_id": "WH-ENG-20260822-AUDIT-E2-03", "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": "2" * 40, "image_digest": "sha256:" + "3" * 64, "contract_sha256": "4" * 64, }, "runner": { "namespace": "whitehat", "service_account": "whitehat-runner", "secret_name": "whitehat-e2-audit-credentials", "mount_root": "/var/run/secrets/whitehat", "manifest_sha256": "5" * 64, }, "window": { "starts_at": "2026-08-22T22:00:00Z", "projection_cutoff": "2026-08-22T22:03:00Z", "expires_at": "2026-08-22T22:15:00Z", }, "authority": { "remote": "railiance01", "registry_path": "platform/workloads/audit-core/senders", "registry_field": "senders.json", "kv_mount": "platform", "kv_prefix": "engagements/WH-ENG-20260822-AUDIT-E2-03/audit-core", "eso_service_account": "external-secrets", "eso_namespace": "external-secrets", }, "identities": [ { "handle": "token-a", "role": "attacker", "sender_name": "whitehat-e2-a-20260822-03", "tenant": "tenant:trial:whitehat-a-20260822-03", "mount_path": "/var/run/secrets/whitehat/token-a", "may_read": True, "may_write": True, }, { "handle": "token-b", "role": "owner", "sender_name": "whitehat-e2-b-20260822-03", "tenant": "tenant:trial:whitehat-b-20260822-03", "mount_path": "/var/run/secrets/whitehat/token-b", "may_read": True, "may_write": True, }, ], } def broker_receipt(contract: dict) -> dict: return { "interface": module.BROKER_INTERFACE, "version": 1, "workplan_id": module.WORKPLAN_ID, "owner": module.BROKER_OWNER, "reviewer": "whitehat-owner", "decision": "approve", "created_at": "2026-08-22T21:50:00Z", "engagement_id": contract["engagement_id"], "target_id": contract["target"]["id"], "projection_contract_digest": module.contract_digest(contract), "projection_receipt_interface": module.PROJECTION_INTERFACE, "interface_artifacts": module.interface_artifacts(), "required_roles": ["attacker", "owner"], "mount_paths": sorted(item["mount_path"] for item in contract["identities"]), "adapter": { "repo": "whitehat-security", "revision": "6" * 40, "path": "src/whitehat_security/platform_custody.py", "sha256": "7" * 64, "tests_passed": True, }, "cleanup_request_supported": True, "secret_values_observed": False, } def projection_receipt(contract: dict) -> dict: base = { "interface": module.PROJECTION_INTERFACE, "version": 1, "workplan_id": module.WORKPLAN_ID, "state": "projected", "lease_id": "custody:" + "8" * 32, "engagement_id": contract["engagement_id"], "target": { "id": contract["target"]["id"], "revision": contract["target"]["revision"], "image_digest": contract["target"]["image_digest"], }, "projection_contract_digest": module.contract_digest(contract), "broker_receipt_digest": module.digest(broker_receipt(contract)), "projected_at": "2026-08-22T22:02:00Z", "expires_at": contract["window"]["expires_at"], "identities": sorted( ( { "handle": item["handle"], "role": item["role"], "sender_name": item["sender_name"], "mount_path": item["mount_path"], } for item in contract["identities"] ), key=lambda item: item["handle"], ), "resources": { "names": module.resource_names(contract), "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:" + module.digest(base)} class CustodyContractTests(unittest.TestCase): def test_contract_validates_and_derives_hash_only_resource_names(self) -> None: contract = projection_contract() self.assertIs(contract, module.validate_projection_contract(contract)) names = module.resource_names(contract) self.assertTrue(all(name.startswith("custody-") for key, name in names.items() if key != "external_secret")) self.assertNotIn("E2-03", json.dumps(names)) self.assertEqual(64, len(module.contract_digest(contract))) def test_contract_rejects_stale_prefix_duplicate_tenant_and_long_window(self) -> None: contract = projection_contract() contract["authority"]["kv_prefix"] = "engagements/WH-ENG-20260822-AUDIT-E2-02/audit-core" with self.assertRaises(module.ContractError): module.validate_projection_contract(contract) contract = projection_contract() contract["identities"][1]["tenant"] = contract["identities"][0]["tenant"] with self.assertRaises(module.ContractError): module.validate_projection_contract(contract) contract = projection_contract() contract["window"]["expires_at"] = "2026-08-22T22:16:00Z" with self.assertRaises(module.ContractError): module.validate_projection_contract(contract) def test_broker_receipt_is_bound_to_contract_and_current_artifacts(self) -> None: contract = projection_contract() receipt = broker_receipt(contract) self.assertIs(receipt, module.validate_broker_receipt(receipt, contract, now=NOW)) stale = copy.deepcopy(receipt) stale["projection_contract_digest"] = "0" * 64 with self.assertRaises(module.ContractError): module.validate_broker_receipt(stale, contract, now=NOW) stale = copy.deepcopy(receipt) stale["interface_artifacts"][next(iter(stale["interface_artifacts"]))] = "0" * 64 with self.assertRaises(module.ContractError): module.validate_broker_receipt(stale, contract, now=NOW) def test_broker_message_requires_whitehat_origin_and_exact_subject(self) -> None: contract = projection_contract() receipt = broker_receipt(contract) message = { "from_agent": module.BROKER_OWNER, "to_agent": "railiance-platform", "subject": module.broker_subject(receipt), "body": module.canonical_json(receipt), } self.assertEqual(receipt, module.parse_broker_message(message, contract, now=NOW)) message["from_agent"] = "coding-agent" self.assertIsNone(module.parse_broker_message(message, contract, now=NOW)) malformed = { "from_agent": module.BROKER_OWNER, "to_agent": "railiance-platform", "subject": module.BROKER_SUBJECT_PREFIX + "/malformed", "body": "{}", } self.assertIsNone(module.parse_broker_message(malformed, contract, now=NOW)) def test_projection_receipt_canonical_id_and_scope_are_enforced(self) -> None: contract = projection_contract() receipt = projection_receipt(contract) self.assertIs(receipt, module.validate_projection_receipt(receipt, contract)) changed = copy.deepcopy(receipt) changed["resources"]["uids"]["mounted_secret"] = "replacement" with self.assertRaises(module.ContractError): module.validate_projection_receipt(changed, contract) value_bearing = copy.deepcopy(receipt) value_bearing["token"] = "never-allowed" with self.assertRaises(module.ContractError): module.validate_projection_receipt(value_bearing, contract) def test_cleanup_receipt_is_bound_to_projection_lease(self) -> None: contract = projection_contract() projection = projection_receipt(contract) cleanup = { "interface": module.CLEANUP_INTERFACE, "version": 1, "workplan_id": module.WORKPLAN_ID, "state": "cleaned", "lease_id": projection["lease_id"], "engagement_id": contract["engagement_id"], "projection_receipt_id": projection["receipt_id"], "cleaned_at": "2026-08-22T22:14:00Z", "removed_resources": ["custody-resource"], "target_ready": True, "secret_values_observed": False, } self.assertIs(cleanup, module.validate_cleanup_receipt(cleanup, projection, contract)) cleanup["lease_id"] = "custody:" + "0" * 32 with self.assertRaises(module.ContractError): module.validate_cleanup_receipt(cleanup, projection, contract) def test_all_published_schemas_are_valid_json(self) -> None: for relative in module.INTERFACE_ARTIFACTS: value = json.loads((ROOT / relative).read_text(encoding="utf-8")) self.assertEqual("https://json-schema.org/draft/2020-12/schema", value["$schema"]) if __name__ == "__main__": unittest.main()