From 7396fe7d612e858689cf6de6628ca2248a27fb9c Mon Sep 17 00:00:00 2001
From: tegwick <bernd.worsch@gmail.com>
Date: Sat, 22 Aug 2026 00:02:22 +0200
Subject: [PATCH] Add bounded audit-core E2 target adapter

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a0260c-4067-7052-9647-ad000d576e38
---
 runners/audit_core_e2.py        | 279 ++++++++++++++++++++++++++++++++
 tests/test_audit_core_runner.py |  32 ++++
 2 files changed, 311 insertions(+)
 create mode 100644 runners/audit_core_e2.py
 create mode 100644 tests/test_audit_core_runner.py

diff --git a/runners/audit_core_e2.py b/runners/audit_core_e2.py
new file mode 100644
index 0000000..b691a3d
--- /dev/null
+++ b/runners/audit_core_e2.py
@@ -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()
+
diff --git a/tests/test_audit_core_runner.py b/tests/test_audit_core_runner.py
new file mode 100644
index 0000000..5cc8eca
--- /dev/null
+++ b/tests/test_audit_core_runner.py
@@ -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
+