fix(model): a guard discharges the obligation it is a guard of

contributor-scan 2.6. Access control entailed 7 of 9 VAmPI handlers,
including three that do authorize and three the OpenAPI spec declares
public. Five separate defects, each invisible to a corpus of one-function
excerpts and each found by asking the query what it actually matched.

**A trailing segment is not a match.** `query.get` was matched by its last
segment, so every `.get(` in the repository became a protected database
operation -- `request.headers.get('Authorization')` and
`request_data.get('username')` included. Four of the seven false positives
came from this, and so did the apparent detection of the known BOLA: it was
entailed because fetching the Authorization header had no dominating guard,
not because of the defect. A qualified token now requires its qualifier.

**Configuration is not authorization.** dominance_specs took every
obligation fact. `security_setting` operations are discharged by a VALUE,
which config_specs already claims them for; pooled in here they made
`app.run(...)` an operation requiring an authorization guard and put
thirteen tokens in BOTH lists -- an obligation that is its own discharge,
the same shape as the taint sink that used to sanitize itself.

**Input validation is not authorization.** `jsonschema.validate` was
counted as a guard, so handlers with no authorization check at all read as
guarded. No operation in the table requires validated_input.

**A guard discharges the obligation it is a guard of.** The largest one.
`orm-read` requires an identity_constraint or an ownership_check;
`token_validator` is a path_guard. Flattened into one list, authenticating
the caller discharged an object-level obligation -- a handler that
authenticates and then looks the record up by a PATH parameter read as
fully guarded. The facts state that relation and the spec was discarding
it; it is now carried and the engine asks per operation.

**A word boundary needs a word.** Found while fixing the above. Guard
matching wrapped every token in `(?<![A-Za-z0-9_])...(?![A-Za-z0-9_])`.
The identity token `['sub']` begins with `[`, and the character before it
in `resp['sub']` is `p` -- so the assertion could never hold. Every
identity token starting with punctuation was unmatchable, and the identity
constraint that separates a correct lookup from an IDOR was invisible.

Also: dominance rows were repository-wide, so the model's unguarded
`delete_user` was attributed to the API handler of the same name. Rows are
scoped to the region's file, which is what the rung's own justification
says -- the evidence is a file contradicting itself.

Measured against VAmPI's OpenAPI `security` blocks, which declare per
operation whether auth is required, and agree with the code:

  vampi precision   28.6% (2/7) -> 50.0% (2/4)
  known found       3 of 3, all model_entailed

And on the labelled access-control pairs, which nobody tuned this against:

  all 9 pairs       pair_correct 3 -> 3, leak 44.4% -> 11.1%
  7 non-VAmPI       pair_correct 2 -> 2, leak 42.9% -> 0.0%

Every detection lost was on a pair whose FIXED side was flagged too, which
scores nothing. No correct detection was traded for the precision.

get_by_title was flagged before it was in the recipe: its code carries the
comment "# Broken Object Level Authorization" and the README lists it, so a
real finding was being scored as a false positive. It is now the recipe's
third known vulnerability.

Both remaining false positives are endpoints the spec declares public.
Recorded as 2.8, with the reason the mapper's access_level cannot simply be
trusted: for decorator frameworks "public" means "no decorator found",
which is the bug itself.

Gates byte-identical.

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

src/openultrasast/cpg/queries/dominance.sc

@@ -18,7 +18,15 @@
 // so it never dominates the operation, but it does control whether the value escapes.
 //
 // Parameters: cpgFile, operations, dischargers, function (optional; reported, never used to filter, because
-// the sibling comparison needs every operation in the CPG).
+// the sibling comparison needs the file's other operations), file (optional: the source file the region
+// belongs to), operationRequires and dischargersByKind (optional JSON, see below).
+//
+// A discharger only discharges the obligation it is a discharger OF. The facts say `orm-read` requires an
+// identity_constraint or an ownership_check, and that `token_validator` is a path_guard; pooling them made
+// authenticating the caller discharge an object-level obligation, so a handler that authenticates and then
+// looks a record up by a PATH parameter read as fully guarded. `operationRequires` maps an operation call to
+// the kinds that would discharge it and `dischargersByKind` maps a kind to its tokens; when both are absent
+// the flat list is used, which is the committed single-request behaviour.
 //
 // Output: a fenced JSON array of {operation, opLine, opMethod, dominatingGuards}.
 
