measure(independent): sink-first selection from source -- the vocabulary misses most real sinks

This is a source-only diagnostic: no engine and no model. At each v2 vulnerable
pin, every call matching the family's shipped sink vocabulary becomes a candidate,
attributed to its enclosing function.
- The declared vulnerable function is a candidate in 2/17 cases (Medplum, GoPay);
  the ranker's top 20 managed 1/17.
- A fix-changed or site file is among the candidates in 8/17.
- Candidates per repository range from 2 to 904 functions.

The misses are vocabulary, not direction. Examples: wp_redirect, download_url,
spawn(shell:true), ky(), requests.Session().get, sympy parse_expr, Jinja
from_string().render, Mongo collection.find, JCR-SQL2 createQuery, TypeORM
andWhere template strings, and CORSMiddleware(allow_origins=["*"]) are not named.
v2 is spent for tuning, so this informs design and qualifies nothing.

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

benchmarks/independent/sink_candidates.py

@@ -0,0 +1,121 @@
+"""Diagnostic: would SINK-FIRST selection, from source alone, put the vulnerable function in scope?
+
+The v2 hunter experiment showed scope selection is the bottleneck: the ranker's top 20 entry-point regions held
+the vulnerable file in 1 of 17 cases. This asks the opposite direction without any engine or model: at each
+case's vulnerable pin, every call matching the family's shipped sink vocabulary (`taint_specs`) in product
+source is a candidate, attributed to its enclosing function by a per-language declaration scan. It reports how
+many candidates a repository yields (the budget a hunter would need) and whether a declared site function, or
+at least a file the fix changes, is among them.
+
+Source only: the checkout is the pin's tree; git is used to export it and, for the score, to read the fix's
+changed files -- never to choose scope.
+
+v2 is spent for tuning; this informs the selection design and qualifies nothing.
+
+Usage: python benchmarks/independent/sink_candidates.py [--population population-v2.toml] [--only ID,...]
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import shutil
+import sys
+from pathlib import Path
+
+HERE = Path(__file__).resolve().parent
+sys.path.insert(0, str(HERE))
+
+import evaluate  # noqa: E402
+
+from openultrasast.model.specs import taint_specs  # noqa: E402
+from openultrasast.preprocess import detect_language, enumerate_source_files  # noqa: E402
+
+DECLARATION = {
+    "python": re.compile(r"^\s*(?:async\s+)?def\s+(\w+)"),
+    "php": re.compile(r"^\s*(?:(?:public|private|protected|static|final|abstract)\s+)*function\s+&?(\w+)"),
+    "javascript": re.compile(r"^\s*(?:export\s+)?(?:async\s+)?function\s*\*?\s*(\w+)|^\s*(?:(?:public|private|protected|static|async)\s+)*(\w+)\s*\([^;]*\)\s*(?::[^{=]+)?\{\s*$|^\s*(?:export\s+)?(?:const|let|var)\s+(\w+)\s*=\s*(?:async\s*)?(?:function|\()"),
+}
+DECLARATION["typescript"] = DECLARATION["javascript"]
+KEYWORDS = {"if", "for", "while", "switch", "catch", "return", "function", "else"}
+
+
+def token(name: str) -> re.Pattern[str]:
+    before = r"(?<![A-Za-z0-9_$])" if name[:1].isalnum() or name[:1] in "_$" else ""
+    return re.compile(before + re.escape(name) + r"\s*\(")
+
+
+def enclosing(lines: list[str], index: int, language: str) -> str:
+    pattern = DECLARATION.get(language)
+    if pattern is None:
+        return ""
+    for back in range(index, -1, -1):
+        match = pattern.match(lines[back])
+        if match:
+            name = next((g for g in match.groups() if g), "")
+            if name and name not in KEYWORDS:
+                return name
+    return "<global>"
+
+
+def candidates(root: Path, family: str) -> dict[tuple[str, str], list[int]]:
+    found: dict[tuple[str, str], list[int]] = {}
+    patterns: dict[str, list[re.Pattern[str]]] = {}
+    for path in enumerate_source_files(root):
+        language = detect_language(path)
+        if language not in patterns:
+            spec = taint_specs(language=language).get(family) if language != "unknown" else None
+            patterns[language] = [token(n) for n in spec.sinks] if spec else []
+        if not patterns[language]:
+            continue
+        try:
+            lines = path.read_text(errors="ignore").splitlines()
+        except OSError:
+            continue
+        relative = path.relative_to(root).as_posix()
+        for index, line in enumerate(lines):
+            stripped = line.strip()
+            if stripped.startswith(("//", "#", "*", "/*")):
+                continue
+            if any(p.search(line) for p in patterns[language]):
+                found.setdefault((relative, enclosing(lines, index, language)), []).append(index + 1)
+    return found
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument("--population", type=Path, default=HERE / "population-v2.toml")
+    parser.add_argument("--only", default="")
+    args = parser.parse_args()
+    evaluate.use(args.population)
+    evaluate.RESULTS = evaluate.RESULTS.with_name(evaluate.RESULTS.name + "-sinkfirst")
+    only = {c.strip() for c in args.only.split(",") if c.strip()}
+    rows = []
+    for case in evaluate.cases():
+        if only and case["id"] not in only:
+            continue
+        checkout = evaluate.export(case, "vulnerable", case["vulnerable"])
+        try:
+            found = candidates(checkout, case["family"])
+        finally:
+            shutil.rmtree(checkout, ignore_errors=True)
+        changed = set(evaluate.hunks(case, "old"))
+        sites = {tuple(site.split("::", 1)) for site in case.get("sites", [])}
+        site_hit = sorted(f"{p}::{f}" for p, f in found if (p, f) in sites)
+        file_hit = sorted({p for p, _ in found if p in changed or p in {s[0] for s in sites}})
+        row = {
+            "id": case["id"], "family": case["family"], "candidate_functions": len(found),
+            "candidate_files": len({p for p, _ in found}), "declared_site_in_candidates": site_hit,
+            "fix_or_site_file_in_candidates": file_hit[:5],
+        }  # fmt: skip
+        rows.append(row)
+        print(json.dumps(row), flush=True)
+    sites_in = sum(bool(r["declared_site_in_candidates"]) for r in rows)
+    files_in = sum(bool(r["fix_or_site_file_in_candidates"]) for r in rows)
+    print(json.dumps({"cases": len(rows), "declared_site_in_candidates": sites_in, "fix_or_site_file_in_candidates": files_in}))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())