Open_Ultra_SAST/tests/test_loo.py

"""corpus-seeded-mechanisms task 3.1: leave-one-out is the corpus's own detection rate (offline)."""

from __future__ import annotations

import json
from pathlib import Path

import pytest

from openultrasast.benchmark import ExpectedFinding
from openultrasast.pairs import PairCase

EXEC_A = (
    "import sqlite3\n\n\ndef lookup(user):\n    c = sqlite3.connect('x').cursor()\n"
    + "    c.execute(\"SELECT * FROM t WHERE u = '%s'\" % user)\n"
)
EXEC_A_FIX = (
    "import sqlite3\n\n\ndef lookup(user):\n    c = sqlite3.connect('x').cursor()\n    c.execute('SELECT * FROM t WHERE u = ?', (user,))\n"
)
EXEC_B = (
    "import sqlite3\n\n\ndef delete(name):\n    cur = sqlite3.connect('y').cursor()\n    cur.execute('DELETE FROM t WHERE n = ' + name)\n"
)
EXEC_B_FIX = (
    "import sqlite3\n\n\ndef delete(name):\n    cur = sqlite3.connect('y').cursor()\n"
    + "    cur.execute('DELETE FROM t WHERE n = ?', (name,))\n"
)
SYS_C = "import os\n\n\ndef ping(host):\n    os.system('ping ' + host)\n"
SYS_C_FIX = "import os\nimport subprocess\n\n\ndef ping(host):\n    subprocess.run(['ping', host])\n"


def _case(tmp_path: Path, name: str, vuln: str, fixed: str, *, function: str, cwe: str, provenance: str = "human") -> PairCase:
    (tmp_path / f"{name}-v.py").write_text(vuln)
    (tmp_path / f"{name}-f.py").write_text(fixed)
    return PairCase(
        name=name,
        slice="toy",
        language="python",
        origin="test",
        vuln_file=tmp_path / f"{name}-v.py",
        fixed_file=tmp_path / f"{name}-f.py",
        relpath="app.py",
        expected=(
            ExpectedFinding(
                cwe=cwe, vulnerability_class="x", path="app.py", evidence="", function=function, mechanism="source_reaches_sink"
            ),
        ),
        min_recall=1.0,
        fix_policy="silent",
        provenance=provenance,
        review_tier="seeded",
    )


def _toy(tmp_path: Path) -> list[PairCase]:
    return [
        _case(tmp_path, "a", EXEC_A, EXEC_A_FIX, function="lookup", cwe="CWE-89"),
        _case(tmp_path, "b", EXEC_B, EXEC_B_FIX, function="delete", cwe="CWE-89", provenance="agent"),
        _case(tmp_path, "c", SYS_C, SYS_C_FIX, function="ping", cwe="CWE-78"),
    ]


def test_leave_one_out_reports_which_pair_taught_which(tmp_path: Path) -> None:
    from openultrasast.semantic.loo import evaluate_loo

    result = evaluate_loo(_toy(tmp_path))
    by_name = {outcome.pair: outcome for outcome in result.outcomes}
    assert set(by_name) == {"a", "b", "c"}
    assert by_name["a"].detected and by_name["a"].silent and by_name["a"].found_by  # taught by b's execute shape
    assert by_name["b"].detected and by_name["b"].silent and any("a" in pairs for _, pairs in by_name["b"].found_by)
    assert not by_name["c"].detected and by_name["c"].found_by == ()  # nobody else teaches os.system: c only tests
    slice_metrics = result.per_slice["toy"]
    assert slice_metrics["pairs"] == 3 and slice_metrics["detected"] == 2 and slice_metrics["silent"] == 3
    assert abs(slice_metrics["recall"] - 2 / 3) < 1e-9 and slice_metrics["silence"] == 1.0 and abs(slice_metrics["youden"] - 2 / 3) < 1e-9
    assert result.per_profile["agent"]["pairs"] == 1 and result.per_profile["human"]["pairs"] == 2
    assert "source_reaches_sink" in result.per_mechanism
    payload = result.to_dict()
    assert set(payload) >= {"per_slice", "per_profile", "per_mechanism", "outcomes"}
    json.dumps(payload)  # artifact-ready


