tegwick
95129d7a35
Encode fail-closed admission, target registrations, and a credential broker that never returns secret values. Calibrate audit-core shaped probes in-process. Send no packets and request no live credentials. Assistant: grok Assistant-Session: 01a02670-3345-76f2-a014-70fde8e2a2bb
234 lines
9.3 KiB
Python
234 lines
9.3 KiB
Python
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.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"
|
|
|
|
|
|
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_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"}
|