import json from datetime import UTC, datetime import pytest from whitehat_security.audit_fixtures import AuditFixture, audit_probe_suite from whitehat_security.differential import execute, execute_authorized from whitehat_security.engagement import AuthorizationError, Engagement from whitehat_security.fixtures import FixtureService, probe_suite from whitehat_security.model import RunReport from whitehat_security.e3 import e3_calibration from whitehat_security.plane import ( KillSwitch, LocalBroker, RateWatcher, UnconnectedCustodyBroker, admit, cleanup, retired_ids, ) from whitehat_security.reporting import queue_risk_nexus from whitehat_security.targets import load_catalog, load_registration NOW = datetime(2026, 8, 22, 12, tzinfo=UTC) def fixture_record(**overrides): data = { "engagement_id": "WH-ENG-FIXTURE-1", "authorization_id": "auth-1", "authorizer": "operator", "approved_at": "2026-08-22T11:00:00Z", "expires_at": "2026-08-22T18:00:00Z", "target": "in-process", "target_id": "fixture-e2", "target_owner": "whitehat-security", "environment": "fixture", "source": "in-process", "approval_class": "fixture-e2", "routes": ["GET /objects/{id}"], "fixture_ids": ["object-a", "object-b"], "credential_lane": "local-broker", "credential_role": "runtime", "credential_max_ttl_seconds": 900, "techniques": ["e2-differential"], "prohibited_techniques": ["saturation"], "rate_limit_per_minute": 10, "max_concurrency": 1, "maximum_requests": 8, "window_start": "2026-08-22T11:00:00Z", "window_end": "2026-08-22T18:00: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", } data.update(overrides) return data def load_engagement(tmp_path, data): path = tmp_path / "engagement.json" path.write_text(json.dumps(data), encoding="utf-8") return Engagement.load(path, now=NOW) def fixture_registration(): return load_registration("targets/fixture-e2.json") def test_catalog_loads_honest_applicability(): catalog = load_catalog("targets") assert catalog["fixture-e2"]["applicability"] == "applicable" assert catalog["audit-core"]["applicability"] == "applicable" assert catalog["tenant-engine"]["applicability"] == "not_applicable" assert catalog["flex-auth"]["applicability"] == "pending" assert catalog["fixture-e3"]["applicability"] == "applicable" assert catalog["fixture-capacity"]["applicability"] == "applicable" assert catalog["platform-pg"]["applicability"] == "pending" assert catalog["shared-substrate"]["applicability"] == "pending" def test_retired_ids_include_cancelled_records(): ids = retired_ids() assert "WH-ENG-20260821-AUDIT-E2" in ids assert "WH-ENG-20260821-TENANT-E2" in ids def test_fixture_plane_admits_and_projects_handles_without_secrets(tmp_path): engagement = load_engagement(tmp_path, fixture_record()) broker = LocalBroker() lease = admit( engagement=engagement, registration=fixture_registration(), broker=broker, kill_switch=KillSwitch(tmp_path / "KILL"), now=NOW, retired=retired_ids(), ) roles = {handle.role for handle in lease.identities} assert roles == {"owner", "attacker"} assert not hasattr(lease.identities[0], "secret") assert not hasattr(lease.identities[1], "secret") rendered = repr(lease) + repr(lease.identities) for value in broker._secrets.values(): assert value.hex() not in rendered assert value not in rendered.encode() report = cleanup(lease, broker) assert report["credential_revocation"] == "revoked" with pytest.raises(AuthorizationError, match="unknown"): broker.revoke(lease.lease_id) def test_not_applicable_target_is_refused(tmp_path): engagement = load_engagement(tmp_path, fixture_record(target_id="tenant-engine")) registration = load_registration("targets/tenant-engine-e2.json") with pytest.raises(AuthorizationError, match="not_applicable"): admit( engagement=engagement, registration=registration, broker=LocalBroker(), kill_switch=KillSwitch(tmp_path / "KILL"), now=NOW, retired=set(), ) def test_retired_id_cannot_be_reused_even_if_fields_are_complete(tmp_path): engagement = load_engagement( tmp_path, fixture_record(engagement_id="WH-ENG-20260821-AUDIT-E2") ) with pytest.raises(AuthorizationError, match="retired"): admit( engagement=engagement, registration=fixture_registration(), broker=LocalBroker(), kill_switch=KillSwitch(tmp_path / "KILL"), now=NOW, retired=retired_ids(), ) def test_kill_switch_fails_closed(tmp_path): path = tmp_path / "KILL" path.write_text("", encoding="utf-8") engagement = load_engagement(tmp_path, fixture_record()) with pytest.raises(AuthorizationError, match="kill switch"): admit( engagement=engagement, registration=fixture_registration(), broker=LocalBroker(), kill_switch=KillSwitch(path), now=NOW, retired=set(), ) def test_live_e2_without_plane_namespace_fails(tmp_path): engagement = load_engagement(tmp_path, fixture_record( approval_class="live-e2", environment="build", target_id="audit-core", routes=["POST /v1/events"], plane_namespace="user-engine", runner_image_digest="sha256:abc", )) registration = load_registration("targets/audit-core-e2.json") with pytest.raises(AuthorizationError, match="whitehat plane namespace"): admit( engagement=engagement, registration=registration, broker=LocalBroker(), kill_switch=KillSwitch(tmp_path / "KILL"), now=NOW, retired=set(), ) def test_unconnected_broker_requests_no_credential(tmp_path): engagement = load_engagement(tmp_path, fixture_record( approval_class="live-e2", environment="build", target_id="audit-core", routes=["POST /v1/events"], plane_namespace="whitehat", runner_image_digest="sha256:abc", )) registration = load_registration("targets/audit-core-e2.json") with pytest.raises(AuthorizationError, match="no credential was requested"): admit( engagement=engagement, registration=registration, broker=UnconnectedCustodyBroker(), kill_switch=KillSwitch(tmp_path / "KILL"), now=NOW, retired=set(), ) def test_e3_is_not_admitted_by_the_e2_plane(tmp_path): engagement = load_engagement(tmp_path, fixture_record( approval_class="e3", techniques=["e3-rls"], )) with pytest.raises(AuthorizationError): admit( engagement=engagement, registration=fixture_registration(), broker=LocalBroker(), kill_switch=KillSwitch(tmp_path / "KILL"), now=NOW, retired=set(), ) def test_fixture_e3_projects_one_runtime_identity(tmp_path): broker = LocalBroker() engagement = load_engagement(tmp_path, fixture_record( target_id="fixture-e3", approval_class="fixture-e3", techniques=["e3-rls"], routes=["conformance"], )) lease = admit( engagement=engagement, registration=load_registration("targets/fixture-e3.json"), broker=broker, kill_switch=KillSwitch(tmp_path / "KILL"), now=NOW, retired=set(), ) assert len(lease.identities) == 1 assert lease.identities[0].role == "runtime" assert cleanup(lease, broker)["credential_revocation"] == "revoked" def test_fixture_capacity_projects_no_identities(tmp_path): broker = LocalBroker() engagement = load_engagement(tmp_path, fixture_record( target_id="fixture-capacity", approval_class="fixture-capacity", techniques=["p1-noisy-neighbour"], routes=["baseline"], )) lease = admit( engagement=engagement, registration=load_registration("targets/fixture-capacity.json"), broker=broker, kill_switch=KillSwitch(tmp_path / "KILL"), now=NOW, retired=set(), ) assert lease.identities == () assert cleanup(lease, broker)["credential_revocation"] == "revoked" def test_pending_e3_target_is_refused(tmp_path): engagement = load_engagement(tmp_path, fixture_record( target_id="platform-pg", approval_class="e3", techniques=["e3-rls"], environment="build", plane_namespace="whitehat", runner_image_digest="sha256:abc", database="platform-pg", routes=["conformance"], )) with pytest.raises(AuthorizationError, match="pending"): admit( engagement=engagement, registration=load_registration("targets/platform-pg-e3.json"), broker=LocalBroker(), kill_switch=KillSwitch(tmp_path / "KILL"), now=NOW, retired=set(), ) def test_live_e3_without_database_fails_closed(tmp_path): registration = load_registration("targets/fixture-e3.json") registration["target_id"] = "fixture-e3-live" registration["approval_classes"] = ["e3"] engagement = load_engagement(tmp_path, fixture_record( target_id="fixture-e3-live", approval_class="e3", techniques=["e3-rls"], environment="build", plane_namespace="whitehat", runner_image_digest="sha256:abc", routes=["conformance"], )) with pytest.raises(AuthorizationError, match="named database"): admit( engagement=engagement, registration=registration, broker=LocalBroker(), kill_switch=KillSwitch(tmp_path / "KILL"), now=NOW, retired=set(), ) def test_e3_calibration_keeps_documented_limit_inconclusive(): report = e3_calibration() assert report["outcome"] == "pass" by_id = {item["probe_id"]: item for item in report["known_good"]} assert by_id["sql-compromise-reset"]["outcome"] == "inconclusive" assert by_id["conformance-view-empty"]["outcome"] == "pass" bad = {item["probe_id"]: item for item in report["known_bad"]} assert bad["conformance-view-empty"]["outcome"] == "finding" assert bad["sql-compromise-reset"]["outcome"] == "inconclusive" def test_rate_and_concurrency_ceilings(tmp_path): watcher = RateWatcher(per_minute=10, max_concurrency=1, max_requests=1) watcher.acquire() with pytest.raises(AuthorizationError, match="concurrency"): watcher.acquire() watcher.release() with pytest.raises(AuthorizationError, match="request ceiling"): watcher.acquire() def test_execute_authorized_requires_plane_lease(tmp_path): engagement = load_engagement(tmp_path, fixture_record()) lease = admit( engagement=engagement, registration=fixture_registration(), broker=LocalBroker(), kill_switch=KillSwitch(tmp_path / "KILL"), now=NOW, retired=set(), ) probe = probe_suite(FixtureService(True))[0] result = execute_authorized( probe, lease=lease, route="GET /objects/{id}", salt=b"test", now=NOW ) assert result.outcome == "pass" def test_fixture_report_is_not_delivered_as_target_assurance(tmp_path): report = RunReport( schema_version="whitehat-run/v1", run_id="run-1", evidence_class="fixture", engagement_id="eng-1", authorization_id="auth-1", target="fixture-e2", target_revision="local", posture_claim="E2", attacker_model="E2", started_at="2026-08-22T00:00:00Z", ended_at="2026-08-22T00:01:00Z", outcome="pass", attempted_operations=1, cleanup="complete", credential_revocation="complete", ) with pytest.raises(AuthorizationError, match="fixture evidence"): queue_risk_nexus(report, tmp_path) def test_target_report_is_queued_without_severity(tmp_path): report = RunReport( schema_version="whitehat-run/v1", run_id="run-2", evidence_class="target", engagement_id="eng-1", authorization_id="auth-1", target="audit-core", target_revision="abc", posture_claim="E2", attacker_model="E2", started_at="2026-08-22T00:00:00Z", ended_at="2026-08-22T00:01:00Z", outcome="pass", attempted_operations=1, cleanup="complete", credential_revocation="complete", ) path = queue_risk_nexus(report, tmp_path) text = path.read_text(encoding="utf-8") assert "**pass**" in text assert "Severity" not in text assert "not proof" in text def test_audit_shaped_probes_calibrate(): good = [execute(probe, salt=b"test") for probe in audit_probe_suite(AuditFixture(True))] bad = [execute(probe, salt=b"test") for probe in audit_probe_suite(AuditFixture(False))] assert {item.probe_id for item in good} == { "audit-event-by-id", "audit-correlation-slice", "audit-append-as-b", } assert {item.outcome for item in good} == {"pass"} assert {item.outcome for item in bad} == {"finding"}