def test_leave_one_out_seeds_only_trusted_pairs_and_skips_known_limit(tmp_path: Path) -> None:
    from openultrasast.semantic.loo import evaluate_loo

    cases = _toy(tmp_path)
    cases[1] = PairCase(**{**cases[1].__dict__, "review_tier": "advisory"})  # b may be tested, never teaches
    cases[2] = PairCase(**{**cases[2].__dict__, "known_limit": "engine"})
    result = evaluate_loo(cases)
    by_name = {outcome.pair: outcome for outcome in result.outcomes}
    assert set(by_name) == {"a", "b"}  # known_limit excluded entirely
    assert not by_name["a"].detected  # b is advisory: it cannot teach a
    assert by_name["b"].detected  # a (seeded) teaches b, which is scored as a held-out target
    assert result.skipped == (("c", "known_limit:engine"),)


def test_cli_pairs_loo_writes_the_artifact_offline(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    import io
    from contextlib import redirect_stdout

    from openultrasast.cli import main

    monkeypatch.chdir(Path.cwd())
    buf = io.StringIO()
    with redirect_stdout(buf):
        assert main(["pairs", "--slice", "vibe-py", "--loo", "--json", "--loo-out", str(tmp_path / "loo.json")]) == 0
    payload = json.loads(buf.getvalue())
    assert payload["per_slice"]["vibe-py"]["pairs"] == 35 and "found_by" in payload["outcomes"][0]
    assert json.loads((tmp_path / "loo.json").read_text()) == payload


HELPER_SINK = "import os\n\n\ndef run(cmd):\n    os.system(cmd)\n\n\ndef ping(host):\n    return run('ping ' + host)\n"
HELPER_SINK_FIX = (
    "import subprocess\n\n\ndef run(cmd):\n    subprocess.run(cmd.split())\n\n\ndef ping(host):\n    return run('ping ' + host)\n"
)


def test_hits_outside_the_labeled_function_or_off_label_never_count(tmp_path: Path) -> None:
    """Round-3 mutants: a hit must sit inside the labeled function and name the labeled mechanism or sink."""
    from openultrasast.semantic.loo import evaluate_loo

    teacher = _case(tmp_path, "t", SYS_C, SYS_C_FIX, function="ping", cwe="CWE-78")
    outside = _case(
        tmp_path, "outside", HELPER_SINK, HELPER_SINK_FIX, function="ping", cwe="CWE-78"
    )  # sink call lives in `run`, not `ping`
    off_label = PairCase(
        **{
            **_case(tmp_path, "offlabel", SYS_C, SYS_C_FIX, function="ping", cwe="CWE-78").__dict__,
            "expected": (
                ExpectedFinding(
                    cwe="CWE-78", vulnerability_class="x", path="app.py", evidence="", function="ping", mechanism="missing_auth_guard"
                ),
            ),
        }
    )
    result = evaluate_loo([teacher, outside, off_label])
    by_name = {outcome.pair: outcome for outcome in result.outcomes}
    assert by_name["outside"].hits_vuln >= 1 and not by_name["outside"].detected
    assert by_name["offlabel"].hits_vuln >= 1 and not by_name["offlabel"].detected


def test_leave_one_out_forwards_search_degradations(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    from openultrasast.semantic.loo import evaluate_loo

    monkeypatch.setenv("OPENULTRASAST_TREE_SITTER_PROBE", "0")
    js = "function run(req) {\n  const cmd = req.query.cmd;\n  exec(cmd);\n}\n"
    (tmp_path / "j-v.js").write_text(js)
    (tmp_path / "j-f.js").write_text(js)
    case = PairCase(
        **{
            **_toy(tmp_path)[0].__dict__,
            "name": "j",
            "language": "javascript",
            "vuln_file": tmp_path / "j-v.js",
            "fixed_file": tmp_path / "j-f.js",
            "relpath": "app.js",
        }
    )
    result = evaluate_loo([*_toy(tmp_path), case])
    assert any(item["reason"] == "variants_language_unsupported" for item in result.degradations)


# --- authorization-obligations 4.2: leave-one-out per obligation kind -------------------------------------------------

_HANDLER = "from flask import request\n\n\n@app.route('/{route}/<key>')\ndef {fn}(key):\n{body}"
OBL_A = _HANDLER.format(route="books", fn="get_book", body="    return Book.query.filter_by(book_title=key).first()\n")
OBL_A_FIX = _HANDLER.format(
    route="books",
    fn="get_book",
    body="    user_id = request.user.id\n    return Book.query.filter_by(user_id=user_id, book_title=key).first()\n",
)
OBL_B = _HANDLER.format(route="orders", fn="get_order", body="    return Order.query.filter_by(ref=key).first()\n")
OBL_B_FIX = _HANDLER.format(
    route="orders", fn="get_order", body="    owner = current_user.id\n    return Order.query.filter_by(owner=owner, ref=key).first()\n"
)
OBL_C = _HANDLER.format(
    route="login", fn="login", body="    resp = make_response('ok')\n    resp.set_cookie('sid', key, samesite='None')\n    return resp\n"
)
OBL_C_FIX = _HANDLER.format(
    route="login",
    fn="login",
    body="    resp = make_response('ok')\n    resp.set_cookie('sid', key, samesite='Strict', httponly=1)\n    return resp\n",
)


def _obligation_case(tmp_path: Path, name: str, vuln: str, fixed: str, *, function: str, obligation: str, mechanism: str) -> PairCase:
    case = _case(tmp_path, name, vuln, fixed, function=function, cwe="CWE-639")
    row = case.expected[0]
    return PairCase(
        **{
            **case.__dict__,
            "slice": "vibe-py",
            "expected": (ExpectedFinding(**{**row.__dict__, "mechanism": mechanism, "obligation": obligation}),),
        }
    )


def _obligation_toy(tmp_path: Path) -> list[PairCase]:
    read = {"obligation": "protected_read", "mechanism": "unconstrained_protected_read"}
    return [
        _obligation_case(tmp_path, "a", OBL_A, OBL_A_FIX, function="get_book", **read),
        _obligation_case(tmp_path, "b", OBL_B, OBL_B_FIX, function="get_order", **read),
        _obligation_case(tmp_path, "c", OBL_C, OBL_C_FIX, function="login", obligation="security_setting", mechanism="permissive_default"),
    ]


def test_leave_one_out_scores_obligation_rows_per_kind_and_names_the_teaching_pair(tmp_path: Path) -> None:
    """Req 7.4: recall, silence and Youden per obligation kind; found_by names the shape record that supplied known_fix."""
    from openultrasast.semantic.loo import evaluate_loo

    result = evaluate_loo(_obligation_toy(tmp_path))
    by_name = {outcome.pair: outcome for outcome in result.outcomes}
    a, b, c = by_name["a"], by_name["b"], by_name["c"]
    assert a.obligations == ("protected_read",) and c.obligations == ("security_setting",)
    assert a.detected and a.silent and [pairs for _, pairs in a.found_by] == [("b",)]  # b taught the identity discharger
    assert b.detected and b.silent and [pairs for _, pairs in b.found_by] == [("a",)]
    assert all(record_id.startswith("corpus:") for record_id, _ in a.found_by)
    assert c.teaches and not c.detected and c.found_by == () and c.silent  # reached and undischarged, but nobody else teaches the kind
    kinds = result.per_obligation_kind
    assert kinds["protected_read"]["pairs"] == 2 and kinds["protected_read"]["detected"] == 2 and kinds["protected_read"]["youden"] == 1.0
    assert kinds["security_setting"]["pairs"] == 1 and kinds["security_setting"]["detected"] == 0
    assert "unconstrained_protected_read" in result.per_mechanism  # flow mechanisms and obligation kinds sit side by side
    payload = result.to_dict()
    assert "per_obligation_kind" in payload and payload["outcomes"][0]["obligations"] == ["protected_read"]
    json.dumps(payload)


def test_obligation_rows_are_not_found_by_the_facts_alone(tmp_path: Path) -> None:
    """A finding without a known fix from the held-out store is the checker's own work, not the corpus teaching."""
    from openultrasast.semantic.loo import evaluate_loo

    (only,) = [outcome for outcome in evaluate_loo(_obligation_toy(tmp_path)[:1]).outcomes]
    assert only.hits_vuln >= 1 and not only.detected and only.found_by == ()  # the leaky read is flagged, but no pair taught it