mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 19:59:26 +00:00
Marius CiepluchandClaude Opus 5.5
d1896cb8e7
learn/schema.py is the closed, typed allow-list of candidate features: 66 features over 11 instruments (language one-hot, quick rules with 23 mechanism buckets and a prior count, engine, repo-facts, source span, entry-point distance, roles, delta, verify, agree, model sinks), two profiles (static 57; plane adds verify, agree and model sinks for 66), schema_version and a feature-set digest. No repository, path, file, function, rule or commit identity is a feature; site_match, in_fix_range, fixed-pin alerts, case.json fields and costs are refused. Missing is not zero: each record carries per-instrument state (ran | none | failed | not_applicable), and every feature of an instrument that did not run is null. learn/features.py builds records on the host (build_for_scan, whole repository or base..head delta); plane/tasks/features.py builds them per case in a Run from the delivered artifacts, dropping site_match and in_fix_range on read and ignoring fixed-pin alerts. The generator adds a model-free `features` entry per case before `remember`, which stores the records as `features` rows (re-validated). memory KINDS gains features, label, decision, experiment, arm_outcome and experiment_result. The committed validation-46 manifests are regenerated (entries appended; the first Run's entries are unchanged). The runner image must be rebuilt and re-pinned before a Run executes the new task. Proof on real artifacts (scratch store, not committed): the recorded validation-46 run gives 75 feature records over 15 cases (43 verify candidates plus vulnerable-pin alerts), each case's vulnerable pin exported and read (0.5k-9k files); remember stored 75 features rows beside 43 verdict, 84 alert, 61 unit_cost, 15 facts and 8 coverage rows. Co-Authored-By: Claude Opus 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo
84 lines
4.4 KiB
Python
84 lines
4.4 KiB
Python
"""The plane task `features` (learned-decision-engine task 1) on validation-46's recorded lmdeploy case
|
|
(``tests/fixtures/plane-remember``), and `remember` storing its records as `features` rows."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import shutil
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from openultrasast.learn.schema import FeatureRecordError, feature_set_digest
|
|
from openultrasast.plane.memory import FileStore
|
|
from openultrasast.plane.tasks import features
|
|
from openultrasast.plane.tasks.remember import Context, remember_run, rows_for
|
|
|
|
FIXTURE = Path(__file__).parent / "fixtures" / "plane-remember" / "validation-46"
|
|
CASE = "lmdeploy-media-url-ssrf"
|
|
|
|
|
|
@pytest.fixture
|
|
def run_dir(tmp_path: Path) -> Path:
|
|
target = tmp_path / "plane" / "validation-46"
|
|
shutil.copytree(FIXTURE, target)
|
|
return target
|
|
|
|
|
|
def _checkout(root: Path) -> Path:
|
|
media = root / "lmdeploy" / "vl" / "media"
|
|
media.mkdir(parents=True)
|
|
(media / "connection.py").write_text("def _is_safe_url(u):\n return True\n\n\ndef _load_http_url(u):\n x = 1\n return x\n")
|
|
return root
|
|
|
|
|
|
def test_features_of_the_recorded_case(run_dir: Path, tmp_path: Path) -> None:
|
|
[rec] = features.recorded_case(run_dir, CASE, workspace=_checkout(tmp_path / "checkout"))
|
|
assert (rec["candidate"], rec["family"], rec["profile"], rec["unit"]) == (
|
|
"lmdeploy/vl/media/connection.py::_load_http_url", "untrusted_destination", "plane", "pin",
|
|
) # fmt: skip
|
|
x, states = rec["x"], {k: v["state"] for k, v in rec["instruments"].items()}
|
|
assert (x["verify.flag_a"], x["verify.flag_b"], x["verify.flag_c"], x["verify.votes"]) == (True, True, None, 2)
|
|
assert x["agree.final"] == "agreed" and x["facts.callers"] == 0 and x["facts.is_global"] is False
|
|
assert x["facts.function_lines"] == 3 and x["lang.python"] is True and x["verify.site_offset"] == 0
|
|
assert states["verify"] == states["agree"] == states["facts"] == states["source"] == "ran"
|
|
assert states["quick"] == states["engine"] == "none", "no alerts task in this Run: no coverage, not zero hits"
|
|
assert rec["instruments"]["verify"]["version"] == "deepseek-flash" and rec["feature_set_digest"] == feature_set_digest()
|
|
assert "site_match" not in json.dumps(rec), "the fixture's site_match: true was dropped on read"
|
|
[without] = features.recorded_case(run_dir, CASE)
|
|
assert without["instruments"]["source"]["state"] == "none" and without["x"]["facts.function_lines"] is None
|
|
|
|
|
|
def test_the_task_writes_records_and_remember_stores_them(run_dir: Path, tmp_path: Path) -> None:
|
|
out = tmp_path / "out"
|
|
env = {
|
|
"OUSAST_OUTPUT_DIR": str(out), "OUSAST_INPUT_FACTS": str(run_dir / f"{CASE}-facts" / "facts.json"),
|
|
"OUSAST_INPUT_AGREED": str(run_dir / f"{CASE}-final" / "agreed.json"),
|
|
**{f"OUSAST_INPUT_PASS_{p.upper()}": str(run_dir / f"{CASE}-v{p}" / "units.jsonl") for p in "abc"},
|
|
**{f"OUSAST_INPUT_SUMMARY_{p.upper()}": str(run_dir / f"{CASE}-v{p}" / "summary.json") for p in "abc"},
|
|
} # fmt: skip
|
|
assert features.main(env) == 0
|
|
summary = json.loads((out / "summary.json").read_text())
|
|
assert (summary["status"], summary["records"], summary["usd"], summary["calls"]) == ("done", 1, 0, 0)
|
|
assert summary["instruments"]["verify"] == {"ran": 1} and summary["instruments"]["quick"] == {"none": 1}
|
|
(run_dir / f"{CASE}-features").mkdir()
|
|
shutil.copy(out / features.OUTPUT, run_dir / f"{CASE}-features" / features.OUTPUT)
|
|
store = FileStore(tmp_path / "memory")
|
|
[result] = remember_run(run_dir, store, population="population-v2", split="validation")
|
|
assert result.kinds == {"facts": 1, "unit_cost": 2, "verdict": 1, "features": 1}
|
|
[row] = store.rows(kind="features")
|
|
assert row.row["candidate"] == "lmdeploy/vl/media/connection.py::_load_http_url" and row.row["family"] == "untrusted_destination"
|
|
assert row.row["repo"] == "example.com/reserved/lmdeploy" and row.row["x"]["agree.final"] == "agreed"
|
|
|
|
|
|
def test_remember_refuses_a_record_outside_the_allow_list(run_dir: Path) -> None:
|
|
[rec] = features.recorded_case(run_dir, CASE)
|
|
rec["x"]["site_match"] = True
|
|
ctx = Context("r", "c-remember", "https://github.com/o/r.git", "0" * 40, "sha256:" + "1" * 64)
|
|
with pytest.raises(FeatureRecordError, match="site_match"):
|
|
rows_for(ctx, facts=None, passes={}, models={}, agreed=None, features=[rec])
|
|
|
|
|
|
def test_the_task_fails_loudly_without_an_output_dir() -> None:
|
|
assert features.main({}) == 2
|