Open_Ultra_SAST/tests/test_candidate_enumeration.py

"""Candidates come from the semantic IR, shaped by family (constrained-detector 1.1, Req 1.1, 1.2, 1.4-1.6).

The detector stops searching. Something deterministic decides what it looks at, which makes that enumerator a hard
ceiling on recall — measured before this was built at 12.1% for the pattern ruleset against 96.6% for the IR. The
ordering is part of the contract: the batches a later task forms, and therefore the sequence of model calls, must
be determined by the input alone.
"""

from __future__ import annotations

import sys
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).parent))
from test_obligation_operations import APP  # noqa: E402

from openultrasast.model.taxonomy import load_families  # noqa: E402
from openultrasast.semantic.extra import has_semantic_extra  # noqa: E402

INJECTION = (
    "from flask import request\n"
    "import os\n"
    "import sqlite3\n\n\n"
    "@app.route('/run')\n"
    "def run():\n"
    "    raw = request.args.get('cmd')\n"
    "    command = 'echo ' + raw\n"
    "    os.system(command)\n"
    "    return 'ok'\n"
)


def _region(path: str = "app.py", function: str | None = "run"):  # type: ignore[no-untyped-def]
    from openultrasast.model.candidates import Region

    return Region(path=path, function=function)


def _write(tmp_path: Path, name: str, text: str) -> Path:
    (tmp_path / name).write_text(text)
    return tmp_path


def test_a_call_site_candidate_carries_what_the_model_needs_to_judge_it(tmp_path: Path) -> None:
    from openultrasast.model.candidates import enumerate_candidates

    root = _write(tmp_path, "app.py", INJECTION)
    result = enumerate_candidates(root, _region(), "injection", taxonomy=load_families())
    assert result.reason == ""
    sites = {item.name: item for item in result.candidates}
    assert "os.system" in sites or "system" in sites, sorted(sites)
    sink = sites.get("os.system") or sites["system"]
    assert sink.kind == "call" and sink.function == "run" and sink.path == "app.py"
    assert sink.line == 10  # `from flask` 1, imports 2-3, blanks, decorator 6, def 7, raw 8, command 9
    assert sink.arg_texts == ("command",)
    # the chain the IR resolved, nearest first: command <- 'echo ' + raw <- request.args.get('cmd')
    assert any("raw" in step for step in sink.binding_chain), sink.binding_chain
    assert any("request" in step for step in sink.binding_chain), sink.binding_chain
    assert sink.id in {"app.py:10:os.system", "app.py:10:system"}


def test_the_family_decides_which_kinds_are_candidates() -> None:
    from openultrasast.model.candidates import FAMILY_SHAPE

    taxonomy = load_families()
    assert set(FAMILY_SHAPE) == {family.id for family in taxonomy.families}, "every family says what carries its bug"
    assert FAMILY_SHAPE["injection"] == ("call",)
    assert FAMILY_SHAPE["access_control"] == ("operation",)
    assert "bind" in FAMILY_SHAPE["config_secrets"]
    assert "call" in FAMILY_SHAPE["prototype"]


@pytest.mark.semantic
@pytest.mark.skipif(not has_semantic_extra(), reason="the guards come from the parsed IR")
def test_access_control_candidates_are_operations_with_the_guards_in_scope(tmp_path: Path) -> None:
    """An absence bug has no sink. Its candidate is the operation, and what the model needs is what guards it."""
    from openultrasast.model.candidates import enumerate_candidates

    root = _write(tmp_path, "app.py", APP)
    guarded = enumerate_candidates(root, _region(function="guarded"), "access_control", taxonomy=load_families())
    assert guarded.reason == "" and guarded.candidates
    assert all(item.kind == "operation" for item in guarded.candidates)
    assert any("login_required" in guard for item in guarded.candidates for guard in item.guards)
    # `leaky` and `constrained` both constrain by `filter_by`; what separates them is where the constraining value
    # came from, and that is the binding chain the IR resolved — the work the model used to do by grepping.
    leaky = enumerate_candidates(root, _region(function="leaky"), "access_control", taxonomy=load_families())
    constrained = enumerate_candidates(root, _region(function="constrained"), "access_control", taxonomy=load_families())
    leaky_site = next(item for item in leaky.candidates if item.name.endswith("filter_by"))
    good_site = next(item for item in constrained.candidates if item.name.endswith("filter_by"))
    assert "user_id=owner" in leaky_site.arg_texts and "user_id=user_id" in good_site.arg_texts
    assert any("request.args.get" in step for step in leaky_site.binding_chain), leaky_site.binding_chain
    assert any("request.user.id" in step for step in good_site.binding_chain), good_site.binding_chain


