Open_Ultra_SAST/tests/test_learn_labels.py

"""Labels from ground truth with provenance (learned-decision-engine task 4; Req 2.1-2.3, 7.1) on fixture
populations, pairs, adjudications and a fixture git history. The fail-closed source list is checked with a
v3-named population that must never be opened."""

from __future__ import annotations

import json
import os
import subprocess
from pathlib import Path
from typing import Any

import pytest

from openultrasast.benchmark import ExpectedFinding
from openultrasast.learn import labels as L
from openultrasast.pairs import PairCase

DAY = 86_400
T0 = 1_600_000_000  # 2020-09-13


def _git(repo: Path, *args: str, when: int | None = None) -> str:
    env = {**os.environ, "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@x", "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@x"}
    if when is not None:
        env |= {"GIT_AUTHOR_DATE": f"{when} +0000", "GIT_COMMITTER_DATE": f"{when} +0000"}
    return subprocess.run(["git", "-C", str(repo), *args], check=True, capture_output=True, text=True, env=env).stdout.strip()


def _commit(repo: Path, files: dict[str, str], message: str, when: int) -> str:
    for name, text in files.items():
        (repo / name).parent.mkdir(parents=True, exist_ok=True)
        (repo / name).write_text(text)
    _git(repo, "add", "-A")
    _git(repo, "commit", "-q", "-m", message, when=when)
    return _git(repo, "rev-parse", "HEAD")


APP = (
    "<?php\nfunction show($id) {\n    $q = 'SELECT * WHERE id=' . $id;\n    return mysqli_query($db, $q);\n}\n"
    "function helper($x) {\n    return $x;\n}\n"
)
FIXED = APP.replace("'SELECT * WHERE id=' . $id", "'SELECT * WHERE id=' . intval($id)")
OTHER = "<?php\nfunction render($a) {\n    return $a;\n}\n"


def _case_repo(cache: Path, case_id: str) -> dict[str, str]:
    """A clone with history: old ordinary commits, a Chinese security commit, the vulnerable and fixed pins, a benign push."""
    repo = cache / "independent" / case_id
    repo.mkdir(parents=True)
    _git(repo, "init", "-q")
    pins: dict[str, str] = {}
    _commit(repo, {"src/app.php": APP, "src/other.php": OTHER, "composer.json": '{"require": {"php": ">=7"}}'}, "initial import", T0)
    pins["ordinary"] = _commit(repo, {"src/other.php": OTHER.replace("$a;", "$a . '';")}, "tidy the render helper", T0 + 10 * DAY)
    pins["chinese"] = _commit(repo, {"src/admin.php": OTHER.replace("render", "grant")}, "修复越权访问问题", T0 + 20 * DAY)
    pins["touched_later"] = _commit(repo, {"src/app.php": APP + "\n"}, "whitespace in app", T0 + 30 * DAY)
    pins["near_fix"] = _commit(repo, {"src/other.php": OTHER}, "reword the renderer", T0 + 190 * DAY)
    pins["vulnerable"] = _commit(repo, {"src/app.php": APP}, "restore app", T0 + 200 * DAY)
    pins["fixed"] = _commit(repo, {"src/app.php": FIXED}, "cast the id", T0 + 260 * DAY)
    pins["benign_base"] = pins["fixed"]
    pins["benign_tip"] = _commit(repo, {"src/app.php": FIXED.replace("return $x;", "return $x; // same")}, "comment", T0 + 270 * DAY)
    return pins


