feat(whitebox): code-graph taint engine + graph-grounded AI triage

.github/workflows/ci.yml

@@ -84,7 +84,7 @@ jobs:
       - name: Byte-compile
         run: python3 -m compileall -q bundle/offat_bundle triager/offat_triage whitebox/offat_wb graybox/offat_gb platform/offat_platform reporter/offat_report
       - name: Install packages
-        run: pip install ./bundle ./triager ./whitebox ./graybox ./platform ./reporter jsonschema pyyaml
+        run: pip install ./bundle ./triager './whitebox[graph]' ./graybox ./platform ./reporter jsonschema pyyaml
       - name: Bundle unit tests
         run: PYTHONPATH=bundle python3 -m unittest discover -s bundle/tests -p 'test_*.py'
       - name: Triager unit tests

docs/ai-roadmap.md

@@ -0,0 +1,144 @@
+# OFFAT-AI: code-graph white-box and AI roadmap
+
+Status: proposal + first implementation (`whitebox/offat_wb/cpg/`), October 2026.
+
+## 1. Where AI sits in OFFAT today
+
+| Stage | What the model does | Limitation |
+|---|---|---|
+| `whitebox` hunt | nothing (regex + semgrep wrapper) | per-line matches, no data flow; noisy |
+| `whitebox` trace | nothing (graft `callers` by *name*, "looks like a handler" regex) | name collisions, no proof the *data* reaches the sink |
+| triage (`offat_triage`) | batched screen + tiered verify of each finding | sees an 800-char snippet; cannot look anything up |
+| `graybox` analyze | same triager, reachable-only | reachability = graft names or "same file as a route" |
+| `platform` test-gen | proposes extra payloads per endpoint (700 tokens) | no idea which sink sits behind the param |
+
+The model is asked to judge exploitability without the one artifact that decides
+it: the path from attacker input to the sink and what happens along it.
+
+## 2. What this change adds
+
+**A code-graph taint engine replaces semgrep as the white-box core.**
+
+- `cpg/graph.py`: a networkx `MultiDiGraph` code property graph. Nodes are
+  functions, params, *versioned* locals, `self` fields, call sites, args, returns
+  and sources. Edges are `flow` (with per-class `sanitizes`), `call`/`ret`
+  labelled by call site, and `calls`. Exportable as JSON or GraphML (Gephi, yEd,
+  Neo4j import).
+- `cpg/py_frontend.py`: stdlib `ast`. Covers Flask, FastAPI, Django/DRF,
+  Tornado, aiohttp and Lambda, `src/` layouts, nested functions, and deferred
+  calls (`queue.enqueue(fn, x)`, `fn.delay(x)`, `Thread(target=fn)`,
+  executors, `BackgroundTasks`).
+- `cpg/js_frontend.py`: tree-sitter JS/TS. Covers Express routers including
+  handler factories and wrappers, `require`/`import`/`export`, TS destructured
+  `req`, and callbacks.
+- `cpg/taint.py`: an interprocedural BFS with a call-site stack, so calls and
+  returns stay matched (`f(tainted)` does not leak into `f("const")`), plus
+  per-class sanitizers and reaching definitions (`x = int(x)` kills taint).
+  BOLA candidates are suppressed when the lookup is scoped to the caller or an
+  ownership check follows. Mass assignment (API3) fires only on whole-body
+  sources.
+- Pattern hits are fused, not discarded: a regex hit on a proven line is
+  dropped as a duplicate. A regex injection hit with no source is kept but
+  demoted, and gains graph reachability.
+- SARIF now carries `codeFlows`, so GitHub code scanning renders the path.
+
+**Three AI features built on the graph:**
+
+1. **Graph-grounded triage**: the batch prompt receives the proven hop-by-hop
+   path, routes and authz context instead of a snippet.
+2. **Graph-tool agent** (`--agent N`, `cpg/agent.py`): a tool-use loop with
+   read-only, repo-confined tools (`get_function`, `callers`, `callees`,
+   `read_lines`, `flows_into`). It tries to refute first, then submits a
+   verdict and a PoC request. Repo content is framed as untrusted data, since
+   comments are a prompt-injection channel. It uses `claude-opus-5-5` with
+   server-side refusal fallback.
+3. **Spec inference** (`--infer-specs`, `cpg/infer.py`): the *taint frontier*
+   is the set of unknown callees that receive attacker data. The model labels
+   each one sink, sanitizer or propagator once per API, and the result is
+   written to `inferred-specs.json` and fed back via `--specs`. This targets
+   the main source of misses (in-house wrappers) and of noise (lookups such as
+   `authenticatedUsers.get(token)` that return server-owned data).
+
+Also fixed: the default screen model `claude-haiku-5` is not a valid model ID,
+so batched screening always failed over to per-finding calls. It is now
+`claude-haiku-5-5`.
+
+## 3. Benchmarks (no AI, graph engine only)
+
+| Target | Size | Time | Entry points | Flows | Notes |
+|---|---|---|---|---|---|
+| fixture (`whitebox/tests/fixtures/taint_app`) | Flask+FastAPI+Express | <1s | 29 | 16 | 16/16 planted vulns found, 0/12 safe twins flagged |
+| OWASP Juice Shop (TS) | 4,260 fns | 6.3s | 91 | 56 (17 in intentionally vulnerable `codefixes/`) | login + search SQLi, SSRF, open redirect, 3× `$where`/Mongo NoSQLi, eval + pug SSTI, basket IDOR, YAML deser; every finding mapped to its route |
+| we45 Vulnerable-Flask-App | 969 fns | 1.1s | 11 | 4 | 4/5 taint-style vulns, 0 FP; miss: SSTI via `str(exception)` |
+| dvpwa (aiohttp) | 978 fns | 1.3s | 8 | 1 | the SQLi |
+| pygoat (Django) | 293 fns | 0.5s | 131 | 25 | SQLi, cmdi, eval, pickle/yaml, SSRF, path traversal, XSS |
+
+Known gaps: taint through exceptions, key-sensitive containers (dict fields),
+second-order (DB write -> read), and `vm` sandbox contexts.
+
+## 4. Roadmap
+
+### P0: close the loop between white-box and DAST
+
+- **Graph finding -> targeted test case.** Each graph finding already has a
+  route, a param and a sink class (plus a PoC from the agent). Emit these into
+  the Bundle as test cases so the Rust engine confirms them against an
+  authorized deployment, and mark them `confirmed-dynamic`. Today test-gen
+  guesses vectors from parameter *names*.
+- **Sink-aware payload generation.** Give `platform/testgen._augment_with_ai`
+  the sink code at the end of the path, not just the endpoint (SQL dialect,
+  shell vs argv, template engine, `$where` vs operator smuggling).
+- **Agent verification for DAST findings.** For a dynamic hit, start the agent
+  from the route's handler node in the graph so it can explain *why* the
+  response looked vulnerable, or refute it.
+
+### P1: precision and coverage where AI is the right tool
+
+- **Authorization model extraction.** Collect per-route decorators/middleware
+  from the graph (`login_required`, `isAuthorized()`, `Depends(current_user)`)
+  into an auth matrix. Let the model flag outliers ("every `/api/Cards/*`
+  route has `isAuthorized` except this one") and feed BOLA/BFLA test-gen with
+  `--identities`.
+- **Business-logic hypotheses.** Hand the model graph slices of
+  state-changing handlers (checkout, coupon, wallet, transfer) and ask for
+  multi-step abuse cases (negative quantities, replayed coupons, race windows),
+  then emit PRG chains.
+- **Second-order sources.** Treat data read back from the DB, and, for
+  security tools and API clients, *responses from the scanned server*, as
+  sources. That would have caught the report XSS in §5.
+- **Key-sensitive containers.** Track `d["field"]` separately and use the
+  agent to refute container over-taint (e.g. `post_test_processor.py:189`).
+- **Shared spec registry.** Cache AI-inferred specs per package and version
+  and share them across runs and projects, with human review.
+- **PR / diff mode.** Slice the graph to changed functions plus everything
+  they reach. Comment with SARIF code flows and the agent's verdict.
+
+### P2: platform and safety of the AI layer
+
+- **Expose the graph to agents over MCP.** The repo already ships Claude Code
+  skills and a plugin. An `offat-graph` MCP server (`callers`, `slice`,
+  `paths_to(sink)`) would let interactive agents use the same graph.
+- **Sanitizer validation by execution.** The model writes a harness, a sandbox
+  runs the project's custom validator against a payload corpus, and the
+  interpreter, not the model, decides whether it holds.
+- **Cost controls.** Prompt-cache the system prompt and tools across agent
+  turns, use the Message Batches API for nightly runs, and set a per-run token
+  budget.
+- **Eval harness.** Juice Shop, pygoat, VFA, dvpwa and OWASP Benchmark with
+  expected findings: track graph precision/recall and AI verdict accuracy per
+  release, including prompt-injection canaries planted in code comments.
+
+## 5. Self-scan of OFFAT
+
+Running the engine on this repository proves flows from `POST /api/v1/scan`
+through the RQ job queue into the classic tester (the job function is reached via
+`task_queue.enqueue(scan_api, ...)`, which the deferred-call modelling resolves).
+Security-relevant results have been reported privately to the maintainers.
+Functional bugs surfaced on the way:
+
+- `src/offat/api/jobs.py:16` calls `generate_and_run_tests()` without the required
+  `http_methods` argument (since `acd733b`), so every REST-API scan job errors out.
+- A non-URL `openapi` string is passed as `spec=` and `BaseParser` calls `.get` on a
+  `str`, so inline specs via the API cannot work.
+- `report/generator.py:47` appends each result once per key (duplicated rows).
+- `post_test_processor.py:184`: `HEADER_REGEX_FILTER` matches the body, not headers.

graybox/offat_gb/analyze.py

@@ -41,6 +41,11 @@ def _gb_block(idx: int, f: Dict[str, Any]) -> str:
     parts.append("reachable_from=" + (", ".join(rf) if rf else "none"))
     parts.append(f"detector_confidence={f.get('confidence', 0)}")
     head = " ".join(parts)
+    path = (f.get("evidence") or {}).get("path") or []
+    if path:   # proven on the code graph: the hops are better evidence than a snippet
+        return head + "\npath:\n" + "\n".join(
+            f"  {h.get('file')}:{h.get('line')} [{h.get('kind')}] {str(h.get('code', ''))[:160]}"
+            for h in path[:20])
     code = str(f.get("code", "") or "")[:800]
     return head + (f"\ncode:\n{code}" if code else "")
 

graybox/offat_gb/cli.py

@@ -18,7 +18,7 @@
 from typing import Any, Dict, List, Optional
 
 from offat_triage import enrich_threats
-from offat_wb import hunt, recon
+from offat_wb import cpg, hunt, recon
 from offat_wb.tools import ToolStatus, have
 
 from . import analyze, graph_map, reachability, report
@@ -37,7 +37,7 @@ def _dedupe(findings: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
 
 def run_pipeline(target: str, classes: Optional[List[str]], *, use_ai: bool, use_semgrep: bool,
                  use_graft: bool, use_cache: bool, out_dir: str,
-                 provider: Optional[str] = None) -> Dict[str, Any]:
+                 provider: Optional[str] = None, use_graph: bool = True) -> Dict[str, Any]:
     status = ToolStatus()
 
     print(f"[recon] scanning {target} ...")
@@ -49,25 +49,37 @@ def run_pipeline(target: str, classes: Optional[List[str]], *, use_ai: bool, use
     print(f"[map] {len(endpoints)} endpoint(s) discovered "
           f"({'graft' if status.available.get('graft_graph') else 'native'})")
 
+    graph = None
+    graph_findings: List[Dict[str, Any]] = []
+    if use_graph and cpg.AVAILABLE:
+        print("[graph] code graph + source-to-sink taint search ...")
+        res = cpg.analyze(target, classes=classes)
+        graph, graph_findings = res.graph, res.findings
+        status.available["code_graph"] = True
+        print(f"[graph] {len(graph_findings)} proven source-to-sink flow(s)")
+
     print("[hunt] built-in pattern hunter ...")
     findings = hunt.builtin_hunt(target, classes)
     if use_semgrep and have("semgrep"):
         status.available["semgrep"] = True
-        print("[hunt] semgrep ...")
+        print("[hunt] semgrep (opt-in) ...")
         findings += hunt.semgrep_hunt(target)
     else:
-        status.available["semgrep"] = have("semgrep")
-        if not have("semgrep"):
-            status.skip("semgrep", "not installed")
+        status.available["semgrep"] = False
     findings += recon_info.pop("dependency_findings", [])
     findings = _dedupe(findings)
+    if graph is not None:
+        findings = cpg.merge_with_patterns(graph, graph_findings, findings)
     print(f"[hunt] {len(findings)} candidate findings")
 
     # Threat mapping: OWASP API Top 10 (2023) + OWASP Top 10 (2021) + CWE.
     enrich_threats(findings)
 
-    print("[reach] linking findings to endpoints ...")
-    findings = reachability.link(target, findings, endpoints, status)
+    if graph is None:
+        print("[reach] linking findings to endpoints ...")
+        findings = reachability.link(target, findings, endpoints, status)
+    else:
+        print("[reach] reachability from the code graph (call graph -> route handlers)")
     reachable = sum(1 for f in findings if f.get("reachable"))
     print(f"[reach] {reachable}/{len(findings)} findings reachable from a mapped endpoint")
 
@@ -109,7 +121,10 @@ def main(argv: Optional[List[str]] = None) -> int:
     ap.add_argument("--no-ai", action="store_true", help="use heuristic triage only (no tokens)")
     ap.add_argument("--provider", default=None,
                     help="AI backend: auto (default), anthropic, claude-code, codex, heuristic")
-    ap.add_argument("--no-semgrep", action="store_true", help="skip semgrep even if installed")
+    ap.add_argument("--semgrep", action="store_true", help="also run semgrep (opt-in)")
+    ap.add_argument("--no-semgrep", action="store_true", help=argparse.SUPPRESS)
+    ap.add_argument("--no-graph", action="store_true",
+                    help="skip the code-graph engine (falls back to graft / same-file reachability)")
     ap.add_argument("--no-graft", action="store_true", help="skip graft; map endpoints natively")
     ap.add_argument("--no-cache", action="store_true", help="do not read/write the AI-verdict cache")
     ap.add_argument("--fail-on", default="",
@@ -124,9 +139,9 @@ def main(argv: Optional[List[str]] = None) -> int:
     classes = [c.strip() for c in args.classes.split(",") if c.strip()] or None
 
     rep = run_pipeline(
-        target, classes, use_ai=not args.no_ai, use_semgrep=not args.no_semgrep,
+        target, classes, use_ai=not args.no_ai, use_semgrep=args.semgrep and not args.no_semgrep,
         use_graft=not args.no_graft, use_cache=not args.no_cache, out_dir=args.out,
-        provider=args.provider,
+        provider=args.provider, use_graph=not args.no_graph,
     )
 
     if args.fail_on:

skills/offat-whitebox/SKILL.md

@@ -11,15 +11,17 @@ description: >
 # OFFAT-AI — White-box (SAST)
 
 Drive the Python pipeline that reviews source code: detects the stack, hunts
-vulnerabilities (built-in multi-language pattern hunter + semgrep when present),
+vulnerabilities (a code-graph taint engine that proves source -> sink paths, plus a
+pattern hunter for secrets/crypto/misconfig; semgrep only with `--semgrep`),
 chains co-located issues, triages them (AI or heuristic) and writes reports.
 
 Scope is **owned or authorized code only** (no live testing of third parties).
 
 ## Steps
 
 1. **Point at the codebase** (a local path). Deepen results by installing the
-   optional scanners when available: `semgrep`, `syft`, `grype` (all optional —
+   graph extra (`pip install './whitebox[graph]'`) and the optional scanners `syft`,
+   `grype` (all optional —
    the pipeline runs without them and notes what was skipped).
 
 2. **Run the pipeline:**

triager/offat_triage/batch.py

@@ -30,7 +30,7 @@
     make_triager,
 )
 
-SCREEN_MODEL = os.environ.get("OFFAT_SCREEN_MODEL", "claude-haiku-5")
+SCREEN_MODEL = os.environ.get("OFFAT_SCREEN_MODEL", "claude-haiku-5-5")
 VERIFY_MODEL = os.environ.get("OFFAT_VERIFY_MODEL", "claude-sonnet-5")
 BATCH_SIZE = int(os.environ.get("OFFAT_BATCH_SIZE", "8") or "8")
 

triager/offat_triage/triager.py

@@ -196,6 +196,11 @@ def _build_prompt(f: Dict[str, Any]) -> str:
     if ev:
         lines.append("Evidence:")
         for k, v in ev.items():
+            if k == "path" and isinstance(v, list):
+                lines.append("- data flow (source -> sink, proven on the code graph):")
+                lines += [f"    {h.get('file')}:{h.get('line')} [{h.get('kind')}] {str(h.get('code', ''))[:160]}"
+                          for h in v[:25]]
+                continue
             if v not in (None, "", 0):
                 lines.append(f"- {k}: {str(v)[:800]}")
     if f.get("code"):