feat(model-grounded-detection): the safe-shape test, without which the model arbitrates nothing

The first live run produced the finding that shapes this design: taint
reachability alone does not distinguish a fix from a bug. On the canonical SQL
pair the flow request.args["name"] -> name -> execute exists on BOTH sides --

  vuln   db.execute("select ... '" + name + "'")   one argument, interpolated
  fixed  db.execute("select ... = %s", (name,))    two arguments, bound

-- so the model entailed the fixed twin exactly as readily as the vulnerable
one, and arbitrated nothing. What separates them is the SHAPE of the sink call,
which is precisely what safe_shape_sinks preserved from the earlier sanitizer
fix. The query now reports sinkArity and sinkArg0Literal, and a flow ending at
a sink written in its declared-safe form is not a vulnerability.

Two bugs in that shape test, both found only against the real engine:
Joern's `call.argument` includes the RECEIVER at argumentIndex 0, so `db` was
counted as an argument -- making a one-argument interpolated call look like a
two-argument bound one, the exact inversion of the test -- and the "first
argument" whose literalness decides a format-string sink was the receiver
rather than the format. Both fixed with argumentIndexGt(0).

The rule is narrow on purpose: only sinks named in safe_shape_sinks are subject
to it (os.system(x) is not safe merely because it takes one argument), and a
row carrying no shape fields is judged on the flow alone -- an unknown shape
must never be read as a safe one.

End to end on the pair, deterministic and with no model call:
  vuln  -> model_entailed  request.args["name"] -> db.execute(...) (line 4)
  fixed -> None

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG

.kiro/specs/model-grounded-detection/tasks.md

@@ -135,7 +135,7 @@ queries in group 4). The gates, `redaction.py`, `pairs.py` overlay/inventory pat
 
 ## Group 2 — The Joern spike on one family (go/no-go)
 
