mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 03:49:26 +00:00
chore(p0): remove the improve lever and the plane improvement loop
Off every active line (PATHFINDER-2026-10-13/03-unified-proposal.md, cuts). Deleted: src/openultrasast/improve/ (1,175 lines), the ousast improve subcommand, plane/tasks/loop.py and loop.yaml with the --loop generation, the memory-snapshot step in memory.seed, pairs.build_pair_signals, and the four improve tests plus the loop cases in other tests. Docs no longer describe the lever. Net: 3,073 lines deleted, 83 added. Targeted tests: 154 passed; 3,436 tests collect. Co-Authored-By: Claude Fable 5.1 <[email protected]>
This commit is contained in:
Marius Ciepluch
co-authored by
Claude Fable 5.1
Claude Fable 5.1
parent
d2010d0693
commit
94e96c6885
@@ -1,69 +0,0 @@
|
||||
"""`ousast improve` CLI: drives the self-improvement loop and the loop-owned ledger."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from openultrasast.cli import main
|
||||
from openultrasast.ruleset import PatternRule, read_rule_ledger, write_ruleset
|
||||
|
||||
|
||||
def _rule(rule_id: str, pattern: str) -> PatternRule:
|
||||
return PatternRule(rule_id=rule_id, title=rule_id, languages=("python",), cwe="CWE-95", tags=("injection",), pattern=pattern)
|
||||
|
||||
|
||||
def _expected_block(rule_id: str) -> str:
|
||||
return (
|
||||
'[[expected]]\ncwe = "CWE-95"\nclass = "code injection"\npath = "app.py"\n'
|
||||
f'line = 2\nrule_id = "{rule_id}"\nsink = "eval"\nevidence = "eval executes input"\n'
|
||||
)
|
||||
|
||||
|
||||
def _scaffold(tmp_path: Path) -> tuple[Path, Path]:
|
||||
repo = tmp_path / "repo"
|
||||
repo.mkdir()
|
||||
(repo / "app.py").write_text("def f(x):\n return eval(x)\ndef g():\n print('hello')\n")
|
||||
ruleset_dir = tmp_path / "ruleset"
|
||||
write_ruleset(ruleset_dir / "python" / "rules.toml", [_rule("good-eval", r"\beval\s*\("), _rule("noisy-print", r"\bprint\s*\(")])
|
||||
manifest = tmp_path / "bench.toml"
|
||||
manifest.write_text('name = "t"\nlanguage = "python_web"\n\n[source]\npath = "repo"\n\n' + _expected_block("good-eval"))
|
||||
return repo, ruleset_dir
|
||||
|
||||
|
||||
def test_improve_accepts_and_writes_loop_owned_ledger(tmp_path: Path) -> None:
|
||||
repo, ruleset_dir = _scaffold(tmp_path)
|
||||
manifest = tmp_path / "bench.toml"
|
||||
|
||||
exit_code = main(["improve", str(manifest), "--ruleset-dir", str(ruleset_dir)])
|
||||
|
||||
assert exit_code == 0
|
||||
# The ledger lands exactly where `scan`/`benchmark` read it for this target.
|
||||
ledger = repo / ".openultrasast" / "calibration" / "rule_policy.json"
|
||||
assert ledger.exists()
|
||||
assert read_rule_ledger(ledger)["noisy-print"]["status"] == "shadow"
|
||||
journal = repo / ".openultrasast" / "calibration" / "improve_journal.json"
|
||||
assert journal.exists()
|
||||
|
||||
|
||||
def test_improve_dry_run_never_touches_the_targets_ledger(tmp_path: Path) -> None:
|
||||
repo, ruleset_dir = _scaffold(tmp_path)
|
||||
manifest = tmp_path / "bench.toml"
|
||||
|
||||
exit_code = main(["improve", str(manifest), "--ruleset-dir", str(ruleset_dir), "--dry-run"])
|
||||
|
||||
assert exit_code == 0
|
||||
assert not (repo / ".openultrasast" / "calibration" / "rule_policy.json").exists()
|
||||
|
||||
|
||||
def test_improve_rejects_missing_manifest(tmp_path: Path) -> None:
|
||||
import pytest
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
main(["improve", str(tmp_path / "nope.toml")])
|
||||
|
||||
|
||||
def test_improve_fails_loud_when_pair_catalog_is_missing(tmp_path: Path) -> None:
|
||||
import pytest
|
||||
|
||||
_, ruleset_dir = _scaffold(tmp_path)
|
||||
manifest = tmp_path / "bench.toml"
|
||||
with pytest.raises(SystemExit, match="pair catalog"):
|
||||
main(["improve", str(manifest), "--ruleset-dir", str(ruleset_dir), "--dry-run", "--pair-catalog", str(tmp_path / "missing.toml")])
|
||||
@@ -132,8 +132,6 @@ GOVERNANCE_SCORING_BENCHMARK_PLANES = (
|
||||
"openultrasast.scoring.project_score",
|
||||
"openultrasast.ruleset",
|
||||
"openultrasast.ruleset.store",
|
||||
"openultrasast.improve",
|
||||
"openultrasast.improve.evolve",
|
||||
"openultrasast.fusion",
|
||||
"openultrasast.skills",
|
||||
"openultrasast.redaction",
|
||||
|
||||
@@ -1,324 +0,0 @@
|
||||
"""Self-improvement loop: bounded levers, validator, gates, accept/revert (Phase 3 task 7)."""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from openultrasast.benchmark import load_benchmark_manifest
|
||||
from openultrasast.improve import (
|
||||
EvolveValidator,
|
||||
PolicyConstantEdit,
|
||||
RuleStatusEdit,
|
||||
StrictValidationError,
|
||||
build_rule_signals,
|
||||
propose_status_edits,
|
||||
run_improvement,
|
||||
run_round,
|
||||
)
|
||||
from openultrasast.policy import load_policy
|
||||
from openultrasast.ruleset import PatternRule, read_rule_ledger, write_ruleset
|
||||
|
||||
POLICY = load_policy()
|
||||
|
||||
|
||||
def _rule(rule_id: str, pattern: str, cwe: str = "CWE-95") -> PatternRule:
|
||||
return PatternRule(rule_id=rule_id, title=rule_id, languages=("python",), cwe=cwe, tags=("injection",), pattern=pattern)
|
||||
|
||||
|
||||
def _expected(rule_id: str) -> str:
|
||||
return (
|
||||
'[[expected]]\ncwe = "CWE-95"\nclass = "code injection"\npath = "app.py"\n'
|
||||
f'line = 2\nrule_id = "{rule_id}"\nsink = "eval"\nevidence = "eval executes input"\n'
|
||||
)
|
||||
|
||||
|
||||
# ---- validator (7.1, 7.2) ---------------------------------------------------
|
||||
|
||||
|
||||
def test_validator_rejects_enabled_to_disabled_jump() -> None:
|
||||
ruleset = {"r": _rule("r", r"\beval\s*\(")}
|
||||
with pytest.raises(StrictValidationError, match="shadow"):
|
||||
EvolveValidator().validate(RuleStatusEdit("r", "enabled", "disabled"), ruleset, POLICY)
|
||||
|
||||
|
||||
def test_validator_accepts_enabled_to_shadow() -> None:
|
||||
ruleset = {"r": _rule("r", r"\beval\s*\(")}
|
||||
EvolveValidator().validate(RuleStatusEdit("r", "enabled", "shadow"), ruleset, POLICY) # no raise
|
||||
|
||||
|
||||
def test_validator_rejects_unknown_rule_and_bad_status() -> None:
|
||||
ruleset = {"r": _rule("r", r"\beval\s*\(")}
|
||||
with pytest.raises(StrictValidationError, match="unknown rule_id"):
|
||||
EvolveValidator().validate(RuleStatusEdit("nope", "enabled", "shadow"), ruleset, POLICY)
|
||||
with pytest.raises(StrictValidationError):
|
||||
EvolveValidator().validate(RuleStatusEdit("r", "enabled", "bogus"), ruleset, POLICY)
|
||||
|
||||
|
||||
def test_validator_never_tunes_severity_and_bounds_constants() -> None:
|
||||
with pytest.raises(StrictValidationError, match="severity is upstream-owned"):
|
||||
EvolveValidator().validate(PolicyConstantEdit("severity", 3, 5), {}, POLICY)
|
||||
with pytest.raises(StrictValidationError):
|
||||
EvolveValidator().validate(PolicyConstantEdit("K", 60, 5000), {}, POLICY) # out of bounds
|
||||
EvolveValidator().validate(PolicyConstantEdit("K", 60, 90), {}, POLICY) # in bounds, no raise
|
||||
|
||||
|
||||
# ---- proposer + signals (7.4, 7.6) -----------------------------------------
|
||||
|
||||
|
||||
def test_propose_status_edits_auto_shadows_precision_draggers_only() -> None:
|
||||
ruleset = {"good": _rule("good", r"\beval\s*\("), "noisy": _rule("noisy", r"\bprint\s*\(")}
|
||||
per_rule = {"good": {"matched": 1, "missed": 0, "false_positives": 0}, "noisy": {"matched": 0, "missed": 0, "false_positives": 2}}
|
||||
|
||||
edits = propose_status_edits(per_rule, ruleset, current_ledger={}, blocked_keys=set())
|
||||
assert [(e.rule_id, e.to_status) for e in edits] == [("noisy", "shadow")]
|
||||
# novelty gate: a previously-reverted edit is not re-proposed.
|
||||
assert propose_status_edits(per_rule, ruleset, {}, {edits[0].key()}) == []
|
||||
# already shadow -> not re-proposed.
|
||||
assert propose_status_edits(per_rule, ruleset, {"noisy": {"status": "shadow"}}, set()) == []
|
||||
|
||||
|
||||
# ---- end-to-end loop (7.5) --------------------------------------------------
|
||||
|
||||
|
||||
def _manifest(tmp_path: Path, expected_block: str) -> tuple[Path, object]:
|
||||
manifest_path = tmp_path / "bench.toml"
|
||||
manifest_path.write_text('name = "t"\nlanguage = "python_web"\n\n[source]\npath = "repo"\n\n' + expected_block)
|
||||
return manifest_path, load_benchmark_manifest(manifest_path)
|
||||
|
||||
|
||||
def _ruleset_dir(tmp_path: Path, rules: list[PatternRule]) -> Path:
|
||||
rd = tmp_path / "ruleset" / "python"
|
||||
write_ruleset(rd / "rules.toml", rules)
|
||||
return tmp_path / "ruleset"
|
||||
|
||||
|
||||
def test_round_auto_shadows_noisy_rule_and_accepts(tmp_path: Path) -> None:
|
||||
repo = tmp_path / "repo"
|
||||
repo.mkdir()
|
||||
(repo / "app.py").write_text("def f(x):\n return eval(x)\ndef g():\n print('hello')\n")
|
||||
ruleset_dir = _ruleset_dir(tmp_path, [_rule("good-eval", r"\beval\s*\("), _rule("noisy-print", r"\bprint\s*\(")])
|
||||
_, manifest = _manifest(tmp_path, _expected("good-eval"))
|
||||
ledger = tmp_path / "rule_policy.json"
|
||||
journal = tmp_path / "journal.json"
|
||||
|
||||
outcome = run_round(repo, manifest, ledger_path=ledger, journal_path=journal, ruleset_dir=ruleset_dir)
|
||||
|
||||
assert outcome.accepted and outcome.reason == "accepted"
|
||||
assert outcome.fp_before > 0.0 and outcome.fp_after == 0.0
|
||||
assert outcome.recall_after == 1.0 # the true positive is preserved
|
||||
assert read_rule_ledger(ledger)["noisy-print"]["status"] == "shadow"
|
||||
assert json.loads(journal.read_text())[0]["outcome"] == "accepted"
|
||||
|
||||
|
||||
def test_round_reverts_byte_for_byte_when_shadowing_would_drop_recall(tmp_path: Path) -> None:
|
||||
repo = tmp_path / "repo"
|
||||
repo.mkdir()
|
||||
(repo / "app.py").write_text("def f(x):\n return eval(x)\ndef g(y):\n return eval(y)\n")
|
||||
ruleset_dir = _ruleset_dir(tmp_path, [_rule("dual-eval", r"\beval\s*\(")])
|
||||
_, manifest = _manifest(tmp_path, _expected("dual-eval"))
|
||||
ledger = tmp_path / "rule_policy.json"
|
||||
journal = tmp_path / "journal.json"
|
||||
|
||||
outcome = run_round(repo, manifest, ledger_path=ledger, journal_path=journal, ruleset_dir=ruleset_dir)
|
||||
|
||||
assert not outcome.accepted and outcome.reason == "reverted"
|
||||
assert outcome.score_after > outcome.score_before # score improved but the recall gate blocks it
|
||||
assert not ledger.exists() # byte-for-byte revert: the ledger was never written
|
||||
assert json.loads(journal.read_text())[0]["outcome"] == "reverted"
|
||||
|
||||
|
||||
def test_run_improvement_converges_after_accepting(tmp_path: Path) -> None:
|
||||
repo = tmp_path / "repo"
|
||||
repo.mkdir()
|
||||
(repo / "app.py").write_text("def f(x):\n return eval(x)\ndef g():\n print('hello')\n")
|
||||
ruleset_dir = _ruleset_dir(tmp_path, [_rule("good-eval", r"\beval\s*\("), _rule("noisy-print", r"\bprint\s*\(")])
|
||||
_, manifest = _manifest(tmp_path, _expected("good-eval"))
|
||||
|
||||
outcomes = run_improvement(
|
||||
repo,
|
||||
manifest,
|
||||
ledger_path=tmp_path / "rule_policy.json",
|
||||
journal_path=tmp_path / "journal.json",
|
||||
ruleset_dir=ruleset_dir,
|
||||
max_rounds=4,
|
||||
)
|
||||
|
||||
assert outcomes[0].accepted # round 1 shadows the noisy rule
|
||||
assert outcomes[-1].reason == "no_proposals" # converges (nothing left to improve)
|
||||
|
||||
|
||||
def test_build_rule_signals_separates_miss_and_fp() -> None:
|
||||
from openultrasast.benchmark import BenchmarkFalsePositive, BenchmarkMetrics, BenchmarkMiss, BenchmarkResult
|
||||
|
||||
result = BenchmarkResult(
|
||||
benchmark_run_id="r",
|
||||
benchmark_name="n",
|
||||
language="python_web",
|
||||
mode="quick",
|
||||
scan_id=None,
|
||||
scan_run_dir=None,
|
||||
metrics=BenchmarkMetrics(1, 1, 0, 1, 1, 0.0, 0.0, None, {}, {}, {}),
|
||||
misses=[BenchmarkMiss("CWE-89", "sqli", "a.py", "ev", "no match", rule_id="sqli")],
|
||||
false_positives=[BenchmarkFalsePositive("noisy:a.py:1", "a.py", 1, "fp", rule_id="noisy")],
|
||||
baseline_deltas=[],
|
||||
calibration_records=[],
|
||||
rule_recommendations=[],
|
||||
)
|
||||
signals = build_rule_signals(result)
|
||||
assert {s["signal"] for s in signals} == {"miss", "fp"}
|
||||
assert {s["rule_id"] for s in signals} == {"sqli", "noisy"}
|
||||
|
||||
|
||||
# ---- pair-corpus-honesty: per-profile holdout clause (Req 7.2-7.4) ----------
|
||||
|
||||
|
||||
def _pair(
|
||||
tmp_path: Path, name: str, vuln_src: str, fixed_src: str, rule_id: str, sink: str, *, provenance: str, review_tier: str = "reviewed"
|
||||
):
|
||||
from openultrasast.benchmark import ExpectedFinding
|
||||
from openultrasast.pairs import PairCase
|
||||
|
||||
vuln = tmp_path / f"{name}-vuln.py"
|
||||
fixed = tmp_path / f"{name}-fixed.py"
|
||||
vuln.write_text(vuln_src)
|
||||
fixed.write_text(fixed_src)
|
||||
return PairCase(
|
||||
name=name,
|
||||
slice="github",
|
||||
language="python",
|
||||
origin="test",
|
||||
vuln_file=vuln,
|
||||
fixed_file=fixed,
|
||||
relpath="app.py",
|
||||
expected=(
|
||||
ExpectedFinding(
|
||||
cwe="CWE-95",
|
||||
vulnerability_class="x",
|
||||
path="app.py",
|
||||
evidence="",
|
||||
rule_id=rule_id,
|
||||
sink=sink,
|
||||
mechanism="source_reaches_sink",
|
||||
),
|
||||
),
|
||||
min_recall=1.0,
|
||||
fix_policy="silent",
|
||||
provenance=provenance,
|
||||
split="holdout",
|
||||
review_tier=review_tier,
|
||||
reviewer="test" if review_tier == "reviewed" else "",
|
||||
)
|
||||
|
||||
|
||||
def test_round_rejects_when_a_profile_regresses_on_holdout(tmp_path: Path) -> None:
|
||||
repo = tmp_path / "repo"
|
||||
repo.mkdir()
|
||||
(repo / "app.py").write_text("def f(x):\n return eval(x)\ndef g():\n print('hello')\n")
|
||||
ruleset_dir = _ruleset_dir(tmp_path, [_rule("good-eval", r"\beval\s*\("), _rule("noisy-print", r"\bprint\s*\(")])
|
||||
_, manifest = _manifest(tmp_path, _expected("good-eval"))
|
||||
ledger = tmp_path / "rule_policy.json"
|
||||
journal = tmp_path / "journal.json"
|
||||
# The agent profile's holdout pair is only detected by the rule the round wants to shadow.
|
||||
agent_pairs = [
|
||||
_pair(tmp_path, f"agent{i}", "def h(y):\n print(y)\n", "def h(y):\n return y\n", "noisy-print", "print", provenance="agent")
|
||||
for i in range(5)
|
||||
]
|
||||
outcome = run_round(repo, manifest, ledger_path=ledger, journal_path=journal, ruleset_dir=ruleset_dir, pair_cases=agent_pairs)
|
||||
assert not outcome.accepted
|
||||
assert outcome.reason == "profile_regression:agent"
|
||||
assert outcome.profile_regressions == ["agent"]
|
||||
assert outcome.per_profile_before["agent"]["pair_correct"] == 5.0 and outcome.per_profile_after["agent"]["pair_correct"] == 0.0
|
||||
assert not ledger.exists()
|
||||
assert json.loads(journal.read_text())[0]["profile_regressions"] == ["agent"]
|
||||
|
||||
|
||||
def test_round_accepts_when_profiles_hold_and_reports_small_profiles(tmp_path: Path) -> None:
|
||||
repo = tmp_path / "repo"
|
||||
repo.mkdir()
|
||||
(repo / "app.py").write_text("def f(x):\n return eval(x)\ndef g():\n print('hello')\n")
|
||||
ruleset_dir = _ruleset_dir(tmp_path, [_rule("good-eval", r"\beval\s*\("), _rule("noisy-print", r"\bprint\s*\(")])
|
||||
_, manifest = _manifest(tmp_path, _expected("good-eval"))
|
||||
ledger = tmp_path / "rule_policy.json"
|
||||
journal = tmp_path / "journal.json"
|
||||
human_pairs = [
|
||||
_pair(tmp_path, f"h{i}", "def h(y):\n return eval(y)\n", "def h(y):\n return y\n", "good-eval", "eval", provenance="human")
|
||||
for i in range(5)
|
||||
]
|
||||
tiny_agent = [
|
||||
_pair(tmp_path, "a0", "def h(y):\n print(y)\n", "def h(y):\n return y\n", "noisy-print", "print", provenance="agent")
|
||||
]
|
||||
outcome = run_round(
|
||||
repo, manifest, ledger_path=ledger, journal_path=journal, ruleset_dir=ruleset_dir, pair_cases=human_pairs + tiny_agent
|
||||
)
|
||||
assert outcome.accepted and outcome.reason == "accepted"
|
||||
assert outcome.profile_regressions == []
|
||||
assert outcome.profiles_under_minimum == ["agent"] # one pair: reported, not gated
|
||||
assert ledger.exists()
|
||||
|
||||
|
||||
def test_profile_gate_ignores_title_and_advisory_tiers(tmp_path: Path) -> None:
|
||||
"""Req 9.3: only seeded and reviewed pairs gate; a regression confined to title pairs never rejects a round."""
|
||||
repo = tmp_path / "repo"
|
||||
repo.mkdir()
|
||||
(repo / "app.py").write_text("def f(x):\n return eval(x)\ndef g():\n print('hello')\n")
|
||||
ruleset_dir = _ruleset_dir(tmp_path, [_rule("good-eval", r"\beval\s*\("), _rule("noisy-print", r"\bprint\s*\(")])
|
||||
_, manifest = _manifest(tmp_path, _expected("good-eval"))
|
||||
ledger = tmp_path / "rule_policy.json"
|
||||
journal = tmp_path / "journal.json"
|
||||
title_pairs = [
|
||||
_pair(
|
||||
tmp_path,
|
||||
f"t{i}",
|
||||
"def h(y):\n print(y)\n",
|
||||
"def h(y):\n return y\n",
|
||||
"noisy-print",
|
||||
"print",
|
||||
provenance="agent",
|
||||
review_tier="title",
|
||||
)
|
||||
for i in range(5)
|
||||
]
|
||||
outcome = run_round(repo, manifest, ledger_path=ledger, journal_path=journal, ruleset_dir=ruleset_dir, pair_cases=title_pairs)
|
||||
assert outcome.accepted and outcome.reason == "accepted"
|
||||
assert outcome.profile_regressions == [] and "agent" not in outcome.per_profile_before
|
||||
ledger.unlink()
|
||||
journal.unlink()
|
||||
reviewed_pairs = [
|
||||
_pair(tmp_path, f"r{i}", "def h(y):\n print(y)\n", "def h(y):\n return y\n", "noisy-print", "print", provenance="agent")
|
||||
for i in range(5)
|
||||
]
|
||||
outcome = run_round(
|
||||
repo, manifest, ledger_path=ledger, journal_path=journal, ruleset_dir=ruleset_dir, pair_cases=reviewed_pairs + title_pairs
|
||||
)
|
||||
assert not outcome.accepted and outcome.reason == "profile_regression:agent"
|
||||
assert outcome.per_profile_before["agent"]["pairs"] == 5.0 # the five title pairs are scored elsewhere, never here
|
||||
|
||||
|
||||
def test_profile_gate_scores_vendored_pairs_only(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Req 10.4 / design: gates call select_vendored; a pointer pair never reaches the improve gate, cached or not."""
|
||||
from openultrasast.improve.evolve import evaluate_profiles
|
||||
from openultrasast.pairs import PairCase
|
||||
from openultrasast.ruleset import load_ruleset
|
||||
|
||||
monkeypatch.setenv("OPENULTRASAST_PAIRS_NETWORK", "1")
|
||||
vendored = _pair(
|
||||
tmp_path, "v", "def h(y):\n return eval(y)\n", "def h(y):\n return y\n", "python-unsafe-eval", "eval", provenance="human"
|
||||
)
|
||||
pointer_src = tmp_path / "cache" / "github" / "p"
|
||||
pointer_src.mkdir(parents=True)
|
||||
(pointer_src / "vuln.py").write_text("def h(y):\n return eval(y)\n")
|
||||
(pointer_src / "fixed.py").write_text("def h(y):\n return y\n")
|
||||
pointer = PairCase(
|
||||
**{
|
||||
**vendored.__dict__,
|
||||
"name": "p",
|
||||
"provenance": "agent",
|
||||
"vendored": False,
|
||||
"vuln_file": pointer_src / "vuln.py",
|
||||
"fixed_file": pointer_src / "fixed.py",
|
||||
"recipe": (("repo", "o/r"), ("parent", "a"), ("commit", "b"), ("path", "app.py"), ("mode", "name")),
|
||||
}
|
||||
)
|
||||
profiles = evaluate_profiles([vendored, pointer], load_ruleset())
|
||||
assert set(profiles) == {"human"} # the agent pointer pair is scored by `pairs --pointers`, never by the gate
|
||||
@@ -1,330 +0,0 @@
|
||||
"""Proposals from plane memory (harnessx-removal task 3, design section 5): rules M1/M2, the advisory M3 signal,
|
||||
the train-on-test guard, provenance, and the unchanged default `improve`."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from openultrasast.benchmark import load_benchmark_manifest
|
||||
from openultrasast.cli import main
|
||||
from openultrasast.improve import EvolveValidator, RuleStatusEdit, run_round
|
||||
from openultrasast.improve.memory import (
|
||||
SIDECAR,
|
||||
Guard,
|
||||
MemoryProposal,
|
||||
disputed_families,
|
||||
outcome_of,
|
||||
outcome_rows,
|
||||
propose_from_memory,
|
||||
read_sidecar,
|
||||
)
|
||||
from openultrasast.pairs import PairCase
|
||||
from openultrasast.plane.memory import FileStore, Record, row_id
|
||||
from openultrasast.ruleset import PatternRule, read_rule_ledger, write_ruleset
|
||||
|
||||
P = {name: ch * 40 for name, ch in (("a", "a"), ("b", "b"), ("c", "c"), ("f", "f"))}
|
||||
NO_GUARD = Guard()
|
||||
|
||||
|
||||
def _rule(rule_id: str, pattern: str = r"\bexec\s*\(", status: str = "enabled") -> PatternRule:
|
||||
return PatternRule(
|
||||
rule_id=rule_id, title=rule_id, languages=("python",), cwe="CWE-95", tags=("injection",), pattern=pattern, status=status
|
||||
)
|
||||
|
||||
|
||||
def _row(kind: str, repo: str, pin: str, subject: str, population: str = "dev", **fields: Any) -> dict[str, Any]:
|
||||
return {
|
||||
"id": row_id(kind, "run-1", f"{repo}-task", subject), "kind": kind, "repo": f"github.com/o/{repo}", "pin": pin,
|
||||
"run": "run-1", "task": f"{repo}-task", "population": population, "split": "validation", "image": "sha256:x", **fields,
|
||||
} # fmt: skip
|
||||
|
||||
|
||||
def verdict(repo: str, pin: str, fn: str, final: str, site_match: bool | None = None, **kw: Any) -> dict[str, Any]:
|
||||
cand = f"app.py::{fn}"
|
||||
return _row("verdict", repo, pin, cand, candidate=cand, family="injection", final=final, site_match=site_match, **kw)
|
||||
|
||||
|
||||
def alert(rule: str, repo: str, pin: str, fn: str, role: str = "vulnerable", status: str = "enabled", **kw: Any) -> dict[str, Any]:
|
||||
fields = {"rule_id": rule, "rule_status": status, "path": "app.py", "line": 3, "function": fn, "pin_role": role}
|
||||
return _row("alert", repo, pin, f"{rule}:{fn}:{role}", **fields, **kw)
|
||||
|
||||
|
||||
def recs(rows: list[dict[str, Any]]) -> list[Record]:
|
||||
return [Record(r, f"repos/{r['repo']}/{r['pin']}.jsonl", None) for r in rows]
|
||||
|
||||
|
||||
def propose(rows: list[dict[str, Any]], rules: list[PatternRule], guard: Guard = NO_GUARD, ledger: Any = None) -> list[MemoryProposal]:
|
||||
return propose_from_memory(recs(rows), {r.rule_id: r for r in rules}, ledger or {}, [], excluded=guard)
|
||||
|
||||
|
||||
def false_alerts(rule: str = "noisy", repos: tuple[str, ...] = ("r1", "r1", "r2"), population: str = "dev") -> list[dict[str, Any]]:
|
||||
rows: list[dict[str, Any]] = []
|
||||
for i, repo in enumerate(repos):
|
||||
rows.append(verdict(repo, P["a"], f"f{i}", "rejected", population=population))
|
||||
rows.append(alert(rule, repo, P["a"], f"f{i}", population=population))
|
||||
return rows
|
||||
|
||||
|
||||
# ---- M1: repeated false alerts -------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_m1_demotes_a_rule_with_false_alerts_across_repositories() -> None:
|
||||
[proposal] = propose(false_alerts(), [_rule("noisy")])
|
||||
assert (proposal.edit.from_status, proposal.edit.to_status) == ("enabled", "shadow")
|
||||
assert proposal.edit.rationale == f"memory:{proposal.proposal_id}"
|
||||
assert proposal.provenance["rule"] == "M1"
|
||||
assert proposal.provenance["counts"] == {"false_alerts": 3, "repos": 2, "agreed_hits": 0}
|
||||
assert len(proposal.provenance["evidence"]) == 6 # three alerts and the three rejected verdicts they sit at
|
||||
|
||||
|
||||
@pytest.mark.parametrize("repos", [("r1", "r2"), ("r1", "r1", "r1")], ids=["two-alerts", "one-repository"])
|
||||
def test_m1_stays_silent_below_threshold(repos: tuple[str, ...]) -> None:
|
||||
assert propose(false_alerts(repos=repos), [_rule("noisy")]) == []
|
||||
|
||||
|
||||
def test_m1_never_demotes_a_rule_with_an_agreed_hit() -> None:
|
||||
rows = false_alerts() + [verdict("r3", P["b"], "hit", "agreed"), alert("noisy", "r3", P["b"], "hit")]
|
||||
assert propose(rows, [_rule("noisy")]) == []
|
||||
site = false_alerts() + [verdict("r3", P["b"], "site", "rejected", site_match=True), alert("noisy", "r3", P["b"], "site")]
|
||||
assert propose(site, [_rule("noisy")]) == []
|
||||
|
||||
|
||||
def test_m1_counts_fixed_pin_alerts_in_the_fix_range_and_ignores_disputes() -> None:
|
||||
rows = [
|
||||
verdict("r1", P["a"], "sink", "agreed", site_match=True),
|
||||
verdict("r2", P["b"], "sink", "agreed", site_match=True),
|
||||
alert("noisy", "r1", P["f"], "sink", role="fixed"), # the declared site's function on the fixed pin
|
||||
alert("noisy", "r2", P["f"], "sink", role="fixed"),
|
||||
alert("noisy", "r2", P["c"], "other", role="fixed", in_fix_range=True),
|
||||
verdict("r1", P["a"], "maybe", "disputed"),
|
||||
alert("noisy", "r1", P["a"], "maybe"), # disputed: evidence in neither direction
|
||||
]
|
||||
[proposal] = propose(rows, [_rule("noisy")])
|
||||
assert proposal.provenance["counts"]["false_alerts"] == 3
|
||||
|
||||
|
||||
def test_m1_skips_rules_the_ruleset_does_not_know_or_that_are_not_enabled() -> None:
|
||||
assert propose(false_alerts(rule="ghost"), [_rule("noisy")]) == []
|
||||
assert propose(false_alerts(), [_rule("noisy")], ledger={"noisy": {"status": "shadow"}}) == []
|
||||
|
||||
|
||||
# ---- M2: missed declared sites -------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def site_hits(rule: str = "quiet") -> list[dict[str, Any]]:
|
||||
return [
|
||||
verdict("r1", P["a"], "s1", "agreed", site_match=True),
|
||||
verdict("r2", P["b"], "s2", "agreed", site_match=True),
|
||||
alert(rule, "r1", P["a"], "s1", status="shadow"),
|
||||
alert(rule, "r2", P["b"], "s2", status="shadow"),
|
||||
]
|
||||
|
||||
|
||||
def test_m2_promotes_a_shadow_rule_that_hits_declared_sites_nothing_else_catches() -> None:
|
||||
[proposal] = propose(site_hits(), [_rule("quiet", status="shadow"), _rule("other")])
|
||||
assert (proposal.edit.from_status, proposal.edit.to_status, proposal.provenance["rule"]) == ("shadow", "enabled", "M2")
|
||||
assert proposal.provenance["counts"]["site_hits"] == 2
|
||||
|
||||
|
||||
def test_m2_stays_silent_below_threshold_or_when_already_caught() -> None:
|
||||
rules = [_rule("quiet", status="shadow"), _rule("other")]
|
||||
assert propose(site_hits()[:3], rules) == [] # one hit
|
||||
assert propose([*site_hits(), alert("other", "r1", P["a"], "s1")], rules) == [] # an enabled rule catches s1
|
||||
|
||||
|
||||
def test_m2_stays_silent_on_a_fixed_pin_alert_or_rejected_candidates() -> None:
|
||||
rules = [_rule("quiet", status="shadow")]
|
||||
assert propose([*site_hits(), alert("quiet", "r1", P["f"], "s1", role="fixed", status="shadow")], rules) == []
|
||||
rejected = [
|
||||
verdict("r3", P["c"], "x1", "rejected"),
|
||||
verdict("r3", P["c"], "x2", "rejected"),
|
||||
alert("quiet", "r3", P["c"], "x1", status="shadow"),
|
||||
alert("quiet", "r3", P["c"], "x2", status="shadow"),
|
||||
]
|
||||
assert propose([*site_hits(), *rejected[::2]], rules) != [] # one rejected alert is tolerated
|
||||
assert propose([*site_hits(), *rejected], rules) == []
|
||||
|
||||
|
||||
# ---- coverage: no rule could fire, so no evidence either way -------------------------------------------------------
|
||||
|
||||
|
||||
def covered(repo: str, pin: str, kind: str, language: str = "python", role: str = "vulnerable") -> dict[str, Any]:
|
||||
return _row("coverage", repo, pin, f"{language}:{role}:{kind}", language=language, coverage=kind, files=3, pin_role=role)
|
||||
|
||||
|
||||
def test_m1_skips_a_repository_and_pin_whose_coverage_for_the_rules_language_is_none() -> None:
|
||||
blind = [*false_alerts(), covered("r2", P["a"], "none")]
|
||||
assert propose(blind, [_rule("noisy")]) == [], "r2's alert could not be evidence: one repository is left"
|
||||
wider = [*false_alerts(repos=("r1", "r1", "r2", "r3")), covered("r3", P["a"], "none")]
|
||||
[proposal] = propose(wider, [_rule("noisy")])
|
||||
assert proposal.provenance["counts"] == {"false_alerts": 3, "repos": 2, "agreed_hits": 0}
|
||||
assert proposal.provenance["excluded"]["coverage_none"] == {f"github.com/o/r3@{P['a']}": 1}
|
||||
|
||||
|
||||
def test_coverage_none_of_another_language_or_beside_engine_coverage_is_no_skip() -> None:
|
||||
other = [*false_alerts(), covered("r2", P["a"], "none", language="php")]
|
||||
assert len(propose(other, [_rule("noisy")])) == 1, "the rule's language (python) was covered there"
|
||||
engine = [*false_alerts(), covered("r2", P["a"], "none"), covered("r2", P["a"], "engine")]
|
||||
assert len(propose(engine, [_rule("noisy")])) == 1, "a later run covered the pin: its alerts are evidence"
|
||||
|
||||
|
||||
def test_m2_skips_a_blind_repository_and_pin() -> None:
|
||||
rules = [_rule("quiet", status="shadow"), _rule("other")]
|
||||
assert propose([*site_hits(), covered("r2", P["b"], "none")], rules) == []
|
||||
|
||||
|
||||
# ---- M3: advisory only ---------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_disputed_families_are_advisory_and_never_reach_the_validator(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
rows = [verdict("r1", P["a"], f"d{i}", "disputed" if i < 2 else "agreed") for i in range(5)]
|
||||
[signal] = disputed_families(recs(rows), NO_GUARD)
|
||||
assert signal["family"] == "injection" and signal["dispute_rate"] == 0.4
|
||||
assert disputed_families(recs(rows[1:]), NO_GUARD) == [] # four candidates: under the minimum
|
||||
assert propose(rows, [_rule("noisy")]) == [] # a dispute is no proposal
|
||||
|
||||
seen: list[object] = []
|
||||
original = EvolveValidator.validate
|
||||
monkeypatch.setattr(EvolveValidator, "validate", lambda self, edit, *a: (seen.append(edit), original(self, edit, *a))[1])
|
||||
repo, rules_dir, manifest = _scaffold(tmp_path)
|
||||
[proposal] = propose(false_alerts(), [_rule("noisy")])
|
||||
run_round(
|
||||
repo, manifest, ledger_path=tmp_path / "l.json", journal_path=tmp_path / "j.json", ruleset_dir=rules_dir, proposals=[proposal]
|
||||
)
|
||||
assert seen and all(isinstance(edit, RuleStatusEdit) for edit in seen)
|
||||
|
||||
|
||||
# ---- the train-on-test guard ---------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_guard_drops_the_qualification_population_and_reports_it() -> None:
|
||||
rows = false_alerts(population="independent-v2/validation") + false_alerts(repos=("r4",), population="dev")
|
||||
guard = Guard.build(populations=["independent-v2/validation"])
|
||||
kept, excluded = guard.apply(recs(rows))
|
||||
assert len(kept) == 2 and excluded["rows"] == 6 and excluded["population"] == {"independent-v2/validation": 6}
|
||||
assert propose(rows, [_rule("noisy")], guard) == [] # the only evidence is excluded: no proposal
|
||||
assert propose(rows, [_rule("noisy")]) != [] # the same rows without the guard would propose
|
||||
|
||||
|
||||
def test_guard_drops_holdout_pair_repositories_and_gated_cases() -> None:
|
||||
holdout = PairCase(
|
||||
name="h", slice="s", language="python", origin="o", vuln_file=Path("v"), fixed_file=Path("f"), relpath="app.py",
|
||||
expected=(), min_recall=1.0, fix_policy="silent", repo="O/R1", split="holdout",
|
||||
) # fmt: skip
|
||||
guard = Guard.build(gated_cases=[("https://github.com/o/r2", P["a"])], holdout_pairs=[holdout])
|
||||
kept, excluded = guard.apply(recs(false_alerts()))
|
||||
assert kept == []
|
||||
assert excluded["holdout_pair"] == {f"github.com/o/r1@{P['a']}": 4}
|
||||
assert excluded["gated_case"] == {f"github.com/o/r2@{P['a']}": 2}
|
||||
[proposal] = propose(false_alerts(repos=("r1", "r2", "r3", "r3", "r4")), [_rule("noisy")], guard)
|
||||
assert {e["repo"] for e in proposal.provenance["evidence"]} == {"github.com/o/r3", "github.com/o/r4"}
|
||||
assert proposal.provenance["excluded"]["rows"] == 4
|
||||
|
||||
|
||||
# ---- provenance, the round, and outcomes ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _scaffold(tmp_path: Path) -> tuple[Path, Path, Any]:
|
||||
repo = tmp_path / "repo"
|
||||
repo.mkdir()
|
||||
(repo / "app.py").write_text("def f(x):\n return eval(x)\n")
|
||||
rules_dir = tmp_path / "ruleset"
|
||||
write_ruleset(rules_dir / "python" / "rules.toml", [_rule("good-eval", r"\beval\s*\("), _rule("noisy")])
|
||||
manifest = tmp_path / "bench.toml"
|
||||
manifest.write_text(
|
||||
'name = "t"\nlanguage = "python_web"\n\n[source]\npath = "repo"\n\n[[expected]]\ncwe = "CWE-95"\n'
|
||||
'class = "code injection"\npath = "app.py"\nline = 2\nrule_id = "good-eval"\nsink = "eval"\nevidence = "eval"\n'
|
||||
)
|
||||
return repo, rules_dir, load_benchmark_manifest(manifest)
|
||||
|
||||
|
||||
def test_provenance_round_trips_journal_sidecar_store(tmp_path: Path) -> None:
|
||||
store = FileStore(tmp_path / "memory")
|
||||
store.ingest_rows("run-1", "t", false_alerts())
|
||||
records = store.rows()
|
||||
[proposal] = propose_from_memory(records, {"noisy": _rule("noisy")}, {}, [], excluded=NO_GUARD, index_digest="idx")
|
||||
repo, rules_dir, manifest = _scaffold(tmp_path)
|
||||
journal, ledger = tmp_path / "cal" / "journal.json", tmp_path / "cal" / "rule_policy.json"
|
||||
|
||||
outcome = run_round(repo, manifest, ledger_path=ledger, journal_path=journal, ruleset_dir=rules_dir, proposals=[proposal])
|
||||
|
||||
assert outcome.accepted and read_rule_ledger(ledger)["noisy"]["status"] == "shadow"
|
||||
[entry] = json.loads(journal.read_text())[0]["edits"]
|
||||
assert entry["evidence"] == proposal.proposal_id and entry["rationale"] == f"memory:{proposal.proposal_id}"
|
||||
side = read_sidecar(journal)[entry["evidence"]]
|
||||
assert side["edit_key"] == entry["key"] and side["index_digest"] == "idx" and side["round"] == 1
|
||||
by_id = {rec.row["id"]: rec for rec in records}
|
||||
for ev in side["evidence"]: # every cited row is in the store, at the object and version recorded
|
||||
assert by_id[ev["id"]].key == ev["key"] and by_id[ev["id"]].version == ev["version"]
|
||||
assert ev["version"].startswith("sha256:")
|
||||
|
||||
rows = outcome_rows(
|
||||
outcome.edits, outcome_of(outcome.reason, outcome.accepted) or "", 1, outcome.reason, {proposal.proposal_id: proposal}, "g"
|
||||
)
|
||||
store.put_rows(rows)
|
||||
back = store.rows(kind="proposal_outcome")
|
||||
assert {r.row["outcome"] for r in back} == {"accepted"} and {r.row["proposal_id"] for r in back} == {proposal.proposal_id}
|
||||
assert len(back) == 2 # one per (repository, pin) of the evidence
|
||||
|
||||
again = run_round(repo, manifest, ledger_path=ledger, journal_path=journal, ruleset_dir=rules_dir, proposals=[proposal])
|
||||
assert again.reason == "no_proposals" # the rule is shadow now: the proposal no longer applies
|
||||
|
||||
|
||||
def test_a_reverted_memory_proposal_is_not_retried_on_the_same_evidence(tmp_path: Path) -> None:
|
||||
repo, rules_dir, manifest = _scaffold(tmp_path)
|
||||
rows = [verdict(r, P["a"], f"f{i}", "rejected") for i, r in enumerate(("r1", "r1", "r2"))]
|
||||
rows += [alert("good-eval", r, P["a"], f"f{i}") for i, r in enumerate(("r1", "r1", "r2"))]
|
||||
[proposal] = propose(rows, [_rule("good-eval", r"\beval\s*\(")])
|
||||
journal = tmp_path / "j.json"
|
||||
first = run_round(repo, manifest, ledger_path=tmp_path / "l.json", journal_path=journal, ruleset_dir=rules_dir, proposals=[proposal])
|
||||
assert first.reason == "reverted" and outcome_of(first.reason, first.accepted) == "reverted" # shadowing drops recall
|
||||
second = run_round(repo, manifest, ledger_path=tmp_path / "l.json", journal_path=journal, ruleset_dir=rules_dir, proposals=[proposal])
|
||||
assert second.reason == "no_proposals"
|
||||
assert outcome_of("validation_failed: x", False) == "rejected" and outcome_of("no_proposals", False) is None
|
||||
|
||||
|
||||
def test_default_round_journal_has_no_evidence_and_no_sidecar(tmp_path: Path) -> None:
|
||||
repo, rules_dir, manifest = _scaffold(tmp_path)
|
||||
(repo / "app.py").write_text("def f(x):\n return eval(x)\ndef g(y):\n exec(y)\n")
|
||||
journal = tmp_path / "j.json"
|
||||
outcome = run_round(repo, manifest, ledger_path=tmp_path / "l.json", journal_path=journal, ruleset_dir=rules_dir)
|
||||
assert outcome.edits # the benchmark's own edit
|
||||
[entry] = json.loads(journal.read_text())[0]["edits"]
|
||||
assert set(entry) == {"key", "lever", "rule_id", "from", "to", "rationale"}
|
||||
assert not (tmp_path / SIDECAR).exists()
|
||||
|
||||
|
||||
# ---- the CLI -------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_default_improve_output_matches_the_recorded_baseline(capsys: pytest.CaptureFixture[str]) -> None:
|
||||
root = Path(__file__).resolve().parents[1]
|
||||
baseline = root / "benchmarks/measurements/2026-09-30-harnessx-removal-baseline/improve/stdout.txt"
|
||||
manifest = "benchmarks/manifests/java-spring-boot-vulnerable.toml"
|
||||
import os
|
||||
|
||||
cwd = Path.cwd()
|
||||
os.chdir(root)
|
||||
try:
|
||||
assert main(["improve", manifest, "--dry-run", "--no-pair-gate"]) == 0
|
||||
finally:
|
||||
os.chdir(cwd)
|
||||
assert capsys.readouterr().out.replace(str(root), "<ROOT>") == baseline.read_text()
|
||||
|
||||
|
||||
def test_improve_memory_with_an_empty_store_has_no_proposals(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
repo, rules_dir, _ = _scaffold(tmp_path)
|
||||
store = tmp_path / "memory"
|
||||
store.mkdir()
|
||||
(store / "index.jsonl").write_text("")
|
||||
catalog = Path(__file__).resolve().parents[1] / "benchmarks/pairs/catalog.toml"
|
||||
args = ["improve", str(tmp_path / "bench.toml"), "--ruleset-dir", str(rules_dir), "--dry-run", "--pair-catalog", str(catalog)]
|
||||
assert main([*args, "--no-pair-gate", "--memory", str(store)]) == 0
|
||||
out = capsys.readouterr().out
|
||||
assert "rows=0" in out and "memory_proposals=0" in out and "round 1: no_proposals" in out
|
||||
with pytest.raises(SystemExit, match="no index.jsonl"):
|
||||
main([*args, "--memory", str(tmp_path / "absent")])
|
||||
@@ -13,12 +13,12 @@ from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from test_plane_loop import LOOP, _templates, _write
|
||||
import yaml
|
||||
|
||||
from openultrasast import cli
|
||||
from openultrasast.plane import engine_alerts
|
||||
from openultrasast.plane.engine_alerts import EngineAlertsError, alerts_engine, engine_rows, parse_site, read_record
|
||||
from openultrasast.plane.generate import CaseInputs, render
|
||||
from openultrasast.plane.manifests import RUN_API_VERSION
|
||||
|
||||
FIXTURE = Path(__file__).parent / "fixtures" / "plane-engine"
|
||||
CASE = "php-probe"
|
||||
@@ -44,17 +44,29 @@ def world(tmp_path: Path) -> dict[str, Any]:
|
||||
(repo / "app" / "index.php").write_text(text)
|
||||
_git(repo, "-c", "user.name=t", "-c", "[email protected]", "commit", "-qam", "fixed")
|
||||
fixed = _git(repo, "rev-parse", "HEAD")
|
||||
record = {"id": CASE, "repo": "https://example.com/reserved/probe", "family": "injection", "vulnerable": vulnerable, "fixed": fixed}
|
||||
candidate = ("app/index.php", "<global>", 3)
|
||||
case = CaseInputs(
|
||||
record, (candidate,), (candidate,), {}, 0.0, ranges={"app/index.php": [(3, 3)]}, fixed_ranges={"app/index.php": [(3, 3)]}
|
||||
)
|
||||
files = render([case], _templates(), "validation-46", "t", population="p", split="s", loop=LOOP)
|
||||
plane = tmp_path / "plane"
|
||||
for name in ("tasks", "workspaces", "runs"):
|
||||
(plane / name).mkdir(parents=True)
|
||||
task = yaml.safe_load((Path(__file__).resolve().parents[1] / "plane/tasks/alerts.yaml").read_text())
|
||||
task["metadata"]["name"] = f"{CASE}-alerts"
|
||||
task["spec"]["env"] = [{"name": f"OUSAST_{role.upper()}_PIN", "value": pin} for role, pin in
|
||||
(("vulnerable", vulnerable), ("fixed", fixed))]
|
||||
task["spec"]["workspaces"] = [{"name": "case", "path": "/workspace/case"}]
|
||||
(plane / "tasks/alerts.yaml").write_text(yaml.safe_dump(task))
|
||||
case = {"family": "injection", "ranges": {"app/index.php": [[3, 3]]}, "fixed_ranges": {"app/index.php": [[3, 3]]}}
|
||||
workspace = {"apiVersion": task["apiVersion"], "kind": "Workspace", "metadata": {"name": "case"}, "spec": {
|
||||
"git": [{"name": "repo", "repo": "https://example.com/reserved/probe", "dir": "repo"}],
|
||||
"files": [{"path": "case.json", "content": json.dumps(case)}],
|
||||
}}
|
||||
(plane / "workspaces/case.yaml").write_text(yaml.safe_dump(workspace))
|
||||
run = plane / "runs/validation-46.yaml"
|
||||
run.write_text(yaml.safe_dump({"apiVersion": RUN_API_VERSION, "kind": "Run", "metadata": {"name": "validation-46"},
|
||||
"spec": {"tasks": [{"name": f"{CASE}-alerts", "task": f"{CASE}-alerts",
|
||||
"outputs": ["alerts.jsonl", "summary.json"], "budget": {"usd": 0, "calls": 0}}]}}))
|
||||
source = tmp_path / "frozen"
|
||||
(source / "src" / "openultrasast").mkdir(parents=True)
|
||||
(source / "benchmarks" / "push").mkdir(parents=True)
|
||||
(source / "benchmarks" / "push" / "finding_dump.py").write_text("")
|
||||
run = _write(tmp_path / "plane", files)
|
||||
return {"run": run, "source": source, "cache": tmp_path / "cache", "pins": (vulnerable, fixed), "results": tmp_path / "results"}
|
||||
|
||||
|
||||
|
||||
@@ -1,313 +0,0 @@
|
||||
"""The improvement loop as a plane Run (harnessx-removal Req 6.3, design section 6).
|
||||
|
||||
The whole graph runs through the reconciler and the fake ax of ``test_plane_reconciler``, whose ``exec`` mode runs the
|
||||
real task modules on the inputs the receiver serves: a recorded case's chain (validation-46's lmdeploy case from
|
||||
``tests/fixtures/plane-remember``, already done) -> ``alerts`` -> ``remember`` -> ``loop-measure`` -> ``loop-propose``
|
||||
-> ``loop-improve``, with ``memory-snapshot`` seeded on the host from a crafted store.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import shutil
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
import yaml
|
||||
from test_plane_reconciler import Fake, fake # noqa: F401 -- `fake` is the fixture
|
||||
|
||||
import openultrasast
|
||||
from openultrasast import cli
|
||||
from openultrasast.plane import reconciler
|
||||
from openultrasast.plane.generate import TEMPLATES, CaseInputs, Loop, fix_ranges, render
|
||||
from openultrasast.plane.manifests import load_manifests
|
||||
from openultrasast.plane.memory import SNAPSHOT_ANNOTATION, FileStore, row_id
|
||||
from openultrasast.plane.tasks import alerts
|
||||
|
||||
FIXTURE = Path(__file__).parent / "fixtures" / "plane-remember" / "validation-46"
|
||||
PLANE = Path(__file__).resolve().parents[1] / "plane"
|
||||
SRC = str(Path(openultrasast.__file__).resolve().parents[1])
|
||||
CASE = "lmdeploy-media-url-ssrf"
|
||||
VULNERABLE = "71648d239c978452720a6d6692fe914efc8a7b94"
|
||||
FIXED = "ab" * 20
|
||||
OTHER = "12" * 20
|
||||
EVAL = "python-unsafe-eval" # a bundled, enabled rule
|
||||
CHAIN = ("facts", "va", "vb", "agree", "vc", "final")
|
||||
LOOP_TASKS = ("memory-snapshot", "loop-measure", "loop-propose", "loop-improve")
|
||||
|
||||
|
||||
def _case(**ranges: Any) -> CaseInputs:
|
||||
record = {
|
||||
"id": CASE, "repo": "https://example.com/reserved/lmdeploy", "family": "untrusted_destination", "vulnerable": VULNERABLE,
|
||||
"fixed": FIXED, "sites": ["lmdeploy/vl/media/connection.py::_load_http_url"],
|
||||
} # fmt: skip
|
||||
candidate = ("lmdeploy/vl/media/connection.py", "_load_http_url", 40)
|
||||
return CaseInputs(record, (candidate,), (candidate,), {}, 0.02, **ranges)
|
||||
|
||||
|
||||
def _templates() -> dict:
|
||||
return load_manifests([PLANE / "tasks" / f"{name}.yaml" for name in TEMPLATES]).tasks
|
||||
|
||||
|
||||
LOOP = Loop(repo="https://example.invalid/openultrasast.git", commit="c" * 40, manifest="bench.toml", catalog="catalog.toml")
|
||||
|
||||
|
||||
def _write(plane: Path, files: dict[str, str]) -> Path:
|
||||
shutil.copytree(PLANE / "models", plane / "models")
|
||||
for relative, text in files.items():
|
||||
(plane / relative).parent.mkdir(parents=True, exist_ok=True)
|
||||
(plane / relative).write_text(text)
|
||||
return plane / "runs" / "validation-46.yaml"
|
||||
|
||||
|
||||
# --- alerts --------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_alerts_attribute_functions_and_mark_the_fix_on_both_pins(tmp_path: Path) -> None:
|
||||
vulnerable, fixed = tmp_path / "v", tmp_path / "f"
|
||||
(vulnerable / "pkg").mkdir(parents=True)
|
||||
(fixed / "pkg").mkdir(parents=True)
|
||||
(vulnerable / "pkg" / "run.py").write_text("import os\n\ndef handle(x):\n y = x\n return eval(y)\n")
|
||||
(fixed / "pkg" / "run.py").write_text("import os\n\ndef handle(x):\n return eval(x, {})\n\nprint(eval('1'))\n")
|
||||
case = tmp_path / "case.json"
|
||||
case.write_text(json.dumps({"ranges": {"pkg/run.py": [[4, 5]]}, "fixed_ranges": {"pkg/run.py": [[4, 4]]}}))
|
||||
out = tmp_path / "out"
|
||||
env = {
|
||||
"OUSAST_OUTPUT_DIR": str(out), "OUSAST_WORKSPACE_DIR": str(vulnerable), "OUSAST_FIXED_DIR": str(fixed),
|
||||
"OUSAST_INPUT_CASE": str(case), "OUSAST_VULNERABLE_PIN": VULNERABLE, "OUSAST_FIXED_PIN": FIXED,
|
||||
} # fmt: skip
|
||||
assert alerts.main(env) == 0
|
||||
rows = [json.loads(line) for line in (out / "alerts.jsonl").read_text().splitlines()]
|
||||
got = [(r["rule_id"], r["pin_role"], r["pin"], r["line"], r["function"], r["in_fix_range"]) for r in rows if r["rule_id"] == EVAL]
|
||||
assert got == [
|
||||
(EVAL, "vulnerable", VULNERABLE, 5, "handle", True),
|
||||
(EVAL, "fixed", FIXED, 4, "handle", True),
|
||||
(EVAL, "fixed", FIXED, 6, "<global>", False),
|
||||
]
|
||||
summary = json.loads((out / "summary.json").read_text())
|
||||
assert (summary["status"], summary["units_done"], summary["usd"], summary["calls"], summary["model"]) == ("done", 2, 0, 0, None)
|
||||
assert summary["read"]["vulnerable"]["files"] == 1 and summary["read"]["vulnerable"]["bytes"] > 0, "the scan read its tree"
|
||||
assert alerts.in_range(None, "pkg/run.py", 4) is None, "no ranges: unknown, not outside"
|
||||
|
||||
empty = tmp_path / "empty"
|
||||
empty.mkdir()
|
||||
assert alerts.main({**env, "OUSAST_FIXED_DIR": str(empty)}) == 2
|
||||
assert "an unread tree is not a clean one" in json.loads((out / "summary.json").read_text())["reason"]
|
||||
|
||||
|
||||
def test_quick_coverage_is_derived_from_the_ruleset_directory(tmp_path: Path) -> None:
|
||||
shipped = alerts.quick_languages()
|
||||
assert {"python", "javascript", "typescript", "java", "c", "groovy", "php"} <= shipped, "PHP has quick rules"
|
||||
assert "php" in alerts.engine_languages(), "the engine's semantic models cover PHP"
|
||||
rules = tmp_path / "ruleset" / "php"
|
||||
rules.mkdir(parents=True)
|
||||
rule = 'rule_id = "php-x"\ntitle = "x"\nlanguages = ["php"]\ncwe = "CWE-89"\ntags = []\npattern = "x"\n'
|
||||
(rules / "rules.toml").write_text(f"[[rule]]\n{rule}")
|
||||
assert alerts.quick_languages(tmp_path / "ruleset") == {"php"}, "a ruleset added for a language covers it"
|
||||
(rules / "rules.toml").write_text(f'[[rule]]\n{rule}status = "disabled"\n')
|
||||
assert alerts.quick_languages(tmp_path / "ruleset") == set(), "a disabled rule fires nowhere"
|
||||
|
||||
|
||||
def test_alerts_on_an_uncovered_language_record_coverage_none_not_a_clean_zero(tmp_path: Path) -> None:
|
||||
for side in ("v", "f"):
|
||||
(tmp_path / side).mkdir()
|
||||
(tmp_path / side / "app.rb").write_text('id = params["id"]\nDB.execute("SELECT " + id)\n')
|
||||
(tmp_path / side / "app.js").write_text("module.exports = 1;\n")
|
||||
out = tmp_path / "out"
|
||||
env = {
|
||||
"OUSAST_OUTPUT_DIR": str(out), "OUSAST_WORKSPACE_DIR": str(tmp_path / "v"), "OUSAST_FIXED_DIR": str(tmp_path / "f"),
|
||||
"OUSAST_VULNERABLE_PIN": VULNERABLE, "OUSAST_FIXED_PIN": FIXED,
|
||||
} # fmt: skip
|
||||
assert alerts.main(env) == 0
|
||||
summary = json.loads((out / "summary.json").read_text())
|
||||
assert (out / "alerts.jsonl").read_text() == "" and summary["alerts"] == {"vulnerable": 0, "fixed": 0}
|
||||
assert summary["coverage"] == {
|
||||
"javascript": {"coverage": "quick", "files": {"fixed": 1, "vulnerable": 1}},
|
||||
"ruby": {"coverage": "none", "files": {"fixed": 1, "vulnerable": 1}},
|
||||
}
|
||||
assert summary["uncovered"] == ["ruby"] and summary["pins"] == {"vulnerable": VULNERABLE, "fixed": FIXED}
|
||||
|
||||
|
||||
def test_php_is_quick_covered_and_alerts_on_a_php_sink(tmp_path: Path) -> None:
|
||||
for side in ("v", "f"):
|
||||
(tmp_path / side).mkdir()
|
||||
(tmp_path / side / "index.php").write_text('<?php\n$id = $_GET["id"];\nmysqli_query($c, "SELECT " . $id);\n')
|
||||
out = tmp_path / "out"
|
||||
env = {
|
||||
"OUSAST_OUTPUT_DIR": str(out), "OUSAST_WORKSPACE_DIR": str(tmp_path / "v"), "OUSAST_FIXED_DIR": str(tmp_path / "f"),
|
||||
"OUSAST_VULNERABLE_PIN": VULNERABLE, "OUSAST_FIXED_PIN": FIXED,
|
||||
} # fmt: skip
|
||||
assert alerts.main(env) == 0
|
||||
summary = json.loads((out / "summary.json").read_text())
|
||||
assert summary["coverage"] == {"php": {"coverage": "quick", "files": {"fixed": 1, "vulnerable": 1}}}
|
||||
assert summary["uncovered"] == [] and summary["alerts"] == {"vulnerable": 1, "fixed": 1}
|
||||
rows = [json.loads(line) for line in (out / "alerts.jsonl").read_text().splitlines()]
|
||||
assert {(r["rule_id"], r["line"]) for r in rows} == {("php-sql-call-composition", 3)}
|
||||
|
||||
|
||||
def test_fix_ranges_new_side_are_the_fixed_pins_lines(tmp_path: Path) -> None:
|
||||
repo = tmp_path / "r"
|
||||
repo.mkdir()
|
||||
git = ["git", "-C", str(repo), "-c", "user.email=t@t", "-c", "user.name=t"]
|
||||
subprocess.run([*git, "init", "-q"], check=True)
|
||||
(repo / "a.py").write_text("a\nb\nc\n")
|
||||
subprocess.run([*git, "add", "."], check=True)
|
||||
subprocess.run([*git, "commit", "-qm", "v"], check=True)
|
||||
old = subprocess.run([*git, "rev-parse", "HEAD"], capture_output=True, text=True, check=True).stdout.strip()
|
||||
(repo / "a.py").write_text("a\nB1\nB2\nB3\nc\n")
|
||||
subprocess.run([*git, "commit", "-qam", "f"], check=True)
|
||||
new = subprocess.run([*git, "rev-parse", "HEAD"], capture_output=True, text=True, check=True).stdout.strip()
|
||||
assert fix_ranges(repo, old, new) == {"a.py": [(2, 2)]}
|
||||
assert fix_ranges(repo, old, new, side="new") == {"a.py": [(2, 4)]}
|
||||
|
||||
|
||||
# --- the generator -------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_every_case_ends_in_remember_and_the_loop_is_model_free(tmp_path: Path) -> None:
|
||||
plain = render([_case()], _templates(), "validation-46", "t", population="population-v2", split="validation")
|
||||
run, _ = reconciler.load_run(_write(tmp_path / "plain", plain))
|
||||
assert [e.name.removeprefix(f"{CASE}-") for e in run.tasks] == [*CHAIN[:4], "vc", "final", "features", "remember"]
|
||||
remember = run.task(f"{CASE}-remember")
|
||||
assert remember.inputs["agreed"] == f"{CASE}-final/agreed.json" and "alerts" not in remember.inputs
|
||||
assert set(remember.producers) == {f"{CASE}-{step}" for step in ("facts", "va", "vb", "vc", "final", "features")}
|
||||
assert set(run.task(f"{CASE}-features").producers) == {f"{CASE}-{step}" for step in ("facts", "va", "vb", "vc", "final")}
|
||||
|
||||
files = render([_case(fixed_ranges={"x.py": [(1, 2)]})], _templates(), "validation-46", "t", population="p", split="s", loop=LOOP)
|
||||
run, manifests = reconciler.load_run(_write(tmp_path / "loop", files))
|
||||
names = [e.name for e in run.tasks]
|
||||
assert names[:6] == [f"{CASE}-{step}" for step in CHAIN]
|
||||
assert names[6:] == [f"{CASE}-alerts", f"{CASE}-features", f"{CASE}-remember", *LOOP_TASKS]
|
||||
assert run.task(f"{CASE}-features").inputs["alerts"] == f"{CASE}-alerts/alerts.jsonl", "features read the alerts"
|
||||
assert run.task(f"{CASE}-remember").inputs["alerts"] == f"{CASE}-alerts/alerts.jsonl"
|
||||
assert run.task(f"{CASE}-remember").inputs["alerts_summary"] == f"{CASE}-alerts/summary.json", "the coverage travels too"
|
||||
measure_inputs = {f"remember_{CASE}": f"{CASE}-remember/memory.jsonl", "snapshot": "memory-snapshot/rows.jsonl"}
|
||||
assert run.task("loop-measure").inputs == measure_inputs
|
||||
assert run.task("loop-improve").producers == ("loop-propose",)
|
||||
assert run.task("loop-propose").producers == ("loop-measure", "memory-snapshot")
|
||||
assert run.task(f"{CASE}-alerts").serialize == run.task("loop-improve").serialize == "engine"
|
||||
fixed = manifests.workspaces[f"{CASE}-fixed"]
|
||||
assert fixed.pins == {"repo": FIXED} and manifests.workspaces["openultrasast-cccccccccccc"].pins == {"repo": "c" * 40}
|
||||
for entry in run.tasks[6:]:
|
||||
task = manifests.tasks[entry.task]
|
||||
assert entry.budget is not None and (entry.budget.usd, entry.budget.calls) == (0, 0), entry.name
|
||||
assert reconciler.MODEL_ANNOTATION not in task.metadata.annotations, f"{entry.name} binds no Model"
|
||||
docs = reconciler.render_task(run, entry, manifests, "http://127.0.0.1:1/")
|
||||
env = {e["name"]: e["value"] for e in docs[-1]["spec"]["env"]}
|
||||
assert "OUSAST_MODEL" not in env and float(env["OUSAST_BUDGET_USD"]) == 0 and env["OUSAST_BUDGET_CALLS"] == "0"
|
||||
guard = json.loads(manifests.tasks["memory-snapshot"].metadata.annotations[SNAPSHOT_ANNOTATION])
|
||||
assert guard == {"catalog": "catalog.toml", "manifest": "bench.toml", "populations": []}
|
||||
inputs = yaml.safe_load((tmp_path / "loop" / "workspaces" / f"{CASE}-inputs.yaml").read_text())
|
||||
case_json = json.loads(next(f["content"] for f in inputs["spec"]["files"] if f["path"] == "case.json"))
|
||||
assert case_json["fixed_ranges"] == {"x.py": [[1, 2]]}
|
||||
|
||||
|
||||
# --- the whole loop through the fake ax ----------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _project(root: Path) -> Path:
|
||||
"""This repository at the commit under test, cut down: a benchmark whose one expected finding the eval rule
|
||||
makes, and an empty pair catalog."""
|
||||
(root / "target").mkdir(parents=True)
|
||||
(root / "target" / "app.py").write_text("def f(x):\n return eval(x)\n")
|
||||
(root / "bench.toml").write_text(
|
||||
'name = "t"\nlanguage = "python_web"\n\n[source]\npath = "target"\n\n[[expected]]\ncwe = "CWE-95"\n'
|
||||
f'class = "code injection"\npath = "app.py"\nline = 2\nrule_id = "{EVAL}"\nsink = "eval"\nevidence = "eval"\n'
|
||||
)
|
||||
(root / "catalog.toml").write_text("")
|
||||
(root / "benchmarks" / "pairs").mkdir(parents=True) # the catalog loader's mechanism vocabulary, by relative path
|
||||
shutil.copy(PLANE.parent / "benchmarks" / "pairs" / "mechanisms.toml", root / "benchmarks" / "pairs")
|
||||
return root
|
||||
|
||||
|
||||
def _older_run(store: FileStore) -> None:
|
||||
"""Two false alerts of the eval rule at rejected candidates of another repository, from an earlier run."""
|
||||
rows = []
|
||||
for fn in ("f1", "f2"):
|
||||
base = {"repo": "example.com/other/app", "pin": OTHER, "run": "older", "task": "app-remember", "population": "p", "split": "s"}
|
||||
base["image"] = "sha256:" + "0" * 64
|
||||
cand = f"app.py::{fn}"
|
||||
rows.append({**base, "id": row_id("verdict", "older", "app-remember", cand), "kind": "verdict", "candidate": cand,
|
||||
"family": "code_injection", "final": "rejected", "site_match": False}) # fmt: skip
|
||||
rows.append({**base, "id": row_id("alert", "older", "app-remember", fn), "kind": "alert", "rule_id": EVAL,
|
||||
"rule_status": "enabled", "path": "app.py", "line": 3, "function": fn, "pin_role": "vulnerable"}) # fmt: skip
|
||||
store.ingest_rows("older", "app-remember", rows)
|
||||
|
||||
|
||||
def test_the_loop_runs_as_one_plane_run_and_the_gate_rejects_a_demotion_that_costs_recall(
|
||||
fake: Fake, # noqa: F811
|
||||
tmp_path: Path,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
project = _project(tmp_path / "project")
|
||||
monkeypatch.chdir(project) # `ousast plane run` from the repository root: the snapshot's guard paths resolve here
|
||||
store = FileStore(tmp_path / "memory")
|
||||
_older_run(store)
|
||||
monkeypatch.setenv("OUSAST_MEMORY", f"file://{store.root}")
|
||||
vulnerable, fixed = tmp_path / "checkout-v", tmp_path / "checkout-f"
|
||||
for root, body in ((vulnerable, "def load(x):\n return x\n"), (fixed, "def load(x):\n return eval(x)\n")):
|
||||
(root / "lmdeploy" / "vl" / "media").mkdir(parents=True)
|
||||
(root / "lmdeploy" / "vl" / "media" / "other.py").write_text(body)
|
||||
case = _case(ranges={"lmdeploy/vl/media/other.py": [(1, 2)]}, fixed_ranges={"lmdeploy/vl/media/other.py": [(1, 2)]})
|
||||
files = render([case], _templates(), "validation-46", "t", population="population-v2", split="validation", loop=LOOP)
|
||||
manifest = _write(tmp_path / "plane", files)
|
||||
inputs = yaml.safe_load((tmp_path / "plane" / "workspaces" / f"{CASE}-inputs.yaml").read_text())
|
||||
(tmp_path / "case.json").write_text(next(f["content"] for f in inputs["spec"]["files"] if f["path"] == "case.json"))
|
||||
|
||||
base = reconciler.run_dir("validation-46") # the recorded chain, done: the loop reuses it and runs only itself
|
||||
shutil.copytree(FIXTURE, base)
|
||||
state = {"run": "validation-46", "tasks": {f"{CASE}-{s}": {"status": "done"} for s in CHAIN}}
|
||||
(base / "state.json").write_text(json.dumps(state))
|
||||
|
||||
project_env = {"PYTHONPATH": SRC, "OUSAST_PROJECT_DIR": str(project)}
|
||||
fake.script({
|
||||
f"{CASE}-alerts": {"exec": {"env": {"PYTHONPATH": SRC, "OUSAST_WORKSPACE_DIR": str(vulnerable), "OUSAST_FIXED_DIR": str(fixed),
|
||||
"OUSAST_INPUT_CASE": str(tmp_path / "case.json")}}},
|
||||
f"{CASE}-features": {"exec": {"env": {"PYTHONPATH": SRC, "OUSAST_WORKSPACE_DIR": str(vulnerable)}}},
|
||||
f"{CASE}-remember": {"exec": {"env": {"PYTHONPATH": SRC}}},
|
||||
"loop-measure": {"exec": {"env": {"PYTHONPATH": SRC}}},
|
||||
"loop-propose": {"exec": {"env": project_env}},
|
||||
"loop-improve": {"exec": {"env": project_env}},
|
||||
}) # fmt: skip
|
||||
|
||||
assert cli.main(["plane", "run", str(manifest), "--ax", str(fake.ax)]) == 0, (fake.root / "ax.log").read_text()[-3000:]
|
||||
|
||||
statuses = {name: info["status"] for name, info in json.loads((base / "state.json").read_text())["tasks"].items()}
|
||||
assert set(statuses.values()) == {"done"} and len(statuses) == 13
|
||||
assert json.loads((base / "memory-snapshot" / "index.json").read_text())["kept"] == 4, "seeded on the host, not run"
|
||||
delivered = [name for _, name in fake.events("delivered")]
|
||||
assert delivered == [f"{CASE}-alerts", f"{CASE}-features", f"{CASE}-remember", "loop-measure", "loop-propose", "loop-improve"]
|
||||
|
||||
measure = json.loads((base / "loop-measure" / "measure.json").read_text())
|
||||
assert measure["run_metrics"]["declared_sites_agreed"] == 1 and measure["run_metrics"]["candidates"] == 1
|
||||
assert measure["run_metrics"]["alerts"] == {EVAL: {"vulnerable": 0, "fixed": 1, "fixed_in_fix_range": 1}}
|
||||
assert measure["union_metrics"]["rejected"] == 2 and measure["rows"]["snapshot"] == 4
|
||||
|
||||
[proposal] = [json.loads(line) for line in (base / "loop-propose" / "proposals.jsonl").read_text().splitlines()]
|
||||
assert proposal["edit"]["rule_id"] == EVAL and (proposal["edit"]["from_status"], proposal["edit"]["to_status"]) == ("enabled", "shadow")
|
||||
assert proposal["provenance"]["counts"] == {"false_alerts": 3, "repos": 2, "agreed_hits": 0}
|
||||
cited = {(e["kind"], e["repo"], e["pin"]) for e in proposal["provenance"]["evidence"]}
|
||||
assert ("alert", "example.com/reserved/lmdeploy", FIXED) in cited, "the fixed-pin alert inside the fix is evidence"
|
||||
assert (base / "loop-propose" / "memory_proposals.jsonl").read_text().count("\n") == 1
|
||||
|
||||
gate = json.loads((base / "loop-improve" / "gate.json").read_text())
|
||||
assert gate["memory_edits"] == [f"rule:{EVAL}:enabled->shadow"] and gate["accepted"] is False and gate["reason"] == "reverted"
|
||||
assert gate["outcome"]["recall_after"] < gate["outcome"]["recall_before"], "demoting the eval rule loses the benchmark's finding"
|
||||
assert not (base / "loop-improve" / "rule_policy.json").exists(), "a ledger only when the gate accepts"
|
||||
assert json.loads((base / "loop-improve" / "journal.json").read_text())[0]["outcome"] == "reverted"
|
||||
assert not (project / "target" / ".openultrasast").exists(), "the repository's own ledger is never touched"
|
||||
|
||||
report = reconciler.attribution("validation-46")
|
||||
for row in report["tasks"]:
|
||||
if row["task"] in (f"{CASE}-alerts", f"{CASE}-features", f"{CASE}-remember", *LOOP_TASKS):
|
||||
assert row["model"] is None and row["usd"] in (0, None) and row["calls"] in (0, None), row
|
||||
ingested = FileStore(store.root)
|
||||
outcomes = ingested.rows(kind="proposal_outcome")
|
||||
assert {r.row["outcome"] for r in outcomes} == {"reverted"} and len(outcomes) == 2, "one per (repo, pin) of the evidence"
|
||||
[fixed_alert] = ingested.rows(repo="example.com/reserved/lmdeploy", pin=FIXED, kind="alert")
|
||||
features = {r.row["candidate"]: r.row for r in ingested.rows(kind="features")}
|
||||
assert set(features) == {"lmdeploy/vl/media/connection.py::_load_http_url"}, "no fixed-pin alert became a candidate"
|
||||
assert all(r["pin"] == VULNERABLE and "in_fix_range" not in json.dumps(r["x"]) for r in features.values())
|
||||
assert fixed_alert.row["in_fix_range"] is True and fixed_alert.row["function"] == "load"
|
||||
@@ -64,32 +64,6 @@ def test_the_removed_modules_are_gone_from_the_tree() -> None:
|
||||
assert not (SRC / relative).exists(), f"{relative} still exists"
|
||||
|
||||
|
||||
def test_the_mechanism_lever_is_gone_from_the_retained_improve_loop() -> None:
|
||||
"""`improve/` survives with its lever removed, not deleted.
|
||||
|
||||
The design's Removed list named `improve/evolve.py` whole, but the file is a mix: the mechanism lever
|
||||
(deleted here) and the deterministic rule-status loop that `ousast improve` still runs from benchmark
|
||||
evidence, which has no LLM in it and is not what this feature supersedes. Deleting the file would have
|
||||
taken a retained capability with it, so the lever is stripped and the loop stays — asserted by symbol so
|
||||
the lever cannot creep back.
|
||||
"""
|
||||
evolve = (SRC / "improve" / "evolve.py").read_text()
|
||||
validator = (SRC / "improve" / "validator.py").read_text()
|
||||
for symbol in (
|
||||
"MechanismEdit",
|
||||
"MechanismStore",
|
||||
"apply_mechanism_edits",
|
||||
"evaluate_mechanism_profiles",
|
||||
"propose_mechanism_edits",
|
||||
"validate_mechanism",
|
||||
"mechanism_store",
|
||||
"mechanism_candidates",
|
||||
):
|
||||
assert symbol not in evolve, f"{symbol} survives in improve/evolve.py"
|
||||
assert symbol not in validator, f"{symbol} survives in improve/validator.py"
|
||||
assert "def run_improvement(" in evolve and "def propose_status_edits(" in evolve, "the rule-status loop must survive"
|
||||
|
||||
|
||||
def test_the_re_homed_modules_are_where_the_model_layer_expects_them() -> None:
|
||||
for relative in ("model/taxonomy.py", "model/candidates.py", "model/endpoint.py", "model/specs.py"):
|
||||
assert (SRC / relative).is_file(), f"{relative} is missing"
|
||||
|
||||
Reference in New Issue
Block a user