tegwick
96d6781036
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02991-be07-7bb3-8b6d-e9701b5621de
320 lines
13 KiB
Python
320 lines
13 KiB
Python
#!/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=request_headers(token, self.args.engagement_id,
|
|
self.args.correlation, payload),
|
|
)
|
|
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 request_headers(token: str, engagement_id: str, correlation: str,
|
|
payload: dict[str, Any] | None) -> dict[str, str]:
|
|
headers = {
|
|
"Authorization": f"Bearer {token}",
|
|
"Content-Type": "application/json",
|
|
"User-Agent": f"whitehat-security/{engagement_id}",
|
|
"X-Correlation-ID": correlation,
|
|
}
|
|
if payload is not None:
|
|
headers["Idempotency-Key"] = str(payload["id"])
|
|
return headers
|
|
|
|
|
|
def event(event_id: str, tenant: str, correlation: str, occurred_at: str) -> dict[str, Any]:
|
|
return {
|
|
"id": event_id,
|
|
"type": "whitehat.fixture",
|
|
"source": "whitehat-security",
|
|
"subject": event_id,
|
|
"tenant": tenant,
|
|
"correlation_id": correlation,
|
|
"occurred_at": occurred_at,
|
|
"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 invocation_fixture_ids(args: argparse.Namespace) -> set[str]:
|
|
"""Exact synthetic identifiers the invocation may address or emit."""
|
|
|
|
return {
|
|
args.tenant_a,
|
|
args.tenant_b,
|
|
args.event_a,
|
|
args.event_b,
|
|
args.absent_event,
|
|
args.correlation,
|
|
args.forged_event,
|
|
}
|
|
|
|
|
|
def build_report(args: argparse.Namespace, *, started: str, outcome: str,
|
|
attempted_operations: int, results: list[dict[str, Any]],
|
|
limitations: list[str]) -> dict[str, Any]:
|
|
return {
|
|
"schema_version": "whitehat-run/v1",
|
|
"run_id": f"{args.engagement_id}-{started}",
|
|
"evidence_class": "target",
|
|
"engagement_id": args.engagement_id,
|
|
"authorization_id": args.authorization_id,
|
|
"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": attempted_operations,
|
|
"cleanup": "named immutable audit fixtures retained by target contract",
|
|
"credential_revocation": "pending orchestrator cleanup",
|
|
"probes": results,
|
|
"limitations": limitations,
|
|
"assurance_statement": ASSURANCE,
|
|
}
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--base-url", required=True)
|
|
parser.add_argument("--engagement-id", required=True)
|
|
parser.add_argument("--authorization-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("--forged-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)
|
|
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,
|
|
args.occurred_at))
|
|
seed_b = client.call(args.token_b_file, "POST", "/v1/events",
|
|
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")
|
|
|
|
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 = args.forged_event
|
|
forged_response = client.observation(client.call(
|
|
args.token_a_file, "POST", "/v1/events",
|
|
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}"
|
|
), (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 = build_report(
|
|
args,
|
|
started=started,
|
|
outcome=outcome,
|
|
attempted_operations=client.requests,
|
|
results=results,
|
|
limitations=limitations,
|
|
)
|
|
print(json.dumps(report, sort_keys=True))
|
|
raise SystemExit(0 if outcome == "pass" else 1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|