@@ -31,13 +39,27 @@
     operations: String = "",
     dischargers: String = "",
     function: String = "",
+    file: String = "",
+    operationRequires: String = "",
+    dischargersByKind: String = "",
     requests: String = ""
 ) = {
   importCpg(cpgFile)
 
   def split(raw: String): List[String] = raw.split(",").map(_.trim).filter(_.nonEmpty).toList
 
-  def rowsFor(operationsS: String, dischargersS: String, functionS: String): List[ujson.Obj] = {
+  def parseMap(raw: String): Map[String, List[String]] =
+    if (raw.isEmpty) Map.empty
+    else ujson.read(raw).obj.map { case (k, v) => k -> v.arr.map(_.str).toList }.toMap
+
+  def rowsFor(
+      operationsS: String,
+      dischargersS: String,
+      functionS: String,
+      fileS: String,
+      requiresS: String,
+      byKindS: String
+  ): List[ujson.Obj] = {
 
   val opNames     = split(operationsS)
   val guardTokens = split(dischargersS)
@@ -46,31 +68,82 @@
   // Word-boundary matching, never substring. A discharger token like `user` matched as a substring hits
   // `users` inside "SELECT * FROM users WHERE id = ?", marking a textbook IDOR as guarded -- a silent false
   // negative on the one family this arbiter exists for.
-  def mentionsGuard(code: String): Boolean =
-    guardTokens.exists(g => java.util.regex.Pattern.compile("(?<![A-Za-z0-9_])" + java.util.regex.Pattern.quote(g) + "(?![A-Za-z0-9_])").matcher(code).find())
+  val requiresByOp = parseMap(requiresS)
+  val tokensByKind = parseMap(byKindS)
+
+  // Word-boundary matching, never substring: a token like `user` matched as a substring hits `users` inside
+  // "SELECT * FROM users WHERE id = ?", marking a textbook IDOR as guarded.
+  //
+  // But the boundary only applies where there IS a word character to bound. `['sub']` begins with `[`, and
+  // the character before it in `resp['sub']` is `p` -- so an unconditional lookbehind made every identity
+  // token that starts with punctuation unmatchable, and the identity constraint that distinguishes a
+  // correct lookup from an IDOR could never be seen.
+  def boundedPattern(token: String): java.util.regex.Pattern = {
+    val before = if (token.headOption.exists(c => c.isLetterOrDigit || c == '_')) "(?<![A-Za-z0-9_])" else ""
+    val after = if (token.lastOption.exists(c => c.isLetterOrDigit || c == '_')) "(?![A-Za-z0-9_])" else ""
+    java.util.regex.Pattern.compile(before + java.util.regex.Pattern.quote(token) + after)
+  }
+
+  def mentions(code: String, tokens: List[String]): Boolean =
+    tokens.exists(g => boundedPattern(g).matcher(code).find())
+
+  def mentionsGuard(code: String): Boolean = mentions(code, guardTokens)
+
+  // The tokens that can discharge THIS operation, rather than every token that can discharge anything.
+  def guardsFor(name: String, code: String): List[String] =
+    if (requiresByOp.isEmpty || tokensByKind.isEmpty) guardTokens
+    else {
+      val kinds = opNames
+        .filter(n => name == n || (n.split("\\.").length > 1 && name == n.split("\\.").last && n.split("\\.").init.forall(code.contains)))
+        .flatMap(n => requiresByOp.getOrElse(n, Nil))
+        .distinct
+      if (kinds.isEmpty) guardTokens else kinds.flatMap(k => tokensByKind.getOrElse(k, Nil)).distinct
+    }
 
   // Match by CALL NAME, never by code containment: the frontend desugars `Note.query.filter_by(x).first()`
   // into a chain of temporaries, and matching on code yields four or five rows for one operation -- some of
   // them the bare `tmp1.filter_by` receiver, which carries none of the guards the real call does. A `leaky`
   // verdict computed over those extra rows would flag a guarded operation. A dotted spec (`query.get`)
   // matches on its trailing segment, which is how the call is named in the graph.
-  def matchesOperation(name: String): Boolean =
-    opNames.exists(n => name == n || name == n.split("\\.").last)
+  //
+  // But the trailing segment ALONE is not enough, and on a repository it is catastrophic: `query.get` ends
+  // in `get`, so it matched `request.headers.get('Authorization')` and `request_data.get('username')` --
+  // ordinary dictionary access read as a protected database operation. Four of seven false positives on the
+  // first repository measured came from exactly that, including three handlers that do call an
+  // authorization guard. A qualified token means the qualification matters, so the receiver must appear
+  // too; the desugared chain still carries it.
+  def matchesOperation(name: String, code: String): Boolean =
+    opNames.exists { n =>
+      if (name == n) true
+      else {
+        val parts = n.split("\\.")
+        parts.length > 1 && name == parts.last && parts.init.forall(q => code.contains(q))
+      }
+    }
 
-  val opCalls = cpg.call.filter(c => matchesOperation(c.name)).l
+  // Scoped to the region's own file. `delete_user` exists in both the API handler and the model of the
+  // repository this was measured on, and unscoped rows attributed the model's unguarded query to the
+  // handler. The sibling comparison is a statement about a FILE contradicting itself, which is what the
+  // rung's justification says, so the file is also the right scope for it.
+  val opCalls = {
+    val all = cpg.call.filter(c => matchesOperation(c.name, c.code)).l
+    if (fileS.isEmpty) all else all.filter(_.method.filename.endsWith(fileS))
+  }
 
   val rows = opCalls.map { op =>
     val method = op.method
 
+    val applicable = guardsFor(op.name, op.code)
+
     // 1. identity constraint: a discharger inside the operation's own arguments.
-    val inArguments = op.argument.code.l.filter(mentionsGuard)
+    val inArguments = op.argument.code.l.filter(c => mentions(c, applicable))
 
     // 2. identity branch: a control structure in this method whose condition names a discharger. The denial
     //    it guards need not dominate the operation -- it controls whether the fetched value escapes.
-    val inCondition = method.controlStructure.condition.code.l.filter(mentionsGuard)
+    val inCondition = method.controlStructure.condition.code.l.filter(c => mentions(c, applicable))
 
     // 3. dominating guard: a call naming a discharger that actually dominates the operation.
-    val dominating = op.dominatedBy.isCall.code.l.filter(mentionsGuard)
+    val dominating = op.dominatedBy.isCall.code.l.filter(c => mentions(c, applicable))
 
     val guards = (inArguments ++ inCondition ++ dominating).distinct.map(_.take(120))
 
@@ -90,11 +163,20 @@
     val parsed = ujson.read(requests).obj
     val answers = parsed.map { case (id, req) =>
       def field(name: String): String = req.obj.get(name).map(_.str).getOrElse("")
-      id -> ujson.Arr(rowsFor(field("operations"), field("dischargers"), field("function")): _*)
+      id -> ujson.Arr(
+        rowsFor(
+          field("operations"),
+          field("dischargers"),
+          field("function"),
+          field("file"),
+          field("operationRequires"),
+          field("dischargersByKind")
+        ): _*
+      )
     }
     println(ujson.write(ujson.Obj.from(answers)))
   } else {
-    println(ujson.write(ujson.Arr(rowsFor(operations, dischargers, function): _*)))
+    println(ujson.write(ujson.Arr(rowsFor(operations, dischargers, function, file, operationRequires, dischargersByKind): _*)))
   }
   println("---OUSAST-CPG-END---")
 }

src/openultrasast/model/dominance.py

@@ -19,6 +19,7 @@ The rungs are deliberately asymmetric:
 
 from __future__ import annotations
 
+import json
 from collections.abc import Mapping, Sequence
 
 from ..cpg.backend import CpgResult
@@ -26,14 +27,29 @@ from .ladder import Rung, Verdict
 from .specs import DominanceSpec
 
 
-def request_params(spec: DominanceSpec, *, function: str = "") -> dict[str, object]:
-    """The query parameters this arbiter sends. Shared with the batcher; see the note in `taint.request_params`."""
-    return {"operations": spec.operations, "dischargers": spec.dischargers, "function": function}
+def request_params(spec: DominanceSpec, *, function: str = "", file: str = "") -> dict[str, object]:
+    """The query parameters this arbiter sends. Shared with the batcher; see the note in `taint.request_params`.
+
+    ``operationRequires`` and ``dischargersByKind`` carry the facts' own relation as JSON, so the engine can
+    ask whether a guard discharges THIS obligation rather than whether it discharges anything. ``file`` scopes
+    the rows: the sibling comparison is a claim about one file contradicting itself, and two functions of the
+    same name in different modules were otherwise pooled into one region's answer.
+    """
+    return {
+        "operations": spec.operations,
+        "dischargers": spec.dischargers,
+        "function": function,
+        "file": file,
+        "operationRequires": json.dumps({k: list(v) for k, v in spec.requirements.items()}, sort_keys=True) if spec.requirements else "",
+        "dischargersByKind": json.dumps({k: list(v) for k, v in spec.dischargers_by_kind.items()}, sort_keys=True)
+        if spec.dischargers_by_kind
+        else "",
+    }
 
 
-def verdict(cpg: CpgResult, spec: DominanceSpec, *, function: str = "") -> Verdict | None:
+def verdict(cpg: CpgResult, spec: DominanceSpec, *, function: str = "", file: str = "") -> Verdict | None:
     """The verdict for the obligated operation in ``function``, or ``None`` when there is nothing to say."""
-    rows = cpg.run("dominance", request_params(spec, function=function))
+    rows = cpg.run("dominance", request_params(spec, function=function, file=file))
     operations = _operations(rows)
     if not operations:
         return None

src/openultrasast/model/pipeline.py

@@ -82,7 +82,7 @@ def _arbitrate(
     pairs read as `both_silent` on the first end-to-end run.
     """
     if isinstance(spec, DominanceSpec):
-        return dominance_verdict(cpg, spec, function=function)
+        return dominance_verdict(cpg, spec, function=function, file=path)
     if isinstance(spec, ConfigSpec):
         return config_verdict(cpg, spec, function=function, file=path)
     return taint_verdict(cpg, spec, function=function, file=path, parameter_sources=parameter_sources, call_depth=call_depth)

src/openultrasast/model/scan.py

@@ -237,7 +237,7 @@ def _grouped(work: Sequence[tuple[str, ScanRegion, ArbiterSpec]]) -> dict[str, d
         function = region.function or ""
         entry = _is_entry_point(region)
         if isinstance(spec, DominanceSpec):
-            kind, params = "dominance", dominance_params(spec, function=function)
+            kind, params = "dominance", dominance_params(spec, function=function, file=region.path)
         elif isinstance(spec, ConfigSpec):
             kind, params = "config", config_params(spec, function=function, file=region.path)
         else:

src/openultrasast/model/specs.py

@@ -19,7 +19,7 @@ from __future__ import annotations
 
 import re
 from collections.abc import Mapping
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 
 from ..semantic.facts import SemanticFacts, load_facts
 from ..semantic.obligations.facts import ObligationFacts, load_obligation_facts
@@ -119,12 +119,24 @@ class DominanceSpec:
                                    evidences nothing and treating it as a guard silences the whole family.
         request_sources            a value from the request discharges nothing, by definition.
         permissive_values          a permissive literal is the bug, not the guard.
+
+    ``dischargers`` is the flat union, kept because a caller may not care which kind discharged what.
+    ``requirements`` and ``dischargers_by_kind`` carry the relation the facts actually state: an operation
+    names the obligation KINDS that would discharge it, and only a discharger of one of those kinds counts.
+
+    Flattening that relation away was a false-negative engine. `orm-read` requires an ``identity_constraint``
+    or an ``ownership_check``; `token_validator` is a ``path_guard``. Pooled together, authenticating the
+    caller discharged an object-level obligation -- so a handler that authenticates and then looks the
+    record up by a PATH parameter read as fully guarded. That is the shape of VAmPI's broken object-level
+    authorization, and of most real IDORs.
     """
 
     family: str
     language: str
     operations: tuple[str, ...]
     dischargers: tuple[str, ...]
+    requirements: Mapping[str, tuple[str, ...]] = field(default_factory=dict)
+    dischargers_by_kind: Mapping[str, tuple[str, ...]] = field(default_factory=dict)
 
 
 @dataclass(frozen=True)
@@ -219,21 +231,56 @@ def taint_specs(
 def dominance_specs(*, language: str, facts: ObligationFacts | None = None) -> Mapping[str, DominanceSpec]:
     """One ``DominanceSpec`` per absence family for this language.
 
-    The obligation facts are authored for authorization, so every obligated operation routes to
-    ``access_control``; the shape generalises when a second absence family is added to the taxonomy.
+    Not every obligation fact is an authorization one, and taking them all was wrong in both directions.
+
+    * ``security_setting`` operations are configuration: a permissive CORS origin or a debug flag is
+      discharged by a VALUE, which is what ``config_specs`` already claims them for and what this arbiter
+      cannot decide. Pooled in here they made `app.run(...)` an operation requiring an authorization guard,
+      and put thirteen tokens in both the operation and discharger lists -- an obligation that is its own
+      discharge, the same shape as the taint sink that used to sanitize itself.
+    * ``validated_input`` dischargers are schema checks. `jsonschema.validate` does not establish who the
+      caller is, and no operation in the table requires it; counting it as a guard marked handlers with no
+      authorization check at all as guarded.
+
+    So both sides are filtered by kind, and the filter is derived from the facts' own ``requires`` relation
+    rather than a list maintained here.
     """
     scoped = (facts if facts is not None else load_obligation_facts()).for_language(language)
     if not scoped.operations:
         return {}
-    operations = tuple(sorted({call for fact in scoped.operations for call in fact.calls}))
+    guarded = tuple(fact for fact in scoped.operations if fact.kind != "security_setting")
+    if not guarded:
+        return {}
+    # What the retained operations actually ask for. A discharger of any other kind discharges nothing here.
+    required = {kind for fact in guarded for kind in fact.requires}
+    dischargers_for = tuple(fact for fact in scoped.dischargers if fact.kind in required)
+    operations = tuple(sorted({call for fact in guarded for call in fact.calls}))
     dischargers = tuple(
         sorted(
-            {call for fact in scoped.dischargers for call in fact.calls}
-            | {decorator for fact in scoped.dischargers for decorator in fact.decorators}
-            | {source for fact in scoped.dischargers for source in fact.identity_sources}
+            {call for fact in dischargers_for for call in fact.calls}
+            | {decorator for fact in dischargers_for for decorator in fact.decorators}
+            | {source for fact in dischargers_for for source in fact.identity_sources}
         )
     )
-    return {"access_control": DominanceSpec(family="access_control", language=language, operations=operations, dischargers=dischargers)}
+    if not operations or not dischargers:
+        return {}
+    # The relation the facts state, kept rather than flattened: which obligation kinds each operation call
+    # would accept as a discharge, and which tokens evidence each kind.
+    requirements = {call: tuple(sorted(set(fact.requires))) for fact in guarded for call in fact.calls}
+    by_kind: dict[str, tuple[str, ...]] = {}
+    for fact in dischargers_for:
+        tokens = set(fact.calls) | set(fact.decorators) | set(fact.identity_sources)
+        by_kind[fact.kind] = tuple(sorted(set(by_kind.get(fact.kind, ())) | tokens))
+    return {
+        "access_control": DominanceSpec(
+            family="access_control",
+            language=language,
+            operations=operations,
+            dischargers=dischargers,
+            requirements=requirements,
+            dischargers_by_kind=by_kind,
+        )
+    }
 
 
 __all__ = [