From c35426e1862dc868b43ad6beffc16ad1383944cc Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Wed, 9 Sep 2026 07:57:35 +0200 Subject: [PATCH] perf(model): one Joern invocation per query kind, not one per region JVM startup dominated a scan: a call per region per family put a ten-line file at 275s over three regions, and would have put a thousand regions at roughly fifty hours. The queries themselves are milliseconds once the CPG is loaded, so the cost was almost entirely process starts. scan_repository() is now collect -> batch -> arbitrate. The three .sc scripts accept a `requests` JSON map and answer all of them in one run, keeping the single-request form so the committed measurements stay reproducible byte-for-byte. Arbiters are untouched: each is handed a CpgResult whose `run` returns its own prefetched rows. The batcher builds parameters by calling request_params() extracted from taint.py, dominance.py and config_value.py rather than reconstructing them. A second copy that drifts from the arbiter's is the exact shape of the five family-assumption bugs the predecessor spec spent a session finding; there is now one definition per family and the batcher borrows it. Regions are also deduplicated by (path, function), collapsing the nameless entry-point rows that were producing `model:injection:?:10`. Measured on the same ten-line file: 275s -> 106s, three duplicate findings -> one `model:injection:ping:10`, rung unchanged at model_entailed. Gates byte-identical. Server mode is recorded as task 2.3, deliberately sequenced after the repository measurement: whether the remaining ~100s floor is worth a process model, a weakened subprocess seam and a Scala-evaluating HTTP endpoint depends on a build-time-to-query-time ratio only 2.2 produces. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG --- .kiro/specs/contributor-scan/tasks.md | 29 +++++- src/openultrasast/cpg/backend.py | 37 +++++++ src/openultrasast/cpg/queries/config.sc | 23 ++++- src/openultrasast/cpg/queries/dominance.sc | 33 +++++- src/openultrasast/cpg/queries/taint.sc | 40 +++++-- src/openultrasast/model/config_value.py | 9 +- src/openultrasast/model/dominance.py | 12 ++- src/openultrasast/model/regions.py | 30 +++--- src/openultrasast/model/scan.py | 115 ++++++++++++++++----- src/openultrasast/model/taint.py | 28 +++-- tests/test_cpg_backend.py | 57 ++++++++++ tests/test_model_regions.py | 34 ++++++ tests/test_model_scan.py | 41 ++++++++ 13 files changed, 412 insertions(+), 76 deletions(-) diff --git a/.kiro/specs/contributor-scan/tasks.md b/.kiro/specs/contributor-scan/tasks.md index 25c0e82..b1f7eb2 100644 --- a/.kiro/specs/contributor-scan/tasks.md +++ b/.kiro/specs/contributor-scan/tasks.md @@ -58,7 +58,7 @@ - _Requirements: 1.1, 1.2, 1.3, 1.4, 1.5, 8.4_ - _Depends: 1.2_ -- [ ] 1.4 Batch the queries — one Joern invocation per scan, not one per region and family +- [x] 1.4 Batch the queries — one Joern invocation per scan, not one per region and family - **Why this is in group 1 rather than group 2.** Wiring 1.3 exposed that the cost model in the design was half right. One CPG per scan was the right call, but every *query* is its own `joern --script` JVM launch of roughly 30 seconds, and the driver issues one per region per family. A single ten-line Python file @@ -88,15 +88,36 @@ - _Depends: 1.3_ - [ ] 2.2 The first repository measurement - - Run the full scan against that checkout and commit the artifact: CPG build time, scan wall-clock, peak - memory, findings by rung, cost, `regions_unjudged`, and **whether the known CVE was found and at which - rung**. If the repository cannot be processed in its envelope, record the stage it failed at. + - Run the full scan against that checkout and commit the artifact: CPG build time, query time and total + scan wall-clock as three separate numbers, peak memory, findings by rung, cost, `regions_unjudged`, + and **whether the known CVE was found and at which rung**. The build/query split is separated because + 2.3 is decided on that ratio and cannot be decided on a single wall-clock figure. If the repository cannot be processed in its envelope, record the stage it failed at. - This is the go/no-go for the shape of the rest: if one CPG per repository is infeasible, the region and budget design changes and phases 3–6 should not be built on it first. - Observable: the artifact exists with every field; the reading states go or no-go and why. - _Requirements: 4.2, 4.3, 4.4, 4.5_ - _Depends: 2.1_ +- [ ] 2.3 Decide on Joern server mode, on 2.2's numbers + - Joern offers `--server` with `--server-host`, `--server-port` and basic auth: one warm JVM with the CPG + loaded once, and queries in milliseconds after that. It would remove the three-to-four JVM starts + (~100s) that remain per scan now that 1.4 has batched the queries. + - **Deliberately sequenced after the measurement, because deciding it now would be optimising before + measuring.** Batching already moved query cost from O(regions) to O(1) per scan; whether the remaining + floor matters is a ratio 2.2 produces and nothing before it does. If a real CPG build takes ten minutes, + a 100s query floor is noise and server mode buys nothing. If the build is thirty seconds, the floor is + 3x the build and server mode is the obvious next move. + - Costs to weigh against that ratio: process lifecycle (start, health, shutdown, orphan reaping -- stray + JVMs had to be killed by hand twice while building the batching); the seam weakening from "the subprocess + is the only Joern touch-point" to "subprocess or HTTP client", which is the promise that has kept the + backend replaceable; and that a CPGQL endpoint evaluates Scala, so it is arbitrary code execution on the + analysis host -- acceptable inside the shipped container, which runs `network_mode: none`, and much less + so as a port listening on a contributor's laptop. + - Observable: a recorded decision that cites 2.2's build-time-to-query-time split. A preference does not + close this task; a ratio does. + - _Requirements: 4.2_ + - _Depends: 2.2_ + ## Group 3 — Ship it - [ ] 3.1 Contributor output diff --git a/src/openultrasast/cpg/backend.py b/src/openultrasast/cpg/backend.py index 00615f6..7f8f737 100644 --- a/src/openultrasast/cpg/backend.py +++ b/src/openultrasast/cpg/backend.py @@ -116,6 +116,43 @@ class JoernBackend: return None return CpgResult(cpg_path=cpg_path, run=lambda query, params: self.query(cpg_path, query, params)) + def query_batch(self, cpg_path: Path, query: str, requests: Mapping[str, Mapping[str, object]]) -> dict[str, list[object]]: + """Run ONE script invocation carrying many requests, keyed back to their ids. + + JVM startup, not CPG construction, is what dominates a repository scan: every ``joern --script`` call + starts a JVM of roughly thirty seconds, and issuing one per region per family put a ten-line Python + file at four minutes and a thousand regions at about fifty hours. The CPG is already built and loaded + by then; the work itself is milliseconds. So the batch carries the whole scan's questions in one + parameter and the script loops over them. + + Fails closed like every other path here: an engine that could not answer returns ``{}``, never an + empty result per request, because "no rows" and "could not ask" must stay distinguishable. + """ + script = self.queries_dir / f"{query}.sc" + if not script.is_file(): + logger.warning("no such cpg query: %s", script) + return {} + payload = json.dumps({rid: {k: _render(v) for k, v in req.items()} for rid, req in requests.items()}) + command = [ + shutil.which("joern") or "joern", + "--script", + str(script), + "--param", + f"cpgFile={cpg_path}", + "--param", + f"requests={payload}", + ] + completed = self._run(command, timeout=self.query_timeout, cwd=cpg_path.parent) + if completed is None or completed.returncode != 0: + detail = (completed.stderr or "")[-400:] if completed is not None else "timeout" + logger.warning("cpg batch %s failed: %s", query, detail) + return {} + parsed = extract_payload(completed.stdout or "") + if not isinstance(parsed, Mapping): + return {} + # A request the engine did not answer is absent, not empty: the caller must be able to tell them apart. + return {str(rid): list(rows) for rid, rows in parsed.items() if isinstance(rows, list)} + def query(self, cpg_path: Path, query: str, params: Mapping[str, object]) -> object | None: """Run a shipped CPGQL script against a built CPG and return its fenced JSON payload.""" script = self.queries_dir / f"{query}.sc" diff --git a/src/openultrasast/cpg/queries/config.sc b/src/openultrasast/cpg/queries/config.sc index 20947a0..28e4aea 100644 --- a/src/openultrasast/cpg/queries/config.sc +++ b/src/openultrasast/cpg/queries/config.sc @@ -7,11 +7,16 @@ // Parameters: cpgFile, settings, function (optional). // Output: fenced JSON array of {setting, line, method, args, literalArgs}. -@main def exec(cpgFile: String, settings: String, function: String = "") = { +// BATCHED, like taint.sc. Single-request form kept for the committed measurements. + +@main def exec(cpgFile: String, settings: String = "", function: String = "", requests: String = "") = { importCpg(cpgFile) def split(raw: String): List[String] = raw.split(",").map(_.trim).filter(_.nonEmpty).toList - val names = split(settings) + + def rowsFor(settingsS: String, functionS: String): List[ujson.Obj] = { + val names = split(settingsS) + val function = functionS def matches(name: String): Boolean = names.exists(n => name == n || name == n.split("\\.").last) @@ -38,7 +43,19 @@ ) } + rows + } + println("---OUSAST-CPG-BEGIN---") - println(ujson.write(ujson.Arr(rows: _*))) + if (requests.nonEmpty) { + 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("settings"), field("function")): _*) + } + println(ujson.write(ujson.Obj.from(answers))) + } else { + println(ujson.write(ujson.Arr(rowsFor(settings, function): _*))) + } println("---OUSAST-CPG-END---") } diff --git a/src/openultrasast/cpg/queries/dominance.sc b/src/openultrasast/cpg/queries/dominance.sc index 072fb52..113219a 100644 --- a/src/openultrasast/cpg/queries/dominance.sc +++ b/src/openultrasast/cpg/queries/dominance.sc @@ -22,13 +22,26 @@ // // Output: a fenced JSON array of {operation, opLine, opMethod, dominatingGuards}. -@main def exec(cpgFile: String, operations: String, dischargers: String, function: String = "") = { +// BATCHED, like taint.sc: `requests` is {id: {operations, dischargers, function}} and the output is +// {id: [row, ...]}. One JVM start answers a whole scan. The single-request form is kept so the committed +// measurements remain reproducible against the same call shape. + +@main def exec( + cpgFile: String, + operations: String = "", + dischargers: String = "", + function: String = "", + requests: String = "" +) = { importCpg(cpgFile) def split(raw: String): List[String] = raw.split(",").map(_.trim).filter(_.nonEmpty).toList - val opNames = split(operations) - val guardTokens = split(dischargers) + def rowsFor(operationsS: String, dischargersS: String, functionS: String): List[ujson.Obj] = { + + val opNames = split(operationsS) + val guardTokens = split(dischargersS) + val function = functionS // 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 @@ -69,7 +82,19 @@ ) } + rows + } + println("---OUSAST-CPG-BEGIN---") - println(ujson.write(ujson.Arr(rows: _*))) + if (requests.nonEmpty) { + 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")): _*) + } + println(ujson.write(ujson.Obj.from(answers))) + } else { + println(ujson.write(ujson.Arr(rowsFor(operations, dischargers, function): _*))) + } println("---OUSAST-CPG-END---") } diff --git a/src/openultrasast/cpg/queries/taint.sc b/src/openultrasast/cpg/queries/taint.sc index bc17391..b95a7e7 100644 --- a/src/openultrasast/cpg/queries/taint.sc +++ b/src/openultrasast/cpg/queries/taint.sc @@ -22,18 +22,29 @@ // Output: a fenced JSON array of {sink, sinkLine, sinkMethod, source, sanitized, length}. The fence exists // because Joern prints a banner, pass logs and a prompt around whatever a script emits. +// BATCHED. `requests` is a JSON object of {id: {sources, sinks, sanitizers, function, parameterSources}} and +// the output is {id: [row, ...]}. One invocation answers a whole scan's questions, because the JVM start +// (~30s) dwarfs the queries themselves once the CPG is loaded -- one call per region per family put a +// ten-line file at four minutes and a thousand regions at roughly fifty hours. +// +// The single-request form is kept below it: the pair harnesses and every measurement committed so far use it, +// and changing their call shape would silently invalidate comparisons against those artifacts. + @main def exec( cpgFile: String, - sources: String, - sinks: String, + sources: String = "", + sinks: String = "", sanitizers: String = "", function: String = "", - parameterSources: String = "false" + parameterSources: String = "false", + requests: String = "" ) = { importCpg(cpgFile) 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): 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 // attacker input. Same bug class as the dominance query's guard matching; fixed in both. @@ -45,9 +56,11 @@ .find() ) - val sourcePatterns = split(sources) - val sinkNames = split(sinks) - val sanitizerNames = split(sanitizers) + val sourcePatterns = split(sourcesS) + val sinkNames = split(sinksS) + val sanitizerNames = split(sanitizersS) + val function = functionS + val parameterSources = paramSrc // A source is any call whose code contains one of the patterns. Matching on `code` rather than a method // name keeps `request.args["x"]` (an indexAccess over a fieldAccess) and `req.body.name` both reachable @@ -126,7 +139,20 @@ } } + rows + } + println("---OUSAST-CPG-BEGIN---") - println(ujson.write(ujson.Arr(rows: _*))) + if (requests.nonEmpty) { + 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("") + val paramSrc = req.obj.get("parameterSources").map(_.str).getOrElse("false") + id -> ujson.Arr(rowsFor(field("sources"), field("sinks"), field("sanitizers"), field("function"), paramSrc): _*) + } + println(ujson.write(ujson.Obj.from(answers))) + } else { + println(ujson.write(ujson.Arr(rowsFor(sources, sinks, sanitizers, function, parameterSources): _*))) + } println("---OUSAST-CPG-END---") } diff --git a/src/openultrasast/model/config_value.py b/src/openultrasast/model/config_value.py index 3f7e50a..d2f080c 100644 --- a/src/openultrasast/model/config_value.py +++ b/src/openultrasast/model/config_value.py @@ -19,9 +19,14 @@ from .ladder import Rung, Verdict from .specs import ConfigSpec +def request_params(spec: ConfigSpec, *, function: str = "") -> dict[str, object]: + """The query parameters this arbiter sends. Shared with the batcher; see the note in `taint.request_params`.""" + return {"settings": spec.settings, "function": function} + + def verdict(cpg: CpgResult, spec: ConfigSpec, *, function: str = "") -> Verdict | None: """The verdict for a security setting in ``function``, or ``None`` when nothing is established.""" - rows = cpg.run("config", {"settings": spec.settings, "function": function}) + rows = cpg.run("config", request_params(spec, function=function)) settings = _settings(rows, spec, function=function) if not settings: return None @@ -79,4 +84,4 @@ def _settings(rows: object, spec: ConfigSpec, *, function: str) -> list[dict[str return sorted(kept, key=lambda row: (str(row["method"]), str(row["line"]), str(row["setting"]))) -__all__ = ["verdict"] +__all__ = ["request_params", "verdict"] diff --git a/src/openultrasast/model/dominance.py b/src/openultrasast/model/dominance.py index 4d104e7..3644613 100644 --- a/src/openultrasast/model/dominance.py +++ b/src/openultrasast/model/dominance.py @@ -26,12 +26,14 @@ 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 verdict(cpg: CpgResult, spec: DominanceSpec, *, function: str = "") -> Verdict | None: """The verdict for the obligated operation in ``function``, or ``None`` when there is nothing to say.""" - rows = cpg.run( - "dominance", - {"operations": spec.operations, "dischargers": spec.dischargers, "function": function}, - ) + rows = cpg.run("dominance", request_params(spec, function=function)) operations = _operations(rows) if not operations: return None @@ -85,4 +87,4 @@ def _operations(rows: object) -> list[dict[str, object]]: return kept -__all__ = ["verdict"] +__all__ = ["request_params", "verdict"] diff --git a/src/openultrasast/model/regions.py b/src/openultrasast/model/regions.py index 41b05a1..844c739 100644 --- a/src/openultrasast/model/regions.py +++ b/src/openultrasast/model/regions.py @@ -64,8 +64,11 @@ def regions_for(entries: Sequence[object], targets: Sequence[object]) -> tuple[S if language: by_path[str(getattr(target, "path", ""))] = language - regions: list[ScanRegion] = [] - covered: set[str] = set() + # One region per (path, function). The mapper can emit several entry points for a single file -- on a + # ten-line Flask file it produced three, one named and two nameless -- and turning each into a region + # scanned the same code repeatedly, multiplying Joern invocations for no new coverage. Where several + # entries agree on a region, the highest access rank wins, because that is the one worth looking at first. + best: dict[tuple[str, str | None], ScanRegion] = {} for entry in entries: path = str(getattr(entry, "path", "")) language = by_path.get(path) @@ -74,17 +77,18 @@ def regions_for(entries: Sequence[object], targets: Sequence[object]) -> tuple[S families = _families(language, has_handler=True) if not families: continue - covered.add(path) - regions.append( - ScanRegion( - path=path, - function=(getattr(entry, "function_name", None) or None), - language=language, - families=families, - rank=_ACCESS_RANK.get(str(getattr(entry, "access_level", "")), 0.5), - source="entry_point", - ) - ) + function = getattr(entry, "function_name", None) or None + rank = _ACCESS_RANK.get(str(getattr(entry, "access_level", "")), 0.5) + key = (path, function) + current = best.get(key) + if current is None or rank > current.rank: + best[key] = ScanRegion(path=path, function=function, language=language, families=families, rank=rank, source="entry_point") + + # A nameless entry point IS the file. Where a named region already covers that file it adds nothing but + # a second full-file pass, so it is dropped -- unless the file has no named region at all. + named_paths = {path for path, function in best if function is not None} + regions = [region for (path, function), region in best.items() if function is not None or path not in named_paths] + covered = {region.path for region in regions} for path, language in by_path.items(): if path in covered: diff --git a/src/openultrasast/model/scan.py b/src/openultrasast/model/scan.py index 9bfbccd..55ed23d 100644 --- a/src/openultrasast/model/scan.py +++ b/src/openultrasast/model/scan.py @@ -24,13 +24,21 @@ from dataclasses import dataclass, field from pathlib import Path from typing import Any +from ..cpg.backend import CpgResult from .candidates import enumerate_candidates +from .config_value import request_params as config_params +from .dominance import request_params as dominance_params from .ladder import Rung from .pipeline import ModelFinding, scan_region from .regions import ScanRegion from .specs import ConfigSpec, DominanceSpec, TaintSpec, config_specs, dominance_specs, taint_specs +from .taint import request_params as taint_params from .taxonomy import load_families +# The three arbiters a region can be routed to. Naming the union keeps the batcher honest: `object` +# would let a fourth spec kind reach `_grouped` and silently fall through to the taint branch. +ArbiterSpec = TaintSpec | DominanceSpec | ConfigSpec + logger = logging.getLogger(__name__) _RUNG_ORDER = {Rung.ENTAILED: 0, Rung.CORROBORATED: 1, Rung.SUSPICION: 2, Rung.EXECUTION_CONFIRMED: -1} @@ -90,36 +98,60 @@ def scan_repository( # spent in belongs to whoever holds the budget. ordered_regions = sorted(regions, key=lambda r: (-r.rank, r.path, r.function or "")) - for region in ordered_regions: - if scanned >= limits.max_regions or (counted is not None and counted.calls >= limits.max_model_calls): - degradations.append({"stage": "model", "reason": "budget_exhausted", "regions_unjudged": len(regions) - scanned}) - break + # Phase 1: collect the whole scan's questions, for at most `max_regions` regions. Nothing is asked yet. + work: list[tuple[str, ScanRegion, ArbiterSpec]] = [] + for region in ordered_regions[: limits.max_regions]: for family in region.families: spec = _spec_for(family, region.language) if spec is None: continue - try: - # Enumerate the region's candidate sites so the judge has something to answer about where the - # model resolves nothing. Without this a repository scan reports only what taint, dominance - # and constant abstraction establish unaided -- and the suspicion band, which supplied 15 of - # 26 findings in the pair measurement, would never be populated at all. - enumerated = enumerate_candidates(root, region, family, taxonomy=taxonomy) - candidates = [{"id": c.id, "text": c.text, "line": c.line} for c in enumerated.candidates] - found = scan_region( - cpg, - spec, - function=region.function or "", - client=counted, - model=model, - candidates=candidates, - parameter_sources=False, # a repository's parameters are not all attacker input - ) - except Exception as exc: # noqa: BLE001 -- one bad region must not end the scan - logger.warning("model scan failed for %s: %s", region.path, exc) - degradations.append({"stage": "model", "reason": "region_failed", "path": region.path}) - continue - collected.extend((finding, region.rank) for finding in found) - scanned += 1 + work.append((f"{len(work)}", region, spec)) + + # Phase 2: ONE invocation per query kind. JVM startup dominates a repository scan -- a call per region + # per family put a ten-line file at four minutes and a thousand regions at roughly fifty hours -- while + # the queries themselves are milliseconds once the CPG is loaded. + rows_by_id: dict[str, list[object]] = {} + for kind, requests in _grouped(work).items(): + batch = getattr(cpg, "run_batch", None) + if callable(batch): + rows_by_id.update(batch(kind, requests) or {}) + else: # a backend without batching still works, one call at a time + for rid, params in requests.items(): + answered = cpg.run(kind, params) + rows_by_id[rid] = list(answered) if isinstance(answered, list) else [] + + # Phase 3: arbitrate from the rows already in hand. The arbiters are unchanged: each is handed a + # CpgResult that simply returns its own prefetched rows. + # + # `judged` counts regions actually ARBITRATED, not regions collected: a region the budget stopped before + # is unjudged, and reporting it as scanned is the silent truncation this exists to prevent. + judged: set[tuple[str, str | None]] = set() + for rid, region, spec in work: + if counted is not None and counted.calls >= limits.max_model_calls: + degradations.append({"stage": "model", "reason": "budget_exhausted", "regions_unjudged": max(len(regions) - len(judged), 0)}) + break + family = getattr(spec, "family", "") + prefetched = CpgResult(cpg_path=cpg.cpg_path, run=_prefetched(rows_by_id.get(rid, []))) + try: + enumerated = enumerate_candidates(root, region, family, taxonomy=taxonomy) + candidates = [{"id": c.id, "text": c.text, "line": c.line} for c in enumerated.candidates] + found = scan_region( + prefetched, + spec, + function=region.function or "", + client=counted, + model=model, + candidates=candidates, + parameter_sources=False, # a repository's parameters are not all attacker input + ) + except Exception as exc: # noqa: BLE001 -- one bad region must not end the scan + logger.warning("model scan failed for %s: %s", region.path, exc) + degradations.append({"stage": "model", "reason": "region_failed", "path": region.path}) + continue + judged.add((region.path, region.function)) + collected.extend((finding, region.rank) for finding in found) + + scanned = len(judged) findings = _ordered(collected) return ModelScanResult( @@ -134,7 +166,36 @@ def scan_repository( ) -def _spec_for(family: str, language: str) -> TaintSpec | DominanceSpec | ConfigSpec | None: +def _prefetched(rows: list[object]): # type: ignore[no-untyped-def] + """A `run` that ignores the question and returns rows already fetched in the batch.""" + + def run(query: str, params: Mapping[str, object]) -> object: + del query, params + return rows + + return run + + +def _grouped(work: Sequence[tuple[str, ScanRegion, ArbiterSpec]]) -> dict[str, dict[str, Mapping[str, object]]]: + """The work, keyed by query kind, with each request built by the arbiter's OWN parameter function. + + Building the parameters here instead would be a second copy that drifts from the arbiter's -- the exact + shape of the five bugs the predecessor spent a session finding. + """ + grouped: dict[str, dict[str, Mapping[str, object]]] = {} + for rid, region, spec in work: + function = region.function or "" + if isinstance(spec, DominanceSpec): + kind, params = "dominance", dominance_params(spec, function=function) + elif isinstance(spec, ConfigSpec): + kind, params = "config", config_params(spec, function=function) + else: + kind, params = "taint", taint_params(spec, function=function, parameter_sources=False) + grouped.setdefault(kind, {})[rid] = params + return grouped + + +def _spec_for(family: str, language: str) -> ArbiterSpec | None: """The spec whose arbiter can decide this family. Routing stays a lookup; dispatch stays in `_arbitrate`.""" if family == "access_control": return dominance_specs(language=language).get(family) diff --git a/src/openultrasast/model/taint.py b/src/openultrasast/model/taint.py index bc61973..ce17457 100644 --- a/src/openultrasast/model/taint.py +++ b/src/openultrasast/model/taint.py @@ -19,6 +19,21 @@ from .ladder import Rung, Verdict from .specs import TaintSpec +def request_params(spec: TaintSpec, *, function: str = "", parameter_sources: bool = False) -> dict[str, object]: + """The query parameters this arbiter sends. Shared with the batcher so there is one definition, not two. + + Two copies of a parameter dict is precisely the duplication that produced five bugs of one shape in the + predecessor: a change made in one place and not its sibling. + """ + return { + "sources": spec.sources, + "sinks": spec.sinks, + "sanitizers": spec.sanitizers, + "function": function, + "parameterSources": "true" if parameter_sources else "false", + } + + def verdict(cpg: CpgResult, spec: TaintSpec, *, function: str = "", parameter_sources: bool = False) -> Verdict | None: """The strongest verdict the taint query supports for ``spec`` in ``function``, or ``None`` to stay at suspicion. @@ -26,16 +41,7 @@ def verdict(cpg: CpgResult, spec: TaintSpec, *, function: str = "", parameter_so for a function-level pair, where the function boundary *is* the trust boundary, and wrong for a whole repository, where most parameters carry internal values — so it is off by default and the caller opts in. """ - rows = cpg.run( - "taint", - { - "sources": spec.sources, - "sinks": spec.sinks, - "sanitizers": spec.sanitizers, - "function": function, - "parameterSources": "true" if parameter_sources else "false", - }, - ) + rows = cpg.run("taint", request_params(spec, function=function, parameter_sources=parameter_sources)) 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 @@ -146,4 +152,4 @@ def _witness(flow: Mapping[str, object]) -> str: return f"{flow['source']} -> {flow['sink']} (line {line}, {flow['length']} steps)" -__all__ = ["verdict"] +__all__ = ["request_params", "verdict"] diff --git a/tests/test_cpg_backend.py b/tests/test_cpg_backend.py index 278ed54..84f7409 100644 --- a/tests/test_cpg_backend.py +++ b/tests/test_cpg_backend.py @@ -105,3 +105,60 @@ def test_the_query_scripts_are_shipped_beside_the_seam() -> None: """The CPGQL is ours and is data: it lives with the package, not inlined in a Python string.""" queries = Path("src/openultrasast/cpg/queries") assert (queries / "taint.sc").is_file(), "the injection taint query must ship with the package" + + +def test_a_batch_is_one_invocation_carrying_many_requests(monkeypatch: pytest.MonkeyPatch) -> None: + """The cost that dominates a repository scan is JVM startup, not CPG construction. + + Every `joern --script` call starts a JVM (~30s). The driver issued one per region per family, so a + ten-line Python file -- six families, one region -- took over four minutes, and 1000 regions would take + ~50 hours. Batching collapses that to one invocation per query kind. + """ + from openultrasast.cpg.backend import BEGIN, END, JoernBackend + + calls: list[list[str]] = [] + + class _Done: + returncode = 0 + stderr = "" + stdout = f'{BEGIN}\n{{"r1": [{{"sink": "os.system(x)"}}], "r2": []}}\n{END}\n' + + def runner(command, **kwargs): # type: ignore[no-untyped-def] + calls.append(list(command)) + return _Done() + + backend = JoernBackend(runner=runner) + rows = backend.query_batch( + Path("/tmp/c.bin"), + "taint", + {"r1": {"sources": ("request.args",), "sinks": ("os.system",)}, "r2": {"sources": ("request.args",), "sinks": ("eval",)}}, + ) + assert len(calls) == 1, "a batch must be ONE invocation, whatever it carries" + assert rows["r1"] and rows["r2"] == [] + + +def test_a_failed_batch_yields_no_rows_for_any_request(monkeypatch: pytest.MonkeyPatch) -> None: + """Fail closed, as everywhere else: an engine that could not answer must not read as 'nothing here'.""" + from openultrasast.cpg.backend import JoernBackend + + class _Bad: + returncode = 1 + stdout = "" + stderr = "boom" + + backend = JoernBackend(runner=lambda c, **k: _Bad()) + assert backend.query_batch(Path("/tmp/c.bin"), "taint", {"r1": {}}) == {} + + +def test_a_batch_result_that_omits_a_request_returns_nothing_for_it() -> None: + """A missing key is not an empty answer: the caller must be able to tell them apart.""" + from openultrasast.cpg.backend import BEGIN, END, JoernBackend + + class _Partial: + returncode = 0 + stderr = "" + stdout = f'{BEGIN}\n{{"r1": []}}\n{END}\n' + + backend = JoernBackend(runner=lambda c, **k: _Partial()) + rows = backend.query_batch(Path("/tmp/c.bin"), "taint", {"r1": {}, "r2": {}}) + assert rows.get("r1") == [] and "r2" not in rows diff --git a/tests/test_model_regions.py b/tests/test_model_regions.py index 1a00626..f7be218 100644 --- a/tests/test_model_regions.py +++ b/tests/test_model_regions.py @@ -116,3 +116,37 @@ def test_an_unsupported_language_yields_no_region() -> None: from openultrasast.model.regions import regions_for assert regions_for([], [_target(path="x.rb", language="ruby")]) == () + + +def test_nameless_entry_points_do_not_duplicate_a_file_already_covered() -> None: + """The mapper can emit several entry points for one file, some with no function name. + + On a ten-line Flask file it produced three: `ping`, and two nameless ones. Turning each into a region + scanned the same file three times and multiplied the Joern invocations by three for no new coverage. + A nameless entry point is the file; if a named region already covers that file, it adds nothing. + """ + from openultrasast.model.regions import regions_for + + entries = [ + _entry(function="ping", access="public"), + _entry(function=None, access="local-only"), + _entry(function=None, access="public"), + ] + regions = regions_for(entries, [_target()]) + assert len(regions) == 1 + assert regions[0].function == "ping" + + +def test_a_file_with_only_nameless_entry_points_still_gets_one_region() -> None: + from openultrasast.model.regions import regions_for + + regions = regions_for([_entry(function=None), _entry(function=None)], [_target()]) + assert len(regions) == 1 and regions[0].function is None + + +def test_two_named_handlers_in_one_file_are_two_regions() -> None: + """Deduping must not collapse genuinely distinct handlers.""" + from openultrasast.model.regions import regions_for + + regions = regions_for([_entry(function="a"), _entry(function="b")], [_target()]) + assert sorted(r.function for r in regions) == ["a", "b"] diff --git a/tests/test_model_scan.py b/tests/test_model_scan.py index 54e4731..aff2e87 100644 --- a/tests/test_model_scan.py +++ b/tests/test_model_scan.py @@ -145,3 +145,44 @@ def test_a_region_is_only_asked_about_families_it_carries() -> None: scan_repository(Path("/repo"), (_region(families=("injection",)),), backend=_B(), client=None, model="") assert "dominance" not in asked and "config" not in asked + + +def test_the_driver_issues_one_query_invocation_per_kind_not_one_per_region(tmp_path) -> None: + """Task 1.4: JVM startup, not CPG construction, is what dominates a repository scan. + + Before batching the driver issued one `joern --script` call per region per family. A ten-line Python file + admits six families, so it cost six JVM launches and over four minutes; a thousand regions would have + taken about fifty hours. The whole scan's questions now go in one invocation per query kind. + """ + from openultrasast.model.scan import scan_repository + + invocations: list[str] = [] + + class _CountingBackend: + def available(self): # type: ignore[no-untyped-def] + return True + + def build(self, root): # type: ignore[no-untyped-def] + from openultrasast.cpg.backend import CpgResult + + def run(query, params): # type: ignore[no-untyped-def] + invocations.append(f"single:{query}") + return [] + + def run_batch(query, requests): # type: ignore[no-untyped-def] + invocations.append(f"batch:{query}") + return {rid: [] for rid in requests} + + result = CpgResult(cpg_path=Path("c.bin"), run=run) + object.__setattr__(result, "run_batch", run_batch) + return result + + source = "import os\n\n\ndef run(cmd):\n os.system(cmd)\n" + for i in range(30): + (tmp_path / f"f{i}.py").write_text(source) + regions = tuple(_region(path=f"f{i}.py", families=("injection", "path")) for i in range(30)) + + scan_repository(tmp_path, regions, backend=_CountingBackend(), client=None, model="") + assert not any(i.startswith("single:") for i in invocations), f"unbatched calls remain: {invocations[:4]}" + # 30 regions x 2 families would have been 60 invocations; batching makes it one per query kind. + assert len(invocations) <= 3, f"expected one invocation per query kind, got {len(invocations)}"