feat(model): a bound on the data lowers the rung, it does not clear it

contributor-scan 2.12. c/injection and c/memory carried zero sanitizers,
so no C flow could ever be discharged: every `argv -> strcpy` entailed
however carefully it was bounded.

Which sinks can be discharged this way is read from the sink fact's own
CWE rather than declared: CWE-121 is a stack buffer overflow, so
gets/sprintf/strcat/strcpy qualify. It is the only overflow CWE in the
fact tables and it is C-only, so no Python or JavaScript sink is touched
and the pair corpus is unaffected -- gates byte-identical.

Not a sanitizer list, because `strlen` appears in the guarded and the
unguarded case alike; the discharge is a guard, not a call.

**The first implementation required the bound to DOMINATE the sink, and it
found nothing.** libpng guards `strcpy(outname+len, ".png")` with
`(len = strlen(inname)) > 250`, but that check sits on an else-if chain
whose sibling branch skips it entirely, and the protection reaches the
sink through an `error` flag: `++error` in the guarded branch, `if
(!error)` around the copy. Joern's dominator set for that sink has 89
calls and correctly excludes the bound; `controlledBy` returns none
either. Verified against the CPG before the rule was changed rather than
after it failed.

So the bound is searched in the method, and the consequence is
corroborated rather than silence. That is the honest rung: a bound on this
variable means the graph has NOT shown the copy is unguarded, and it has
not shown it is guarded either -- off-by-one being how bounds fail. The
judge is asked. Proving the flag-mediated case needs path-sensitive
reasoning this arbiter does not do.

  libpng entailed   3 -> 1

The survivor is also a false positive: `strcpy(output + len, ".icc")` in
iccfrompng.c, where `output = malloc(len + 5)` makes it fit exactly. That
is a different discharge shape -- allocated to size, not bounds-checked --
and it is recorded as 2.13 rather than chased. Both shapes were found on
one repository, and extending a rule until that repository is clean is the
overfitting Req 8 exists to prevent; 2.13 requires a second C checkout
with a genuine CWE-121 so the change can be measured for what it silences
as well as what it clears.

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

src/openultrasast/cpg/queries/taint.sc

@@ -19,7 +19,8 @@
 //     closure lexically nested inside it -- see `nestedInLabeled`),
 //   parameterSources ("true" to treat the labeled function's parameters as untrusted -- see below),
 //   file (optional: restrict sinks to this source file),
-//   callDepth (optional: how many call-graph levels below the labeled method may hold a sink)
+//   callDepth (optional: how many call-graph levels below the labeled method may hold a sink),
+//   boundedSinks (optional: sinks whose danger is a LENGTH, which a dominating bound discharges)
 //
 // `file` is what makes a region with no enclosing function askable at all. Without it such a region sends
 //   function="" and matches every sink in the repository, and the driver attributes that whole answer to
@@ -45,6 +46,7 @@
     function: String = "",
     file: String = "",
     callDepth: String = "0",
