Open_Ultra_SAST/tests/test_cpg_backend.py

"""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
import subprocess
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"


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": {}}) is None, (
        "None means COULD NOT ASK; {} would mean the engine answered and found nothing, and a caller that "
        "cannot tell them apart reports a failed scan as a clean repository"
    )


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


def test_a_built_cpg_offers_the_batch_path_to_its_driver(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    """The capability must be reachable through the object the driver is handed, not merely present.

    ``query_batch`` was implemented, tested and committed while ``build`` returned a result carrying only
    ``run``. The driver discovers batching by looking for ``run_batch`` on that result, found nothing, and
    fell back to one JVM per region -- silently, because every test called ``query_batch`` directly. A
    repository scan is what made it visible: ~14s per region-family, one ``--param function=`` invocation at
    a time.
    """
    from openultrasast.cpg.backend import BEGIN, END, JoernBackend

    scripts: list[list[str]] = []

    def runner(command, **kwargs):  # type: ignore[no-untyped-def]
        class _Done:
            returncode = 0
            stderr = ""
            stdout = f'{BEGIN}\n{{"r1": [{{"sink": "os.system(x)"}}]}}\n{END}\n'

        if "--output" in command:  # the parse step; make the artifact it promises
            Path(command[command.index("--output") + 1]).write_text("cpg")
        else:
            scripts.append(list(command))
        return _Done()

    monkeypatch.setenv("OPENULTRASAST_JOERN_PROBE", "on")
    result = JoernBackend(runner=runner).build(tmp_path)

    assert result is not None
    assert result.run_batch is not None, "a backend that can batch must expose it, or the driver cannot find it"
    rows = result.run_batch("taint", {"r1": {"sources": ("request.args",), "sinks": ("os.system",)}})
    assert rows["r1"], "the batch path must reach the engine, not just exist"
    assert len(scripts) == 1, "one invocation for the batch"
    assert not any(arg.startswith("function=") for arg in scripts[0]), "a batch carries requests, not one function"


def test_the_engine_runs_under_a_bounded_heap(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    """A JVM with no -Xmx claims a quarter of physical RAM.

    That is not a tuning preference. The first repository measurements were killed under memory pressure at
    different points, so their numbers described the host rather than the engine, and on a contributor's
    laptop an unbounded heap is the difference between a scan running in the background and a scan the
    machine notices.
    """
    from openultrasast.cpg.backend import JoernBackend

    commands: list[list[str]] = []

    def runner(command, **kwargs):  # type: ignore[no-untyped-def]
        class _Done:
            returncode = 0
            stderr = ""
            stdout = ""

        commands.append(list(command))
        if "--output" in command:
            Path(command[command.index("--output") + 1]).write_text("cpg")
        return _Done()

    monkeypatch.setenv("OPENULTRASAST_JOERN_PROBE", "on")
    monkeypatch.delenv("OPENULTRASAST_CPG_HEAP_MB", raising=False)
    result = JoernBackend(runner=runner).build(tmp_path)
    assert result is not None
    result.run("taint", {})

    assert commands, "nothing ran"
    for command in commands:
        assert any(arg.startswith("-J-Xmx") for arg in command), f"unbounded heap in {command[0]}"


def test_the_operator_can_raise_the_heap(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    """A bound that cannot be lifted is a bound that gets removed. A big repository may need a big heap."""
    from openultrasast.cpg.backend import JoernBackend

    commands: list[list[str]] = []

    def runner(command, **kwargs):  # type: ignore[no-untyped-def]
        class _Done:
            returncode = 0
            stderr = ""
            stdout = ""

        commands.append(list(command))
        if "--output" in command:
            Path(command[command.index("--output") + 1]).write_text("cpg")
        return _Done()

    monkeypatch.setenv("OPENULTRASAST_JOERN_PROBE", "on")
    monkeypatch.setenv("OPENULTRASAST_CPG_HEAP_MB", "6144")
    JoernBackend(runner=runner).build(tmp_path)

    assert "-J-Xmx6144m" in commands[0]


def test_a_failed_joern_parse_retries_through_the_language_frontend(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    """Measured on a 637-file WordPress plugin: `joern-parse` threw after applying overlays while `php2cpg`
    on the same tree produced a 4.3MB CPG with no errors. Joern's own message recommends the direct route for
    a large codebase, so this is the documented fallback rather than a workaround."""
    from openultrasast.cpg.backend import JoernBackend

    tried: list[str] = []

    def runner(command, **kwargs):  # type: ignore[no-untyped-def]
        class _Result:
            stderr = ""
            stdout = ""

        tried.append(Path(command[0]).name)
        if Path(command[0]).name == "joern-parse":
            _Result.returncode = 1
            return _Result()
        _Result.returncode = 0
        Path(command[command.index("-o") + 1]).write_text("cpg")
        return _Result()

    monkeypatch.setenv("OPENULTRASAST_JOERN_PROBE", "on")
    monkeypatch.setattr("shutil.which", lambda name: f"/opt/joern/{name}")

    result = JoernBackend(runner=runner).build(tmp_path, language="php")

    assert result is not None, "the frontend succeeded where joern-parse did not"
    assert tried == ["joern-parse", "php2cpg"]


def test_an_unknown_language_has_nothing_to_retry_with(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    """No guessing at a frontend name: a language with no mapping fails closed, as it did before."""
    from openultrasast.cpg.backend import JoernBackend

    def runner(command, **kwargs):  # type: ignore[no-untyped-def]
        class _Result:
            returncode = 1
            stderr = ""
            stdout = ""

        return _Result()

    monkeypatch.setenv("OPENULTRASAST_JOERN_PROBE", "on")
    assert JoernBackend(runner=runner).build(tmp_path, language="cobol") is None
    assert JoernBackend(runner=runner).build(tmp_path) is None


def test_a_batch_goes_through_a_file_not_the_command_line(tmp_path: Path) -> None:
    """Linux caps a SINGLE argument at MAX_ARG_STRLEN (128KB) whatever ARG_MAX says.

    2500 taint requests over a 637-file WordPress plugin is a 1.49MB payload. As `--param requests=<json>`
    execve returned E2BIG, the runner caught the OSError, query_batch returned {} -- and {} reads as "no rows
    found". The scan reported 500 regions examined, nothing found, and 0.05 seconds of taint query: a quiet
    failure the rung ladder cannot catch, because no verdict was ever produced to label.
    """
    from openultrasast.cpg.backend import BEGIN, END, JoernBackend

    commands: list[list[str]] = []

    class _Done:
        returncode = 0
        stderr = ""
        stdout = f'{BEGIN}\n{{"r0": []}}\n{END}\n'

    def runner(command, **kwargs):  # type: ignore[no-untyped-def]
        commands.append(list(command))
        return _Done()

    cpg = tmp_path / "cpg.bin"
    cpg.write_text("cpg")
    big = {f"r{i}": {"sources": ("request.args",) * 40, "sinks": ("execute",) * 40} for i in range(2000)}

    JoernBackend(runner=runner).query_batch(cpg, "taint", big)

    argv = commands[0]
    assert not any(len(arg) > 131072 for arg in argv), "no argument may approach MAX_ARG_STRLEN"
    staged = [arg for arg in argv if arg.startswith("requestsFile=")]
    assert staged, "the batch is staged to a file"
    written = Path(staged[0].split("=", 1)[1])
    assert written.is_file() and len(written.read_text()) > 200_000, "and the file holds the whole payload"


def test_a_file_the_frontend_dropped_is_carried_off_the_build() -> None:
    """A Joern frontend fails a file without failing the build, and exits 0 with the rest of the graph.

    php2cpg 4.0.623 does exactly this: one oversized file corrupts the batched parser stream, that file is
    logged at WARN and dropped, and the CPG comes back missing it. A scan over the remainder that says
    nothing is a clean bill of health over code nobody looked at.
    """
    warning = (
        "2026-09-09 17:46:21.892 WARN  AstCreationPass  Failed to process '/src/class.memberorder.php'\n"
        "2026-09-09 17:46:21.892 WARN  SymbolSummaryPass  Failed to process '/src/class.memberorder.php'\n"
        "2026-09-09 17:46:21.893 WARN  AstCreationPass  Failed to process '/src/rest-api.php'\n"
    )
    from openultrasast.cpg.backend import _unparsed_files

    completed = subprocess.CompletedProcess(args=[], returncode=0, stdout="", stderr=warning)
    assert _unparsed_files(completed) == ("/src/class.memberorder.php", "/src/rest-api.php")


def test_a_build_with_nothing_dropped_reports_no_unparsed_files() -> None:
    from openultrasast.cpg.backend import _unparsed_files

    completed = subprocess.CompletedProcess(args=[], returncode=0, stdout="Successfully wrote graph", stderr="")
    assert _unparsed_files(completed) == ()
    assert _unparsed_files(None) == ()


def test_a_joern_log_line_inside_the_fence_does_not_destroy_the_payload() -> None:
    """Joern logs to stdout while the script runs, so a log line can land ahead of the payload.

    `[INFO ] Attempting to determine flows from empty list of sources.` did, and because the batch is one
    JSON document the driver lost all 205 answers in it and reported `query_failed` for every one.
    """
    from openultrasast.cpg.backend import BEGIN, END, extract_payload

    noisy = f'{BEGIN}\n[INFO ] Attempting to determine flows from empty list of sources.\n{{"0": [], "1": [1]}}\n{END}\n'
    assert extract_payload(noisy) == {"0": [], "1": [1]}