feat(AUDIT-WP-0008): add T02 synthetic load driver

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a02991-be07-7bb3-8b6d-e9701b5621de

docs/recovery-synthetic-load-driver.md

@@ -0,0 +1,39 @@
+# T02 synthetic-load driver
+
+`scripts/t02_synthetic_load_driver.py` implements the hash-bound
+`railiance.synthetic-load-driver` v1 contract for
+`RAILIANCE-WP-0024-T02-SYNTHETIC-01`. It is a candidate execution component,
+not authorization to revoke a database lease or create a temporary identity.
+
+The attended operator supplies these non-secret settings to the driver process:
+
+- `AUDIT_T02_BASE_URL`: the approved direct audit-core URL;
+- `AUDIT_T02_TENANT`, `AUDIT_T02_SOURCE`, and `AUDIT_T02_FIXTURE_ID`: the exact
+  approved synthetic identity scope and immutable fixture;
+- `AUDIT_T02_OCCURRED_AT`: the fixed RFC3339 event time used by every phase;
+- optionally `AUDIT_T02_MAX_ATTEMPTS` (1–30, default 6) and
+  `AUDIT_T02_RETRY_INTERVAL_SECONDS` (0–10, default 1).
+
+`AUDIT_T02_TOKEN_FILE` is different: it names the approved private mounted
+bearer file. The driver requires a regular file with no group/world permission,
+reads one bounded line inside its own process, and never returns the bearer,
+request body, response body, database credential, Secret data, or OpenBao
+output. Do not put the bearer in an environment variable or parent shell.
+
+The four invocations are:
+
+```bash
+scripts/t02_synthetic_load_driver.py baseline --contract-id RAILIANCE-WP-0024-T02-SYNTHETIC-01
+scripts/t02_synthetic_load_driver.py expect-unavailable --contract-id RAILIANCE-WP-0024-T02-SYNTHETIC-01
+scripts/t02_synthetic_load_driver.py expect-recovered --contract-id RAILIANCE-WP-0024-T02-SYNTHETIC-01
+scripts/t02_synthetic_load_driver.py cleanup --contract-id RAILIANCE-WP-0024-T02-SYNTHETIC-01
+```
+
+`baseline` commits or reconciles one immutable synthetic event.
+`expect-unavailable` accepts only HTTP 503 as the required retryable evidence
+and never retries HTTP 400, 401, 403, or 409. `expect-recovered` retries bounded
+transport/500/503 failures until the same event is accepted or reconciled as a
+duplicate. `cleanup` owns no server-side deletion: it stops no persistent
+process and reports its ephemeral driver runtime clean, while the accepted
+audit fixture remains as evidence. Temporary sender identity cleanup is a
+separate attended `railiance-platform` custody action.

scripts/t02_synthetic_load_driver.py

