Add bounded audit-core E2 target adapter
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a0260c-4067-7052-9647-ad000d576e38
This commit is contained in:
parent
3770b41b74
commit
7396fe7d61
2 changed files with 311 additions and 0 deletions
279
runners/audit_core_e2.py
Normal file
279
runners/audit_core_e2.py
Normal file
|
|
@ -0,0 +1,279 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Bounded audit-core E2 adapter; intended to run inside the attributed pod.
|
||||
|
||||
Token values are read from mounted files, used only as Authorization headers,
|
||||
and never included in stdout, exceptions, or evidence.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any
|
||||
|
||||
ASSURANCE = (
|
||||
"Pass means only that the attacks attempted in this run did not work; "
|
||||
"it is not proof that the tenant boundary always holds."
|
||||
)
|
||||
|
||||
|
||||
class Abort(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
def now() -> str:
|
||||
return datetime.now(UTC).isoformat().replace("+00:00", "Z")
|
||||
|
||||
|
||||
def canonical(value: Any) -> bytes:
|
||||
return json.dumps(value, sort_keys=True, separators=(",", ":")).encode()
|
||||
|
||||
|
||||
def values(value: Any) -> list[str]:
|
||||
if isinstance(value, dict):
|
||||
return [item for child in value.values() for item in values(child)]
|
||||
if isinstance(value, list):
|
||||
return [item for child in value for item in values(child)]
|
||||
return [] if value is None else [str(value)]
|
||||
|
||||
|
||||
def shape(value: Any, prefix: str = "$") -> list[str]:
|
||||
if isinstance(value, dict):
|
||||
result = [prefix]
|
||||
for key in sorted(value):
|
||||
result.extend(shape(value[key], f"{prefix}.{key}"))
|
||||
return result
|
||||
if isinstance(value, list):
|
||||
result = [f"{prefix}[]"]
|
||||
if value:
|
||||
result.extend(shape(value[0], f"{prefix}[]"))
|
||||
return result
|
||||
return [f"{prefix}:{type(value).__name__}"]
|
||||
|
||||
|
||||
class Client:
|
||||
def __init__(self, args: argparse.Namespace) -> None:
|
||||
self.args = args
|
||||
self.salt = os.urandom(32)
|
||||
self.last_request = 0.0
|
||||
self.latencies: list[float] = []
|
||||
self.requests = 0
|
||||
|
||||
def call(self, token_file: str, method: str, path: str,
|
||||
payload: dict[str, Any] | None = None) -> tuple[int, Any, str]:
|
||||
delay = 60.0 / self.args.rate - (time.monotonic() - self.last_request)
|
||||
if delay > 0:
|
||||
time.sleep(delay)
|
||||
if self.requests >= self.args.max_requests:
|
||||
raise Abort("request ceiling reached")
|
||||
token = open(token_file, encoding="utf-8").read().strip()
|
||||
if not token:
|
||||
raise Abort("mounted credential is empty")
|
||||
raw = canonical(payload) if payload is not None else None
|
||||
request = urllib.request.Request(
|
||||
self.args.base_url + path,
|
||||
data=raw,
|
||||
method=method,
|
||||
headers={
|
||||
"Authorization": f"Bearer {token}",
|
||||
"Content-Type": "application/json",
|
||||
"User-Agent": f"whitehat-security/{self.args.engagement_id}",
|
||||
"X-Correlation-ID": self.args.correlation,
|
||||
},
|
||||
)
|
||||
started = time.monotonic()
|
||||
try:
|
||||
with urllib.request.urlopen(request, timeout=self.args.timeout) as response:
|
||||
status = response.status
|
||||
content_type = response.headers.get("Content-Type", "application/octet-stream")
|
||||
body = response.read(524_289)
|
||||
except urllib.error.HTTPError as error:
|
||||
status = error.code
|
||||
content_type = error.headers.get("Content-Type", "application/octet-stream")
|
||||
body = error.read(524_289)
|
||||
except (urllib.error.URLError, TimeoutError, OSError) as error:
|
||||
raise Abort(f"transport failure: {type(error).__name__}") from None
|
||||
finally:
|
||||
self.last_request = time.monotonic()
|
||||
elapsed = (time.monotonic() - started) * 1000
|
||||
self.latencies.append(elapsed)
|
||||
self.requests += 1
|
||||
if len(body) > 524_288:
|
||||
raise Abort("response exceeds evidence-safe maximum")
|
||||
if status >= 500:
|
||||
raise Abort("target returned a 5xx response")
|
||||
if _p95(self.latencies) > self.args.abort_p95_ms:
|
||||
raise Abort("probe p95 latency exceeded the engagement threshold")
|
||||
try:
|
||||
parsed = json.loads(body) if body else {}
|
||||
except ValueError:
|
||||
raise Abort("target returned non-JSON on an in-scope JSON route") from None
|
||||
return status, parsed, content_type
|
||||
|
||||
def observation(self, response: tuple[int, Any, str], markers: tuple[str, ...]) -> dict:
|
||||
status, body, content_type = response
|
||||
scalars = values(body)
|
||||
count = len(body) if isinstance(body, list) else 1
|
||||
if isinstance(body, dict):
|
||||
for key in ("events", "items", "results", "rows"):
|
||||
if isinstance(body.get(key), list):
|
||||
count = len(body[key])
|
||||
break
|
||||
return {
|
||||
"status": status,
|
||||
"content_type": content_type.split(";", 1)[0].lower(),
|
||||
"count": count,
|
||||
"schema": shape(body),
|
||||
"run_digest": hashlib.sha256(self.salt + canonical(body)).hexdigest(),
|
||||
"fixture_match_count": sum(any(marker in value for value in scalars)
|
||||
for marker in markers),
|
||||
}
|
||||
|
||||
|
||||
def _p95(samples: list[float]) -> float:
|
||||
ordered = sorted(samples)
|
||||
return ordered[max(0, math.ceil(len(ordered) * .95) - 1)]
|
||||
|
||||
|
||||
def event(event_id: str, tenant: str, correlation: str) -> dict[str, Any]:
|
||||
return {
|
||||
"id": event_id,
|
||||
"type": "whitehat.fixture",
|
||||
"source": "whitehat-security",
|
||||
"subject": event_id,
|
||||
"tenant": tenant,
|
||||
"correlation_id": correlation,
|
||||
"occurred_at": now(),
|
||||
"data": {"fixture_id": event_id},
|
||||
}
|
||||
|
||||
|
||||
def equivalent(left: dict, right: dict) -> bool:
|
||||
return all(left[key] == right[key] for key in (
|
||||
"status", "content_type", "count", "schema", "run_digest"
|
||||
))
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--base-url", required=True)
|
||||
parser.add_argument("--engagement-id", required=True)
|
||||
parser.add_argument("--target-revision", required=True)
|
||||
parser.add_argument("--token-a-file", required=True)
|
||||
parser.add_argument("--token-b-file", required=True)
|
||||
parser.add_argument("--tenant-a", required=True)
|
||||
parser.add_argument("--tenant-b", required=True)
|
||||
parser.add_argument("--event-a", required=True)
|
||||
parser.add_argument("--event-b", required=True)
|
||||
parser.add_argument("--absent-event", required=True)
|
||||
parser.add_argument("--correlation", required=True)
|
||||
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)
|
||||
parser.add_argument("--abort-p95-ms", type=float, default=500)
|
||||
args = parser.parse_args()
|
||||
started = now()
|
||||
client = Client(args)
|
||||
results: list[dict[str, Any]] = []
|
||||
outcome = "pass"
|
||||
limitations: list[str] = []
|
||||
try:
|
||||
seed_a = client.call(args.token_a_file, "POST", "/v1/events",
|
||||
event(args.event_a, args.tenant_a, args.correlation))
|
||||
seed_b = client.call(args.token_b_file, "POST", "/v1/events",
|
||||
event(args.event_b, args.tenant_b, args.correlation))
|
||||
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")
|
||||
|
||||
owner = client.observation(
|
||||
client.call(args.token_b_file, "GET", f"/v1/events/{args.event_b}"),
|
||||
(args.event_b, args.tenant_b),
|
||||
)
|
||||
absent = client.observation(
|
||||
client.call(args.token_a_file, "GET", f"/v1/events/{args.absent_event}"),
|
||||
(args.event_b, args.tenant_b),
|
||||
)
|
||||
attacker = client.observation(
|
||||
client.call(args.token_a_file, "GET", f"/v1/events/{args.event_b}"),
|
||||
(args.event_b, args.tenant_b),
|
||||
)
|
||||
reasons = []
|
||||
if attacker["fixture_match_count"]:
|
||||
reasons.append("tenant-A response contains tenant-B fixture marker")
|
||||
if not equivalent(attacker, absent):
|
||||
reasons.append("tenant-A response differs from absent-event reference")
|
||||
results.append({"probe_id": "audit-event-by-id", "operation": "read",
|
||||
"outcome": "finding" if reasons else "pass", "reasons": reasons,
|
||||
"observations": {"owner": owner, "attacker": attacker,
|
||||
"absent": absent}})
|
||||
|
||||
owner_corr = client.observation(client.call(
|
||||
args.token_b_file, "GET", f"/v1/events?correlation_id={args.correlation}"
|
||||
), (args.event_b, args.tenant_b))
|
||||
attacker_corr = client.observation(client.call(
|
||||
args.token_a_file, "GET", f"/v1/events?correlation_id={args.correlation}"
|
||||
), (args.event_b, args.tenant_b))
|
||||
reasons = (["tenant-A correlation slice contains tenant-B fixture marker"]
|
||||
if attacker_corr["fixture_match_count"] else [])
|
||||
results.append({"probe_id": "audit-correlation-slice", "operation": "read",
|
||||
"outcome": "finding" if reasons else "pass", "reasons": reasons,
|
||||
"observations": {"owner": owner_corr, "attacker": attacker_corr}})
|
||||
|
||||
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)
|
||||
), (forged, args.tenant_b))
|
||||
forged_state = client.observation(client.call(
|
||||
args.token_b_file, "GET", f"/v1/events/{forged}"
|
||||
), (forged, args.tenant_b))
|
||||
absent_b = client.observation(client.call(
|
||||
args.token_b_file, "GET", f"/v1/events/{args.absent_event}"
|
||||
), (forged, args.tenant_b))
|
||||
reasons = []
|
||||
if forged_response["status"] in (200, 201, 202):
|
||||
reasons.append("tenant-A credential accepted an event attributed to tenant B")
|
||||
if not equivalent(forged_state, absent_b):
|
||||
reasons.append("forged tenant-B event exists after tenant-A submission")
|
||||
results.append({"probe_id": "audit-append-as-b", "operation": "create",
|
||||
"outcome": "finding" if reasons else "pass", "reasons": reasons,
|
||||
"observations": {"attacker": forged_response,
|
||||
"state_after": forged_state, "absent": absent_b}})
|
||||
if any(result["outcome"] == "finding" for result in results):
|
||||
outcome = "finding"
|
||||
except Abort as error:
|
||||
outcome = "aborted"
|
||||
limitations.append(str(error))
|
||||
report = {
|
||||
"schema_version": "whitehat-run/v1",
|
||||
"run_id": f"{args.engagement_id}-{started}",
|
||||
"evidence_class": "target",
|
||||
"engagement_id": args.engagement_id,
|
||||
"authorization_id": "operator-session-2026-08-21-e2-approval",
|
||||
"target": "audit-core",
|
||||
"target_revision": args.target_revision,
|
||||
"posture_claim": "implemented E2; currently evidenced E1",
|
||||
"attacker_model": "E2-authenticated-tenant-a",
|
||||
"started_at": started,
|
||||
"ended_at": now(),
|
||||
"outcome": outcome,
|
||||
"attempted_operations": client.requests,
|
||||
"cleanup": "named immutable audit fixtures retained by target contract",
|
||||
"credential_revocation": "pending orchestrator cleanup",
|
||||
"probes": results,
|
||||
"limitations": limitations,
|
||||
"assurance_statement": ASSURANCE,
|
||||
}
|
||||
print(json.dumps(report, sort_keys=True))
|
||||
raise SystemExit(0 if outcome == "pass" else 1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
32
tests/test_audit_core_runner.py
Normal file
32
tests/test_audit_core_runner.py
Normal file
|
|
@ -0,0 +1,32 @@
|
|||
import importlib.util
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
SPEC = importlib.util.spec_from_file_location(
|
||||
"audit_core_e2_runner", Path("runners/audit_core_e2.py")
|
||||
)
|
||||
runner = importlib.util.module_from_spec(SPEC)
|
||||
assert SPEC.loader is not None
|
||||
SPEC.loader.exec_module(runner)
|
||||
|
||||
|
||||
def test_equivalent_denial_includes_digest():
|
||||
base = {
|
||||
"status": 404, "content_type": "application/json", "count": 1,
|
||||
"schema": ["$", "$.error:str"], "run_digest": "same",
|
||||
}
|
||||
assert runner.equivalent(base, dict(base))
|
||||
changed = dict(base, run_digest="different")
|
||||
assert not runner.equivalent(base, changed)
|
||||
|
||||
|
||||
def test_event_is_synthetic_and_correlation_bound():
|
||||
payload = runner.event("event-a", "tenant-a", "corr")
|
||||
assert payload["id"] == payload["data"]["fixture_id"]
|
||||
assert payload["tenant"] == "tenant-a"
|
||||
assert payload["correlation_id"] == "corr"
|
||||
|
||||
|
||||
def test_p95_is_conservative_for_small_runs():
|
||||
assert runner._p95([10, 20, 30]) == 30
|
||||
|
||||
Loading…
Add table
Add a link
Reference in a new issue