From 9c4af4f54660fcba4327747e04cc9e1e1a169c5d Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Sun, 21 Jun 2026 20:44:08 +0200 Subject: [PATCH] feat(calibration): automatic in-pipeline false-positive feedback Close the HarnessX self-improving loop without an agent in the loop. Every scan now feeds its verifier outcomes back into the next scan's ranking: - New `calibrate` stage (after rank, before findings) loads a persistent ledger of scoped false-positive learnings and demotes every previously-rejected scope (`calibrate_rankings`), writing `calibration/applied_calibrations.json`. - New `record_calibration` stage (after verify) converts this run's non-accepted verifier outcomes into scoped `FalsePositiveLearning` records (`learnings_from_verifications`) and merges them into `.openultrasast/calibration/false_positive_learnings.json`, de-duplicated by finding ID so a repeat rejection does not compound without bound. - Lower the calibration floor (MIN_CALIBRATED_PRIORITY = 0.1) so accumulated rejections can demote a scope below the natural ranking minimum. Demotion is scoped and reversible: a rejection in one path never touches an accepted finding in another, and priority is demoted rather than the finding deleted so the audit trail explains every downgrade. First run is a no-op (empty ledger), so existing single-run behavior is unchanged. Adds tests/test_pipeline_calibration.py (two-run demotion, order-independent), documents the realized loop in the README, and checks off the standard_security_harness milestone in tasks.md. Co-Authored-By: Claude Opus 4.8 --- .kiro/specs/openrouter-sast-harness/tasks.md | 2 +- README.md | 54 ++++++------- src/openultrasast/calibration.py | 79 +++++++++++++++++++- src/openultrasast/cli.py | 43 ++++++++++- tests/test_pipeline_calibration.py | 65 ++++++++++++++++ 5 files changed, 213 insertions(+), 30 deletions(-) create mode 100644 tests/test_pipeline_calibration.py diff --git a/.kiro/specs/openrouter-sast-harness/tasks.md b/.kiro/specs/openrouter-sast-harness/tasks.md index 40b8e81..1c50fc9 100644 --- a/.kiro/specs/openrouter-sast-harness/tasks.md +++ b/.kiro/specs/openrouter-sast-harness/tasks.md @@ -128,7 +128,7 @@ Do not report OpenUltraSAST as a working security harness until `usable_harness_ - [x] Standard mode uses ranked hunter scheduling, retrieval packages, selected skill snippets, and independent verification. - [x] Every hunter trajectory is persisted and linked to finding artifacts. -- [ ] False-positive learning updates ranking calibration from verifier outcomes. +- [x] False-positive learning updates ranking calibration from verifier outcomes. - [ ] Fusion can be triggered for high-impact, contradictory, or difficult findings. - [ ] Standard mode uses real language-aware LLM hunters instead of reusing quick static findings. diff --git a/README.md b/README.md index 338a572..343b331 100644 --- a/README.md +++ b/README.md @@ -219,37 +219,41 @@ vulnerability OpenUltraSAST missed (or vice-versa). Each scan is a composed harness of typed processors with read/write **state contracts** (strict mode fails the scan on a violation; warn mode records a -degradation) and a full event trace. That trace plus the verifier and benchmark -outputs are the feedback that tunes the *next* run — and **OpenCode is the agent -that steers it**: running the triage and fix skills, it reads the harness's -signals and applies the adjustments back into config. +degradation) and a full event trace. The loop is now **closed automatically +inside the pipeline** — no agent required: every scan persists its verifier +rejections as scoped learnings and the next scan loads them and demotes those +scopes before findings are produced. ``` - ┌──────────────────── OpenCode steers (triage / fix skills) ─────┐ + ┌──────────── automatic ledger (no agent in the loop) ───────────┐ ▼ │ - scan ──▶ traces (events.jsonl) + verifier decisions + benchmark misses │ - │ │ - ├─ rejected findings ─▶ scoped false-positive learnings ─────────┤ - ├─ verifier outcomes ─▶ ranking calibration / demotion ──────────┤ applied - ├─ benchmark misses ─▶ calibration records (next candidate) ─────┤ back - └─ ranking metrics ─▶ retrieval filters · prompt constraints ───┘ into - · skill routing · variant seeds config + scan ──▶ verify ──▶ non-accepted findings ──▶ scoped learnings ─────────┤ + │ (.openultrasast/calibration/ │ loaded by + │ false_positive_learnings.json) │ the next + next scan ──▶ rank ──▶ calibrate (demote rejected scopes) ──▶ findings ──┘ scan ``` -What the harness exposes today (✅): the event/trace model and processor -contracts; the evidence-ladder verifier; the scoped false-positive learning -taxonomy with ranking calibration and prompt/retrieval-filter adjustments -(`calibration.py`); ranking metrics by tier; and benchmark miss records. +What runs automatically each scan (✅, `calibration.py` + `cli.py`): -**How the loop is closed today** — OpenCode is the controller. Invoked from the -command line on the repo, it runs a scan, reads the verifier decisions and -benchmark `calibration_records.json`, and via `openultrasast-triage` records the -scoped learnings and recalibrates ranking, then re-runs the benchmark gate to -confirm the change helped without regressing other findings. So the -self-improving cycle is real now with OpenCode as the steering agent; making the -harness apply that feedback **autonomously inside the pipeline** — recalibrating -on the following run without an agent in the loop — is the next milestone -(`standard_security_harness`). +1. **`calibrate` stage** (after `rank`, before findings) loads the persistent + ledger and demotes the priority of every scope that previously produced a + rejected finding (`calibrate_rankings`), writing `applied_calibrations.json`. +2. **`record_calibration` stage** (after `verify`) turns this run's non-accepted + verifier outcomes into scoped `FalsePositiveLearning` records + (`learnings_from_verifications`) and merges them into the ledger, de-duplicated + by finding ID so a repeat rejection does not compound without bound. + +Demotion is **scoped and reversible**: a rejection in `lib.py` demotes only that +scope; an accepted (reachable) finding in `app.py` is never touched; and because +priority is demoted rather than the finding deleted, the audit trail +(`applied_calibrations.json`, the ledger) shows exactly why a surface lost +attention. Covered by `tests/test_pipeline_calibration.py`. + +Beyond the automatic ranking loop, **OpenCode still steers the richer +adjustments** — prompt constraints, retrieval filters, skill routing, and +benchmark-miss triage — via `openultrasast-triage`; those `RankingCalibration` +fields are already produced and will be consumed directly once the language-aware +LLM hunters land (`standard_security_harness`). ## Development diff --git a/src/openultrasast/calibration.py b/src/openultrasast/calibration.py index 396d5cd..6226769 100644 --- a/src/openultrasast/calibration.py +++ b/src/openultrasast/calibration.py @@ -7,6 +7,11 @@ from pathlib import Path from .findings import StaticFinding from .rank import RankingScore +from .verification import VerificationResult, VerificationStatus + +# Calibration may push a repeatedly-rejected scope below the natural ranking +# minimum (1.0) so accumulated false positives keep losing priority over runs. +MIN_CALIBRATED_PRIORITY = 0.1 class FalsePositiveReason(StrEnum): @@ -139,7 +144,7 @@ def calibrate_ranking(ranking: RankingScore, learnings: list[FalsePositiveLearni prompt_constraints: list[str] = [] retrieval_adjustments: dict[str, str] = {} for learning in applicable: - calibrated = max(1.0, calibrated - learning.demotion) + calibrated = max(MIN_CALIBRATED_PRIORITY, calibrated - learning.demotion) prompt_constraints.append( f"Do not repeat unsupported {learning.vulnerability_class} claim for scope {learning.scope} without new evidence." ) @@ -171,7 +176,7 @@ def apply_ranking_calibration(ranking: RankingScore, calibration: RankingCalibra surface=ranking.surface, influence=ranking.influence, reachability=ranking.reachability, - priority=max(1.0, min(ranking.priority, calibration.calibrated_priority)), + priority=max(MIN_CALIBRATED_PRIORITY, min(ranking.priority, calibration.calibrated_priority)), rationale=f"{ranking.rationale}; calibrated by false-positive learning" if calibration.applied_learning_ids else ranking.rationale, model_id=ranking.model_id, static_boosts=ranking.static_boosts, @@ -183,6 +188,76 @@ def write_false_positive_learnings(learnings: list[FalsePositiveLearning], path: path.write_text(json.dumps({"learnings": [asdict(learning) for learning in learnings]}, indent=2, sort_keys=True) + "\n") +def write_ranking_calibrations(calibrations: list[RankingCalibration], path: Path) -> None: + applied = [calibration for calibration in calibrations if calibration.applied_learning_ids] + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps({"calibrations": [asdict(calibration) for calibration in applied]}, indent=2, sort_keys=True) + "\n") + + +def load_false_positive_learnings(path: Path) -> list[FalsePositiveLearning]: + if not path.exists(): + return [] + payload = json.loads(path.read_text()) + learnings: list[FalsePositiveLearning] = [] + for item in payload.get("learnings", []): + if not isinstance(item, dict): + continue + learnings.append( + FalsePositiveLearning( + finding_id=str(item["finding_id"]), + path=str(item["path"]), + vulnerability_class=str(item["vulnerability_class"]), + reason=FalsePositiveReason(item["reason"]), + evidence=str(item["evidence"]), + scope=str(item["scope"]), + demotion=float(item.get("demotion", 0.5)), + ) + ) + return learnings + + +def merge_false_positive_learnings( + existing: list[FalsePositiveLearning], + new: list[FalsePositiveLearning], +) -> list[FalsePositiveLearning]: + by_id: dict[str, FalsePositiveLearning] = {learning.finding_id: learning for learning in existing} + for learning in new: + by_id[learning.finding_id] = learning + return [by_id[finding_id] for finding_id in sorted(by_id)] + + +def learnings_from_verifications( + findings: list[StaticFinding], + verifications: list[VerificationResult], +) -> list[FalsePositiveLearning]: + finding_by_id = {finding.finding_id: finding for finding in findings} + learnings: list[FalsePositiveLearning] = [] + for result in verifications: + if result.status == VerificationStatus.ACCEPTED: + continue + finding = finding_by_id.get(result.finding_id) + if finding is None: + continue + outcome = FindingOutcome.UNVERIFIED if result.status == VerificationStatus.NEEDS_EVIDENCE else FindingOutcome.REJECTED + learning = learning_from_finding_outcome( + finding, + outcome=outcome, + evidence=f"{result.counter_case} {result.required_next_step}".strip(), + reason=_reason_from_verification(finding, result.status), + ) + if learning is not None: + learnings.append(learning) + return learnings + + +def _reason_from_verification(finding: StaticFinding, status: VerificationStatus) -> FalsePositiveReason: + if finding.reachability_status == "unknown": + return FalsePositiveReason.UNREACHABLE_PATH + if status == VerificationStatus.NEEDS_EVIDENCE: + return FalsePositiveReason.UNVERIFIED + return FalsePositiveReason.CONTRADICTED + + def _vulnerability_class(finding: StaticFinding) -> str: if finding.tags: return finding.tags[0] diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py index 6748ddd..7d5dd83 100644 --- a/src/openultrasast/cli.py +++ b/src/openultrasast/cli.py @@ -15,6 +15,15 @@ from .benchmark import ( resolve_benchmark_source, write_benchmark_artifacts, ) +from .calibration import ( + FalsePositiveLearning, + calibrate_rankings, + learnings_from_verifications, + load_false_positive_learnings, + merge_false_positive_learnings, + write_false_positive_learnings, + write_ranking_calibrations, +) from .config import load_config from .findings import StaticFinding, quick_scan_findings, write_findings from .harness import HarnessRuntime, HarnessTraceWriter, write_harness_config @@ -24,8 +33,10 @@ from .mapping import analyze_entry_points, attach_reachability_hints, ingest_sar from .preprocess import preprocess_repository, write_preprocess_artifact from .rank import rank_targets, write_rankings from .reports import scan_exit_code, write_manifest, write_markdown_report, write_sarif_report -from .run import create_scan_run -from .verification import verify_findings, write_verification_results +from .run import ScanRun, create_scan_run +from .verification import VerificationResult, verify_findings, write_verification_results + +CALIBRATION_DIR = ".openultrasast/calibration" @dataclass(frozen=True) @@ -35,6 +46,7 @@ class ScanOutcome: file_target_count: int ranked_target_count: int finding_count: int + calibrations_applied: int exit_code: int @@ -100,7 +112,12 @@ 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] write_rankings(rankings, run.root / "rank" / "ranking.json") + write_ranking_calibrations(calibrations, run.root / "calibration" / "applied_calibrations.json") if mode == "standard": hunter_result = runtime.run_stage("hunter_pool", lambda: run_hunter_pool(run.target, targets, rankings, scan_id=run.scan_id)) findings = hunter_result.findings @@ -119,6 +136,10 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut write_hunter_trajectories(trajectories, trajectories_path) verifications = runtime.run_stage("verify", lambda: verify_findings(findings)) write_verification_results(verifications, verification_path) + runtime.run_stage( + "record_calibration", + lambda: _persist_calibration_feedback(run, ledger_path, prior_learnings, findings, verifications), + ) runtime.run_stage("report", lambda: write_markdown_report(findings, markdown_path, verifications)) runtime.run_stage("sarif", lambda: write_sarif_report(findings, verifications, sarif_path)) runtime.run_stage( @@ -144,16 +165,34 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut file_target_count=len(targets), ranked_target_count=len(rankings), finding_count=len(findings), + calibrations_applied=len(applied_calibrations), exit_code=scan_exit_code(findings, verifications, fail_on), ) +def _persist_calibration_feedback( + run: ScanRun, + ledger_path: Path, + prior_learnings: list[FalsePositiveLearning], + findings: list[StaticFinding], + verifications: list[VerificationResult], +) -> list[FalsePositiveLearning]: + # Turn this run's non-accepted verifier outcomes into scoped learnings and + # merge them into the persistent ledger so the next scan demotes those scopes. + new_learnings = learnings_from_verifications(findings, verifications) + merged = merge_false_positive_learnings(prior_learnings, new_learnings) + write_false_positive_learnings(merged, ledger_path) + write_false_positive_learnings(new_learnings, run.root / "calibration" / "false_positive_learnings.json") + return merged + + def _print_scan_outcome(outcome: ScanOutcome) -> None: print(f"scan_id={outcome.scan_id}") print(f"run_dir={outcome.run_dir}") print(f"file_targets={outcome.file_target_count}") print(f"ranked_targets={outcome.ranked_target_count}") print(f"findings={outcome.finding_count}") + print(f"calibrations_applied={outcome.calibrations_applied}") def _load_static_hints(sarif_paths: tuple[str, ...]) -> list[object]: diff --git a/tests/test_pipeline_calibration.py b/tests/test_pipeline_calibration.py new file mode 100644 index 0000000..1c12a27 --- /dev/null +++ b/tests/test_pipeline_calibration.py @@ -0,0 +1,65 @@ +"""Automatic in-pipeline calibration feedback. + +A scan that rejects a finding (e.g. an unreachable pattern match) must persist a +scoped false-positive learning, and the *next* scan on the same target must load +that learning and demote the rejected scope's ranking without any agent in the +loop. Accepted (reachable) findings must not be demoted. +""" + +import json +from pathlib import Path + +import pytest + +from openultrasast.cli import main + + +def _run_dirs(repo: Path) -> set[Path]: + runs = repo / ".runs" + return set(runs.iterdir()) if runs.exists() else set() + + +def _ranking_priority(run_dir: Path, path: str) -> float: + rankings = json.loads((run_dir / "rank" / "ranking.json").read_text())["rankings"] + return next(r["priority"] for r in rankings if r["path"] == path) + + +def _finding_priority(run_dir: Path, path: str) -> float: + findings = json.loads((run_dir / "findings.json").read_text())["findings"] + return next(f["ranking_priority"] for f in findings if f["path"] == path) + + +def test_rejected_finding_demotes_its_scope_on_the_next_scan(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + repo = tmp_path / "repo" + repo.mkdir() + # Reachable route eval -> accepted (never demoted). + (repo / "app.py").write_text("@app.route('/x')\ndef view():\n return eval(request.data)\n") + # Helper eval with no entry point -> unknown reachability -> rejected -> learning. + (repo / "lib.py").write_text("def helper(x):\n return eval(x)\n") + monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs") + ledger = repo / ".openultrasast" / "calibration" / "false_positive_learnings.json" + + # --- First scan: produces the learning, applies nothing yet --------------- + before = _run_dirs(repo) + assert main(["scan", str(repo), "--mode", "quick"]) == 0 + (run1,) = _run_dirs(repo) - before + learnings = json.loads(ledger.read_text())["learnings"] + lib_learnings = [item for item in learnings if item["path"] == "lib.py"] + assert lib_learnings and lib_learnings[0]["reason"] == "unreachable_path" + assert json.loads((run1 / "calibration" / "applied_calibrations.json").read_text())["calibrations"] == [] + lib_priority_1 = _ranking_priority(run1, "lib.py") + lib_finding_1 = _finding_priority(run1, "lib.py") + app_priority_1 = _ranking_priority(run1, "app.py") + + # --- Second scan: loads the ledger and demotes the rejected scope --------- + before = _run_dirs(repo) + assert main(["scan", str(repo), "--mode", "quick"]) == 0 + (run2,) = _run_dirs(repo) - before + applied = json.loads((run2 / "calibration" / "applied_calibrations.json").read_text())["calibrations"] + applied_paths = {item["path"] for item in applied} + + assert "lib.py" in applied_paths + assert "app.py" not in applied_paths # accepted finding is never demoted + assert _ranking_priority(run2, "lib.py") < lib_priority_1 + assert _finding_priority(run2, "lib.py") < lib_finding_1 + assert _ranking_priority(run2, "app.py") == app_priority_1 # scoped: unrelated path untouched