Build authorization-gated tenancy evidence harness

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a0260c-4067-7052-9647-ad000d576e38

tests/test_differential.py

@@ -0,0 +1,37 @@
+from whitehat_security.cli import fixture_calibration
+from whitehat_security.differential import execute
+from whitehat_security.fixtures import FixtureService, probe_suite
+
+
+def test_every_probe_passes_known_good_fixture():
+    results = [execute(probe, salt=b"test") for probe in probe_suite(FixtureService(True))]
+    assert results
+    assert {result.outcome for result in results} == {"pass"}
+
+
+def test_every_probe_detects_known_bad_fixture():
+    results = [execute(probe, salt=b"test") for probe in probe_suite(FixtureService(False))]
+    assert results
+    assert {result.outcome for result in results} == {"finding"}
+
+
+def test_evidence_never_contains_fixture_body_values():
+    result = execute(probe_suite(FixtureService(False))[0], salt=b"test")
+    rendered = repr(result)
+    assert "fixture-b" not in rendered
+    assert "tenant-b" not in rendered
+    assert result.observations["attacker"].fixture_match_count > 0
+
+
+def test_fixture_marker_is_found_when_embedded_in_a_value():
+    service = FixtureService(False)
+    service.objects["object-b"]["label"] = "prefix-fixture-b-suffix"
+    result = execute(probe_suite(service)[0], salt=b"test")
+    assert result.outcome == "finding"
+
+
+def test_calibration_distinguishes_good_and_bad():
+    report = fixture_calibration()
+    assert report["outcome"] == "pass"
+    assert all(item["outcome"] == "pass" for item in report["known_good"])
+    assert all(item["outcome"] == "finding" for item in report["known_bad"])

tests/test_e3_capacity_reporting.py

@@ -0,0 +1,50 @@
+from whitehat_security.capacity import CapacitySample, characterize
+from whitehat_security.e3 import PROBES, evaluate
+from whitehat_security.model import RunReport
+from whitehat_security.reporting import risk_nexus_message
+
+
+def test_e3_expected_failures_are_findings_and_limit_is_not():
+    conformance = next(probe for probe in PROBES if probe.probe_id == "conformance-view-empty")
+    boundary = next(probe for probe in PROBES if probe.probe_id == "sql-compromise-reset")
+    assert evaluate(conformance, rows=1).outcome == "finding"
+    assert evaluate(conformance, rows=0).outcome == "pass"
+    assert evaluate(boundary, rows=1).outcome == "inconclusive"
+
+
+def test_capacity_records_neighbour_degradation():
+    result = characterize(
+        baseline=[CapacitySample("n", 10, 0, 100)],
+        loaded=[CapacitySample("n", 15, .01, 75)],
+        governor_bound=True, aggressor_peak=5, aggressor_ceiling=5,
+    )
+    assert result.outcome == "pass"
+    assert result.neighbour_degradation["n"] == {
+        "latency_increase_percent": 50.0,
+        "error_rate_increase_points": .01,
+        "throughput_decrease_percent": 25.0,
+    }
+
+
+def test_capacity_unbound_governor_is_finding():
+    result = characterize(
+        baseline=[], loaded=[], governor_bound=False,
+        aggressor_peak=7, aggressor_ceiling=5,
+    )
+    assert result.outcome == "aborted"
+    assert len(result.reasons) == 2
+
+
+def test_risk_message_contains_pass_and_no_severity():
+    report = RunReport(
+        schema_version="whitehat-run/v1", run_id="run-1", evidence_class="target",
+        engagement_id="eng-1", authorization_id="auth-1", target="service",
+        target_revision="abc", posture_claim="E2", attacker_model="E2",
+        started_at="2026-08-21T00:00:00Z", ended_at="2026-08-21T00:01:00Z",
+        outcome="pass", attempted_operations=1, cleanup="complete",
+        credential_revocation="complete",
+    )
+    message = risk_nexus_message(report)
+    assert "**pass**" in message
+    assert "Severity" not in message
+    assert "not proof" in message

tests/test_engagement.py

@@ -0,0 +1,63 @@
+import json
+from datetime import UTC, datetime
+
+import pytest
+
+from whitehat_security.engagement import AuthorizationError, Engagement
+
+
+def record():
+    return {
+        "engagement_id": "WH-ENG-1", "authorization_id": "auth-1",
+        "authorizer": "operator", "approved_at": "2026-08-21T08:00:00Z",
+        "expires_at": "2026-08-21T12:00:00Z", "target": "https://fixture.invalid",
+        "target_owner": "target-repo", "environment": "build", "source": "runner",
+        "routes": ["GET /objects/{id}"], "fixture_ids": ["object-a", "object-b"],
+        "credential_lane": "openbao", "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-21T08:00:00Z",
+        "window_end": "2026-08-21T10:00:00Z", "operator_contact": "operator",
+        "abort_contact": "target-owner", "posture_claim": "E2",
+        "attacker_model": "E2-authenticated-tenant-a", "finding_destination": "risk-nexus",
+        "target_owner_acknowledged_at": "2026-08-21T08:01:00Z",
+    }
+
+
+def load(tmp_path, data):
+    path = tmp_path / "engagement.json"
+    path.write_text(json.dumps(data), encoding="utf-8")
+    return Engagement.load(path, now=datetime(2026, 8, 21, 9, tzinfo=UTC))
+
+
+def test_complete_current_record_is_accepted(tmp_path):
+    engagement = load(tmp_path, record())
+    engagement.permits(technique="e2-differential", route="GET /objects/{id}")
+
+
+@pytest.mark.parametrize("field", ["target_owner_acknowledged_at", "abort_contact", "fixture_ids"])
+def test_incomplete_record_fails_closed(tmp_path, field):
+    data = record()
+    del data[field]
+    with pytest.raises(AuthorizationError):
+        load(tmp_path, data)
+
+
+def test_expired_record_fails_closed(tmp_path):
+    with pytest.raises(AuthorizationError, match="outside"):
+        Engagement.load(
+            _write(tmp_path, record()), now=datetime(2026, 8, 21, 13, tzinfo=UTC)
+        )
+
+
+def test_unauthorized_route_fails_closed(tmp_path):
+    engagement = load(tmp_path, record())
+    with pytest.raises(AuthorizationError, match="route"):
+        engagement.permits(technique="e2-differential", route="DELETE /objects/{id}")
+
+
+def _write(tmp_path, data):
+    path = tmp_path / "engagement.json"
+    path.write_text(json.dumps(data), encoding="utf-8")
+    return path
+