+    boundedSinks: String = "",
     parameterSources: String = "false",
     requests: String = ""
 ) = {
@@ -52,7 +54,16 @@
 
   def split(raw: String): List[String] = raw.split(",").map(_.trim).filter(_.nonEmpty).toList
 
-  def rowsFor(sourcesS: String, sinksS: String, sanitizersS: String, functionS: String, paramSrc: String, fileS: String, depthS: String): List[ujson.Obj] = {
+  def rowsFor(
+      sourcesS: String,
+      sinksS: String,
+      sanitizersS: String,
+      functionS: String,
+      paramSrc: String,
+      fileS: String,
+      depthS: String,
+      boundedS: String
+  ): List[ujson.Obj] = {
 
   // Word-boundary matching, never substring. `resolveUrl` contains `resolve`, so a bare-substring sanitizer
   // test marked the VULNERABLE flow sanitized and the verdict fell back to a captured `res` that names no
@@ -71,6 +82,12 @@
   val sourcePatterns = split(sourcesS)
   val sinkNames      = split(sinksS)
   val sanitizerNames = split(sanitizersS)
+  val boundedNames   = split(boundedS)
+
+  // A relational comparison against an integer literal: `len > 250`, `n <= sizeof(buf)`. Equality and null
+  // checks are deliberately excluded -- `if (buf)` and `p == NULL` govern whether the sink RUNS, not how
+  // much it copies, and counting them would discharge every guarded-against-null overflow there is.
+  val boundShape = java.util.regex.Pattern.compile("(?:<=|>=|<|>)\\s*[0-9]+|[0-9]+\\s*(?:<=|>=|<|>)")
   val function       = functionS
   val parameterSources = paramSrc
 
@@ -188,6 +205,29 @@
     val arity       = realArgs.size
     val arg0Literal = realArgs.headOption.map(a => a.isLiteral).getOrElse(false)
 
+    // Is there a bound on the data reaching this sink? Only asked of sinks whose CWE says the danger is a
+    // length, and the bound must mention a name the sink actually uses -- an unrelated `i < 10` loop in the
+    // same function discharges nothing.
+    //
+    // Scoped to the METHOD rather than to the sink's dominators, and that is deliberate. libpng guards
+    // `strcpy(outname+len, ".png")` with `(len = strlen(inname)) > 250`, but the check sits on an else-if
+    // chain whose sibling branch skips it entirely, and the protection is carried to the sink through an
+    // `error` flag: `++error` in the guarded branch, `if (!error)` around the copy. So the bound does NOT
+    // dominate the sink -- Joern is right about that -- and requiring dominance found nothing.
+    //
+    // Which is exactly why this lowers the rung instead of clearing the finding. A bound on this variable
+    // somewhere in this function means the graph CANNOT ENTAIL: it has not shown the copy is unguarded. It
+    // has not shown it is guarded either, so the judge is asked. Proving the flag-mediated case needs
+    // path-sensitive reasoning this arbiter does not do.
+    val boundedSink = boundedNames.exists(n => sink.name == n)
+    val argNames    = sink.argument.ast.isIdentifier.name.l.distinct
+    val bounds =
+      if (!boundedSink || argNames.isEmpty) List.empty
+      else
+        sink.method.ast.isCall.code.l.filter(c =>
+          boundShape.matcher(c).find() && argNames.exists(a => mentionsToken(c, List(a)))
+        )
+
     flows.map { flow =>
       val elements = flow.elements.map(_.code).l
       val sanitized = sanitizerNames.nonEmpty && elements.exists(code => mentionsToken(code, sanitizerNames))
@@ -203,6 +243,8 @@
         "length"          -> elements.size,
         "sinkArity"       -> arity,
         "sinkArg0Literal" -> arg0Literal,
+        "bounded"         -> bounds.nonEmpty,
+        "bound"            -> bounds.headOption.getOrElse("").take(120),
         "inLabeledScope"  -> (function.isEmpty || nestedInLabeled(sink.method) || reachableMethods.contains(sink.method.fullName))
       )
     }
@@ -217,11 +259,24 @@
     val answers = parsed.map { case (id, req) =>
       def field(name: String): String = req.obj.get(name).map(_.str).getOrElse("")
       val paramSrc = req.obj.get("parameterSources").map(_.str).getOrElse("false")
-      id -> ujson.Arr(rowsFor(field("sources"), field("sinks"), field("sanitizers"), field("function"), paramSrc, field("file"), field("callDepth")): _*)
+      id -> ujson.Arr(
+        rowsFor(
+          field("sources"),
+          field("sinks"),
+          field("sanitizers"),
+          field("function"),
+          paramSrc,
+          field("file"),
+          field("callDepth"),
+          field("boundedSinks")
+        ): _*
+      )
     }
     println(ujson.write(ujson.Obj.from(answers)))
   } else {
-    println(ujson.write(ujson.Arr(rowsFor(sources, sinks, sanitizers, function, parameterSources, file, callDepth): _*)))
+    println(
+      ujson.write(ujson.Arr(rowsFor(sources, sinks, sanitizers, function, parameterSources, file, callDepth, boundedSinks): _*))
+    )
   }
   println("---OUSAST-CPG-END---")
 }

src/openultrasast/model/specs.py

@@ -74,6 +74,14 @@ def classify_guard_text(text: str) -> str:
 # --- The specs the CPG queries are parameterised by ------------------------------------------------------
 
 
+# Buffer-overflow classes: the danger is a length, so a bound that dominates the sink discharges it. Only
+# CWE-121 is in the fact tables today, and it is C-only -- naming the family keeps a later fact from having
+# to remember this rule exists.
+OVERFLOW_CWES = frozenset(
+    {"CWE-119", "CWE-120", "CWE-121", "CWE-122", "CWE-124", "CWE-126", "CWE-127", "CWE-787", "CWE-788", "CWE-805", "CWE-806"}
+)
+
+
 @dataclass(frozen=True)
 class TaintSpec:
     """What a flow family's source-to-sink-minus-sanitizer query matches on, for one language.
@@ -97,6 +105,12 @@ class TaintSpec:
     sources: tuple[str, ...]
     sinks: tuple[str, ...]
     sanitizers: tuple[str, ...]
+    # Sinks whose danger is a LENGTH, and which a bound that dominates them therefore discharges. Derived
+    # from the sink fact's own CWE rather than asserted here: CWE-121 is a stack buffer overflow, and a
+    # `strcpy` a `len > 250` check governs is not the same claim as one nothing governs. A sanitizer list
+    # cannot express this -- `strlen` appears in the guarded and unguarded case alike -- which is why the
+    # discharge is a guard and not a call.
+    bounded_sinks: tuple[str, ...] = ()
     safe_shape_sinks: tuple[str, ...] = ()
 
 
@@ -210,11 +224,16 @@ def taint_specs(
     sanitizers = tuple(sorted({call for fact in scoped.sanitizers if not _is_shape(fact) for call in fact.calls}))
     safe_shapes = tuple(sorted({call for fact in scoped.sanitizers if _is_shape(fact) for call in fact.calls}))
     by_family: dict[str, set[str]] = {}
+    bounded: dict[str, set[str]] = {}
     for sink in scoped.sinks:
         family = families.family_of_cwe(sink.cwe)
         if family is None:
             continue
         by_family.setdefault(family.id, set()).update(sink.calls)
+        # A sink whose CWE is a buffer-overflow class is dischargeable by a bound that dominates it. The CWE
+        # is on the fact already, so which sinks these are is read from the data rather than declared here.
+        if sink.cwe in OVERFLOW_CWES:
+            bounded.setdefault(family.id, set()).update(sink.calls)
     return {
         family_id: TaintSpec(
             family=family_id,
@@ -222,6 +241,7 @@ def taint_specs(
             sources=sources,
             sinks=tuple(sorted(calls)),
             sanitizers=sanitizers,
+            bounded_sinks=tuple(sorted(bounded.get(family_id, ()))),
             safe_shape_sinks=safe_shapes,
         )
         for family_id, calls in sorted(by_family.items())

src/openultrasast/model/taint.py

@@ -40,6 +40,7 @@ def request_params(
         "file": file,
         "parameterSources": "true" if parameter_sources else "false",
         "callDepth": str(call_depth),
+        "boundedSinks": spec.bounded_sinks,
     }
 
 
@@ -76,10 +77,18 @@ def verdict(
     # still vulnerable on the strength of a flow that names no attacker input.
     modelled = [flow for flow in flows if _from_modelled_source(flow, spec)]
     considered = modelled or flows
-    unsanitized = [flow for flow in considered if not flow["sanitized"]]
+    # A flow the graph cannot see a discharge for is the finding; a sanitized or BOUNDED one is a question.
+    # libpng bounds `strcpy(outname+len, ".png")` with `(len = strlen(inname)) > 250` against a char[256],
+    # and entailing it is a claim the code contradicts. But the bound reaches the sink through an `error`
+    # flag rather than by dominating it, so the graph has not shown the copy is guarded either -- and a
+    # bound is no proof of safety in any case, off-by-one being the classic way they fail. Corroborated is
+    # the honest rung: the model cannot assert this, so the judge is asked instead of the site being
+    # silenced.
+    open_flows = [flow for flow in considered if not flow["sanitized"] and not flow["bounded"]]
     # Deterministic by construction: a total order, then take the first. Nothing depends on engine ordering.
-    chosen = min(unsanitized or considered, key=lambda flow: _rank(flow, spec))
-    rung = Rung.ENTAILED if not chosen["sanitized"] else Rung.CORROBORATED
+    chosen = min(open_flows or considered, key=lambda flow: _rank(flow, spec))
+    settled = bool(chosen["sanitized"]) or bool(chosen["bounded"])
+    rung = Rung.CORROBORATED if settled else Rung.ENTAILED
     return Verdict(rung=rung, family=spec.family, witness=_witness(chosen), location=_location(chosen))
 
 
@@ -102,6 +111,8 @@ def _flows(rows: object, *, function: str) -> list[dict[str, object]]:
                 "sinkLine": str(row.get("sinkLine", "")),
                 "sinkMethod": str(row.get("sinkMethod", "")),
                 "sinkFile": str(row.get("sinkFile", "")),
+                "bounded": bool(row.get("bounded", False)),
+                "bound": str(row.get("bound", "")),
                 "source": str(row.get("source", "")),
                 "sanitized": bool(row.get("sanitized", False)),
                 "length": _as_int(row.get("length")),
@@ -172,7 +183,9 @@ def _witness(flow: Mapping[str, object]) -> str:
     line = flow.get("sinkLine") or "?"
     where = str(flow.get("sinkFile") or "")
     at = f" in {where}" if where else ""
-    return f"{flow['source']} -> {flow['sink']}{at} (line {line}, {flow['length']} steps)"
+    bound = str(flow.get("bound") or "")
+    governed = f", governed by `{bound}`" if flow.get("bounded") and bound else ""
+    return f"{flow['source']} -> {flow['sink']}{at} (line {line}, {flow['length']} steps{governed})"
 
 
 def _location(flow: Mapping[str, object]) -> str: