feat(model): one finding per sink site, and WordPress's registration model

contributor-scan 5.4. Both halves, and the second found a frontend blocker.

**One verdict per region was right for a pair and wrong for a file.** A
pair is one labelled function with one bug, so one answer is correct. A
region spanning a PHP file is not: the probe's SQL injection on line 5 and
command injection on line 13 produced ONE injection verdict, `system` won
it on flow length, and the SQL injection went unreported by construction
rather than by any failure of the analysis.

`taint.verdicts()` now returns one verdict per distinct sink site,
strongest first, and its first element is exactly what `verdict()`
returns -- so the pair path and every committed measurement resting on it
are unchanged, and the gates are byte-identical. Both entry points share
`_usable_flows` so neither can drift from the other.

  php probe   2 of 3 -> 3 of 3
  vampi       6 entailed, unchanged, now with lines on access-control
              findings from task 2.7 (users.py:187, books.py:51 -- the two
              documented BOLAs at their exact lines)
  libpng      1 -> 2 entailed; the second was hidden by the collapse.
              Neither is validated and both are in contrib/.

**WordPress registers handlers rather than decorating them, and the hook
name is the declaration.** `wp_ajax_nopriv_*` is WordPress saying
logged-out callers reach this, so it is public; `wp_ajax_*` fires only for
logged-in ones; `permission_callback => __return_true` declares a REST
route open. That is the declared-provenance signal task 2.8 needs,
arriving from a second framework. Plus every top-level PHP function as its
own region, without which a file is one region however many functions it
holds: 1 file-level region became 10 function-scoped ones.

**And a blocker, recorded as 5.5.** php2cpg emits a CPG containing only
ONE PHP file, the last alphabetically. Reproduced twice on independent
samples; both files parse cleanly under `php -l` and under PHP-Parser
directly, and joern-parse reports success either way. php2cpg logs "Found
warning in PHP-Parser JSON output" naming the survivor, and that warning is
PHP-Parser's own `====> File X:` banner appearing in what php2cpg expects
to be pure JSON.

Not isolated yet: this machine has no PHP interpreter, so php2cpg was
driven through a container shim. The banner comes from PHP-Parser rather
than the shim, which makes a shim-only cause unlikely but not excluded, and
rerunning with the native php-cli in the shipped image is the first step.
It blocks WordPress outright -- a plugin is many files, and a scan that
silently covers one of them and reports success is the quiet failure Req
5.6 was approved for.

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

src/openultrasast/mapping.py

