feat(hardening): Phase 15 — redaction, retry/backoff, budgets, threat-model docs

Release-readiness hardening.

- redaction.py (zero-dep): redact_secrets masks provider API keys, AWS/GitHub/Slack/
  Google tokens, bearer tokens, URL-embedded credentials, PEM private keys, and
  sensitive key=value assignments. Wired into the trace writer and the markdown report
  (on by default), so scan artifacts never leak a credential quoted from source.
- provider/openrouter.py: call_with_retry — exponential backoff on transient failures
  (HTTP 429/5xx, connection errors, timeouts); non-transient errors fail fast. Both the
  chat and embedding clients use it (max_attempts/retry_base_delay configurable).
- config [hardening]: redact_secrets (default true) + max_findings (bounded-CI output
  cap, 0 = unlimited; truncation is severity-ordered and disclosed as a `budget`
  degradation). Agentic spend stays bounded by [harnessx] max_cost_usd/token_threshold.
- docs/threat-model.md (trust boundaries, controls, sandbox limits) + docs/examples.md
  (six end-to-end walkthroughs); README "Security & hardening" section links both.
- CI lint/typecheck/test/detection-gate/dagger gates already existed.

tests/test_redaction.py, test_provider_retry.py, test_hardening_pipeline.py + config
tests. 206 passed, 4 skipped; ruff + mypy clean; detection gate PASS.

Co-Authored-By: Claude Opus 4.8 <[email protected]>

src/openultrasast/cli.py

@@ -149,7 +149,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut
     runtime = HarnessRuntime(
         scan_id=run.scan_id,
         config=config,
-        trace_writer=HarnessTraceWriter(run.root / "trace" / "events.jsonl"),
+        trace_writer=HarnessTraceWriter(run.root / "trace" / "events.jsonl", redact=config.hardening.redact_secrets),
     )
     write_harness_config(config=config, processors=[], contract_mode="strict", path=run.root / "harness.json")
     runtime.start(mode=mode, target=run.target)
@@ -219,6 +219,14 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut
     if shadow_findings:
         findings = [finding for finding in findings if finding.status != "shadow"]
         write_findings(shadow_findings, run.root / "shadow_findings.json")
+    # Bounded-CI budget: cap the reported finding count (0 = unlimited). Findings are
+    # severity/priority-sorted, so truncation keeps the most severe and is disclosed.
+    max_findings = config.hardening.max_findings
+    if max_findings and len(findings) > max_findings:
+        runtime.state["degradations"].append(
+            {"stage": "budget", "reason": "max_findings_exceeded", "requested": max_findings, "actual": len(findings)}
+        )
+        findings = findings[:max_findings]
     findings_path = run.root / "findings.json"
     verification_path = run.root / "verification.json"
     markdown_path = run.root / "report.md"
@@ -291,7 +299,9 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut
     )
     score_path = run.root / "score.json"
     score_path.write_text(json.dumps(score_artifact.to_dict(), indent=2, sort_keys=True) + "\n")
-    runtime.run_stage("report", lambda: write_markdown_report(findings, markdown_path, verifications))
+    runtime.run_stage(
+        "report", lambda: write_markdown_report(findings, markdown_path, verifications, redact=config.hardening.redact_secrets)
+    )
     runtime.run_stage("sarif", lambda: write_sarif_report(findings, verifications, sarif_path))
     runtime.run_stage(
         "manifest",

src/openultrasast/config.py

@@ -77,6 +77,14 @@ class HarnessxConfig:
     token_threshold: int = 120_000
 
 
+@dataclass(frozen=True)
+class HardeningConfig:
+    # Release-readiness controls. redact_secrets masks credentials in traces/reports;
+    # max_findings caps the reported finding count for bounded CI runs (0 = unlimited).
+    redact_secrets: bool = True
+    max_findings: int = 0
+
+
 @dataclass(frozen=True)
 class FusionConfig:
     # Two-panel fusion adjudication for triggered findings (standard mode). Runs
@@ -100,6 +108,7 @@ class ResolvedConfig:
     ruleset: RulesetConfig = RulesetConfig()
     harnessx: HarnessxConfig = HarnessxConfig()
     fusion: FusionConfig = FusionConfig()
+    hardening: HardeningConfig = HardeningConfig()
     runs_dir: str = ".openultrasast/runs"
 
 
@@ -120,6 +129,7 @@ def load_config(config_path: Path | None = None) -> ResolvedConfig:
         ruleset=_load_ruleset_config(data.get("ruleset", {})),
         harnessx=_load_harnessx(data.get("harnessx", {})),
         fusion=_load_fusion(data.get("fusion", {})),
+        hardening=_load_hardening(data.get("hardening", {})),
         runs_dir=os.environ.get("OPENULTRASAST_RUNS_DIR", ".openultrasast/runs"),
     )
 
