#!/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()