@@ -187,9 +187,99 @@ def _language_entry_points(target: FileTarget, text: str) -> list[EntryPointReco
         return _c_entry_points(target, text)
     if target.language == "solidity":
         return _solidity_entry_points(target, text)
+    if target.language == "php":
+        return _php_entry_points(target, text)
     return []
 
 
+# WordPress registers handlers rather than decorating them, and the hook name states the access level:
+# `wp_ajax_nopriv_*` is admin-ajax for logged-OUT users and is therefore declared public, while `wp_ajax_*`
+# fires only for logged-in ones. That is a DECLARED access level, not a guess from an absent decorator, which
+# is the distinction task 2.8 turns on.
+_PHP_REGISTRATION = re.compile(
+    r"""\b(?P<call>add_action|add_filter|add_shortcode)\s*\(\s*['"](?P<hook>[^'"]+)['"]\s*,\s*['"](?P<handler>[A-Za-z_][A-Za-z0-9_]*)['"]""",
+    re.VERBOSE,
+)
+_PHP_REST_ROUTE = re.compile(r"\bregister_rest_route\s*\(")
+_PHP_CALLBACK = re.compile(r"""['"]callback['"]\s*=>\s*['"](?P<handler>[A-Za-z_][A-Za-z0-9_]*)['"]""")
+_PHP_PERMISSION = re.compile(r"""['"]permission_callback['"]\s*=>\s*(?P<value>[^,\)]+)""")
+_PHP_FUNCTION = re.compile(r"^\s*function\s+(?P<name>[A-Za-z_][A-Za-z0-9_]*)\s*\(")
+
+
+def _php_entry_points(target: FileTarget, text: str) -> list[EntryPointRecord]:
+    """WordPress's registration model, plus every top-level function as a region of its own.
+
+    The functions matter as much as the hooks. Without them a PHP file is ONE file-level region, and a region
+    that spans a file attributes every finding to the file rather than to the function that holds it.
+    """
+    records: list[EntryPointRecord] = []
+    lines = text.splitlines()
+    bounds = _php_function_bounds(lines)
+
+    for number, line in enumerate(lines, start=1):
+        for match in _PHP_REGISTRATION.finditer(line):
+            hook, handler = match.group("hook"), match.group("handler")
+            access, evidence = _wordpress_hook_access(match.group("call"), hook)
+            start, end = bounds.get(handler, (number, number))
+            records.append(_entry(target, start, end, handler, f"wp:{hook}", "route", access, "http_request", evidence, []))
+        if _PHP_REST_ROUTE.search(line):
+            window = "\n".join(lines[number - 1 : number + 12])
+            callback = _PHP_CALLBACK.search(window)
+            if callback is not None:
+                handler = callback.group("handler")
+                permission = _PHP_PERMISSION.search(window)
+                access, evidence = _rest_route_access(permission.group("value").strip() if permission else None)
+                start, end = bounds.get(handler, (number, number))
+                records.append(_entry(target, start, end, handler, "wp:rest_route", "route", access, "http_request", evidence, []))
+
+    registered = {record.function_name for record in records}
+    for name, (start, end) in sorted(bounds.items()):
+        if name not in registered:
+            records.append(
+                _entry(target, start, end, name, f"php:{name}", "function", "review-required", "call", ["php function declaration"], [])
+            )
+    return records
+
+
+def _wordpress_hook_access(call: str, hook: str) -> tuple[AccessLevel, list[str]]:
+    """The hook name is the declaration. `nopriv` is WordPress saying "logged-out callers reach this"."""
+    if hook.startswith("wp_ajax_nopriv_"):
+        return "public", [f"{call}('{hook}') is admin-ajax for logged-out callers"]
+    if hook.startswith("wp_ajax_"):
+        return "authenticated", [f"{call}('{hook}') fires only for logged-in callers"]
+    if call == "add_shortcode":
+        return "public", [f"add_shortcode('{hook}') renders in page content"]
+    return "review-required", [f"{call}('{hook}')"]
+
+
+def _rest_route_access(permission: str | None) -> tuple[AccessLevel, list[str]]:
+    """`permission_callback` is the REST contract. `__return_true` declares the route open."""
+    if permission is None:
+        return "review-required", ["register_rest_route with no permission_callback"]
+    if "__return_true" in permission:
+        return "public", ["permission_callback => __return_true declares the route public"]
+    return "authenticated", [f"permission_callback => {permission[:60]}"]
+
+
+def _php_function_bounds(lines: Sequence[str]) -> dict[str, tuple[int, int]]:
+    """Line ranges of top-level PHP functions, by brace depth. Good enough to scope a region."""
+    bounds: dict[str, tuple[int, int]] = {}
+    name: str | None = None
+    start = 0
+    depth = 0
+    for number, line in enumerate(lines, start=1):
+        if name is None:
+            match = _PHP_FUNCTION.match(line)
+            if match is not None:
+                name, start, depth = match.group("name"), number, 0
+        if name is not None:
+            depth += line.count("{") - line.count("}")
+            if depth <= 0 and "{" in "".join(lines[start - 1 : number]):
+                bounds[name] = (start, number)
+                name = None
+    return bounds
+
+
 def _python_entry_points(target: FileTarget, text: str) -> list[EntryPointRecord]:
     records: list[EntryPointRecord] = []
     try:

src/openultrasast/model/pipeline.py

@@ -34,6 +34,7 @@ from .dominance import verdict as dominance_verdict
 from .ladder import Rung, Verdict
 from .specs import ConfigSpec, DominanceSpec, TaintSpec
 from .taint import verdict as taint_verdict
+from .taint import verdicts as taint_verdicts
 
 logger = logging.getLogger(__name__)
 
@@ -65,6 +66,30 @@ class ModelFinding:
     contradiction: str = ""
 
 
+def _arbitrate_all(
+    cpg: CpgResult,
+    spec: TaintSpec | DominanceSpec | ConfigSpec,
+    *,
+    path: str,
+    function: str,
+    parameter_sources: bool,
+    call_depth: int,
+) -> list[Verdict]:
+    """Every verdict this region admits, strongest first.
+
+    One verdict per region was the right answer for a pair -- one labelled function, one bug -- and the wrong
+    one for a region that spans a file. A PHP file with an SQL injection on line 6 and a command injection on
+    line 13 produced ONE injection verdict, and `system` won it on flow length, so the SQL injection went
+    unreported by construction. Only taint can report several sites today; dominance and config still answer
+    once per region, which is correct for dominance (its claim is about the region's consistency) and a
+    known gap for config.
+    """
+    if isinstance(spec, (DominanceSpec, ConfigSpec)):
+        answer = _arbitrate(cpg, spec, path=path, function=function, parameter_sources=parameter_sources, call_depth=call_depth)
+        return [answer] if answer is not None else []
+    return taint_verdicts(cpg, spec, function=function, file=path, parameter_sources=parameter_sources, call_depth=call_depth)
+
+
 def _arbitrate(
     cpg: CpgResult,
     spec: TaintSpec | DominanceSpec | ConfigSpec,
@@ -101,12 +126,18 @@ def scan_region(
     call_depth: int = 0,
 ) -> list[ModelFinding]:
     """Findings for one region and family, ordered strongest first."""
-    answer = _arbitrate(cpg, spec, path=path, function=function, parameter_sources=parameter_sources, call_depth=call_depth)
+    answers = _arbitrate_all(cpg, spec, path=path, function=function, parameter_sources=parameter_sources, call_depth=call_depth)
 
-    if answer is not None and answer.rung is Rung.ENTAILED:
-        # The graph decided. No model call, and no candidate needs asking about: this region has its finding.
-        return [ModelFinding(site=_site(path, function, answer), family=spec.family, rung=Rung.ENTAILED, witness=answer.witness)]
+    entailed = [answer for answer in answers if answer.rung is Rung.ENTAILED]
+    if entailed:
+        # The graph decided. No model call, and no candidate needs asking about: this region has its findings,
+        # one per site rather than one per region.
+        return [
+            ModelFinding(site=_site(path, function, answer), family=spec.family, rung=Rung.ENTAILED, witness=answer.witness)
+            for answer in entailed
+        ]
 
+    answer = answers[0] if answers else None
     if answer is not None:
         # A real flow through a sanitizer. Whether it suffices is the residual, and the model HAS coverage
         # here — so a "no" from the judge is a genuine contradiction and the claim is dropped.

src/openultrasast/model/taint.py

@@ -59,24 +59,9 @@ def verdict(
     function-local question; above zero is what lets a handler's parameter reach a sink in another module,
     which is the shape VAmPI's SQL injection has and the shape a function-scoped question cannot see.
     """
-    rows = cpg.run(
-        "taint",
-        request_params(spec, function=function, file=file, parameter_sources=parameter_sources, call_depth=call_depth),
-    )
-    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:
+    considered = _usable_flows(cpg, spec, function=function, file=file, parameter_sources=parameter_sources, call_depth=call_depth)
+    if not considered:
         return None
-    # Parameter sources are a FALLBACK, never an override. When a modelled source reaches the sink, those
-    # flows are the evidence: a captured `res` parameter also reaching it unsanitized must not report a fix as
-    # 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
     # 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`
@@ -87,9 +72,79 @@ def verdict(
     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(open_flows or considered, key=lambda flow: _rank(flow, spec))
-    settled = bool(chosen["sanitized"]) or bool(chosen["bounded"])
+    return _verdict_for(chosen, spec)
+
+
+def _usable_flows(
+    cpg: CpgResult,
+    spec: TaintSpec,
+    *,
+    function: str,
+    file: str,
+    parameter_sources: bool,
+    call_depth: int,
+) -> list[dict[str, object]]:
+    """The flows both entry points reason over, so neither can drift from the other.
+
+    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 a flow reaching them is not a
+    vulnerability even though its taint path is identical to the vulnerable twin's. Dropping those 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 arbitrated nothing.
+
+    Parameter sources are then a FALLBACK, never an override. When a modelled source reaches the sink those
+    flows are the evidence, so a captured `res` parameter also reaching it unsanitized cannot report a fix as
+    still vulnerable on the strength of a flow that names no attacker input.
+    """
+    rows = cpg.run(
+        "taint",
+        request_params(spec, function=function, file=file, parameter_sources=parameter_sources, call_depth=call_depth),
+    )
+    flows = [flow for flow in _flows(rows, function=function) if not _has_safe_shape(flow, spec)]
+    modelled = [flow for flow in flows if _from_modelled_source(flow, spec)]
+    return modelled or flows
+
+
+def verdicts(
+    cpg: CpgResult,
+    spec: TaintSpec,
+    *,
+    function: str = "",
+    file: str = "",
+    parameter_sources: bool = False,
+    call_depth: int = 0,
+) -> list[Verdict]:
+    """One verdict per distinct SINK SITE, strongest first.
+
+    ``verdict`` answers "what is the strongest thing to say about this region", which is the right question
+    for a pair: one labelled function, one bug, one answer. It is the wrong question for a region that spans
+    a file. A PHP file with an SQL injection on line 6 and a command injection on line 13 got ONE injection
+    verdict, and `system` won it on flow length -- so the SQL injection went unreported, by construction
+    rather than by any failure of the analysis.
+
+    The first element is exactly what ``verdict`` returns, so the pair path and the committed measurements
+    that rest on it are unchanged.
+    """
+    flows = _usable_flows(cpg, spec, function=function, file=file, parameter_sources=parameter_sources, call_depth=call_depth)
+    if not flows:
+        return []
+    by_site: dict[str, list[Mapping[str, object]]] = {}
+    for flow in flows:
+        by_site.setdefault(_location(flow) or str(flow.get("sink", "")), []).append(flow)
+
+    chosen: list[Mapping[str, object]] = []
+    for site_flows in by_site.values():
+        open_flows = [flow for flow in site_flows if not flow["sanitized"] and not flow["bounded"]]
+        chosen.append(min(open_flows or site_flows, key=lambda flow: _rank(flow, spec)))
+    # Same total order the single-verdict path uses, so `verdicts(...)[0] == verdict(...)`.
+    chosen.sort(key=lambda flow: (bool(flow["sanitized"]) or bool(flow["bounded"]), _rank(flow, spec)))
+    return [_verdict_for(flow, spec) for flow in chosen]
+
+
+def _verdict_for(flow: Mapping[str, object], spec: TaintSpec) -> Verdict:
+    settled = bool(flow["sanitized"]) or bool(flow["bounded"])
     rung = Rung.CORROBORATED if settled else Rung.ENTAILED
-    return Verdict(rung=rung, family=spec.family, witness=_witness(chosen), location=_location(chosen))
+    return Verdict(rung=rung, family=spec.family, witness=_witness(flow), location=_location(flow))
 
 
 def _flows(rows: object, *, function: str) -> list[dict[str, object]]: