diff --git a/.kiro/specs/harnessx-self-improving-rulesets/spec.json b/.kiro/specs/harnessx-self-improving-rulesets/spec.json index 4821e9e..19924bb 100644 --- a/.kiro/specs/harnessx-self-improving-rulesets/spec.json +++ b/.kiro/specs/harnessx-self-improving-rulesets/spec.json @@ -1,7 +1,7 @@ { "feature_name": "harnessx-self-improving-rulesets", "created_at": "2026-06-22T10:07:18Z", - "updated_at": "2026-06-22T10:45:00Z", + "updated_at": "2026-06-22T13:45:00Z", "language": "en", "phase": "implementation", "approvals": { diff --git a/.kiro/specs/harnessx-self-improving-rulesets/tasks.md b/.kiro/specs/harnessx-self-improving-rulesets/tasks.md index 19d23e9..cbc2a62 100644 --- a/.kiro/specs/harnessx-self-improving-rulesets/tasks.md +++ b/.kiro/specs/harnessx-self-improving-rulesets/tasks.md @@ -90,22 +90,22 @@ This plan splits the fused regex layer into three governed owners (rules detect, - _Boundary: PolicyScoringProcessor, RulesetStore, PolicyStore_ - _Depends: 3.1, 3.2, 1.2_ -- [ ] 4. Make benchmark output an actionable per-rule signal -- [ ] 4.1 Attribute misses and false positives to rules +- [x] 4. Make benchmark output an actionable per-rule signal +- [x] 4.1 Attribute misses and false positives to rules - Attribute every recorded miss and every recorded false positive to a rule identifier and aggregate per-rule precision and recall in the benchmark metrics. - Observable completion: a benchmark run reports per-rule precision/recall and every miss and false positive carries a rule identifier, verified by a metrics test. - _Requirements: 4.1_ - _Boundary: Benchmark Harness_ - _Depends: 3.1_ -- [ ] 4.2 Emit a per-rule recommendation delta artifact +- [x] 4.2 Emit a per-rule recommendation delta artifact - Replace the inert improvement-candidate stub with a per-rule recommendation artifact recommending enable, disable, shadow, threshold, or evidence-floor changes per rule. - Observable completion: a benchmark run writes a recommendation artifact with per-rule suggested actions alongside its records, verified against the documented delta schema. - _Requirements: 4.2_ - _Boundary: Benchmark Harness_ - _Depends: 4.1_ -- [ ] 4.3 Introduce the loop-owned ledger and re-key calibration to rules +- [x] 4.3 Introduce the loop-owned ledger and re-key calibration to rules - Add the loop-owned overlay ledger that records per-rule status, evidence floor, threshold, and precision, applied over the ruleset at load time. - Re-key calibration from the first-tag key to the rule identifier and write status flips into the ledger rather than only demoting ranking. - Observable completion: an empty ledger leaves behavior byte-identical to the prior phase while a ledger entry overlays a rule's status at load, proven by an integration test. @@ -113,7 +113,7 @@ This plan splits the fused regex layer into three governed owners (rules detect, - _Boundary: RulesetStore, Calibration subsystem_ - _Depends: 3.1, 3.2_ -- [ ] 4.4 Convert false-positive confirmation into a reachability knob +- [x] 4.4 Convert false-positive confirmation into a reachability knob - When confirming a false positive, lower the finding's effective reachability multiplier rather than delete the rule, so recall stays high while the score stops over-penalizing. - Observable completion: confirming a false positive lowers the finding's effective reachability multiplier and the rule still fires, proven by a calibration test. - _Requirements: 5.6_ diff --git a/src/openultrasast/benchmark.py b/src/openultrasast/benchmark.py index 96cbcb0..bef8418 100644 --- a/src/openultrasast/benchmark.py +++ b/src/openultrasast/benchmark.py @@ -60,6 +60,7 @@ class BenchmarkMiss: path: str evidence: str reason: str + rule_id: str | None = None @dataclass(frozen=True) @@ -79,6 +80,7 @@ class BenchmarkCalibrationRecord: failed_stage: str next_improvement_candidate: str reason: str + rule_id: str | None = None @dataclass(frozen=True) @@ -87,6 +89,17 @@ class BenchmarkFalsePositive: path: str line: int | None reason: str + rule_id: str = "" + + +@dataclass(frozen=True) +class RuleRecommendation: + rule_id: str + action: str # keep | shadow | loosen + matched: int + missed: int + false_positives: int + rationale: str @dataclass(frozen=True) @@ -101,6 +114,7 @@ class BenchmarkMetrics: runtime_seconds: float | None model_usage: dict[str, object] artifact_links: dict[str, str] + per_rule: dict[str, dict[str, int]] @dataclass(frozen=True) @@ -123,6 +137,7 @@ class BenchmarkResult: false_positives: list[BenchmarkFalsePositive] baseline_deltas: list[BenchmarkBaselineDelta] calibration_records: list[BenchmarkCalibrationRecord] + rule_recommendations: list[RuleRecommendation] def load_benchmark_manifest(path: Path) -> BenchmarkManifest: @@ -224,6 +239,8 @@ def evaluate_benchmark( false_positives = [_false_positive(finding) for index, finding in enumerate(findings) if index not in matched_finding_indexes] baseline_deltas = _baseline_deltas(run.manifest.expected, baseline_findings or []) calibration_records = [_calibration_record(miss) for miss in misses] + per_rule = _per_rule_metrics(findings, matched_finding_indexes, misses) + rule_recommendations = _rule_recommendations(per_rule) expected_total = len(run.manifest.expected) actual_total = len(findings) missed_total = len(misses) @@ -249,11 +266,13 @@ def evaluate_benchmark( runtime_seconds=runtime_seconds, model_usage={}, artifact_links=_artifact_links(scan_run_dir), + per_rule=per_rule, ), misses=misses, false_positives=false_positives, baseline_deltas=baseline_deltas, calibration_records=calibration_records, + rule_recommendations=rule_recommendations, ) @@ -266,6 +285,9 @@ def write_benchmark_artifacts(run: BenchmarkRun, manifest_path: Path, result: Be (run.root / "external_baseline_deltas.json").write_text( json.dumps(_jsonable([asdict(delta) for delta in result.baseline_deltas]), indent=2, sort_keys=True) + "\n" ) + (run.root / "rule_policy_delta.json").write_text( + json.dumps(_jsonable([asdict(rec) for rec in result.rule_recommendations]), indent=2, sort_keys=True) + "\n" + ) def _matching_finding_index(expected: ExpectedFinding, findings: list[StaticFinding], used_indexes: set[int]) -> int | None: @@ -301,6 +323,7 @@ def _miss(expected: ExpectedFinding) -> BenchmarkMiss: path=expected.path, evidence=expected.evidence, reason="no OpenUltraSAST finding matched the expected benchmark vulnerability", + rule_id=expected.rule_id, ) @@ -310,9 +333,57 @@ def _false_positive(finding: StaticFinding) -> BenchmarkFalsePositive: path=finding.path, line=finding.line, reason="finding did not match any expected benchmark vulnerability", + rule_id=_rule_id_of(finding.finding_id), ) +def _rule_id_of(finding_id: str) -> str: + return finding_id.split(":", 1)[0] + + +def _per_rule_metrics( + findings: list[StaticFinding], + matched_finding_indexes: set[int], + misses: list[BenchmarkMiss], +) -> dict[str, dict[str, int]]: + per_rule: dict[str, dict[str, int]] = {} + + def bucket(rule_id: str) -> dict[str, int]: + return per_rule.setdefault(rule_id, {"matched": 0, "missed": 0, "false_positives": 0}) + + for index, finding in enumerate(findings): + rule_id = _rule_id_of(finding.finding_id) + key = "matched" if index in matched_finding_indexes else "false_positives" + bucket(rule_id)[key] += 1 + for miss in misses: + if miss.rule_id: + bucket(miss.rule_id)["missed"] += 1 + return dict(sorted(per_rule.items())) + + +def _rule_recommendations(per_rule: dict[str, dict[str, int]]) -> list[RuleRecommendation]: + recommendations: list[RuleRecommendation] = [] + for rule_id, counts in per_rule.items(): + matched, missed, fps = counts["matched"], counts["missed"], counts["false_positives"] + if fps > 0: + action, rationale = "shadow", "rule produced false positives; stage it before it drags precision" + elif missed > 0: + action, rationale = "loosen", "rule missed expected findings; widen or lower its evidence floor" + else: + action, rationale = "keep", "rule matched expected findings with no false positives" + recommendations.append( + RuleRecommendation( + rule_id=rule_id, + action=action, + matched=matched, + missed=missed, + false_positives=fps, + rationale=rationale, + ) + ) + return recommendations + + def _baseline_deltas(expected_findings: list[ExpectedFinding], baseline_findings: list[BaselineFinding]) -> list[BenchmarkBaselineDelta]: tools = sorted({finding.tool for finding in baseline_findings}) deltas: list[BenchmarkBaselineDelta] = [] @@ -359,14 +430,18 @@ def _baseline_matches_expected(expected: ExpectedFinding, finding: BaselineFindi def _calibration_record(miss: BenchmarkMiss) -> BenchmarkCalibrationRecord: + # A miss tied to a known rule points the loop at that rule; an unattributed + # miss (no rule fired for the class) is a coverage gap for a language hunter. + candidate = f"rule:{miss.rule_id}" if miss.rule_id else "language_hunter" return BenchmarkCalibrationRecord( cwe=miss.cwe, vulnerability_class=miss.vulnerability_class, path=miss.path, evidence=miss.evidence, failed_stage="benchmark_ground_truth_matching", - next_improvement_candidate="rules_or_language_hunter", + next_improvement_candidate=candidate, reason=miss.reason, + rule_id=miss.rule_id, ) diff --git a/src/openultrasast/calibration.py b/src/openultrasast/calibration.py index b304794..1735db7 100644 --- a/src/openultrasast/calibration.py +++ b/src/openultrasast/calibration.py @@ -249,6 +249,30 @@ def learnings_from_verifications( return learnings +# A confirmed false positive lowers a finding's effective reachability multiplier +# by this factor in the project score (it does not delete the rule). +FP_REACHABILITY_FACTOR = 0.4 +_BASE_REACH_MULT = {"reachable": 1.0, "inferred-file-surface": 0.6, "unknown": 0.4} + + +def fp_reachability_overrides( + findings: list[StaticFinding], + learnings: list[FalsePositiveLearning], +) -> dict[str, float]: + """Lowered score multipliers for findings whose scope has a confirmed FP. + + Keyed by ``finding_id``; consumed by the scoring model so a confirmed false + positive reduces the score penalty instead of the rule being deleted. + """ + overrides: dict[str, float] = {} + for finding in findings: + vulnerability_class = _vulnerability_class(finding) + if any(_learning_applies(finding.path, learning) and learning.vulnerability_class == vulnerability_class for learning in learnings): + base = _BASE_REACH_MULT.get(finding.reachability_status, _BASE_REACH_MULT["unknown"]) + overrides[finding.finding_id] = round(base * FP_REACHABILITY_FACTOR, 3) + return overrides + + def _reason_from_verification(finding: StaticFinding, status: VerificationStatus) -> FalsePositiveReason: if finding.reachability_status == "unknown": return FalsePositiveReason.UNREACHABLE_PATH diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py index a6c845b..61e0e61 100644 --- a/src/openultrasast/cli.py +++ b/src/openultrasast/cli.py @@ -18,6 +18,7 @@ from .benchmark import ( from .calibration import ( FalsePositiveLearning, calibrate_rankings, + fp_reachability_overrides, learnings_from_verifications, load_false_positive_learnings, merge_false_positive_learnings, @@ -105,7 +106,9 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut write_harness_config(config=config, processors=[], contract_mode="strict", path=run.root / "harness.json") runtime.start(mode=mode, target=run.target) policy = runtime.run_stage("policy_load", load_policy) - ruleset = runtime.run_stage("ruleset_load", lambda: load_ruleset(DEFAULT_RULESET_DIR)) + ledger_path = run.target / CALIBRATION_DIR / "false_positive_learnings.json" + rule_ledger_path = run.target / CALIBRATION_DIR / "rule_policy.json" + ruleset = runtime.run_stage("ruleset_load", lambda: load_ruleset(DEFAULT_RULESET_DIR, rule_ledger_path)) runtime.run_stage("policy_check", lambda: assert_rules_resolve(ruleset, policy)) static_hints = runtime.run_stage("static_mapping", lambda: _load_static_hints(config.static_analysis.sarif_paths)) write_static_hints(static_hints, run.root / "mapping" / "static_hints.json") @@ -118,7 +121,6 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut targets = attach_reachability_hints(targets, entry_points) write_preprocess_artifact(snapshot, targets, run.root / "preprocess" / "file_targets.json") rankings = runtime.run_stage("rank", lambda: rank_targets(targets)) - ledger_path = run.target / CALIBRATION_DIR / "false_positive_learnings.json" prior_learnings = load_false_positive_learnings(ledger_path) rankings, calibrations = runtime.run_stage("calibrate", lambda: calibrate_rankings(rankings, prior_learnings)) applied_calibrations = [calibration for calibration in calibrations if calibration.applied_learning_ids] @@ -168,6 +170,9 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut ) cwe_by_rule = {rule.rule_id: rule.cwe for rule in ruleset} rule_cwe_by_id = {finding.finding_id: cwe_by_rule.get(finding.finding_id.split(":", 1)[0], "") for finding in findings} + # A confirmed false positive (prior-scan learning) lowers a finding's effective + # reachability multiplier in the score instead of deleting the rule. + reachability_override = fp_reachability_overrides(findings, prior_learnings) score_artifact = runtime.run_stage( "score", lambda: build_score_artifact( @@ -178,6 +183,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut min_score=config.score.min_score, block_severity_reachable=config.score.block_severity_reachable, blocking=config.score.blocking, + reachability_override=reachability_override, ), ) score_path = run.root / "score.json" diff --git a/src/openultrasast/ruleset/__init__.py b/src/openultrasast/ruleset/__init__.py index 79709fc..840605d 100644 --- a/src/openultrasast/ruleset/__init__.py +++ b/src/openultrasast/ruleset/__init__.py @@ -5,6 +5,8 @@ from .store import ( PatternRule, RulesetError, load_ruleset, + read_rule_ledger, + write_rule_ledger, write_ruleset, ) @@ -13,5 +15,7 @@ __all__ = [ "PatternRule", "RulesetError", "load_ruleset", + "read_rule_ledger", + "write_rule_ledger", "write_ruleset", ] diff --git a/src/openultrasast/ruleset/store.py b/src/openultrasast/ruleset/store.py index c0fd1a3..33ccf9d 100644 --- a/src/openultrasast/ruleset/store.py +++ b/src/openultrasast/ruleset/store.py @@ -81,6 +81,17 @@ def write_ruleset(path: Path, rules: Iterable[PatternRule]) -> None: path.write_text("\n\n".join(blocks) + "\n") +def read_rule_ledger(path: Path) -> dict[str, dict[str, object]]: + """Read the loop-owned ``rule_policy.json`` overlay (status/floors per rule_id).""" + return _load_ledger(path) + + +def write_rule_ledger(path: Path, entries: dict[str, dict[str, object]]) -> None: + """Write the loop-owned ``rule_policy.json`` overlay (used by the evolve loop).""" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(entries, indent=2, sort_keys=True) + "\n") + + def _rule_from_dict(item: dict[str, object], path: Path) -> PatternRule: try: rule_id = str(item["rule_id"]) diff --git a/src/openultrasast/scoring/project_score.py b/src/openultrasast/scoring/project_score.py index e07d13c..44a7394 100644 --- a/src/openultrasast/scoring/project_score.py +++ b/src/openultrasast/scoring/project_score.py @@ -29,15 +29,26 @@ class ScoreArtifact: return asdict(self) -def finding_penalty(finding: StaticFinding, rule_cwe: str | None, policy: Mapping[str, CwePolicy]) -> float: +def finding_penalty( + finding: StaticFinding, + rule_cwe: str | None, + policy: Mapping[str, CwePolicy], + reachability_override: Mapping[str, float] | None = None, +) -> float: """Penalty contributed by one finding: severity weight × reachability multiplier. - Dynamic-only or unmapped CWEs contribute zero (report-only, never scored). + Dynamic-only or unmapped CWEs contribute zero (report-only, never scored). A + confirmed false positive can supply a lowered effective reachability multiplier + via ``reachability_override`` (keyed by ``finding_id``) instead of deleting the rule. """ pol = policy.get(rule_cwe) if rule_cwe else None if pol is None or not pol.static: return 0.0 - return SEV_WEIGHT[pol.severity] * REACH_MULT.get(finding.reachability_status, REACH_MULT["unknown"]) + if reachability_override and finding.finding_id in reachability_override: + multiplier = reachability_override[finding.finding_id] + else: + multiplier = REACH_MULT.get(finding.reachability_status, REACH_MULT["unknown"]) + return SEV_WEIGHT[pol.severity] * multiplier def project_score( @@ -45,13 +56,14 @@ def project_score( rule_cwe_by_id: Mapping[str, str], policy: Mapping[str, CwePolicy], k: float = 60.0, + reachability_override: Mapping[str, float] | None = None, ) -> int: """Compute the 0-100 project score (shadow findings excluded by the caller). Exponential decay of total penalty: no penalty → 100; one severity-5 reachable finding → ~43. """ - total = sum(finding_penalty(finding, rule_cwe_by_id.get(finding.finding_id), policy) for finding in findings) + total = sum(finding_penalty(finding, rule_cwe_by_id.get(finding.finding_id), policy, reachability_override) for finding in findings) return round(100 * math.exp(-total / k)) @@ -92,8 +104,9 @@ def build_score_artifact( min_score: int = 80, block_severity_reachable: int = 5, blocking: bool = False, + reachability_override: Mapping[str, float] | None = None, ) -> ScoreArtifact: - score = project_score(findings, rule_cwe_by_id, policy, k=k) + score = project_score(findings, rule_cwe_by_id, policy, k=k, reachability_override=reachability_override) penalty_total = 0.0 max_severity = 0 by_category: dict[str, int] = {} @@ -109,7 +122,7 @@ def build_score_artifact( if pol.dynamic: out_of_scope_dynamic_only += 1 continue - penalty_total += finding_penalty(finding, cwe, policy) + penalty_total += finding_penalty(finding, cwe, policy, reachability_override) max_severity = max(max_severity, pol.severity) by_category[pol.flaw_category] = by_category.get(pol.flaw_category, 0) + 1 return ScoreArtifact( diff --git a/tests/test_benchmark.py b/tests/test_benchmark.py index a74cef5..bc4cd99 100644 --- a/tests/test_benchmark.py +++ b/tests/test_benchmark.py @@ -322,15 +322,82 @@ evidence = "memcpy copies attacker-controlled input" "cwe": "CWE-120", "evidence": "memcpy copies attacker-controlled input", "failed_stage": "benchmark_ground_truth_matching", - "next_improvement_candidate": "rules_or_language_hunter", + "next_improvement_candidate": "rule:c-unsafe-memory-copy", "path": "src/vuln.cpp", "reason": "no OpenUltraSAST finding matched the expected benchmark vulnerability", + "rule_id": "c-unsafe-memory-copy", "vulnerability_class": "unsafe memory copy", } ] assert deltas == [{"extra_findings_total": 1, "matched_expected_total": 1, "missed_expected_total": 0, "tool": "clearwing"}] +def test_benchmark_attributes_per_rule_signals_and_recommendations(tmp_path: Path) -> None: + manifest_path = tmp_path / "benchmark.toml" + repo = tmp_path / "repo" + repo.mkdir() + manifest_path.write_text( + """ +name = "cpp-smoke" +language = "c_cpp" + +[source] +path = "repo" + +[[expected]] +cwe = "CWE-120" +class = "buffer overflow" +path = "src/vuln.cpp" +line = 9 +rule_id = "c-unsafe-memory-copy" +sink = "memcpy" +evidence = "memcpy copies attacker-controlled input" + +[[expected]] +cwe = "CWE-78" +class = "command injection" +path = "src/vuln.cpp" +line = 20 +rule_id = "c-shell-exec" +sink = "system" +evidence = "system runs attacker input" +""".strip() + + "\n" + ) + manifest = load_benchmark_manifest(manifest_path) + run = create_benchmark_run(repo, manifest) + findings = [ + _finding( + finding_id="c-unsafe-memory-copy:src/vuln.cpp:9", + path="src/vuln.cpp", + line=9, + rationale="Static pattern c-unsafe-memory-copy matched memcpy with attacker-controlled input.", + ), + _finding( + finding_id="js-eval:src/app.js:3", + path="src/app.js", + line=3, + rationale="Static pattern js-eval matched eval.", + ), + ] + + result = evaluate_benchmark(run=run, mode="quick", findings=findings, scan_id="scan-1", scan_run_dir=repo / ".runs" / "scan-1") + write_benchmark_artifacts(run, manifest_path, result) + + per_rule = result.metrics.per_rule + assert per_rule["c-unsafe-memory-copy"] == {"matched": 1, "missed": 0, "false_positives": 0} + assert per_rule["c-shell-exec"]["missed"] == 1 + assert per_rule["js-eval"]["false_positives"] == 1 + + actions = {rec.rule_id: rec.action for rec in result.rule_recommendations} + assert actions["c-unsafe-memory-copy"] == "keep" + assert actions["c-shell-exec"] == "loosen" + assert actions["js-eval"] == "shadow" + + delta = json.loads((run.root / "rule_policy_delta.json").read_text()) + assert {item["rule_id"] for item in delta} == {"c-unsafe-memory-copy", "c-shell-exec", "js-eval"} + + def _finding( *, finding_id: str, diff --git a/tests/test_ruleset.py b/tests/test_ruleset.py new file mode 100644 index 0000000..49c6b47 --- /dev/null +++ b/tests/test_ruleset.py @@ -0,0 +1,67 @@ +"""Rules-as-data store and the loop-owned ledger overlay (Phase 2 tasks 3.1, 4.3).""" + +import json +from pathlib import Path + +import pytest + +from openultrasast.cli import main +from openultrasast.ruleset import ( + DEFAULT_RULESET_DIR, + PatternRule, + load_ruleset, + read_rule_ledger, + write_rule_ledger, + write_ruleset, +) + + +def test_default_ruleset_loads_as_data() -> None: + rules = load_ruleset(DEFAULT_RULESET_DIR) + assert len(rules) == 32 + assert all(rule.status == "enabled" for rule in rules) + assert all(not hasattr(rule, "severity") for rule in rules) # severity is policy-governed, not rule-local + + +def test_empty_ledger_is_byte_identical() -> None: + assert load_ruleset(DEFAULT_RULESET_DIR) == load_ruleset(DEFAULT_RULESET_DIR, None) + + +def test_ledger_overlay_changes_only_targeted_rule(tmp_path: Path) -> None: + ledger = tmp_path / "rule_policy.json" + ledger.write_text(json.dumps({"python-unsafe-eval": {"status": "disabled", "precision_estimate": 0.3}})) + + rules = {rule.rule_id: rule for rule in load_ruleset(DEFAULT_RULESET_DIR, ledger)} + + assert rules["python-unsafe-eval"].status == "disabled" + assert rules["python-unsafe-eval"].precision_estimate == 0.3 + assert rules["c-shell-exec"].status == "enabled" # untargeted rule untouched + + +def test_rule_ledger_round_trips(tmp_path: Path) -> None: + ledger = tmp_path / "rule_policy.json" + entries = {"python-unsafe-eval": {"status": "shadow", "precision_estimate": 0.41}} + write_rule_ledger(ledger, entries) + assert read_rule_ledger(ledger) == entries + + +def test_write_ruleset_round_trips(tmp_path: Path) -> None: + rule = PatternRule(rule_id="x-demo", title="Demo", languages=("python",), cwe="CWE-89", tags=("injection",), pattern=r"\bfoo\(") + write_ruleset(tmp_path / "r" / "rules.toml", [rule]) + assert load_ruleset(tmp_path / "r")[0] == rule + + +def test_disabled_rule_via_ledger_does_not_fire_in_scan(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + repo = tmp_path / "repo" + repo.mkdir() + (repo / "app.py").write_text("def f(x):\n return eval(x)\n") + ledger_dir = repo / ".openultrasast" / "calibration" + ledger_dir.mkdir(parents=True) + (ledger_dir / "rule_policy.json").write_text(json.dumps({"python-unsafe-eval": {"status": "disabled"}})) + monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs") + + assert main(["scan", str(repo), "--mode", "quick"]) == 0 + + run_dir = sorted((repo / ".runs").iterdir())[-1] + findings = json.loads((run_dir / "findings.json").read_text())["findings"] + assert all(not finding["finding_id"].startswith("python-unsafe-eval:") for finding in findings) diff --git a/tests/test_scoring.py b/tests/test_scoring.py index 5aa4c3c..833164e 100644 --- a/tests/test_scoring.py +++ b/tests/test_scoring.py @@ -1,3 +1,4 @@ +from openultrasast.calibration import FalsePositiveReason, fp_reachability_overrides, record_false_positive_learning from openultrasast.findings import StaticFinding from openultrasast.policy import CwePolicy from openultrasast.scoring import build_score_artifact, gate, project_score @@ -62,6 +63,38 @@ def test_hard_gate_fails_on_severity5_reachable_only() -> None: assert ok["passed"] is True # inferred reachability does not trip the hard gate +def test_confirmed_false_positive_lowers_penalty_without_deleting_rule() -> None: + finding = StaticFinding( + finding_id="python-os-command:app.py:1", + path="app.py", + title="t", + severity="critical", + confidence="medium", + evidence_level="static_corroboration", + rationale="r", + line=1, + function_name=None, + reachability_status="reachable", + reachability_evidence=[], + reachability_conditions=[], + tags=["syscall_entry"], + ranking_priority=3.0, + ) + learning = record_false_positive_learning( + finding, + reason=FalsePositiveReason.MISSING_ATTACKER_CONTROL, + evidence="input is a constant, not attacker-controlled", + ) + overrides = fp_reachability_overrides([finding], [learning]) + rule_cwe = {finding.finding_id: "CWE-78"} + + assert overrides[finding.finding_id] < 1.0 # reachable multiplier lowered from 1.0 + base = project_score([finding], rule_cwe, POLICY) + with_fp = project_score([finding], rule_cwe, POLICY, reachability_override=overrides) + assert with_fp > base # confirmed FP reduces the penalty, raising the score + assert finding.finding_id in rule_cwe # the rule still fires; it was not deleted + + def test_score_gate_blocks_only_when_blocking_enabled() -> None: finding = _finding("sql:a.py:1", "inferred-file-surface") rule_cwe = {"sql:a.py:1": "CWE-89"}