@@ -223,6 +233,14 @@ def _load_harnessx(value: object) -> HarnessxConfig:
     )
 
 
+def _load_hardening(value: object) -> HardeningConfig:
+    data = _section(value)
+    return HardeningConfig(
+        redact_secrets=bool(data.get("redact_secrets", True)),
+        max_findings=_int_value(data.get("max_findings"), 0),
+    )
+
+
 def _load_fusion(value: object) -> FusionConfig:
     data = _section(value)
     return FusionConfig(

src/openultrasast/harness.py

@@ -62,13 +62,19 @@ class SerializedHarnessConfig:
 
 
 class HarnessTraceWriter:
-    def __init__(self, path: Path) -> None:
+    def __init__(self, path: Path, *, redact: bool = True) -> None:
         self.path = path
+        self.redact = redact
         self.path.parent.mkdir(parents=True, exist_ok=True)
 
     def append(self, event: HarnessEvent) -> None:
+        line = json.dumps(asdict(event), sort_keys=True)
+        if self.redact:
+            from .redaction import redact_secrets
+
+            line = redact_secrets(line)
         with self.path.open("a") as handle:
-            handle.write(json.dumps(asdict(event), sort_keys=True) + "\n")
+            handle.write(line + "\n")
 
 
 class HarnessRuntime:

src/openultrasast/provider/openrouter.py

@@ -3,19 +3,56 @@ from __future__ import annotations
 import json
 import os
 import re
+import time
 import urllib.error
 import urllib.request
+from collections.abc import Callable
 from dataclasses import dataclass
+from typing import TypeVar, cast
 
 
 class OpenRouterError(RuntimeError):
     """Raised when an OpenRouter request or response is invalid."""
 
 
+_TRANSIENT_STATUS = {429, 500, 502, 503, 504}
+_T = TypeVar("_T")
+
+
+def _is_transient(exc: Exception) -> bool:
+    """A network failure worth retrying (rate-limit, 5xx, connection error, timeout)."""
+    if isinstance(exc, urllib.error.HTTPError):
+        return exc.code in _TRANSIENT_STATUS
+    return isinstance(exc, urllib.error.URLError | TimeoutError)
+
+
+def call_with_retry(
+    operation: Callable[[], _T],
+    *,
+    attempts: int = 3,
+    base_delay: float = 0.5,
+    sleep: Callable[[float], None] = time.sleep,
+) -> _T:
+    """Run ``operation`` with exponential backoff on transient network failures."""
+    last: Exception | None = None
+    for attempt in range(attempts):
+        try:
+            return operation()
+        except Exception as exc:  # noqa: BLE001 — retry only transient failures; re-raise the rest
+            last = exc
+            if attempt + 1 < attempts and _is_transient(exc):
+                sleep(base_delay * (2**attempt))
+                continue
+            raise
+    raise last if last is not None else OpenRouterError("retry exhausted")
+
+
 @dataclass(frozen=True)
 class OpenRouterChatClient:
     api_key: str
     base_url: str = "https://openrouter.ai/api/v1"
+    max_attempts: int = 3
+    retry_base_delay: float = 0.5
 
     @classmethod
     def from_env(cls) -> OpenRouterChatClient:
@@ -35,9 +72,13 @@ class OpenRouterChatClient:
             },
             method="POST",
         )
-        try:
+
+        def _do() -> dict[str, object]:
             with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
-                response_payload = json.loads(response.read().decode())
+                return cast(dict[str, object], json.loads(response.read().decode()))
+
+        try:
+            response_payload = call_with_retry(_do, attempts=self.max_attempts, base_delay=self.retry_base_delay)
         except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc:
             raise OpenRouterError(f"OpenRouter chat request failed: {exc}") from exc
 
@@ -49,6 +90,8 @@ class OpenRouterChatClient:
 class OpenRouterEmbeddingClient:
     api_key: str
     base_url: str = "https://openrouter.ai/api/v1"
+    max_attempts: int = 3
+    retry_base_delay: float = 0.5
 
     @classmethod
     def from_env(cls) -> OpenRouterEmbeddingClient:
@@ -68,9 +111,13 @@ class OpenRouterEmbeddingClient:
             },
             method="POST",
         )