def test_a_side_the_parser_rejects_yields_no_candidates_and_says_why(tmp_path: Path) -> None:
    from openultrasast.model.candidates import enumerate_candidates

    root = _write(tmp_path, "app.py", "def broken(:\n    pass\n")
    result = enumerate_candidates(root, _region(function="broken"), "injection", taxonomy=load_families())
    assert result.candidates == () and result.reason == "unsupported_language"


def test_a_labeled_function_with_nothing_to_judge_says_no_candidate(tmp_path: Path) -> None:
    from openultrasast.model.candidates import enumerate_candidates

    root = _write(tmp_path, "app.py", "def empty():\n    return 1\n")
    result = enumerate_candidates(root, _region(function="empty"), "injection", taxonomy=load_families())
    assert result.candidates == () and result.reason == "no_candidate"


def test_the_bound_is_a_constant_and_the_overflow_is_counted(tmp_path: Path) -> None:
    from openultrasast.model.candidates import MAX_CANDIDATES_PER_CALL, enumerate_candidates

    # one call per line, so the arithmetic is the bound and not the fixture
    body = "\n".join(f"    sink_{index}(value)" for index in range(MAX_CANDIDATES_PER_CALL + 5))
    root = _write(tmp_path, "app.py", f"def many(value):\n{body}\n")
    result = enumerate_candidates(root, _region(function="many"), "injection", taxonomy=load_families())
    assert len(result.candidates) == MAX_CANDIDATES_PER_CALL
    assert result.discarded == 5


def test_enumerating_twice_returns_an_identical_sequence(tmp_path: Path) -> None:
    """The batches, and therefore the sequence of model calls, are determined by the input alone (Req 2.4)."""
    from openultrasast.model.candidates import enumerate_candidates

    root = _write(tmp_path, "app.py", APP)
    taxonomy = load_families()
    first = enumerate_candidates(root, _region(function="constrained"), "injection", taxonomy=taxonomy)
    second = enumerate_candidates(root, _region(function="constrained"), "injection", taxonomy=taxonomy)
    assert [item.id for item in first.candidates] == [item.id for item in second.candidates]
    assert [item.id for item in first.candidates] == sorted({item.id for item in first.candidates}, key=_order(first))


def _order(result):  # type: ignore[no-untyped-def]
    index = {item.id: (item.path, item.line, item.name) for item in result.candidates}
    return lambda identifier: index[identifier]


def test_a_region_without_a_function_enumerates_the_whole_file(tmp_path: Path) -> None:
    from openultrasast.model.candidates import enumerate_candidates

    root = _write(tmp_path, "app.py", APP)
    whole = enumerate_candidates(root, _region(function=None), "injection", taxonomy=load_families())
    named = enumerate_candidates(root, _region(function="leaky"), "injection", taxonomy=load_families())
    assert len(whole.candidates) >= len(named.candidates)
    assert {item.function for item in whole.candidates} > {"leaky"}


@pytest.mark.parametrize("family", ["injection", "path", "deserialization", "untrusted_destination", "config_secrets"])
def test_every_family_shape_enumerates_something_on_a_file_that_has_it(tmp_path: Path, family: str) -> None:
    from openultrasast.model.candidates import enumerate_candidates

    root = _write(tmp_path, "app.py", INJECTION)
    result = enumerate_candidates(root, _region(), family, taxonomy=load_families())
    assert result.reason in {"", "no_candidate"}
    if family in {"injection", "path", "deserialization", "untrusted_destination"}:
        assert result.candidates, f"{family} takes call sites and this file has three"


