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
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": {}}) == {}


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