whitehat-security/tests/test_plane.py

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, ReceiptBroker, 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"] == "not_applicable"
    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
    assert "WH-ENG-20260822-AUDIT-E2-01" in ids
    assert "WH-ENG-20260822-AUDIT-E2-02" in ids
    assert "WH-ENG-20260822-AUDIT-E2-03" 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_receipt_broker_issues_lease_without_secret_values(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",
    ))
    receipt = {
        "engagement_id": "WH-ENG-FIXTURE-1",
        "projected_at": "2026-08-22T12:00:00Z",
        "expires_at": "2099-01-01T00:00:00Z",
        "identities": ["whitehat-e2-a", "whitehat-e2-b"],
        "mounted_secret": "whitehat/whitehat-e2-audit-credentials",
        "mounted_keys": ["token-a", "token-b"],
        "target_ready": True,
        "secret_values_observed": False,
    }
    lease = admit(
        engagement=engagement, registration=load_registration("targets/audit-core-e2.json"),
        broker=ReceiptBroker(receipt), kill_switch=KillSwitch(tmp_path / "KILL"),
        now=NOW, retired=set(),
    )
    assert {handle.role for handle in lease.identities} == {"owner", "attacker"}
    rendered = repr(lease) + repr(lease.identities) + json.dumps(receipt)
    assert "token_urlsafe" not in rendered
    with pytest.raises(AuthorizationError, match="custody must revoke"):
        cleanup(lease, ReceiptBroker(receipt))


def test_receipt_broker_rejects_secret_material(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",
    ))
    receipt = {
        "engagement_id": "WH-ENG-FIXTURE-1",
        "projected_at": "2026-08-22T12:00:00Z",
        "expires_at": "2099-01-01T00:00:00Z",
        "identities": ["whitehat-e2-a", "whitehat-e2-b"],
        "mounted_secret": "whitehat/whitehat-e2-audit-credentials",
        "mounted_keys": ["token-a", "token-b"],
        "target_ready": True,
        "secret_values_observed": False,
        "token": "must-not-appear",
    }
    with pytest.raises(AuthorizationError, match="secret material"):
        admit(
            engagement=engagement, registration=load_registration("targets/audit-core-e2.json"),
            broker=ReceiptBroker(receipt), 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_not_applicable_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="not_applicable"):
        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"}