# --- the ceiling (task 1.2) ---------------------------------------------------


def test_the_ceiling_counts_labeled_functions_with_at_least_one_candidate(tmp_path: Path) -> None:
    from openultrasast.benchmark import ExpectedFinding
    from openultrasast.model.candidates import ceiling
    from openultrasast.pairs import PairCase

    def case(name: str, body: str, *, slice_name: str, family: str) -> PairCase:
        (tmp_path / f"{name}.py").write_text(body)
        return PairCase(
            name=name,
            slice=slice_name,
            language="python",
            origin="test",
            vuln_file=tmp_path / f"{name}.py",
            fixed_file=tmp_path / f"{name}.py",
            relpath="app.py",
            expected=(
                ExpectedFinding(
                    cwe="CWE-78",
                    vulnerability_class="x",
                    path="app.py",
                    evidence="",
                    function="run",
                    mechanism="other",
                    family=family,
                ),
            ),
            min_recall=1.0,
            fix_policy="silent",
            review_tier="seeded",
        )

    cases = [
        case("has", "import os\n\n\ndef run(cmd):\n    os.system(cmd)\n", slice_name="vibe-py", family="injection"),
        case("empty", "def run():\n    return 1\n", slice_name="vibe-py", family="injection"),
        case("real", "import os\n\n\ndef run(cmd):\n    os.system(cmd)\n", slice_name="agent-vfc", family="injection"),
    ]
    report = ceiling(cases, taxonomy=load_families())
    assert report.per_family["injection"]["pairs"] == 3
    assert report.per_family["injection"]["with_candidate"] == 2
    assert report.per_family["injection"]["ceiling"] == pytest.approx(2 / 3)
    assert report.per_slice["vibe-py"]["ceiling"] == pytest.approx(0.5)
    assert report.per_slice["agent-vfc"]["ceiling"] == 1.0
    assert ("vibe-py", "empty", "no_candidate") in report.gaps
    assert report.per_family["injection"]["sites"]["median"] >= 1


def test_the_ruleset_generator_is_measured_too_so_the_comparison_is_reproducible(tmp_path: Path) -> None:
    """12.1% against 96.6% is why the ruleset is not the generator. A remembered number is not a measurement."""
    from openultrasast.model.candidates import ceiling
    from openultrasast.pairs import DEFAULT_CATALOG, load_pair_catalog, select_slice, select_vendored

    cases = [c for c in select_vendored(select_slice(load_pair_catalog(DEFAULT_CATALOG), "vibe-py")) if not c.unscorable][:6]
    ir = ceiling(cases, taxonomy=load_families(), generator="ir")
    ruleset = ceiling(cases, taxonomy=load_families(), generator="ruleset")
    assert ir.generator == "ir" and ruleset.generator == "ruleset"
    assert sum(b["with_candidate"] for b in ir.per_family.values()) >= sum(b["with_candidate"] for b in ruleset.per_family.values())


def test_the_command_runs_offline_and_writes_a_stable_artifact(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    from openultrasast.cli import main

    for name in ("OPENULTRASAST_HUNTER_CLIENT", "DEEPSEEK_API_KEY", "OPENROUTER_API_KEY"):
        monkeypatch.delenv(name, raising=False)
    out = tmp_path / "model"
    assert main(["model", "candidates", "--slice", "vibe-py", "--out", str(out)]) == 0
    artifact = out / "candidate-ceiling.json"
    assert artifact.is_file()
    first = artifact.read_bytes()
    assert main(["model", "candidates", "--slice", "vibe-py", "--out", str(out)]) == 0
    assert artifact.read_bytes() == first, "the ceiling is a property of the corpus, not of when it was measured"
    import json

    payload = json.loads(first)
    assert payload["generator"] == "ir" and payload["per_family"] and payload["per_slice"]