def _layout(tmp_path: Path) -> tuple[Path, Path, dict[str, str]]:
    """A repository root with a v2-shaped population, a v3-named reserved population beside it, results files, and
    the sources.toml that lists the first and excludes the second."""
    root, cache = tmp_path / "repo", tmp_path / "cache"
    pins = _case_repo(cache, "acme-sqli")
    ind = root / "benchmarks" / "independent"
    ind.mkdir(parents=True)
    population = (
        '[[case]]\nid = "acme-sqli"\nrepo = "https://github.com/Acme/Shop.git"\nfamily = "injection"\nlanguage = "php"\n'
        f'advisory = ["CVE-2024-0001"]\nvulnerable = "{pins["vulnerable"]}"\nfixed = "{pins["fixed"]}"\n'
        'sites = ["src/app.php::show", "src/app.php::helper"]\n'
        f'benign = {{ base = "{pins["benign_base"]}", tip = "{pins["benign_tip"]}" }}\n'
    )
    (ind / "population-v2.toml").write_text(population)
    (ind / "population-v3-php.toml").write_text('[[case]]\nid = "secret"\n')
    (ind / "results-v2.json").write_text(
        json.dumps(
            {
                "precision_sample": [
                    {
                        "case": "acme-sqli",
                        "site": "src/other.php:3:render",
                        "family": "injection",
                        "verdict": "False, guarded",
                        "reason": "cast",
                    },
                    {
                        "case": "acme-sqli",
                        "site": "src/app.php:4:",
                        "family": "injection",
                        "verdict": "True, privileged",
                        "reason": "admin only",
                    },
                ]
            }
        )  # fmt: skip
    )
    sources = (
        'version = 1\n\n[[source]]\nid = "population-v2"\nkind = "population"\nfile = "benchmarks/independent/population-v2.toml"\n'
        'evaluation = ["benchmarks/independent/results-v2.json"]\nrecorded = "2026-09-28"\nstatus = "spent"\nmatching = "declared_sites"\n'
        'design_informed = true\ncache = "independent"\n\n'
        '[[source]]\nid = "adjudications"\nkind = "adjudications"\nfiles = ["benchmarks/independent/results-v2.json#precision_sample"]\n'
        'evaluation = []\nrecorded = "2026-09-28"\nstatus = "recorded"\n\n'
        '[[source]]\nid = "assumed_benign"\nkind = "history"\nrepositories_from = ["population-v2"]\nrecorded = "2026-09-30"\n'
        'status = "assumption"\nweight = 0.5\nmin_days_from_fix = 90\nscan_commits = 100\nmax_commits_per_repository = 10\n'
        'max_files_per_commit = 20\n\n[[excluded]]\nfile = "population-v3-php.toml"\nreason = "reserved"\n'
    )
    (root / "sources.toml").write_text(sources)
    return root, cache, pins


def _build(tmp_path: Path, monkeypatch: pytest.MonkeyPatch | None = None) -> tuple[L.Build, dict[str, Any], list[str], dict[str, str]]:
    root, cache, pins = _layout(tmp_path)
    opened: list[str] = []
    if monkeypatch is not None:
        real_read = Path.read_text
        real_open = open

        def read_text(self: Path, *a: Any, **k: Any) -> str:
            opened.append(self.name)
            return real_read(self, *a, **k)

        def tracking_open(file: Any, *a: Any, **k: Any) -> Any:
            opened.append(Path(str(file)).name)
            return real_open(file, *a, **k)

        monkeypatch.setattr(Path, "read_text", read_text)
        monkeypatch.setattr("builtins.open", tracking_open)
    build, record = L.build_labels(root, sources_path=root / "sources.toml", cache=cache, created="2026-09-30")
    return build, record, opened, pins


def test_population_positives_negatives_and_the_untouched_function(tmp_path: Path) -> None:
    build, _, _, pins = _build(tmp_path)
    index = L.LabelIndex(build.labels)
    show = index.lookup("src/app.php::show", "injection")
    assert show is not None and show.label == 1 and show.design_informed and show.group == "acme/shop" and show.pin == pins["vulnerable"]
    negative = index.lookup("src/app.php::show", "injection", pin_role="fixed")
    assert negative is not None and negative.label == 0 and negative.pin == pins["fixed"]
    assert index.lookup("src/app.php::helper", "injection").label == 1  # type: ignore[union-attr]
    assert index.lookup("src/app.php::helper", "injection", pin_role="fixed") is None, "the fix did not touch helper: not a negative"
    assert index.lookup("src/app.php::nowhere", "injection") is None, "an unmatched candidate is unlabelled, never negative"
    deltas = {(r.candidate, r.direction, r.label) for r in build.labels if r.unit == "delta" and r.source == "population-v2"}
    assert ("src/app.php::show", "introduce", 1) in deltas and ("src/app.php::show", "repair", 0) in deltas
    benign = [r for r in build.labels if r.source == "benign_control"]
    assert [r.candidate for r in benign] == ["src/app.php::helper"] and benign[0].conditional == L.CONDITIONAL
    assert all(r.frameworks == () for r in build.labels if r.source == "population-v2"), "composer.json names no framework"


def test_adjudications_map_verdicts_and_resolve_a_missing_function(tmp_path: Path) -> None:
    build, _, _, _ = _build(tmp_path)
    rows = {r.candidate: r for r in build.labels if r.source == "adjudications"}
    assert rows["src/other.php::render"].label == 0 and "guarded" in rows["src/other.php::render"].evidence
    assert rows["src/app.php::show"].label == 1 and rows["src/app.php::show"].privileged, "the line resolved to its function"


def test_v3_is_never_opened_and_an_unlisted_source_cannot_label(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    build, _, opened, _ = _build(tmp_path, monkeypatch)
    assert build.labels and "population-v2.toml" in opened
    assert "population-v3-php.toml" not in opened, "the reserved population must never be read"
    sources = L.load_sources(tmp_path / "repo" / "sources.toml")
    with pytest.raises(L.LabelSourceError, match="not a label source"):
        sources.get("population-v3")
    guard = L.SourceGuard(sources, tmp_path / "repo")
    with pytest.raises(L.LabelSourceError, match="excluded"):
        guard.text("benchmarks/independent/population-v3-php.toml")
    with pytest.raises(L.LabelSourceError, match="not a listed label source"):
        guard.text("benchmarks/independent/results-v1.json")
    assert "population-v3-php.toml" not in opened


def test_a_source_naming_an_excluded_file_is_refused(tmp_path: Path) -> None:
    (tmp_path / "s.toml").write_text(
        '[[source]]\nid = "v3"\nkind = "population"\nfile = "b/population-v3-php.toml"\nrecorded = "x"\nstatus = "x"\n'
        '[[excluded]]\nfile = "population-v3-php.toml"\n'
    )
    with pytest.raises(L.LabelSourceError, match="excluded"):
        L.load_sources(tmp_path / "s.toml")


def test_a_source_without_its_evaluation_record_cannot_label(tmp_path: Path) -> None:
    root, cache, _ = _layout(tmp_path)
    (root / "benchmarks" / "independent" / "results-v2.json").unlink()
    with pytest.raises(L.LabelSourceError, match="evaluation record"):
        L.build_labels(root, sources_path=root / "sources.toml", cache=cache)


def test_assumed_benign_is_mined_separately_and_the_filter_is_multilingual(tmp_path: Path) -> None:
    build, record, _, pins = _build(tmp_path)
    mined = [r for r in build.labels if r.source == "assumed_benign"]
    assert {r.pin for r in mined} == {pins["ordinary"]}, "only the ordinary commit far from the fix, whose files no later fix touched"
    assert all(r.weight == 0.5 and r.conditional == L.CONDITIONAL and r.label == 0 for r in mined)
    history = record["assumed_benign_history"]
    assert history["rejected"]["other_language"] == 1, "the Chinese 'fix IDOR' commit is caught with no English word"
    assert history["rejected"]["near_fix"] >= 1 and history["rejected"]["later fix touches its files"] >= 1
    assert not [r for r in L.verified(build.labels) if r.source == "assumed_benign"], "never merged with verified negatives"
    assert "assumed_benign" in record["conditional_negatives"] and "assumed_benign" not in record["verified_pin_labels"]["by_source"]


@pytest.mark.parametrize(
    "message,reason",
    [
        ("修复越权访问问题", "other_language"),
        ("修复 IDOR", "english"),
        ("脆弱性の修正", "other_language"),
        ("보안 취약점 수정", "other_language"),
        ("исправлена уязвимость", "other_language"),
        ("Sicherheitslücke geschlossen", "other_language"),
        ("corrige une faille", "other_language"),
        ("Corrige una falla de seguridad", "other_language"),
        ("Fix CVE-2024-1234", "reference"),
        ("see GHSA-abcd-efgh-ijkl", "reference"),
        ("Escape the title (CWE-79)", "reference"),
        ("Prevent XSS in comments", "english"),
        ("sanitize input", "english"),
        ("Bump lodash from 4.17.20 to 4.17.21", "dependency"),
        ("fix: validate log file name more thoroughly", "english"),
        ("add password masking to the keyboard", "english"),
        ("增加文件名校验", "other_language"),
        ("Update README", None),
        ("修复拼写错误", None),
        ("重构代码结构", None),
        ("Añadir traducción al español", None),
        ("Refactor the settings page", None),
    ],
)
def test_security_filter(message: str, reason: str | None) -> None:
    assert L.security_reason(message) == reason


def test_repository_groups_merge_urls_forks_and_advisories() -> None:
    assert L.repo_name("https://github.com/Acme/Shop.git") == L.repo_name("[email protected]:acme/shop") == "acme/shop"
    assert L.repo_name("github.com/acme/shop/") == "acme/shop"
    groups = L.Groups({"someone/renamed": "acme/shop"})
    groups.add("https://github.com/acme/shop", advisory_ids={"CVE-2024-0001"}, commits=("a" * 40,))
    groups.add("https://github.com/fork/shop-mirror", advisory_ids={"CVE-2024-0001"})
    groups.add("https://gitlab.com/other/copy", commits=("a" * 40,))
    groups.add("someone/renamed")
    groups.add("unrelated/shop")
    assert groups.group("fork/shop-mirror") == groups.group("other/copy") == groups.group("someone/renamed") == groups.group("acme/shop")
    assert groups.group("unrelated/shop") != groups.group("acme/shop"), "a shared name alone does not merge"
    assert groups.merged == {"advisory": 1, "commit": 1, "alias": 1} and groups.count() == 2
    assert L.advisories("fix for cve-2024-0001 and GHSA-mh7h-gpmx-fggj") == {"CVE-2024-0001", "GHSA-MH7H-GPMX-FGGJ"}


def _pair(name: str, *, tier: str = "advisory", unscorable: str | None = None, function: str | None = "f", repo: str = "") -> PairCase:
    expected = (ExpectedFinding(cwe="CWE-89", vulnerability_class="sqli", path="a.py", evidence="e", function=function),)
    return PairCase(
        name=name, slice="vfc", language="python", origin="o", vuln_file=Path("v"), fixed_file=Path("f"), relpath="a.py",
        expected=expected, min_recall=1.0, fix_policy="p", repo=repo, cve="CVE-2024-0001", review_tier=tier, unscorable=unscorable,
    )  # fmt: skip


def test_pairs_exclude_unscorable_and_title_and_group_by_repository(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    (tmp_path / "pairs").mkdir()
    (tmp_path / "pairs" / "catalog.toml").write_text("")
    (tmp_path / "s.toml").write_text(
        '[[source]]\nid = "pairs"\nkind = "pairs"\nfile = "pairs/catalog.toml"\nrecorded = "x"\nstatus = "x"\nexclude_tiers = ["title"]\n'
    )
    cases = (
        _pair("ok", repo="https://github.com/o/r"), _pair("twin", unscorable="identical_twin"), _pair("titled", tier="title"),
        _pair("nofn", function=None), _pair("fork", repo="https://github.com/x/r-fork"),
    )  # fmt: skip
    monkeypatch.setattr("openultrasast.pairs.load_pair_catalog", lambda path: cases)
    build, record = L.build_labels(tmp_path, sources_path=tmp_path / "s.toml", cache=tmp_path, assumed_benign=False)
    refs = {r.source_ref for r in build.labels}
    assert refs == {"ok", "fork"}
    assert {r.group for r in build.labels} == {"o/r"}, "the fork shares the CVE, so it is one repository group"
    assert {(r.unit, r.label, r.pin_role) for r in build.labels if r.source_ref == "ok"} == {
        ("pin", 1, "vulnerable"),
        ("delta", 1, "vulnerable"),
        ("pin", 0, "fixed"),
        ("delta", 0, "fixed"),
    }
    assert record["skipped"] == {"pairs: label names no function": 1, "pairs: title tier (sensitivity arm)": 1, "pairs: unscorable pair": 1}
    family = record["verified_pin_labels"]["by_family"]["injection"]
    assert family["positive_groups"] == family["negative_groups"] == 1 and family["model"] == "insufficient"
    titled, _ = L.build_labels(tmp_path, sources_path=tmp_path / "s.toml", cache=tmp_path, assumed_benign=False, include_title=True)
    assert "titled" in {r.source_ref for r in titled.labels}


def test_the_snapshot_carries_counts_only(tmp_path: Path) -> None:
    _, record, _, pins = _build(tmp_path)
    text = json.dumps(record)
    for identity in ("acme", "shop", "show", "helper", "src/app.php", *pins.values()):
        assert identity not in text, f"{identity} leaked into the committed snapshot"
    assert record["verified_pin_labels"]["by_family"]["injection"]["positive_groups"] == 1


def test_frameworks_from_manifests_and_plugin_header() -> None:
    assert L.frameworks_from({"requirements.txt": "Django==4.2\nrequests"}) == ("django", "requests")
    assert L.frameworks_from({}, ["<?php\n/*\n * Plugin Name: Shop\n */"]) == ("wordpress",)
    assert L.frameworks_from({"package.json": '{"dependencies": {"express-session": "1"}}'}) == ()