From 2b973dc0db6998290c8f344863c35f39394321d0 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Tue, 8 Sep 2026 12:23:40 +0200 Subject: [PATCH] feat(model-grounded-detection): tasks 2.1-2.2, the Joern seam and the taint verdict The whole engine integration is one subprocess boundary: joern-parse builds a cpg.bin, joern --script runs a shipped CPGQL query that prints its JSON between fence markers, because Joern's stdout carries a banner and a REPL prompt around whatever a script emits. Nothing is imported in-process, so the core install keeps dependencies = [] -- a test walks the seam's AST to assert it imports only the standard library and names no JVM binding. Fail closed everywhere. A build or query that errors, times out, or prints something unparseable yields no verdict and the caller reports suspicion. The taint verdict distinguishes this from absence deliberately: a query returning None means "the engine could not decide", which is not "there is no flow", and conflating them is how a missing engine silently becomes a clean bill of health. The verdict is where the ladder earns its keep. An unsanitized source->sink path is model_entailed -- the graph establishes it and the LLM's claim is no longer load-bearing. A path a sanitizer may break is model_corroborated: the flow is real, but whether the sanitizer suffices is the semantic residual the LLM answers, checked against this verdict rather than averaged over runs. The chosen flow is picked by a total order (shortest, then lexicographic), so two runs over one CPG cannot disagree -- Req 6.4 asserted directly. The query itself is the part our flat IR could not do: reachableByFlows walks the PDG, so `q = "..." + request.args["x"]; execute(q)` is found, which is exactly the one-hop limit that held measured entailment at 2.2%. Capability detection reuses semantic/engines.joern_available rather than growing a second notion of "installed". The audit manifest is regenerated; the new modules are named by Req 4/5/6 rather than counted as orphans. Task 2.3, the go/no-go measurement, needs Joern installed and has not run. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01T7XD7x5KonMdmGRkp63EQG --- .../measurements/2026-09-08-module-audit.json | 54 +++++- src/openultrasast/cpg/__init__.py | 3 + src/openultrasast/cpg/backend.py | 168 ++++++++++++++++++ src/openultrasast/cpg/capability.py | 18 ++ src/openultrasast/cpg/queries/taint.sc | 61 +++++++ src/openultrasast/model/audit.py | 5 + src/openultrasast/model/ladder.py | 48 +++++ src/openultrasast/model/taint.py | 73 ++++++++ tests/test_cpg_backend.py | 94 ++++++++++ tests/test_model_taint.py | 113 ++++++++++++ 10 files changed, 632 insertions(+), 5 deletions(-) create mode 100644 src/openultrasast/cpg/__init__.py create mode 100644 src/openultrasast/cpg/backend.py create mode 100644 src/openultrasast/cpg/capability.py create mode 100644 src/openultrasast/cpg/queries/taint.sc create mode 100644 src/openultrasast/model/ladder.py create mode 100644 src/openultrasast/model/taint.py create mode 100644 tests/test_cpg_backend.py create mode 100644 tests/test_model_taint.py diff --git a/benchmarks/measurements/2026-09-08-module-audit.json b/benchmarks/measurements/2026-09-08-module-audit.json index 7cefec7..e2f5a5b 100644 --- a/benchmarks/measurements/2026-09-08-module-audit.json +++ b/benchmarks/measurements/2026-09-08-module-audit.json @@ -1,13 +1,13 @@ { "counts": { - "load_bearing": 74, + "load_bearing": 79, "standalone_capability": 7 }, - "generated": "2026-09-08", + "generated": "2026-09-08 (regenerated after group 2 added the cpg/ seam)", "kept_with_a_lever_stripped": { "improve/evolve.py + improve/validator.py": "the design's Removed list named these whole, but each is a mix: the mechanism lever (removed) and the deterministic rule-status improve loop that `ousast improve` still runs from benchmark evidence, which has no LLM in it and is not what this feature supersedes. Deleting the files would have taken a retained capability with them." }, - "modules": 81, + "modules": 86, "note": "Classification of every source module after the group-1 subtraction. load_bearing = reachable from a gate, the CLI, or named by a surviving requirement; standalone_capability = its own entry point or protocol, deliberately off the scan path; orphaned = no importer, no entry point, no surviving requirement -- removed, never carried.", "orphaned": [], "removed_in_this_pass": { @@ -124,6 +124,31 @@ "module": "config", "reason": "reachable from gate, map_gate, pair_gate or the CLI" }, + { + "classification": "load_bearing", + "importers": [ + "cpg.backend", + "model.taint" + ], + "module": "cpg", + "reason": "model-grounded-detection Req 4: the CPG seam package" + }, + { + "classification": "load_bearing", + "importers": [ + "model.taint" + ], + "module": "cpg.backend", + "reason": "model-grounded-detection Req 4.1/4.5: the single Joern subprocess boundary" + }, + { + "classification": "load_bearing", + "importers": [ + "cpg.backend" + ], + "module": "cpg.capability", + "reason": "model-grounded-detection Req 4.2/4.3: the engine probe that degrades to suspicion" + }, { "classification": "load_bearing", "importers": [ @@ -295,7 +320,8 @@ "classification": "load_bearing", "importers": [ "cli", - "model.specs" + "model.specs", + "model.taint" ], "module": "model", "reason": "reachable from gate, map_gate, pair_gate or the CLI" @@ -322,10 +348,26 @@ }, { "classification": "load_bearing", - "importers": [], + "importers": [ + "model.taint" + ], + "module": "model.ladder", + "reason": "model-grounded-detection Req 5: the evidence ladder every finding carries" + }, + { + "classification": "load_bearing", + "importers": [ + "model.taint" + ], "module": "model.specs", "reason": "model-grounded-detection Req 7.2-7.4: the security vocabularies the CPG queries are parameterised by" }, + { + "classification": "load_bearing", + "importers": [], + "module": "model.taint", + "reason": "model-grounded-detection Req 6.1: taint reachability over the CPG, the flow-family arbiter" + }, { "classification": "load_bearing", "importers": [ @@ -598,6 +640,7 @@ "classification": "load_bearing", "importers": [ "cli", + "cpg.capability", "hunter_tools", "model.candidates", "model.specs", @@ -627,6 +670,7 @@ { "classification": "load_bearing", "importers": [ + "cpg.capability", "semantic", "semantic.overlay" ], diff --git a/src/openultrasast/cpg/__init__.py b/src/openultrasast/cpg/__init__.py new file mode 100644 index 0000000..0b73e52 --- /dev/null +++ b/src/openultrasast/cpg/__init__.py @@ -0,0 +1,3 @@ +"""The Code Property Graph seam: the adopted engine the model layer arbitrates over.""" + +from __future__ import annotations diff --git a/src/openultrasast/cpg/backend.py b/src/openultrasast/cpg/backend.py new file mode 100644 index 0000000..e6671f5 --- /dev/null +++ b/src/openultrasast/cpg/backend.py @@ -0,0 +1,168 @@ +"""The Joern subprocess seam — the single place this codebase touches a CPG engine (Req 4.1, 4.4, 4.5). + +Joern is JVM/Scala. Everything crosses one subprocess boundary: ``joern-parse`` builds a ``cpg.bin`` from a +source tree, then ``joern --script`` runs a CPGQL script that prints a JSON payload. Nothing is imported +in-process, so the core install keeps ``dependencies = []`` and an absent engine costs a recorded reason +rather than a traceback. + +Two rules hold everywhere below: + +* **Fail closed.** A build or query that errors, times out, or prints something unparseable yields *no* + verdict. The model layer then reports ``suspicion``. A wrong verdict is worse than no verdict — that is the + whole reason this arbiter exists. +* **Fence the payload.** Joern prints a banner, compilation notices and a REPL prompt on stdout. The script + brackets its JSON between markers so parsing can never swallow the noise around it. + +We do not build a control-flow or dataflow engine here (Req 4.4): the CFG, PDG and call graph our flat IR +lacks are exactly what is being adopted. +""" + +from __future__ import annotations + +import json +import logging +import os +import shutil +import subprocess +import tempfile +from collections.abc import Callable, Mapping +from dataclasses import dataclass, field +from pathlib import Path +from typing import Protocol + +logger = logging.getLogger(__name__) + +QUERIES_DIR = Path(__file__).resolve().parent / "queries" + +# The payload fence. Joern's stdout carries a banner and a prompt; the script prints these around its JSON. +BEGIN = "---OUSAST-CPG-BEGIN---" +END = "---OUSAST-CPG-END---" + +BUILD_TIMEOUT_SECONDS = 900 # a large tree takes minutes; the go/no-go records what it actually cost +QUERY_TIMEOUT_SECONDS = 300 + +Runner = Callable[..., subprocess.CompletedProcess[str]] + + +def extract_payload(stdout: str) -> object | None: + """The JSON between the fence markers, or ``None`` when it is absent or malformed.""" + start = stdout.find(BEGIN) + end = stdout.find(END, start + 1) if start >= 0 else -1 + if start < 0 or end < 0: + return None + body = stdout[start + len(BEGIN) : end].strip() + try: + payload: object = json.loads(body) + except json.JSONDecodeError: + logger.debug("cpg query payload was not valid JSON") + return None + return payload + + +@dataclass(frozen=True) +class CpgResult: + """A built CPG and the way to query it. Opaque: only ``model/`` reads through ``run``.""" + + cpg_path: Path + run: Callable[[str, Mapping[str, object]], object | None] + + +class CpgBackend(Protocol): + def available(self) -> bool: ... + + def build(self, root: Path) -> CpgResult | None: ... + + +@dataclass +class NullBackend: + """No engine. Every verdict degrades to ``suspicion`` and the caller records ``cpg_unavailable``.""" + + reason: str = "cpg_unavailable" + + def available(self) -> bool: + return False + + def build(self, root: Path) -> CpgResult | None: + del root + return None + + +@dataclass +class JoernBackend: + """``joern-parse`` then ``joern --script``, both out of process and both timeout-bounded.""" + + runner: Runner | None = None + build_timeout: int = BUILD_TIMEOUT_SECONDS + query_timeout: int = QUERY_TIMEOUT_SECONDS + queries_dir: Path = field(default_factory=lambda: QUERIES_DIR) + + def available(self) -> bool: + from .capability import has_cpg + + return has_cpg() + + def build(self, root: Path) -> CpgResult | None: + """Build a CPG for ``root``. ``None`` on any failure — the caller degrades, never guesses.""" + parse = shutil.which("joern-parse") + if parse is None and os.environ.get("OPENULTRASAST_JOERN_PROBE", "").strip().lower() not in {"1", "on", "true", "yes"}: + return None + scratch = Path(tempfile.mkdtemp(prefix="ousast-cpg-")) + cpg_path = scratch / "cpg.bin" + command = [parse or "joern-parse", str(root), "--output", str(cpg_path)] + completed = self._run(command, timeout=self.build_timeout) + if completed is None or completed.returncode != 0 or not cpg_path.is_file(): + detail = (completed.stderr or completed.stdout or "")[-400:] if completed is not None else "timeout" + logger.warning("cpg build failed for %s: %s", root, detail) + return None + return CpgResult(cpg_path=cpg_path, run=lambda query, params: self.query(cpg_path, query, params)) + + 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" + if not script.is_file(): + logger.warning("no such cpg query: %s", script) + return None + command = [shutil.which("joern") or "joern", "--script", str(script), "--param", f"cpgFile={cpg_path}"] + for key, value in sorted(params.items()): + command += ["--param", f"{key}={_render(value)}"] + completed = self._run(command, timeout=self.query_timeout) + if completed is None or completed.returncode != 0: + detail = (completed.stderr or "")[-400:] if completed is not None else "timeout" + logger.warning("cpg query %s failed: %s", query, detail) + return None + return extract_payload(completed.stdout or "") + + def _run(self, command: list[str], *, timeout: int) -> subprocess.CompletedProcess[str] | None: + run = self.runner if self.runner is not None else subprocess.run + try: + return run(command, capture_output=True, text=True, timeout=timeout, check=False) + except (OSError, subprocess.SubprocessError): + return None + + +def _render(value: object) -> str: + """CPGQL parameters are strings; a sequence becomes a comma-separated list the script splits.""" + if isinstance(value, str): + return value + if isinstance(value, (list, tuple, set, frozenset)): + return ",".join(sorted(str(item) for item in value)) + return str(value) + + +def resolve_cpg_backend(*, runner: Runner | None = None) -> CpgBackend: + """The backend for this machine: Joern when it is installed, else the null one.""" + from .capability import has_cpg + + return JoernBackend(runner=runner) if has_cpg() else NullBackend() + + +__all__ = [ + "BEGIN", + "END", + "CpgBackend", + "CpgResult", + "JoernBackend", + "NullBackend", + "extract_payload", + "resolve_cpg_backend", +] diff --git a/src/openultrasast/cpg/capability.py b/src/openultrasast/cpg/capability.py new file mode 100644 index 0000000..7637c01 --- /dev/null +++ b/src/openultrasast/cpg/capability.py @@ -0,0 +1,18 @@ +"""Is a CPG engine available? (Req 4.2, 4.3) + +One probe, not two: ``semantic/engines.joern_available`` already answers this and already honours the +``OPENULTRASAST_JOERN_PROBE`` override that the test matrix uses to force the absent case. The model layer +reuses it rather than growing a second, subtly different notion of "installed". +""" + +from __future__ import annotations + +from ..semantic.engines import joern_available + + +def has_cpg() -> bool: + """True when a CPG can be built. False means every verdict degrades to ``suspicion`` with a reason.""" + return joern_available() + + +__all__ = ["has_cpg"] diff --git a/src/openultrasast/cpg/queries/taint.sc b/src/openultrasast/cpg/queries/taint.sc new file mode 100644 index 0000000..0e1ab41 --- /dev/null +++ b/src/openultrasast/cpg/queries/taint.sc @@ -0,0 +1,61 @@ +// Taint reachability for one flow family (model-grounded-detection Req 6.1). +// +// The question this answers is the one our flat IR could not: does an untrusted source reach the labeled sink +// *through resolved dataflow*, and is there a sanitizer on that path. Joern's `reachableByFlows` walks the +// PDG, so a value bound three statements upstream — `q = "..." + request.args["x"]; execute(q)` — is found, +// which is exactly the one-hop limit that held our measured entailment at 2.2%. +// +// Parameters (all comma-separated where plural): +// cpgFile the built cpg.bin +// sources source matchers, e.g. request.args,request.form +// sinks sink call names, e.g. execute,os.system +// sanitizers sanitizer call names; a flow through one of these is corroborated, not entailed +// function optional: restrict sinks to this enclosing method +// +// Output: a fenced JSON array of {sink, sinkLine, sinkMethod, source, sanitized, length}. +// The fence exists because Joern prints a banner and a prompt around whatever a script emits. + +@main def exec(cpgFile: String, sources: String, sinks: String, sanitizers: String = "", function: String = "") = { + importCpg(cpgFile) + + def split(raw: String): List[String] = + raw.split(",").map(_.trim).filter(_.nonEmpty).toList + + val sourcePatterns = split(sources) + val sinkNames = split(sinks) + val sanitizerNames = split(sanitizers) + + // A source is any call or identifier whose code matches one of the source patterns. Matching on `code` + // rather than on a method name keeps `request.args["x"]` and `req.body.name` both reachable without + // per-framework special cases. + def sourceNodes = cpg.call.filter(c => sourcePatterns.exists(p => c.code.contains(p))).l ++ + cpg.identifier.filter(i => sourcePatterns.exists(p => i.code.contains(p))).l + + // A sink is a call whose name (or full code) names one of the family's sinks, optionally inside `function`. + def sinkNodes = { + val all = cpg.call.filter(c => sinkNames.exists(n => c.name == n || c.code.startsWith(n) || c.methodFullName.contains(n))).l + if (function.isEmpty) all else all.filter(_.method.name == function) + } + + def isSanitized(flowCode: List[String]): Boolean = + sanitizerNames.nonEmpty && flowCode.exists(code => sanitizerNames.exists(s => code.contains(s))) + + val rows = sinkNodes.flatMap { sink => + val flows = sourceNodes.reachableByFlows(sink).l + flows.map { flow => + val elements = flow.elements.map(_.code).l + ujson.Obj( + "sink" -> sink.code.take(200), + "sinkLine" -> sink.lineNumber.getOrElse(-1).toString, + "sinkMethod" -> sink.method.name, + "source" -> elements.headOption.getOrElse("").take(200), + "sanitized" -> isSanitized(elements), + "length" -> elements.size + ) + } + } + + println("---OUSAST-CPG-BEGIN---") + println(ujson.write(ujson.Arr(rows: _*))) + println("---OUSAST-CPG-END---") +} diff --git a/src/openultrasast/model/audit.py b/src/openultrasast/model/audit.py index 38a32b8..264cd70 100644 --- a/src/openultrasast/model/audit.py +++ b/src/openultrasast/model/audit.py @@ -31,6 +31,11 @@ SPEC_OWNED = { "model.specs": "model-grounded-detection Req 7.2-7.4: the security vocabularies the CPG queries are parameterised by", "model.endpoint": "model-grounded-detection Req 8: the judge client, wired by model/judge.py in group 3", "model.audit": "model-grounded-detection Req 3: this classifier; run by the maintainer, not by the scan", + "model.ladder": "model-grounded-detection Req 5: the evidence ladder every finding carries", + "model.taint": "model-grounded-detection Req 6.1: taint reachability over the CPG, the flow-family arbiter", + "cpg": "model-grounded-detection Req 4: the CPG seam package", + "cpg.backend": "model-grounded-detection Req 4.1/4.5: the single Joern subprocess boundary", + "cpg.capability": "model-grounded-detection Req 4.2/4.3: the engine probe that degrades to suspicion", } STANDALONE = { diff --git a/src/openultrasast/model/ladder.py b/src/openultrasast/model/ladder.py new file mode 100644 index 0000000..1e85fb7 --- /dev/null +++ b/src/openultrasast/model/ladder.py @@ -0,0 +1,48 @@ +"""The evidence ladder: what established a finding (model-grounded-detection Req 5). + +Every finding carries the rung that established it, so a deterministic model result is never confused with an +LLM suspicion. The two middle rungs are the point of this feature: + + suspicion a site was enumerated and something flagged it + model_corroborated the LLM's claim is consistent with the CPG -- the source it names reaches the sink it + names, or a guard its siblings carry is absent + model_entailed the CPG establishes it alone, independent of trusting the LLM + execution_confirmed a sandbox reproduced it (deferred tier, memory/C first) + +`model_corroborated` and above are established with no model call, and deterministically: run the checker +twice over the same CPG and the verdict sequence is identical. That is what removes the run-to-run noise the +prior architecture tried to average away. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import StrEnum + + +class Rung(StrEnum): + SUSPICION = "suspicion" + CORROBORATED = "model_corroborated" + ENTAILED = "model_entailed" + EXECUTION_CONFIRMED = "execution_confirmed" + + +# Ascending strength; `_ORDER.index` is the comparison, so a new rung cannot be silently mis-ranked. +_ORDER = (Rung.SUSPICION, Rung.CORROBORATED, Rung.ENTAILED, Rung.EXECUTION_CONFIRMED) + + +def at_or_above(rung: Rung, threshold: Rung) -> bool: + return _ORDER.index(rung) >= _ORDER.index(threshold) + + +@dataclass(frozen=True) +class Verdict: + """What the model established, and the evidence it established it from.""" + + rung: Rung + family: str + witness: str = "" # the taint path or the dominance fact; empty only at `suspicion` + contradiction: str = "" # why an LLM claim was dropped, when it was + + +__all__ = ["Rung", "Verdict", "at_or_above"] diff --git a/src/openultrasast/model/taint.py b/src/openultrasast/model/taint.py new file mode 100644 index 0000000..3c1b688 --- /dev/null +++ b/src/openultrasast/model/taint.py @@ -0,0 +1,73 @@ +"""Taint reachability over the CPG: the arbiter for the flow families (Req 6.1). + +The verdict is the static analogue of a reproduced crash. An unsanitized source→sink path is `ENTAILED` — the +graph establishes it and the LLM's claim is no longer load-bearing. A path a sanitizer may break is +`CORROBORATED`: the flow is real, but whether the sanitizer is *sufficient* is a semantic judgement the model +cannot make, and that residual is the LLM's — checked against this verdict rather than averaged over runs. + +No path at all yields `None`, not a negative verdict: the caller reports `suspicion`. And a query that failed +yields `None` too, because "the engine could not decide" is not "there is no flow" — conflating them is how a +missing engine would silently become a clean bill of health. +""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence + +from ..cpg.backend import CpgResult +from .ladder import Rung, Verdict +from .specs import TaintSpec + + +def verdict(cpg: CpgResult, spec: TaintSpec, *, function: str = "") -> Verdict | None: + """The strongest verdict the taint query supports for ``spec`` in ``function``, or ``None`` to stay at suspicion.""" + rows = cpg.run( + "taint", + {"sources": spec.sources, "sinks": spec.sinks, "sanitizers": spec.sanitizers, "function": function}, + ) + flows = _flows(rows, function=function) + if not flows: + return None + unsanitized = [flow for flow in flows if not flow["sanitized"]] + # Deterministic by construction: sort, then take the first. Nothing here depends on engine ordering. + chosen = min(unsanitized or flows, key=_rank) + rung = Rung.ENTAILED if not chosen["sanitized"] else Rung.CORROBORATED + return Verdict(rung=rung, family=spec.family, witness=_witness(chosen)) + + +def _flows(rows: object, *, function: str) -> list[dict[str, object]]: + """The query's rows, filtered to the labeled function. A non-list (a failure) yields nothing.""" + if not isinstance(rows, Sequence) or isinstance(rows, (str, bytes)): + return [] + kept: list[dict[str, object]] = [] + for row in rows: + if not isinstance(row, Mapping): + continue + if function and str(row.get("sinkMethod", "")) != function: + continue + kept.append({ + "sink": str(row.get("sink", "")), + "sinkLine": str(row.get("sinkLine", "")), + "sinkMethod": str(row.get("sinkMethod", "")), + "source": str(row.get("source", "")), + "sanitized": bool(row.get("sanitized", False)), + "length": _as_int(row.get("length")), + }) + return kept + + +def _as_int(value: object) -> int: + return int(value) if isinstance(value, int | float) else 0 + + +def _rank(flow: Mapping[str, object]) -> tuple[int, str, str]: + """Shortest flow first, then lexicographic — a total order, so two runs cannot disagree.""" + return (_as_int(flow["length"]), str(flow["source"]), str(flow["sink"])) + + +def _witness(flow: Mapping[str, object]) -> str: + line = flow.get("sinkLine") or "?" + return f"{flow['source']} -> {flow['sink']} (line {line}, {flow['length']} steps)" + + +__all__ = ["verdict"] diff --git a/tests/test_cpg_backend.py b/tests/test_cpg_backend.py new file mode 100644 index 0000000..a5d2d60 --- /dev/null +++ b/tests/test_cpg_backend.py @@ -0,0 +1,94 @@ +"""model-grounded-detection task 2.1: the Joern seam, out of process and capability-detected. + +Joern is a JVM tool. The whole integration is one subprocess boundary so that the core install stays +zero-dependency and an absent engine degrades to `suspicion` with a recorded reason rather than a traceback. +Nothing here imports a JVM binding, and every test runs with Joern absent. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + + +def test_without_joern_the_resolved_backend_is_the_null_one(monkeypatch: pytest.MonkeyPatch) -> None: + from openultrasast.cpg.backend import NullBackend, resolve_cpg_backend + + monkeypatch.setenv("OPENULTRASAST_JOERN_PROBE", "off") + backend = resolve_cpg_backend() + assert isinstance(backend, NullBackend) + assert backend.available() is False + assert backend.build(Path(".")) is None, "an absent engine yields no CPG, and the caller degrades" + + +def test_the_capability_probe_reuses_the_existing_joern_probe(monkeypatch: pytest.MonkeyPatch) -> None: + """`semantic/engines.joern_available` already exists; the model layer must not grow a second probe.""" + from openultrasast.cpg.capability import has_cpg + + monkeypatch.setenv("OPENULTRASAST_JOERN_PROBE", "off") + assert has_cpg() is False + monkeypatch.setenv("OPENULTRASAST_JOERN_PROBE", "on") + assert has_cpg() is True + + +def test_the_seam_imports_nothing_outside_the_standard_library() -> None: + """Req 4.2: the core install has no dependencies, so the seam may not import one at module load.""" + import ast + + source = Path("src/openultrasast/cpg/backend.py").read_text() + imported = set() + for node in ast.walk(ast.parse(source)): + if isinstance(node, ast.ImportFrom) and node.module and node.level == 0: + imported.add(node.module.split(".")[0]) + elif isinstance(node, ast.Import): + imported.update(alias.name.split(".")[0] for alias in node.names) + stdlib = {"__future__", "json", "os", "re", "shutil", "subprocess", "tempfile", "dataclasses", "pathlib", "typing", + "collections", "logging", "hashlib"} + assert imported <= stdlib, f"the seam imports outside the standard library: {sorted(imported - stdlib)}" + assert "import jpype" not in source and "py4j" not in source, "Joern is reached by subprocess, never in-process" + + +def test_a_scripted_backend_parses_a_query_result_without_joern(tmp_path: Path) -> None: + """The parsing is exercised by canned JSON, so the query contract is testable with no engine installed.""" + from openultrasast.cpg.backend import CpgResult + + payload = [{"sink": "os.system", "source": "request.args", "path": ["a", "b"], "sanitized": False}] + + def run(query: str, params: dict[str, object]) -> object: + assert query == "taint" + return payload + + result = CpgResult(cpg_path=tmp_path / "cpg.bin", run=run) + assert result.run("taint", {}) == payload + + +def test_joern_output_is_read_from_a_delimited_block_not_from_noisy_stdout() -> None: + """Joern prints a banner and a REPL prompt; the payload is fenced so the parse cannot swallow either.""" + from openultrasast.cpg.backend import BEGIN, END, extract_payload + + noisy = f"Compiling...\nJoern v4.0\n{BEGIN}\n{json.dumps([{'sink': 'eval'}])}\n{END}\njoern> \n" + assert extract_payload(noisy) == [{"sink": "eval"}] + assert extract_payload("no markers at all") is None + assert extract_payload(f"{BEGIN}\nnot json\n{END}") is None + + +def test_a_failing_or_slow_engine_fails_closed(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Fail closed: a build or query that errors yields no verdict, never a wrong one.""" + import subprocess + + from openultrasast.cpg.backend import JoernBackend + + def boom(*args: object, **kwargs: object) -> object: + raise subprocess.TimeoutExpired(cmd="joern-parse", timeout=1) + + backend = JoernBackend(runner=boom) + monkeypatch.setenv("OPENULTRASAST_JOERN_PROBE", "on") + assert backend.build(tmp_path) is None + + +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" diff --git a/tests/test_model_taint.py b/tests/test_model_taint.py new file mode 100644 index 0000000..bb0fbb6 --- /dev/null +++ b/tests/test_model_taint.py @@ -0,0 +1,113 @@ +"""model-grounded-detection task 2.2: the taint verdict, and what separates entailed from corroborated. + +An `ENTAILED` verdict is the static analogue of a reproduced crash: the CPG establishes the flow on its own and +the LLM's assertion is no longer load-bearing. `CORROBORATED` means a path exists but a sanitizer may break it, +so the judgement of sufficiency is the LLM's residual — checked against the graph, never averaged over runs. +""" + +from __future__ import annotations + +from pathlib import Path + + +def _cpg(rows: list[dict[str, object]]): # type: ignore[no-untyped-def] + from openultrasast.cpg.backend import CpgResult + + return CpgResult(cpg_path=Path("cpg.bin"), run=lambda query, params: rows) + + +def _spec(): # type: ignore[no-untyped-def] + from openultrasast.model.specs import TaintSpec + + return TaintSpec( + family="injection", + language="python", + sources=("request.args",), + sinks=("os.system", "execute"), + sanitizers=("escape", "quote"), + ) + + +def test_an_unsanitized_source_to_sink_path_is_entailed() -> None: + from openultrasast.model.ladder import Rung + from openultrasast.model.taint import verdict + + rows = [ + {"sink": "os.system(cmd)", "sinkLine": "4", "sinkMethod": "run", "source": "request.args['x']", "sanitized": False, "length": 3} + ] + answer = verdict(_cpg(rows), _spec(), function="run") + assert answer is not None and answer.rung is Rung.ENTAILED + assert answer.family == "injection" + assert "request.args" in answer.witness and "os.system" in answer.witness, "the witness names the flow it found" + + +def test_a_path_through_a_sanitizer_is_corroborated_not_entailed() -> None: + """The model saw the flow but cannot judge whether the sanitizer is sufficient. That residual is the LLM's.""" + from openultrasast.model.ladder import Rung + from openultrasast.model.taint import verdict + + rows = [{"sink": "execute(q)", "sinkLine": "9", "sinkMethod": "run", "source": "request.args['q']", "sanitized": True, "length": 4}] + answer = verdict(_cpg(rows), _spec(), function="run") + assert answer is not None and answer.rung is Rung.CORROBORATED + + +def test_no_path_yields_no_verdict_so_the_caller_reports_suspicion() -> None: + from openultrasast.model.taint import verdict + + assert verdict(_cpg([]), _spec(), function="run") is None + + +def test_a_query_that_failed_is_not_read_as_an_absence_of_flow() -> None: + """`None` from the engine means "could not decide", which is not the same as "no path" — Req 4.3.""" + from openultrasast.cpg.backend import CpgResult + from openultrasast.model.taint import verdict + + broken = CpgResult(cpg_path=Path("cpg.bin"), run=lambda query, params: None) + assert verdict(broken, _spec(), function="run") is None + + +def test_the_verdict_is_restricted_to_the_labeled_function() -> None: + from openultrasast.model.taint import verdict + + rows = [ + { + "sink": "os.system(x)", "sinkLine": "40", "sinkMethod": "elsewhere", + "source": "request.args['x']", "sanitized": False, "length": 2, + } + ] + assert verdict(_cpg(rows), _spec(), function="run") is None, "a flow in another function does not judge this one" + + +def test_the_shortest_unsanitized_flow_wins_and_the_sequence_is_deterministic() -> None: + """Req 6.4: two runs over one CPG give an identical verdict sequence, so there is nothing to average.""" + from openultrasast.model.ladder import Rung + from openultrasast.model.taint import verdict + + rows = [ + {"sink": "os.system(a)", "sinkLine": "7", "sinkMethod": "run", "source": "request.args['b']", "sanitized": True, "length": 9}, + {"sink": "os.system(a)", "sinkLine": "7", "sinkMethod": "run", "source": "request.args['a']", "sanitized": False, "length": 2}, + ] + first = verdict(_cpg(rows), _spec(), function="run") + second = verdict(_cpg(rows), _spec(), function="run") + assert first == second + assert first is not None and first.rung is Rung.ENTAILED and "request.args['a']" in first.witness + + +def test_the_spec_drives_the_query_parameters() -> None: + """The security modelling is ours; the traversal is the engine's. The spec is what crosses the seam.""" + from openultrasast.cpg.backend import CpgResult + from openultrasast.model.taint import verdict + + seen: dict[str, object] = {} + + def run(query: str, params: dict[str, object]) -> object: + seen["query"] = query + seen.update(params) + return [] + + verdict(CpgResult(cpg_path=Path("cpg.bin"), run=run), _spec(), function="run") + assert seen["query"] == "taint" + assert seen["sources"] == ("request.args",) + assert seen["sinks"] == ("os.system", "execute") + assert seen["sanitizers"] == ("escape", "quote") + assert seen["function"] == "run"