Files
Open_Ultra_SAST/tests/test_learn_labels.py
T
Marius CiepluchandClaude Opus 5.5 658f2d5e39 fix(learn): fixed-side negatives only where the fixed side declares the function
A pair now gives a fixed-side negative only when the fixed side's source declares the
labelled function (the excerpt builder's matcher; a name on a code line where the matcher
resolves neither side), or when a deterministic rename at a distinct fix commit replaces it
(one new function, same parameter list, changed body: provenance fixed_side_moved). Anything
else is counted under pair_fixed_side.absent_fixed_side, not labelled.

108 pair labels lose their negative: 80 seeded pointer pairs whose fixed side is a guard
helper of the same commit, and 28 Real-Vuln-Benchmark traps whose fixed side is a sibling
handler of the same snapshot (4 of those handlers are labelled positives themselves). The
memory build no longer substitutes the catalog's fix_function on the fixed side. The pair
cache is an allowed label-source tree so pointer sides are read through the guard.

Snapshot: benchmarks/measurements/2026-10-01-decision-engine-labels (counts only).

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo
2026-10-01 11:03:54 +02:00

394 lines
21 KiB
Python

"""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"}
VULN_PY = "def f(q):\n return db.execute('SELECT ' + q)\n"
FIXED_PY = "def f(q):\n return db.execute('SELECT ?', (q,))\n"
def _pair(
name: str,
*,
tier: str = "advisory",
unscorable: str | None = None,
function: str | None = "f",
repo: str = "",
root: Path | None = None,
vuln: str = VULN_PY,
fixed: str = FIXED_PY,
parent: str = "a" * 40,
commit: str = "b" * 40,
) -> PairCase:
expected = (ExpectedFinding(cwe="CWE-89", vulnerability_class="sqli", path="a.py", evidence="e", function=function),)
folder = (root or Path("/nonexistent")) / "pairs" / name
if root is not None:
folder.mkdir(parents=True, exist_ok=True)
header = f"# Provenance: fixture\n# commit: {commit}\n# parent: {parent}\n# function: {function}\n"
(folder / "vuln.py").write_text(header + vuln)
(folder / "fixed.py").write_text(header + fixed)
return PairCase(
name=name, slice="vfc", language="python", origin="o", vuln_file=folder / "vuln.py", fixed_file=folder / "fixed.py",
relpath="a.py", expected=expected, min_recall=1.0, fix_policy="p", repo=repo, commit=commit, 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", root=tmp_path), _pair("twin", unscorable="identical_twin"),
_pair("titled", tier="title", root=tmp_path), _pair("nofn", function=None),
_pair("fork", repo="https://github.com/x/r-fork", root=tmp_path),
) # 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 _pair_build(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, *cases: PairCase) -> tuple[L.Build, dict[str, Any]]:
(tmp_path / "pairs").mkdir(exist_ok=True)
(tmp_path / "pairs" / "catalog.toml").write_text("")
source = '[[source]]\nid = "pairs"\nkind = "pairs"\nfile = "pairs/catalog.toml"\nrecorded = "x"\nstatus = "x"\n'
(tmp_path / "s.toml").write_text(source)
monkeypatch.setattr("openultrasast.pairs.load_pair_catalog", lambda path: cases)
return L.build_labels(tmp_path, sources_path=tmp_path / "s.toml", cache=tmp_path, assumed_benign=False)
def _fixed_rows(build: L.Build, name: str) -> list[L.Label]:
return [r for r in build.labels if r.source_ref == name and r.pin_role == "fixed"]
GATE = "def _gate(q):\n return bool(q)\n" # a guard helper of the same snapshot, same parameter list as the labelled f
def test_a_fixed_side_without_the_function_gives_no_negative(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
snapshot = "c" * 40
cases = (
_pair("gate", root=tmp_path, fixed=GATE, parent=snapshot, commit=snapshot), # the kolega-ai-dev pointer shape
_pair("other", root=tmp_path, fixed="def g(a, b):\n return a\n"), # a real fix, but no renamed equivalent
_pair("unread", root=None),
)
build, record = _pair_build(tmp_path, monkeypatch, *cases)
for name in ("gate", "other", "unread"):
assert _fixed_rows(build, name) == [], name
assert {(r.unit, r.label) for r in build.labels if r.source_ref == name} == {("pin", 1), ("delta", 1)}, "positives stay"
assert record["pair_fixed_side"] == {"absent_fixed_side": {"labels": 3, "by_family": {"injection": 3}}, "fixed_side_moved": 0}
assert record["skipped"] == {
f"pairs: no fixed-side negative ({L.ABSENT_SAME_SNAPSHOT})": 1,
f"pairs: no fixed-side negative ({L.ABSENT_NO_EQUIVALENT})": 1,
f"pairs: no fixed-side negative ({L.ABSENT_UNREAD})": 1,
}
assert record["verified_pin_labels"]["by_family"]["injection"]["negative_groups"] == 0
def test_a_renamed_function_at_a_real_fix_is_a_moved_negative(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
moved = "def f_safe(q):\n return db.execute('SELECT ?', (q,))\n"
build, record = _pair_build(tmp_path, monkeypatch, _pair("moved", root=tmp_path, fixed=moved, repo="https://github.com/o/r"))
rows = _fixed_rows(build, "moved")
assert {(r.unit, r.label, r.candidate, r.provenance) for r in rows} == {
("pin", 0, "a.py::f_safe", L.FIXED_MOVED),
("delta", 0, "a.py::f_safe", L.FIXED_MOVED),
}
assert all("f -> f_safe" in r.evidence for r in rows)
assert record["pair_fixed_side"]["fixed_side_moved"] == 1 and record["pair_fixed_side"]["absent_fixed_side"]["labels"] == 0
# the same rename in one snapshot (no fix commit) is a sibling function, never a negative
snapshot = "d" * 40
same, _ = _pair_build(tmp_path, monkeypatch, _pair("sibling", root=tmp_path, fixed=moved, parent=snapshot, commit=snapshot))
assert _fixed_rows(same, "sibling") == []
# two candidate renames are ambiguous: no negative
two = moved + "\n\ndef f_other(q):\n return q\n"
ambiguous, _ = _pair_build(tmp_path, monkeypatch, _pair("two", root=tmp_path, fixed=two))
assert _fixed_rows(ambiguous, "two") == []
def test_ordinary_pairs_and_matcher_limits_keep_their_negative(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
js_vuln, js_fixed = (
"module.exports = {\n ps: function(pid) {\n exec('ps ' + pid)\n }\n}\n",
("module.exports = {\n ps: function(pid) {\n execFile('ps', [pid])\n }\n}\n"),
)
cache = tmp_path / "pointer-cache"
monkeypatch.setenv("OPENULTRASAST_PAIR_CACHE", str(cache))
js = _pair("js", root=cache, function="ps", vuln=js_vuln, fixed=js_fixed)
js = PairCase(**{**js.__dict__, "language": "javascript"})
build, record = _pair_build(tmp_path, monkeypatch, _pair("plain", root=tmp_path), js)
assert {(r.unit, r.candidate, r.provenance) for r in _fixed_rows(build, "plain")} == {("pin", "a.py::f", ""), ("delta", "a.py::f", "")}
assert {r.candidate for r in _fixed_rows(build, "js")} == {"a.py::ps"}, "the pointer file in the pair cache was read"
assert record["pair_fixed_side"]["absent_fixed_side"]["labels"] == 0
assert record["verified_pin_labels"]["by_family"]["injection"]["negative_rows"] == 2
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"}}'}) == ()