coulomb/whitehat-security
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Make audit fixture retries deterministic

Browse source 
Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a0260c-4067-7052-9647-ad000d576e38
This commit is contained in:
tegwick 2026-08-22 00:06:35 +02:00
parent 8a97fac17f
commit cf696f09d5
3 changed files with 18 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

14
runners/audit_core_e2.py
View file

@ -150,7 +150,7 @@ def request_headers(token: str, engagement_id: str, correlation: str,
return headers
def event(event_id: str, tenant: str, correlation: str) -> dict[str, Any]:
def event(event_id: str, tenant: str, correlation: str, occurred_at: str) -> dict[str, Any]:
return {
"id": event_id,
"type": "whitehat.fixture",
@ -158,7 +158,7 @@ def event(event_id: str, tenant: str, correlation: str) -> dict[str, Any]:
"subject": event_id,
"tenant": tenant,
"correlation_id": correlation,
"occurred_at": now(),
"occurred_at": occurred_at,
"data": {"fixture_id": event_id},
}
@ -182,6 +182,8 @@ def main() -> None:
parser.add_argument("--event-b", required=True)
parser.add_argument("--absent-event", required=True)
parser.add_argument("--correlation", required=True)
parser.add_argument("--occurred-at", required=True,
help="fixed RFC3339 fixture time; makes retries idempotent")
parser.add_argument("--rate", type=int, default=10)
parser.add_argument("--max-requests", type=int, default=30)
parser.add_argument("--timeout", type=float, default=5)
@ -194,9 +196,11 @@ def main() -> None:
limitations: list[str] = []
try:
seed_a = client.call(args.token_a_file, "POST", "/v1/events",
event(args.event_a, args.tenant_a, args.correlation))
event(args.event_a, args.tenant_a, args.correlation,
args.occurred_at))
seed_b = client.call(args.token_b_file, "POST", "/v1/events",
event(args.event_b, args.tenant_b, args.correlation))
event(args.event_b, args.tenant_b, args.correlation,
args.occurred_at))
if seed_a[0] not in (200, 202) or seed_b[0] not in (200, 202):
raise Abort("fixture seeding did not reach accepted/duplicate state")
@ -237,7 +241,7 @@ def main() -> None:
forged = "whitehat-e2-forged-b-20260821"
forged_response = client.observation(client.call(
args.token_a_file, "POST", "/v1/events",
event(forged, args.tenant_b, args.correlation)
event(forged, args.tenant_b, args.correlation, args.occurred_at)
), (forged, args.tenant_b))
forged_state = client.observation(client.call(
args.token_b_file, "GET", f"/v1/events/{forged}"

3
tests/test_audit_core_runner.py
View file

@ -21,10 +21,11 @@ def test_equivalent_denial_includes_digest():
def test_event_is_synthetic_and_correlation_bound():
payload = runner.event("event-a", "tenant-a", "corr")
payload = runner.event("event-a", "tenant-a", "corr", "2026-08-21T23:00:00Z")
assert payload["id"] == payload["data"]["fixture_id"]
assert payload["tenant"] == "tenant-a"
assert payload["correlation_id"] == "corr"
assert payload["occurred_at"] == "2026-08-21T23:00:00Z"
def test_p95_is_conservative_for_small_runs():

8
tests/test_engagement.py
View file

@ -56,8 +56,14 @@ def test_unauthorized_route_fails_closed(tmp_path):
engagement.permits(technique="e2-differential", route="DELETE /objects/{id}")
def test_pending_owner_acknowledgement_fails_closed(tmp_path):
data = record()
data["target_owner_acknowledged_at"] = None
with pytest.raises(AuthorizationError, match="pending"):
load(tmp_path, data)
def _write(tmp_path, data):
path = tmp_path / "engagement.json"
path.write_text(json.dumps(data), encoding="utf-8")
return path