-- [ ] 2.1 The CpgBackend seam, out of process
+- [x] 2.1 The CpgBackend seam, out of process
   - `src/openultrasast/cpg/capability.py::has_cpg()` — Joern on PATH (probe `joern-parse --help`), the
     `has_semantic_extra` pattern exactly. `cpg/backend.py`: `CpgResult` (a built `cpg.bin` path + a `run(query,
     params)->JSON` callable); `JoernBackend` (subprocess: `joern-parse <root>` then `joern --script
@@ -148,7 +148,7 @@ queries in group 4). The gates, `redaction.py`, `pairs.py` overlay/inventory pat
   - _Requirements: 4.1, 4.2, 4.3, 4.5, 11.3_
   - _Depends: 1.6_
 
-- [ ] 2.2 The injection taint query and its verdict
+- [x] 2.2 The injection taint query and its verdict
   - `cpg/queries/taint.sc` — a CPGQL script parameterised by a `TaintSpec` (sources, sinks, sanitizers) that
     emits, per candidate, the JSON `{sink, source, path, sanitized: bool}`. `model/taint.py::verdict(cpg,
     spec, candidate)` → `ENTAILED` when a source→sink path exists with `sanitized=false`, `CORROBORATED` when

src/openultrasast/cpg/queries/taint.sc

@@ -45,17 +45,31 @@
 
   val rows = sinkCalls.l.flatMap { sink =>
     // Data flows into the arguments; asking the call node itself finds nothing.
-    val flows = sink.argument.reachableByFlows(sourceNodes).l
+    // `call.argument` includes the RECEIVER at argumentIndex 0 (`db` in `db.execute(...)`), so counting it
+    // makes a one-argument interpolated call look like a two-argument bound one -- the exact inversion of the
+    // test. Real arguments start at index 1.
+    val realArgs = sink.argument.argumentIndexGt(0).l
+    val flows    = sink.argument.reachableByFlows(sourceNodes).l
+
+    // The SHAPE of the sink call, which is what distinguishes a fix from a bug when the fix is a safe form
+    // rather than a sanitizing call: `execute(sql, params)` binds where `execute(sql + x)` interpolates, and
+    // `printf("literal", x)` is safe where `printf(userFmt)` is not. The taint path is identical in both, so
+    // without these two fields the model entails a pair's fixed side exactly as readily as its vulnerable one.
+    val arity       = realArgs.size
+    val arg0Literal = realArgs.headOption.map(a => a.isLiteral).getOrElse(false)
+
     flows.map { flow =>
       val elements = flow.elements.map(_.code).l
       val sanitized = sanitizerNames.nonEmpty && elements.exists(code => sanitizerNames.exists(s => code.contains(s)))
       ujson.Obj(
-        "sink"       -> sink.code.take(200),
-        "sinkLine"   -> sink.lineNumber.getOrElse(-1).toString,
-        "sinkMethod" -> sink.method.name,
-        "source"     -> elements.headOption.getOrElse("").take(200),
-        "sanitized"  -> sanitized,
-        "length"     -> elements.size
+        "sink"            -> sink.code.take(200),
+        "sinkLine"        -> sink.lineNumber.getOrElse(-1).toString,
+        "sinkMethod"      -> sink.method.name,
+        "source"          -> elements.headOption.getOrElse("").take(200),
+        "sanitized"       -> sanitized,
+        "length"          -> elements.size,
+        "sinkArity"       -> arity,
+        "sinkArg0Literal" -> arg0Literal
       )
     }
   }

src/openultrasast/model/taint.py

@@ -26,6 +26,12 @@ def verdict(cpg: CpgResult, spec: TaintSpec, *, function: str = "") -> Verdict |
         {"sources": spec.sources, "sinks": spec.sinks, "sanitizers": spec.sanitizers, "function": function},
     )
     flows = _flows(rows, function=function)
+    # A sink whose *shape* is safe is the fix, not the bug. `execute(sql, params)` binds rather than
+    # interpolates and `printf("literal", x)` has a constant format string, so the flow that reaches them is
+    # not a vulnerability -- even though the taint path is identical to the vulnerable twin's. Dropping these
+    # is what lets the model distinguish a pair at all: measured on the injection slice, taint reachability
+    # alone entailed both sides of the canonical SQL pair and therefore arbitrated nothing.
+    flows = [flow for flow in flows if not _has_safe_shape(flow, spec)]
     if not flows:
         return None
     unsanitized = [flow for flow in flows if not flow["sanitized"]]
@@ -47,6 +53,8 @@ def _flows(rows: object, *, function: str) -> list[dict[str, object]]:
             continue
         kept.append({
             "sink": str(row.get("sink", "")),
+            "sinkArity": row.get("sinkArity"),
+            "sinkArg0Literal": row.get("sinkArg0Literal"),
             "sinkLine": str(row.get("sinkLine", "")),
             "sinkMethod": str(row.get("sinkMethod", "")),
             "source": str(row.get("source", "")),
@@ -56,6 +64,27 @@ def _flows(rows: object, *, function: str) -> list[dict[str, object]]:
     return kept
 
 
+def _has_safe_shape(flow: Mapping[str, object], spec: TaintSpec) -> bool:
+    """Does this flow end at a sink call written in the form the spec declares safe?
+
+    Only sinks named in ``safe_shape_sinks`` are subject to the test -- ``os.system(x)`` is not safe merely
+    because it takes one argument. A row that carries no shape fields (an engine that could not report arity)
+    is judged on the flow alone: unknown shape must never be read as a safe one.
+    """
+    if not spec.safe_shape_sinks:
+        return False
+    sink = str(flow.get("sink", ""))
+    if not any(name in sink for name in spec.safe_shape_sinks):
+        return False
+    arity = flow.get("sinkArity")
+    literal = flow.get("sinkArg0Literal")
+    if arity is None and literal is None:
+        return False  # shape unknown -> judge on the flow, never assume safety
+    bound = isinstance(arity, int | float) and int(arity) >= 2
+    constant_format = literal is True
+    return bound or constant_format
+
+
 def _as_int(value: object) -> int:
     return int(value) if isinstance(value, int | float) else 0
 

tests/test_model_taint.py

@@ -111,3 +111,74 @@ def test_the_spec_drives_the_query_parameters() -> None:
     assert seen["sinks"] == ("os.system", "execute")
     assert seen["sanitizers"] == ("escape", "quote")
     assert seen["function"] == "run"
+
+
+# --- the safe-shape test (task 2.3 finding) ------------------------------------------------------------
+#
+# Taint reachability alone does not distinguish a parameterised fix from an interpolated bug: the flow
+# `request.args["name"] -> name -> execute` exists on BOTH sides of the canonical SQL pair.
+#
+#     vuln   db.execute("select ... '" + name + "'")      one argument, interpolated
+#     fixed  db.execute("select ... = %s", (name,))       two arguments, bound
+#
+# What separates them is the SHAPE of the sink call, which is exactly what `TaintSpec.safe_shape_sinks`
+# preserves. Without this test the model entails both sides and arbitrates nothing.
+
+
+def _shape_spec():  # type: ignore[no-untyped-def]
+    from openultrasast.model.specs import TaintSpec
+
+    return TaintSpec(
+        family="injection",
+        language="python",
+        sources=("request.args",),
+        sinks=("execute",),
+        sanitizers=(),
+        safe_shape_sinks=("execute",),
+    )
+
+
+def test_a_parameterised_sink_call_is_not_a_vulnerability_even_though_the_flow_reaches_it() -> None:
+    from openultrasast.model.taint import verdict
+
+    rows = [{
+        "sink": 'db.execute("select ... = %s", (name,))', "sinkLine": "5", "sinkMethod": "run",
+        "source": "request.args['name']", "sanitized": False, "length": 4, "sinkArity": 2, "sinkArg0Literal": True,
+    }]
+    assert verdict(_cpg(rows), _shape_spec(), function="run") is None, "a bound query is the fix, not the bug"
+
+
+def test_an_interpolated_sink_call_of_the_same_name_is_entailed() -> None:
+    from openultrasast.model.ladder import Rung
+    from openultrasast.model.taint import verdict
+
+    rows = [{
+        "sink": 'db.execute("select ... \'" + name + "\'")', "sinkLine": "5", "sinkMethod": "run",
+        "source": "request.args['name']", "sanitized": False, "length": 4, "sinkArity": 1, "sinkArg0Literal": False,
+    }]
+    answer = verdict(_cpg(rows), _shape_spec(), function="run")
+    assert answer is not None and answer.rung is Rung.ENTAILED
+
+
+def test_the_shape_test_only_applies_to_sinks_declared_safe_in_that_shape() -> None:
+    """`os.system(x)` with one argument is not "safe because it has one argument" -- the rule is per sink."""
+    from openultrasast.model.ladder import Rung
+    from openultrasast.model.taint import verdict
+
+    rows = [{
+        "sink": "os.system(full)", "sinkLine": "7", "sinkMethod": "run",
+        "source": "request.args['cmd']", "sanitized": False, "length": 5, "sinkArity": 1, "sinkArg0Literal": False,
+    }]
+    answer = verdict(_cpg(rows), _spec(), function="run")  # os.system is not in safe_shape_sinks
+    assert answer is not None and answer.rung is Rung.ENTAILED
+
+
+def test_a_row_without_shape_fields_is_judged_on_the_flow_alone() -> None:
+    """Older rows, or an engine that could not report arity, must not be silently treated as safe."""
+    from openultrasast.model.ladder import Rung
+    from openultrasast.model.taint import verdict
+
+    rows = [{"sink": "db.execute(q)", "sinkLine": "5", "sinkMethod": "run", "source": "request.args['n']",
+             "sanitized": False, "length": 3}]
+    answer = verdict(_cpg(rows), _shape_spec(), function="run")
+    assert answer is not None and answer.rung is Rung.ENTAILED