-        try:
+
+        def _do() -> dict[str, object]:
             with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
-                response_payload = json.loads(response.read().decode())
+                return cast(dict[str, object], json.loads(response.read().decode()))
+
+        try:
+            response_payload = call_with_retry(_do, attempts=self.max_attempts, base_delay=self.retry_base_delay)
         except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc:
             raise OpenRouterError(f"OpenRouter embedding request failed: {exc}") from exc
         return parse_embedding_response(response_payload)

src/openultrasast/redaction.py

@@ -0,0 +1,45 @@
+"""Secret redaction for traces and reports (Phase 15 hardening).
+
+Scan artifacts (trace events, reports) can quote source code that contains live
+credentials. This module masks well-known secret shapes before they are persisted, so
+a finding's evidence never leaks a key into an artifact an analyst later shares.
+Conservative by design: it targets recognizable secret formats, not arbitrary text.
+"""
+
+from __future__ import annotations
+
+import re
+
+REDACTED = "***REDACTED***"
+
+# (compiled pattern, replacement). Replacements keep any non-secret prefix group so the
+# redaction is legible (e.g. `api_key = ***REDACTED***`, `Authorization: Bearer ***REDACTED***`).
+_PATTERNS: tuple[tuple[re.Pattern[str], str], ...] = (
+    # PEM private key blocks.
+    (re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----.*?-----END [A-Z ]*PRIVATE KEY-----", re.DOTALL), REDACTED),
+    # Authorization: Bearer <token>
+    (re.compile(r"(Bearer\s+)[A-Za-z0-9._\-]{12,}"), r"\1" + REDACTED),
+    # Provider API keys: OpenAI/Anthropic sk-..., AWS AKIA..., GitHub gh*_..., Slack xox*, Google AIza...
+    (re.compile(r"\bsk-[A-Za-z0-9._\-]{16,}"), REDACTED),
+    (re.compile(r"\bAKIA[0-9A-Z]{16}\b"), REDACTED),
+    (re.compile(r"\bgh[pousr]_[A-Za-z0-9]{20,}"), REDACTED),
+    (re.compile(r"\bxox[baprs]-[A-Za-z0-9-]{10,}"), REDACTED),
+    (re.compile(r"\bAIza[A-Za-z0-9_\-]{20,}"), REDACTED),
+    # Credentials embedded in a URL: scheme://user:secret@host
+    (re.compile(r"([a-zA-Z][a-zA-Z0-9+.\-]*://[^\s:/@]+:)[^\s:/@]{3,}(@)"), r"\1" + REDACTED + r"\2"),
+    # Generic `key = "value"` assignments for sensitive names.
+    (
+        re.compile(r"(?i)\b(api[_-]?key|secret|token|password|passwd|pwd|access[_-]?key)\b(\s*[=:]\s*)(['\"]?)[A-Za-z0-9/+_\-]{8,}\3"),
+        r"\1\2\3" + REDACTED + r"\3",
+    ),
+)
+
+
+def redact_secrets(text: str) -> str:
+    """Return ``text`` with recognizable secrets masked. Idempotent and pattern-scoped."""
+    for pattern, replacement in _PATTERNS:
+        text = pattern.sub(replacement, text)
+    return text
+
+
+__all__ = ["REDACTED", "redact_secrets"]

src/openultrasast/reports.py

@@ -9,7 +9,9 @@ from .run import ScanRun
 from .verification import VerificationResult
 
 
-def write_markdown_report(findings: list[StaticFinding], path: Path, verifications: list[VerificationResult] | None = None) -> None:
+def write_markdown_report(
+    findings: list[StaticFinding], path: Path, verifications: list[VerificationResult] | None = None, *, redact: bool = True
+) -> None:
     path.parent.mkdir(parents=True, exist_ok=True)
     verification_by_id = _verification_by_id(verifications or [])
     lines = ["# OpenUltraSAST Report", "", f"Findings: {len(findings)}", ""]
@@ -38,7 +40,12 @@ def write_markdown_report(findings: list[StaticFinding], path: Path, verificatio
                 "",
             ]
         )
-    path.write_text("\n".join(lines).rstrip() + "\n")
+    report = "\n".join(lines).rstrip() + "\n"
+    if redact:
+        from .redaction import redact_secrets
+
+        report = redact_secrets(report)
+    path.write_text(report)
 
 
 def write_sarif_report(findings: list[StaticFinding], verifications: list[VerificationResult], path: Path) -> None: