Merge branch 'joern-leak-audit'

tests/test_joern_leak_audit.py

@@ -0,0 +1,161 @@
+"""Synthetic, offline result snapshots; no analyzer, model or corpus access."""
+
+import importlib
+import json
+from pathlib import Path
+
+import pytest
+
+
+@pytest.fixture
+def audit_module(monkeypatch):
+    monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[1] / "benchmarks/learn"))
+    return importlib.import_module("joern_leak_audit")
+
+
+def units(count=12):
+    return [
+        {"unit": f"{pair}-{label}", "pair": str(pair), "label": label, "family": "injection", "group": "g", "fold": "f"}
+        for pair in range(count)
+        for label in (1, 0)
+    ]
+
+
+def write_result(root, module, rows, name="pin", *, done=True, version=None):
+    path = root / f"{name}.json"
+    path.write_text(
+        json.dumps(
+            {
+                "repo": "repo",
+                "pin": name,
+                "done": done,
+                "materialization_version": module.materialization_version() if version is None else version,
+                "units": rows,
+            }
+        )
+    )
+    return path
+
+
+def result(unit, status):
+    return {
+        **unit,
+        "source": "authored",
+        "status": status,
+        "pin_role": "vulnerable" if unit["label"] else "fixed",
+        "traces": [{"steps": [{"roles": ["source"]}, {"roles": ["sink"]}], "sanitized": False, "guard": {"text": "", "bounded": False}}]
+        if status == "path"
+        else [],
+        "witness_rows": [{"sink": "execute(x)"}] if status == "path" else [],
+    }
+
+
+def run(module, root, population):
+    return module.audit(population, root, expected_rows=len(population), expected_pairs=len(population) // 2)
+
+
+def test_skips_stale_unfinished_and_proves_read(audit_module, tmp_path):
+    population = units(2)
+    paths = [
+        write_result(tmp_path, audit_module, [result(population[0], "path")], "valid"),
+        write_result(tmp_path, audit_module, [result(population[1], "path")], "stale", version=-1),
+        write_result(tmp_path, audit_module, [result(population[2], "path")], "unfinished", done=False),
+        write_result(tmp_path, audit_module, [result(population[3], "path")], "both", version=-1, done=False),
+    ]
+    report = run(audit_module, tmp_path, population)
+    evidence = report["inputs"]
+    assert (evidence["stale"], evidence["unfinished"], evidence["skipped"]) == (2, 2, 3)
+    assert evidence["files_read"] == 4 and evidence["accepted_files"] == evidence["result_rows_read"] == 1
+    assert evidence["bytes_read"] == sum(p.stat().st_size for p in paths)
+    assert report["matched_rows"] == 1
+    assert report["overall"]["coverage"]["missing"] == 3
+    assert report["by_source"]["unknown"]["rows"] == 3
+    assert report["overall"]["signals"]["has_path"]["rows"] == 1
+    assert report["overall"]["both_asked_signals"]["has_path"]["pooled_auc"] is None
+
+
+def test_asymmetry_and_honest_denominator(audit_module, tmp_path):
+    population = units(6)
+    statuses = [
+        "path",
+        "asked-nothing",
+        "asked-nothing",
+        "path",
+        "path",
+        "path",
+        "asked-nothing",
+        "asked-nothing",
+        "path",
+        "failed",
+        "unsupported",
+        "timeout",
+    ]
+    write_result(tmp_path, audit_module, [result(u, s) for u, s in zip(population, statuses, strict=True)])
+    report = run(audit_module, tmp_path, population)
+    overall = report["overall"]
+    assert overall["pair_asymmetry"] == {"vulnerable_only": 1, "fixed_only": 1, "both": 1, "neither": 1, "not_both_asked": 2}
+    assert overall["both_asked_pairs"] == 4
+    assert overall["both_asked_signals"]["asked"]["within_pair_accuracy"] == 0.5
+    assert overall["coverage"] == {"path": 5, "asked-nothing": 4, "failed": 1, "timeout": 1, "unsupported": 1, "missing": 0}
+    assert report["by_family"]["injection"] == overall == report["by_source"]["authored"]
+
+
+@pytest.mark.parametrize("balanced", [False, True])
+def test_constructed_leak_and_balanced_set(audit_module, tmp_path, balanced):
+    population = units()
+    rows = [result(u, "path" if u["label"] == (int(u["pair"]) % 2 if balanced else 1) else "asked-nothing") for u in population]
+    write_result(tmp_path, audit_module, rows)
+    overall = run(audit_module, tmp_path, population)["overall"]
+    for denominator in ("signals", "both_asked_signals"):
+        signals = overall[denominator]
+        for name in ("has_path", "shortest_path_steps"):
+            assert signals[name]["flagged"] is not balanced
+            assert signals[name]["within_pair_accuracy"] == signals[name]["pooled_auc"] == (0.5 if balanced else 1)
+        assert signals["sink_kind"]["execute"]["flagged"] is not balanced
+        assert not signals["asked"]["flagged"]
+        assert not signals["sanitized"]["flagged"]
+        assert not signals["guard_present"]["flagged"]
+
+
+def test_minimum_samples_reverse_and_source_proxy(audit_module, tmp_path):
+    population = units(2)
+    write_result(tmp_path, audit_module, [result(u, "path" if not u["label"] else "asked-nothing") for u in population])
+    assert not run(audit_module, tmp_path, population)["overall"]["signals"]["has_path"]["flagged"]
+    population = units(12)
+    rows = []
+    for u in population:
+        row = result(u, "path" if not u["label"] else "unsupported")
+        row["source"] = "positive" if u["label"] else "negative"
+        rows.append(row)
+    write_result(tmp_path, audit_module, rows)
+    report = run(audit_module, tmp_path, population)
+    assert report["overall"]["signals"]["asked"]["flagged"]
+    assert report["overall"]["signals"]["asked"]["pooled_auc"] == 0
+    assert report["overall"]["both_asked_pairs"] == 0
+    assert report["by_source"]["positive"]["signals"]["asked"]["pooled_auc"] is None
+
+
+def test_shortest_trace_evidence_and_missing_trace(audit_module):
+    row = result(units(1)[0], "path")
+    row["traces"].insert(0, {"steps": [{}] * 5, "sanitized": False})
+    row["witness_rows"].insert(0, {"sink": "longer(x)"})
+    row["traces"][1].update(sanitized=True, guard={"bounded": True, "text": "off-path bound"})
+    features = audit_module.features(row)
+    assert (features["shortest_path_steps"], features["sanitized"], features["guard_present"], features["sink_kind"]) == (
+        2,
+        1,
+        1,
+        "execute",
+    )
+    row["traces"] = []
+    assert audit_module.features(row)["shortest_path_steps"] is None
+
+
+def test_bad_inputs_fail_loudly(audit_module, tmp_path):
+    population = units(1)
+    write_result(tmp_path, audit_module, [result(population[0], "path")] * 2)
+    with pytest.raises(ValueError, match="Duplicate"):
+        run(audit_module, tmp_path, population)
+    write_result(tmp_path, audit_module, [{**result(population[0], "path"), "label": 0}])
+    with pytest.raises(ValueError, match="metadata mismatch"):
+        run(audit_module, tmp_path, population)