@@ -0,0 +1,252 @@
+#!/usr/bin/env python3
+"""Value-safe synthetic sender for the attended T02 lease-recovery exercise.
+
+The driver implements the revision-pinned ``railiance.synthetic-load-driver``
+interface.  It reads a short-lived bearer only from a mounted file, sends one
+immutable synthetic event, and emits only the exact evidence shape required by
+the selected phase.  It never prints request or response bodies.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import stat
+import sys
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Callable, Mapping
+
+
+CONTRACT_ID = "RAILIANCE-WP-0024-T02-SYNTHETIC-01"
+PHASES = ("baseline", "expect-unavailable", "expect-recovered", "cleanup")
+TERMINAL_HTTP = {400, 401, 403, 409}
+RETRYABLE_HTTP = {0, 500, 503}
+MAX_RESPONSE_BYTES = 64 * 1024
+
+
+class DriverError(RuntimeError):
+    """A value-safe, operator-facing driver failure."""
+
+
+@dataclass(frozen=True)
+class Config:
+    base_url: str
+    token_file: Path
+    tenant: str
+    source: str
+    fixture_id: str
+    occurred_at: str
+    max_attempts: int
+    retry_interval: float
+
+
+def _required(env: Mapping[str, str], name: str) -> str:
+    value = env.get(name, "").strip()
+    if not value:
+        raise DriverError(f"required configuration is missing: {name}")
+    return value
+
+
+def _bounded_int(env: Mapping[str, str], name: str, default: int) -> int:
+    try:
+        value = int(env.get(name, str(default)))
+    except ValueError as exc:
+        raise DriverError(f"invalid numeric configuration: {name}") from exc
+    if not 1 <= value <= 30:
+        raise DriverError(f"configuration is outside its safe bound: {name}")
+    return value
+
+
+def _bounded_float(env: Mapping[str, str], name: str, default: float) -> float:
+    try:
+        value = float(env.get(name, str(default)))
+    except ValueError as exc:
+        raise DriverError(f"invalid numeric configuration: {name}") from exc
+    if not 0 <= value <= 10:
+        raise DriverError(f"configuration is outside its safe bound: {name}")
+    return value
+
+
+def load_config(env: Mapping[str, str]) -> Config:
+    base_url = _required(env, "AUDIT_T02_BASE_URL").rstrip("/")
+    parsed = urllib.parse.urlsplit(base_url)
+    if (
+        parsed.scheme not in {"http", "https"}
+        or not parsed.netloc
+        or parsed.username is not None
+        or parsed.password is not None
+        or parsed.query
+        or parsed.fragment
+    ):
+        raise DriverError("AUDIT_T02_BASE_URL must be a credential-free HTTP(S) URL")
+    token_file = Path(_required(env, "AUDIT_T02_TOKEN_FILE"))
+    return Config(
+        base_url=base_url,
+        token_file=token_file,
+        tenant=_required(env, "AUDIT_T02_TENANT"),
+        source=_required(env, "AUDIT_T02_SOURCE"),
+        fixture_id=_required(env, "AUDIT_T02_FIXTURE_ID"),
+        occurred_at=_required(env, "AUDIT_T02_OCCURRED_AT"),
+        max_attempts=_bounded_int(env, "AUDIT_T02_MAX_ATTEMPTS", 6),
+        retry_interval=_bounded_float(env, "AUDIT_T02_RETRY_INTERVAL_SECONDS", 1.0),
+    )
+
+
+def load_cleanup_fixture(env: Mapping[str, str]) -> str:
+    """Cleanup owns no credential or server-side object; only the id is needed."""
+    return _required(env, "AUDIT_T02_FIXTURE_ID")
+
+
+def read_bearer(path: Path) -> str:
+    try:
+        details = path.stat()
+    except OSError as exc:
+        raise DriverError("approved mounted bearer file is unavailable") from exc
+    mode = stat.S_IMODE(details.st_mode)
+    if not stat.S_ISREG(details.st_mode) or mode & 0o077:
+        raise DriverError("approved mounted bearer file must be private and regular")
+    try:
+        raw = path.read_text(encoding="utf-8")
+    except OSError as exc:
+        raise DriverError("approved mounted bearer file cannot be read") from exc
+    lines = raw.splitlines()
+    bearer = lines[0].strip() if len(lines) == 1 else ""
+    if not bearer or len(bearer) > 4096:
+        raise DriverError("approved mounted bearer file has an invalid shape")
+    return bearer
+
+
+def event_payload(config: Config) -> dict[str, object]:
+    return {
+        "id": config.fixture_id,
+        "type": "audit.recovery.synthetic",
+        "source": config.source,
+        "subject": config.fixture_id,
+        "tenant": config.tenant,
+        "correlation_id": f"{config.fixture_id}:t02",
+        "occurred_at": config.occurred_at,
+        "data": {"exercise": "RAILIANCE-WP-0024-T02", "synthetic": True},
+    }
+
+
+def request_event(config: Config) -> tuple[int, str | None]:
+    bearer = read_bearer(config.token_file)
+    payload = json.dumps(event_payload(config), separators=(",", ":")).encode("utf-8")
+    request = urllib.request.Request(
+        f"{config.base_url}/v1/events", data=payload, method="POST"
+    )
+    request.add_header("Authorization", f"Bearer {bearer}")
+    request.add_header("Content-Type", "application/json")
+    request.add_header("Idempotency-Key", config.fixture_id)
+    try:
+        with urllib.request.urlopen(request, timeout=10) as response:
+            status = response.status
+            body = response.read(MAX_RESPONSE_BYTES + 1)
+    except urllib.error.HTTPError as exc:
+        exc.close()
+        return exc.code, None
+    except (OSError, TimeoutError, urllib.error.URLError):
+        return 0, None
+    if len(body) > MAX_RESPONSE_BYTES:
+        raise DriverError("target response exceeded the safe size bound")
+    try:
+        value = json.loads(body or b"{}")
+    except json.JSONDecodeError as exc:
+        raise DriverError("target returned invalid JSON") from exc
+    application_status = value.get("status") if isinstance(value, dict) else None
+    return status, application_status if isinstance(application_status, str) else None
+
+
+def _attempt(
+    config: Config,
+    phase: str,
+    requester: Callable[[Config], tuple[int, str | None]],
+    sleeper: Callable[[float], None],
+) -> dict[str, object]:
+    last_status = 0
+    for attempt in range(1, config.max_attempts + 1):
+        http_status, application_status = requester(config)
+        last_status = http_status
+        if http_status in TERMINAL_HTTP:
+            raise DriverError(f"{phase} received terminal HTTP status {http_status}")
+
+        if phase == "expect-unavailable" and http_status == 503:
+            return {
+                "contract_id": CONTRACT_ID,
+                "fixture_id": config.fixture_id,
+                "status": "retryable_unavailable",
+                "http_status": 503,
+                "attempts": attempt,
+                "secret_values_observed": False,
+            }
+
+        if http_status in {200, 202}:
+            if application_status not in {"accepted", "duplicate"}:
+                raise DriverError(f"{phase} returned an unexpected application status")
+            if phase == "expect-unavailable":
+                raise DriverError("expect-unavailable observed a successful accept")
+            if phase == "baseline":
+                return {
+                    "contract_id": CONTRACT_ID,
+                    "fixture_id": config.fixture_id,
+                    "status": "ready",
+                    "secret_values_observed": False,
+                }
+            return {
+                "contract_id": CONTRACT_ID,
+                "fixture_id": config.fixture_id,
+                "status": application_status,
+                "http_status": http_status,
+                "attempts": attempt,
+                "secret_values_observed": False,
+            }
+
+        if http_status not in RETRYABLE_HTTP:
+            raise DriverError(f"{phase} received unexpected HTTP status {http_status}")
+        if attempt < config.max_attempts:
+            sleeper(config.retry_interval)
+    raise DriverError(f"{phase} exhausted its bounded attempts at HTTP status {last_status}")
+
+
+def run_phase(
+    phase: str,
+    contract_id: str,
+    *,
+    env: Mapping[str, str] = os.environ,
+    requester: Callable[[Config], tuple[int, str | None]] = request_event,
+    sleeper: Callable[[float], None] = time.sleep,
+) -> dict[str, object]:
+    if contract_id != CONTRACT_ID:
+        raise DriverError("requested contract id is not approved")
+    if phase == "cleanup":
+        return {
+            "contract_id": CONTRACT_ID,
+            "fixture_id": load_cleanup_fixture(env),
+            "status": "clean",
+            "secret_values_observed": False,
+        }
+    return _attempt(load_config(env), phase, requester, sleeper)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("phase", choices=PHASES)
+    parser.add_argument("--contract-id", required=True)
+    args = parser.parse_args(argv)
+    try:
+        result = run_phase(args.phase, args.contract_id)
+    except DriverError as exc:
+        print(f"synthetic-load driver refused: {exc}", file=sys.stderr)
+        return 2
+    print(json.dumps(result, sort_keys=True, separators=(",", ":")))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_t02_synthetic_load_driver.py

@@ -0,0 +1,129 @@
+import importlib.util
+import sys
+from pathlib import Path
+
+import pytest
+
+
+SCRIPT = Path(__file__).parents[1] / "scripts" / "t02_synthetic_load_driver.py"
+SPEC = importlib.util.spec_from_file_location("t02_synthetic_load_driver", SCRIPT)
+assert SPEC and SPEC.loader
+DRIVER = importlib.util.module_from_spec(SPEC)
+sys.modules[SPEC.name] = DRIVER
+SPEC.loader.exec_module(DRIVER)
+
+
+def environment(tmp_path):
+    token = tmp_path / "token"
+    token.write_text("opaque-test-value\n", encoding="utf-8")
+    token.chmod(0o600)
+    return {
+        "AUDIT_T02_BASE_URL": "http://audit-core.test:8080",
+        "AUDIT_T02_TOKEN_FILE": str(token),
+        "AUDIT_T02_TENANT": "tenant:trial:recovery-20260822",
+        "AUDIT_T02_SOURCE": "audit-core-recovery",
+        "AUDIT_T02_FIXTURE_ID": "audit-t02-recovery-20260822",
+        "AUDIT_T02_OCCURRED_AT": "2026-08-22T18:30:00Z",
+        "AUDIT_T02_MAX_ATTEMPTS": "4",
+        "AUDIT_T02_RETRY_INTERVAL_SECONDS": "0",
+    }
+
+
+def test_baseline_returns_exact_value_safe_shape(tmp_path):
+    result = DRIVER.run_phase(
+        "baseline",
+        DRIVER.CONTRACT_ID,
+        env=environment(tmp_path),
+        requester=lambda config: (202, "accepted"),
+    )
+    assert result == {
+        "contract_id": DRIVER.CONTRACT_ID,
+        "fixture_id": "audit-t02-recovery-20260822",
+        "status": "ready",
+        "secret_values_observed": False,
+    }
+
+
+def test_unavailable_retries_only_retryable_statuses(tmp_path):
+    replies = iter([(500, None), (503, None)])
+    result = DRIVER.run_phase(
+        "expect-unavailable",
+        DRIVER.CONTRACT_ID,
+        env=environment(tmp_path),
+        requester=lambda config: next(replies),
+    )
+    assert result["status"] == "retryable_unavailable"
+    assert result["http_status"] == 503
+    assert result["attempts"] == 2
+    assert set(result) == {
+        "contract_id", "fixture_id", "status", "http_status", "attempts",
+        "secret_values_observed",
+    }
+
+
+@pytest.mark.parametrize("status", [400, 401, 403, 409])
+def test_terminal_status_is_never_retried(tmp_path, status):
+    attempts = 0
+
+    def terminal(config):
+        nonlocal attempts
+        attempts += 1
+        return status, None
+
+    with pytest.raises(DRIVER.DriverError, match="terminal HTTP status"):
+        DRIVER.run_phase(
+            "expect-unavailable",
+            DRIVER.CONTRACT_ID,
+            env=environment(tmp_path),
+            requester=terminal,
+        )
+    assert attempts == 1
+
+
+def test_recovered_reports_duplicate_after_retry(tmp_path):
+    replies = iter([(503, None), (200, "duplicate")])
+    result = DRIVER.run_phase(
+        "expect-recovered",
+        DRIVER.CONTRACT_ID,
+        env=environment(tmp_path),
+        requester=lambda config: next(replies),
+    )
+    assert result["status"] == "duplicate"
+    assert result["http_status"] == 200
+    assert result["attempts"] == 2
+
+
+def test_cleanup_needs_no_bearer_and_retains_immutable_fixture(tmp_path):
+    env = {"AUDIT_T02_FIXTURE_ID": "audit-t02-recovery-20260822"}
+    result = DRIVER.run_phase("cleanup", DRIVER.CONTRACT_ID, env=env)
+    assert result == {
+        "contract_id": DRIVER.CONTRACT_ID,
+        "fixture_id": "audit-t02-recovery-20260822",
+        "status": "clean",
+        "secret_values_observed": False,
+    }
+
+
+def test_private_mounted_bearer_is_required(tmp_path):
+    env = environment(tmp_path)
+    Path(env["AUDIT_T02_TOKEN_FILE"]).chmod(0o644)
+    config = DRIVER.load_config(env)
+    with pytest.raises(DRIVER.DriverError, match="private and regular"):
+        DRIVER.request_event(config)
+
+
+def test_payload_is_synthetic_and_never_contains_bearer(tmp_path):
+    env = environment(tmp_path)
+    config = DRIVER.load_config(env)
+    payload = DRIVER.event_payload(config)
+    assert payload["data"] == {
+        "exercise": "RAILIANCE-WP-0024-T02",
+        "synthetic": True,
+    }
+    assert "opaque-test-value" not in repr(payload)
+
+
+def test_wrong_contract_is_refused_without_reading_token(tmp_path):
+    env = {"AUDIT_T02_FIXTURE_ID": "audit-t02-recovery-20260822"}
+    with pytest.raises(DRIVER.DriverError, match="not approved"):
+        DRIVER.run_phase("cleanup", "some-other-contract", env=env)