From 431ab1e738568b4d8099181965dc993f61892fb1 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sun, 4 Oct 2026 08:19:46 +0200 Subject: [PATCH] learn: engine_trace materialises catalog pairs without commits (dedented parseable excerpts, in-function scope), protects labelled files in exports, scopes large repos, records frontend diagnostics, retries the jssrc2cpg engine bug, writes progress.json Implemented with Codex. Co-Authored-By: Claude Opus 5.5 --- benchmarks/learn/engine_trace.py | 210 +++++++++++++++++-- benchmarks/learn/engine_trace_worker.py | 79 ++++++- tests/test_engine_trace.py | 261 ++++++++++++++++++++++++ 3 files changed, 526 insertions(+), 24 deletions(-) diff --git a/benchmarks/learn/engine_trace.py b/benchmarks/learn/engine_trace.py index baffc74..e4d0a7d 100644 --- a/benchmarks/learn/engine_trace.py +++ b/benchmarks/learn/engine_trace.py @@ -7,6 +7,7 @@ is git archive HEAD src plus a hashed snapshot of this uncommitted tool. from __future__ import annotations import argparse +import ast import fcntl import hashlib import json @@ -15,6 +16,7 @@ import re import shutil import subprocess import tempfile +import textwrap import time import tomllib from collections import Counter, defaultdict @@ -41,7 +43,7 @@ SNAPSHOT_FILES = ( "src/openultrasast/model/taint.py", "src/openultrasast/cpg/queries/taint.sc", ) -MATERIALIZATION_VERSION = 2 +MATERIALIZATION_VERSION = 3 EXPORT_CAP = 20 * 1024 * 1024 STATUSES = ("path", "asked-nothing", "failed", "timeout", "unsupported") @@ -268,7 +270,8 @@ class Inputs: def export_local(self, clone, pin, target): with tempfile.TemporaryDirectory(prefix="trace-export-", dir=target.parent) as scratch: export_pin(clone, pin["pin"], Path(scratch)) - proof = export_sources(Path(scratch), target, pin["units"]) + roots = sorted({source_root(r["file"], r["language"]) for r in pin["units"]}) + proof = export_sources(Path(scratch), target, pin["units"], roots) return {**proof, "mode": "source", "commit": pin["pin"]} def materialize(self, pin, target): @@ -279,7 +282,43 @@ class Inputs: raise ValueError(row["mapping_error"]) if any(r.get("excerpt") or r["source"].startswith("advisory-fixes") for r in rows): resolved = [self.pair_for(row) for row in rows] - identities = {self.source_identity(case, side) for case, side in resolved} + identities = set() + catalog_files = [] + for case, side in resolved: + _, item = self.catalog_for(case) + parent, commit = item.get("parent", ""), item.get("commit", "") + if parent != commit and all(re.fullmatch(r"[0-9a-fA-F]{40}", sha) for sha in (parent, commit)): + identities.add(self.source_identity(case, side)) + else: + file = case.fixed_file if side == "fixed" else case.vuln_file + data = file.read_text() + if is_catalog_excerpt(data): + data = parseable_excerpt(file, data) + mode = "excerpt-parseable" + else: + # Validate BOTH twins before treating this as a complete-file pair. + for twin in (case.vuln_file, case.fixed_file): + verify_catalog_file(twin) + mode = "catalog-file" + catalog_files.append((case, side, mode, data)) + if catalog_files: + if identities: + raise ValueError("mixed source and catalog-file materialization at one pair pin") + for case, side, mode, data in catalog_files: + _materialize_side(target, case, side=side) + if mode == "excerpt-parseable": + safe_path(target, case.relpath).write_text(data) + verify_functions(target, rows) + modes = {case.relpath: mode for case, _, mode, _ in catalog_files} + for row in rows: + if modes[row["file"]] == "excerpt-parseable": + row["evidence_scope"] = "in-function-only" + return { + "mode": "excerpt-parseable" if "excerpt-parseable" in modes.values() else "catalog-file", + "files": sum(p.is_file() for p in target.rglob("*")), + "bytes": sum(p.stat().st_size for p in target.rglob("*") if p.is_file()), + "roots": sorted({source_root(r["file"], r["language"]) for r in rows}), + } if len(identities) != 1: raise ValueError("ambiguous source commits at one pair pin") repo, commit = identities.pop() @@ -295,7 +334,7 @@ class Inputs: _materialize_side(target, case, side=side) return {"mode": "excerpt-fallback", "reason": str(exc), "repo": repo, "commit": commit} verify_functions(target, rows) - return {**proof, "mode": "source", "repo": repo, "commit": commit, "roots": roots} + return {**proof, "mode": "source", "repo": repo, "commit": commit} if all(r["source"] == "dev-php" for r in rows): # Export the exact requested commit from the local recipe checkout. for recipe in (ROOT / "benchmarks/repos").glob("*.toml"): @@ -310,43 +349,168 @@ class Inputs: raise ValueError(f"no local clone contains {pin['repo']}@{pin['pin']}") +def is_catalog_excerpt(data): + # Older Python exports have this header but no upstream_start field. + return re.match(r"\s*(?://|#)\s*Provenance:", data) is not None + + +def parseable_excerpt(path, data): + """Dedent the code independently of its unindented provenance comments.""" + language = detect_language(path) + lines = data.splitlines(keepends=True) + start = 0 + while start < len(lines) and (not lines[start].strip() or lines[start].lstrip().startswith(("#", "//"))): + start += 1 + header = "".join(lines[:start]) + if language == "python": + header = re.sub(r"(?m)^(\s*)//", r"\1#", header) + data = header + textwrap.dedent("".join(lines[start:])) + if not "".join(lines[start:]).strip(): + raise ValueError(f"excerpt does not parse: empty code ({path.name})") + if language == "python": + try: + ast.parse(data, filename=str(path)) + except SyntaxError as exc: + raise ValueError(f"excerpt does not parse: {exc}") from exc + else: + from openultrasast.semantic.cst import parse_with_cst + + parsed = parse_with_cst(str(path), data, language) + if parsed is None: + raise ValueError(f"excerpt does not parse: no cheap parser for {language} ({path.name})") + if not parsed.parse_ok: + raise ValueError(f"excerpt does not parse: {language} {parsed.reason} ({path.name})") + return data + + +def verify_catalog_file(path): + """Admit complete source only; syntactically valid provenance excerpts still fail.""" + data = path.read_text() + if is_catalog_excerpt(data): + raise ValueError(f"catalog side is an upstream excerpt without distinct commits: {path.name}") + language = detect_language(path) + if not data.strip(): + raise ValueError(f"catalog side is empty: {path.name}") + if language == "python": + try: + ast.parse(data, filename=str(path)) + except SyntaxError as exc: + raise ValueError(f"catalog side is not complete Python source: {path.name}: {exc.msg}") from exc + else: + from openultrasast.semantic.extra import grammar_for + + grammar = grammar_for(language) + if grammar is None: + raise ValueError(f"cannot verify complete catalog source for {language}: {path.name}") + from tree_sitter import Parser + + if Parser(grammar).parse(data.encode()).root_node.has_error: + raise ValueError(f"catalog side is not complete {language} source: {path.name}") + + def source_root(file, language): parts = Path(file).parts if language == "java": for i in range(len(parts) - 2): if parts[i : i + 3] == ("src", "main", "java"): return "/".join(parts[: i + 3]) + # Monorepo and source-layout containers are not themselves a package. + if len(parts) > 2 and parts[0] in {"packages", "apps", "src", "web"}: + return "/".join(parts[:2]) return parts[0] if len(parts) > 1 else "." +def local_imports(source, rows): + """Cheap one-level Python imports from labelled files, without importing code.""" + found = set() + for row in rows: + if row["language"] != "python": + continue + file = safe_path(source, row["file"]) + try: + tree = ast.parse(file.read_bytes()) + except (SyntaxError, ValueError): + continue + for node in ast.walk(tree): + if isinstance(node, ast.Import): + modules, bases = [alias.name for alias in node.names], [source, source / "src"] + elif isinstance(node, ast.ImportFrom): + modules = [node.module or ""] + modules += [".".join(filter(None, (node.module, alias.name))) for alias in node.names if alias.name != "*"] + bases = [file.parents[node.level - 1]] if node.level and node.level <= len(file.parents) else [source, source / "src"] + else: + continue + for base in bases: + for module in modules: + path = base.joinpath(*module.split(".")) + for candidate in (path.with_suffix(".py"), path / "__init__.py"): + if candidate.is_relative_to(source) and candidate.is_file() and not candidate.is_symlink(): + rel = candidate.relative_to(source).as_posix() + if not any(part in EXCLUDED_DIRS for part in Path(rel).parts): + found.add(rel) + return found + + def export_sources(source, target, rows, roots=(".",), cap=EXPORT_CAP): - """Preserve labelled paths first; cap context and exclude non-source trees.""" + """Copy labels intact, narrowing oversized roots before bounding context bytes.""" labelled = {r["file"] for r in rows} languages = {r["language"] for r in rows} files = set() + protected = {parent.as_posix() for rel in labelled for parent in Path(rel).parents} for directory, dirs, names in os.walk(source): - dirs[:] = sorted(d for d in dirs if d not in EXCLUDED_DIRS and d != ".git" and not (Path(directory) / d).is_symlink()) + dirs[:] = sorted( + d + for d in dirs + if d != ".git" + and not (Path(directory) / d).is_symlink() + and (d not in EXCLUDED_DIRS or (Path(directory) / d).relative_to(source).as_posix() in protected) + ) for name in names: file = Path(directory) / name rel = file.relative_to(source).as_posix() - if ( - not file.is_symlink() - and detect_language(file) in languages - and any(root == "." or rel.startswith(root + "/") for root in roots) + excluded = any(part in EXCLUDED_DIRS for part in Path(rel).parts[:-1]) + if not file.is_symlink() and ( + rel in labelled + or ( + not excluded + and detect_language(file) in languages + and any(root == "." or rel == root or rel.startswith(root + "/") for root in roots) + ) ): files.add(rel) missing = labelled - files if missing: raise ValueError(f"labelled files missing or excluded from export: {sorted(missing)}") - proof = {"bytes": 0, "files": 0, "cap_bytes": cap, "cap_hit": False, "excludes": sorted(EXCLUDED_DIRS)} + imports = local_imports(source, rows) + files.update(imports) + sizes = {rel: safe_path(source, rel).stat().st_size for rel in files} + original_roots = sorted(roots) + roots = original_roots + # Avoid lexicographically truncating a whole large package such as vllm. + if sum(sizes.values()) > cap: + roots = sorted({str(Path(rel).parent) for rel in labelled}) + files = { + rel for rel in files if rel in labelled or rel in imports or any(root == "." or rel.startswith(root + "/") for root in roots) + } + proof = { + "bytes": 0, + "files": 0, + "cap_bytes": cap, + "cap_hit": False, + "excludes": sorted(EXCLUDED_DIRS), + "roots": roots, + "requested_roots": original_roots, + "scope_narrowed": roots != original_roots, + "import_files": sorted(imports), + } for rel in sorted(files, key=lambda rel: (rel not in labelled, rel)): file = safe_path(source, rel) - size = file.stat().st_size + size = sizes[rel] + # The cap bounds context only. No labelled source is ever cut or omitted. if proof["bytes"] + size > cap: proof["cap_hit"] = True - if rel in labelled: - raise ValueError("labelled source exceeds 20 MB export cap") - continue + if rel not in labelled: + continue dest = safe_path(target, rel) dest.parent.mkdir(parents=True, exist_ok=True) dest.write_bytes(file.read_bytes()) @@ -431,7 +595,7 @@ def prepare_questions(pin, root, question_deadline): specs = taint_specs(language="javascript" if u["language"] == "typescript" else u["language"]) if not declares(safe_path(root, u["file"]).read_text(errors="replace").splitlines(), u["file"], u["function"]): raise ValueError(f"labelled function not declared: {u['file']}::{u['function']}") - excerpt = pin.get("materialization", {}).get("mode") == "excerpt-fallback" + excerpt = u.get("evidence_scope") == "in-function-only" or pin.get("materialization", {}).get("mode") == "excerpt-fallback" params = request_params( specs[u["family"]], file=u["file"], @@ -573,7 +737,7 @@ def main(): parser.add_argument("--limit", type=int, default=0) parser.add_argument("--only-family") parser.add_argument("--only-source", default="", help="comma-separated source catalog IDs") - parser.add_argument("--rerun-status", default="failed", help="re-run completed pins with any selected unit in these statuses") + parser.add_argument("--rerun-status", default="failed,timeout", help="comma-separated statuses to re-run (including timeout)") parser.add_argument( "--allow-excerpt-fallback", action="store_true", help="allow fragments only if fetching/exporting resolved source fails" ) @@ -652,6 +816,9 @@ def main(): with tempfile.TemporaryDirectory(prefix="trace-source-", dir=args.out) as scratch: source = Path(scratch) / "frozen" provenance = freeze_source(source) + progress = {"pins_done": 0, "pins_total": len(plan), "last_pin": None, "statuses": {}} + counts = Counter() + write_json(args.out / "progress.json", progress) for pin in plan: if (args.out / "STOP").exists(): break @@ -660,8 +827,13 @@ def main(): args.out / pin_name(pin), {**pin, "done": True, "units": [unit_record(u, "unsupported", u["reason"]) for u in pin["units"]]}, ) - continue - failures |= run_pin(pin, source, provenance, inputs, args.out, args) + else: + failures |= run_pin(pin, source, provenance, inputs, args.out, args) + result = json.loads((args.out / pin_name(pin)).read_text()) + selected = {u["unit"] for u in pin["units"]} + counts.update(u["status"] for u in result["units"] if u["unit"] in selected) + progress.update(pins_done=progress["pins_done"] + 1, last_pin={"repo": pin["repo"], "pin": pin["pin"]}, statuses=dict(counts)) + write_json(args.out / "progress.json", progress) return 2 if failures else 0 diff --git a/benchmarks/learn/engine_trace_worker.py b/benchmarks/learn/engine_trace_worker.py index 33a9733..3717b62 100644 --- a/benchmarks/learn/engine_trace_worker.py +++ b/benchmarks/learn/engine_trace_worker.py @@ -14,11 +14,12 @@ import os import selectors import signal import subprocess +import tempfile import time from collections import Counter from pathlib import Path -from openultrasast.cpg.backend import JoernBackend, _render +from openultrasast.cpg.backend import DATAFLOW_OVERLAY, JoernBackend, _render from openultrasast.model.trace import parse_trace @@ -39,6 +40,7 @@ def safe_path(root, name): FRONTENDS = {"php2cpg", "pysrc2cpg", "jssrc2cpg", "javasrc2cpg", "c2cpg", "joern-parse"} MIN_CPG_BYTES = 1024 HEAP_MB = 2560 +ENGINE_BUG = "engine bug: jssrc2cpg ObjectPropertyCallLinker" EXCLUDED_DIRS = frozenset({"tests", "test", "docs", "node_modules", "vendor", "dist", "build", "templates", "migrations"}) @@ -74,6 +76,61 @@ class MeasuredBackend(JoernBackend): self.completion_counts = Counter() self.timed_out = False self.fatal = "" + self.labelled_files = [] + self.retry_pending = False + self.retries = [] + + def _apply_overlays(self, cpg_path, scratch): + result = super()._apply_overlays(cpg_path, scratch) + # An optional overlay failure must not prevent the taint-only attempt. + if self.fatal == ENGINE_BUG and not self.timed_out: + self.retry_pending = True + self.fatal = "" + return result + + def _batch_once(self, cpg_path, query, requests): + if self.fatal or self.timed_out: + return None + if not self.retry_pending: + answers = super()._batch_once(cpg_path, query, requests) + if self.fatal != ENGINE_BUG: + return answers + if query != "taint" or self.timed_out or self.retries: + self.fatal = ENGINE_BUG + return None + self.retry_pending = False + self.fatal = "" + retry = {"reason": ENGINE_BUG, "mode": "taint-only", "completed": False} + self.retries.append(retry) + # Do not reuse rows/markers from the failed attempt as completion evidence. + self.asked.clear() + self.answers.clear() + self.completion_counts.clear() + original = self.queries_dir + with tempfile.TemporaryDirectory(prefix="trace-taint-", dir=cpg_path.parent) as scratch: + script = (original / "taint.sc").read_text() + marker = "importCpg(cpgFile)" + if script.count(marker) != 1: + self.fatal = ENGINE_BUG + return None + # Loading without enhancement skips the failing JS postprocessing passes. + # A raw graph without dataflow cannot establish a negative answer. + script = script.replace( + marker, + "importCpg(cpgFile, enhance = false)\n" + f' require(cpg.metaData.overlays.l.contains("{DATAFLOW_OVERLAY}"), ' + '"taint-only retry requires saved dataflowOss overlay")', + ) + self.queries_dir = Path(scratch) + (self.queries_dir / "taint.sc").write_text(script) + try: + answers = super()._batch_once(cpg_path, query, requests) + finally: + self.queries_dir = original + retry["completed"] = answers is not None and not self.fatal and not self.timed_out + if not retry["completed"]: + self.fatal = ENGINE_BUG + return answers def _run(self, command, *, timeout, cwd=None): if self.fatal or self.timed_out: @@ -140,6 +197,12 @@ class MeasuredBackend(JoernBackend): "exit": process.returncode, "oom": b"OutOfMemoryError" in output, } + lines = output.decode(errors="replace").splitlines() + stage["labelled_file_logs"] = { + file: [line for line in lines if file in line or Path(file).name in line] + for file in self.labelled_files + if any(file in line or Path(file).name in line for line in lines) + } self.stages.append(stage) if query and process.returncode != 0: # stderr is merged into stdout above; drop JVM frames before taking @@ -162,6 +225,9 @@ class MeasuredBackend(JoernBackend): self.fatal = "frontend wrote no CPG" elif elapsed < 5 and not self.fatal: self.fatal = f"implausibly fast JVM step: {command[0]} in {elapsed:.3f}s (<5s)" + if b"ObjectPropertyCallLinker failed" in output and b"Assignment statement with 3 arguments" in output: + self.fatal = ENGINE_BUG + stage["engine_bug"] = ENGINE_BUG if b"OutOfMemoryError" in output: self.fatal = "JVM out of memory" self.checkpoint() @@ -219,6 +285,8 @@ def worker_language(pin, root, out, deadline, question_deadline, persist=write_j backend = MeasuredBackend(deadline, question_deadline, checkpoint) proof["jvm"] = backend.stages + proof["retries"] = getattr(backend, "retries", []) + backend.labelled_files = sorted({u["file"] for u in pin["units"] if u["supported"]}) cpg = None try: active = [u for u in pin["units"] if u["supported"]] @@ -281,13 +349,14 @@ def worker_language(pin, root, out, deadline, question_deadline, persist=write_j extra.update(witness_rows=raw_flows, traces=[parse_trace(r) for r in raw_flows]) if shard_incomplete: extra["questions_completed"] = [] - dropped = any(p.endswith(unit["file"]) for p in cpg.unparsed) names = census[0].get("file_names", []) if census else [] file_missing = not any(str(p).removeprefix("./") == unit["file"] or str(p).endswith("/" + unit["file"]) for p in names) - if file_missing and rows is not None: - dropped = True + dropped = file_missing and rows is not None if failure or dropped: status, reason = "failed", failure or "labelled file was dropped or absent from query census" + if dropped: + logs = [line for stage in backend.stages for line in stage.get("labelled_file_logs", {}).get(unit["file"], [])] + reason += "\n" + ("\n".join(dict.fromkeys(logs)) if logs else "frontend emitted no log lines mentioning labelled file") elif rows is None or shard_incomplete: status, reason = ("timeout" if backend.timed_out else "failed"), "question did not complete" elif not rows and rid not in backend.asked: @@ -319,7 +388,7 @@ def worker_language(pin, root, out, deadline, question_deadline, persist=write_j if stage["command"] == "joern" and stage.get("exit") and "output_tail" in stage: detail = f"joern query exited {stage['exit']}: {stage['output_tail'] or 'no output'}" for row in record["units"]: - if row["supported"]: + if row["supported"] and row["status"] in {"failed", "timeout"}: row["reason"] = f"{row['reason']}\n{detail}".strip() record["questions_asked"] = list(dict.fromkeys(backend.asked)) record["questions_completed"] = [rid for u in record["units"] for rid in u["questions_completed"]] diff --git a/tests/test_engine_trace.py b/tests/test_engine_trace.py index 4e4a6c6..a4f99df 100644 --- a/tests/test_engine_trace.py +++ b/tests/test_engine_trace.py @@ -530,6 +530,8 @@ def test_trace_real_unsupported_run_then_typescript_opt_in_replans(tmp_path, mon result_file = out / trace.pin_name(u) first = json.loads(result_file.read_text()) assert first["done"] and first["units"][0]["status"] == "unsupported" + progress = json.loads((out / "progress.json").read_text()) + assert progress == {"pins_done": 1, "pins_total": 1, "last_pin": {"repo": u["repo"], "pin": u["pin"]}, "statuses": {"unsupported": 1}} assert trace.make_plan(trace.join_units([u], [example]), out) == [] called = [] @@ -676,6 +678,7 @@ def pair_inputs(tmp_path): @pytest.mark.parametrize("side,sha", [("vuln", "a" * 40), ("fixed", "b" * 40)]) def test_pair_materializes_catalog_side_not_header(tmp_path, monkeypatch, side, sha): inputs, case = pair_inputs(tmp_path) + case.vuln_file.write_text("# Provenance: example\n def run(self): pass\n") fetched = tmp_path / "fetched" (fetched / "pkg").mkdir(parents=True) (fetched / case.relpath).write_text("class Handler:\n def run(self): pass\n") @@ -897,3 +900,261 @@ def test_sparse_fetch_from_local_git_without_network(tmp_path, monkeypatch): proof = trace.export_sources(clone, target, [labelled], ["module/src/main/java"]) trace.verify_functions(target, [labelled]) assert proof["files"] == 1 and proof["bytes"] > 0 + + +@pytest.mark.parametrize("side", ["vuln", "fixed"]) +def test_complete_catalog_twins_and_context_need_no_commits(tmp_path, monkeypatch, side): + inputs, case = pair_inputs(tmp_path) + inputs.catalog_rows[case.name][0][1].update(parent="", commit="") + fixed = tmp_path / "fixed.py" + fixed.write_text("def run(): return 'fixed'\n") + case.fixed_file = fixed + context = tmp_path / "context.py" + context.write_text("registered = True\n") + case.context_files = (("pkg/context.py", context, context),) + monkeypatch.setattr(inputs, "pair_for", lambda row: (case, side)) + monkeypatch.setattr(inputs, "fetch_source", lambda *args: pytest.fail("catalog files must not fetch")) + target = tmp_path / "export" + proof = inputs.materialize({"units": [unit(file=case.relpath)]}, target) + assert proof["mode"] == "catalog-file" + assert (target / case.relpath).read_bytes() == (fixed if side == "fixed" else case.vuln_file).read_bytes() + assert (target / "pkg/context.py").read_bytes() == context.read_bytes() + + +@pytest.mark.parametrize( + "body,reason", + [ + ("# Provenance: example upstream_start: 25\n def run():\n return (\n", "excerpt does not parse:.*never closed"), + ("def run(:\n", "not complete Python"), + ], +) +def test_incomplete_catalog_without_commits_fails_specifically(tmp_path, monkeypatch, body, reason): + inputs, case = pair_inputs(tmp_path) + inputs.catalog_rows[case.name][0][1].update(parent="", commit="") + case.vuln_file.write_text(body) + monkeypatch.setattr(inputs, "pair_for", lambda row: (case, "vuln")) + with pytest.raises(ValueError, match=reason): + inputs.materialize({"units": [unit(file=case.relpath)]}, tmp_path / "export") + + +@pytest.mark.parametrize("comment,newline", [("#", "\n"), ("//", "\n"), ("#", "\r\n")]) +@pytest.mark.parametrize("upstream", [True, False]) +def test_parseable_python_method_excerpt_is_in_function_only(tmp_path, monkeypatch, comment, newline, upstream): + inputs, case = pair_inputs(tmp_path) + inputs.catalog_rows[case.name][0][1].update(parent="", commit="") + header = f"{comment} Provenance: example\n" + (f"{comment} upstream_start: 25\n" if upstream else "") + body = header + "\n def run(self, cmd):\n os.system(cmd)\n" + case.vuln_file.write_bytes(body.replace("\n", newline).encode()) + monkeypatch.setattr(inputs, "pair_for", lambda row: (case, "vuln")) + monkeypatch.setattr(inputs, "fetch_source", lambda *args: pytest.fail("excerpt attempted fetch")) + pin = {"units": [unit(file=case.relpath)]} + target = tmp_path / "export" + pin["materialization"] = inputs.materialize(pin, target) + assert pin["materialization"]["mode"] == "excerpt-parseable" + assert (target / case.relpath).read_text() == header.replace("//", "#") + "\ndef run(self, cmd):\n os.system(cmd)\n" + assert trace.unit_record(pin["units"][0], "path")["evidence_scope"] == "in-function-only" + params = trace.prepare_questions(pin, target, 60)["a"] + assert params["callDepth"] == "0" and params["parameterSources"] == "true" + + +def test_java_fragment_is_not_parseable_excerpt(tmp_path, monkeypatch): + inputs, case = pair_inputs(tmp_path) + inputs.catalog_rows[case.name][0][1].update(parent="", commit="") + case.relpath = "pkg/App.java" + case.vuln_file = case.fixed_file = tmp_path / "excerpt.java" + case.vuln_file.write_text("// Provenance: example\n// upstream_start: 25\n public void run() {\n execute(cmd);\n") + inputs.catalog_rows[case.name][0][1]["relpath"] = case.relpath + monkeypatch.setattr(inputs, "pair_for", lambda row: (case, "vuln")) + with pytest.raises(ValueError, match="excerpt does not parse:"): + inputs.materialize({"units": [unit(file=case.relpath, language="java")]}, tmp_path / "export") + + +def test_excerpt_without_cheap_parser_is_rejected(tmp_path, monkeypatch): + monkeypatch.setattr("openultrasast.semantic.extra.grammar_for", lambda language: None) + with pytest.raises(ValueError, match="excerpt does not parse: no cheap parser for javascript"): + trace.parseable_excerpt(tmp_path / "app.js", "// Provenance: example upstream_start: 25\nfunction run() {}\n") + + +def test_parseable_javascript_excerpt_uses_repository_parser(tmp_path): + from openultrasast.semantic.extra import grammar_for + + if grammar_for("javascript") is None: + pytest.skip("optional JavaScript grammar unavailable") + assert ( + trace.parseable_excerpt(tmp_path / "app.js", "// Provenance: example\n function run(cmd) { execute(cmd); }\n") + == "// Provenance: example\nfunction run(cmd) { execute(cmd); }\n" + ) + + +def test_export_keeps_labels_over_cap_and_excluded_ancestors(tmp_path): + source, target = tmp_path / "source", tmp_path / "export" + for rel in ("templates/preferences.php", "templates/other.php", "other.php"): + path = source / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(" proof["cap_bytes"] + + +def test_oversized_package_narrows_before_truncation_and_keeps_import(tmp_path): + source, target = tmp_path / "source", tmp_path / "export" + for rel, body in { + "pkg/sub/app.py": "from helper import value\ndef run(): pass\n", + "pkg/sub/context.py": "x = 1\n", + "pkg/huge.py": "#" * 500, + "helper.py": "value = 1\n", + }.items(): + path = source / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(body) + proof = trace.export_sources(source, target, [unit(file="pkg/sub/app.py")], ["pkg"], cap=100) + assert proof["roots"] == ["pkg/sub"] and proof["requested_roots"] == ["pkg"] + assert proof["scope_narrowed"] and not proof["cap_hit"] + assert (target / "helper.py").exists() and (target / "pkg/sub/context.py").exists() + assert not (target / "pkg/huge.py").exists() + + +def test_local_repository_export_scopes_packages(tmp_path, monkeypatch): + inputs, _ = pair_inputs(tmp_path) + + def export(clone, pin, dest): + for rel in ("packages/server/run.js", "packages/other/skip.js"): + path = dest / rel + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("function run() {}\n") + + monkeypatch.setattr(trace, "export_pin", export) + proof = inputs.export_local( + tmp_path, {"pin": "sha", "units": [unit(file="packages/server/run.js", language="javascript")]}, tmp_path / "out" + ) + assert proof["roots"] == ["packages/server"] + assert not (tmp_path / "out/packages/other/skip.js").exists() + + +def test_census_membership_is_uncapped_and_warning_is_not_absence(tmp_path, monkeypatch): + class LargeBackend(FakeBackend): + def build(self, root, language): + cpg = super().build(root, language) + original = cpg.run_batch_once + + def batch(query, requests): + result = original(query, requests) + result["__census__"] = [{"files": "3001", "file_names": [f"f{i}.py" for i in range(3000)] + ["/case/app.py"]}] + return result + + return replace(cpg, run_batch_once=batch, unparsed=("app.py",)) + + result = run_fake(tmp_path, monkeypatch, LargeBackend) + assert [u["status"] for u in result["units"]] == ["path", "asked-nothing"] + for name in ("census.sc", "taint.sc"): + assert '"file_names" -> ujson.Arr.from(cpg.file.name.l)' in (trace.ROOT / "src/openultrasast/cpg/queries" / name).read_text() + + +def test_absent_label_includes_frontend_parse_log(tmp_path, monkeypatch): + class MissingBackend(FakeBackend): + def build(self, root, language): + cpg = super().build(root, language) + self.stages[0]["labelled_file_logs"] = {"app.py": ["Failed to parse app.py: invalid syntax"]} + + def batch(query, requests): + self.asked.extend(requests) + return {**dict.fromkeys(requests, []), "__census__": [{"files": "1", "file_names": [""]}]} + + return replace(cpg, run_batch_once=batch) + + result = run_fake(tmp_path, monkeypatch, MissingBackend) + assert all(u["status"] == "failed" and "Failed to parse app.py: invalid syntax" in u["reason"] for u in result["units"]) + + +def test_frontend_retains_matching_log_lines(tmp_path): + launcher = tmp_path / "pysrc2cpg" + launcher.write_text(f"#!{sys.executable}\nprint('Failed to parse /case/pkg/app.py: syntax')\n") + launcher.chmod(0o755) + backend = worker.MeasuredBackend(30, 10, lambda: None) + backend.labelled_files = ["pkg/app.py"] + backend._run([str(launcher)], timeout=10) + assert backend.stages[0]["labelled_file_logs"]["pkg/app.py"] == ["Failed to parse /case/pkg/app.py: syntax"] + + +@pytest.mark.parametrize("success", [True, False]) +def test_engine_bug_retries_taint_only_once(tmp_path, monkeypatch, success): + backend = worker.MeasuredBackend(900, 120, lambda: None) + calls = [] + + def batch(self, graph, query, requests): + calls.append((self.queries_dir / "taint.sc").read_text()) + if len(calls) == 1: + self.fatal = worker.ENGINE_BUG + self.asked.append("stale") + self.answers["stale"] = [] + return None + assert not self.asked and not self.answers + assert "enhance = false" in calls[-1] + assert "requires saved dataflowOss overlay" in calls[-1] + if success: + return {"a": [], "__census__": [{"files": "1", "file_names": ["app.py"]}]} + self.fatal = "joern query exited 1" + return None + + monkeypatch.setattr(JoernBackend, "_batch_once", batch) + result = backend._batch_once(tmp_path / "cpg.bin", "taint", {"a": {}}) + assert len(calls) == 2 and backend.retries[0]["completed"] == success + assert (result is not None) == success + assert backend.fatal == ("" if success else worker.ENGINE_BUG) + if not success: + backend._batch_once(tmp_path / "cpg.bin", "taint", {"a": {}}) + assert len(backend.retries) == 1 + + +def test_named_engine_bug_from_process_output(tmp_path): + launcher = tmp_path / "joern" + launcher.write_text( + f"#!{sys.executable}\nprint('Pass io.joern.x2cpg.frontendspecific.jssrc2cpg.ObjectPropertyCallLinker failed')\n" + "print('java.lang.RuntimeException: Assignment statement with 3 arguments')\nraise SystemExit(1)\n" + ) + launcher.chmod(0o755) + backend = worker.MeasuredBackend(30, 10, lambda: None) + backend._run([str(launcher), "--script", "query.sc"], timeout=10) + assert backend.fatal == worker.ENGINE_BUG + + +def test_rerun_failed_and_timeout_only_preserves_completed_v2(tmp_path): + units = [unit(status, pin=status) for status in trace.STATUSES] + for pin in trace.make_plan(units, tmp_path): + trace.save_pin(tmp_path / trace.pin_name(pin), {**pin, "done": True, "units": [trace.unit_record(pin["units"][0], pin["pin"])]}) + assert {p["pin"] for p in trace.make_plan(units, tmp_path, rerun_status={"failed", "timeout"})} == {"failed", "timeout"} + + +def test_successful_retry_keeps_result_and_records_attempt(tmp_path, monkeypatch): + class RetriedBackend(FakeBackend): + def __init__(self, *args): + super().__init__(*args) + self.retries = [{"reason": worker.ENGINE_BUG, "mode": "taint-only", "completed": True}] + self.stages.append({"command": "joern", "seconds": 20, "exit": 1, "output_tail": worker.ENGINE_BUG}) + + result = run_fake(tmp_path, monkeypatch, RetriedBackend) + assert [u["status"] for u in result["units"]] == ["path", "asked-nothing"] + assert all(u["reason"] == "" for u in result["units"]) + assert result["instrument"]["retries"][0]["completed"] + + +def test_overlay_engine_bug_allows_one_taint_only_attempt(tmp_path, monkeypatch): + backend = worker.MeasuredBackend(900, 120, lambda: None) + + def overlay(self, graph, scratch): + self.fatal = worker.ENGINE_BUG + return False + + monkeypatch.setattr(JoernBackend, "_apply_overlays", overlay) + assert not backend._apply_overlays(tmp_path / "cpg.bin", tmp_path) + assert backend.retry_pending and not backend.fatal + + def batch(self, graph, query, requests): + assert "enhance = false" in (self.queries_dir / "taint.sc").read_text() + return {"a": []} + + monkeypatch.setattr(JoernBackend, "_batch_once", batch) + assert backend._batch_once(tmp_path / "cpg.bin", "taint", {"a": {}}) == {"a": []} + assert len(backend.retries) == 1