From 5519413c82e14037122fb959d17850d1425f7e99 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Thu, 18 Jun 2026 09:59:28 +0200 Subject: [PATCH] feat(verification): add evidence ladder verifier --- .kiro/specs/openrouter-sast-harness/tasks.md | 22 +-- src/openultrasast/cli.py | 3 + src/openultrasast/verification.py | 131 +++++++++++++++++ tests/test_verification.py | 141 +++++++++++++++++++ 4 files changed, 286 insertions(+), 11 deletions(-) create mode 100644 src/openultrasast/verification.py create mode 100644 tests/test_verification.py diff --git a/.kiro/specs/openrouter-sast-harness/tasks.md b/.kiro/specs/openrouter-sast-harness/tasks.md index 8dd0f90..cb1d54d 100644 --- a/.kiro/specs/openrouter-sast-harness/tasks.md +++ b/.kiro/specs/openrouter-sast-harness/tasks.md @@ -9,7 +9,7 @@ Checklist completion is not product completion. Each phase now carries a functio - `integrated`: the feature participates in the harness runtime, traces, evidence, and reports. - `production`: the feature is robust for real repositories, CI budgets, and analyst audit. -Current verified gate: `scaffold_quick_scan`. +Current verified gate: `usable_harness_mvp` partial. Do not report OpenUltraSAST as a working security harness until `usable_harness_mvp` is checked. @@ -77,7 +77,7 @@ Do not report OpenUltraSAST as a working security harness until `usable_harness_ - [x] Quick scan is uv-managed and covered by unit tests. - [ ] Quick scan includes real LLM hunter execution. This is intentionally deferred to standard mode. - [ ] Quick scan includes SARIF output. This is intentionally deferred to Phase 9. -- [ ] Quick scan includes enforced verifier decisions. This is intentionally deferred to Phase 6. +- [x] Quick scan includes enforced verifier decisions. ## Gate: `usable_harness_mvp` @@ -85,10 +85,10 @@ Do not report OpenUltraSAST as a working security harness until `usable_harness_ - [x] Event traces are emitted for every scan stage. - [x] Static mapping ingestion feeds FileTarget hints and verifier evidence candidates. - [x] Entry-point reachability feeds FileTarget reachability hints and ranking priority. -- [ ] Evidence ladder state machine prevents invalid verified statuses. -- [ ] Independent verifier runs on fixture findings without hunter reasoning. +- [x] Evidence ladder state machine prevents invalid verified statuses. +- [x] Independent verifier runs on fixture findings without hunter reasoning. - [ ] JSON, Markdown, SARIF, and manifest artifacts share finding IDs and evidence references. -- [ ] A fixture scan demonstrates one accepted static-corroborated finding and one rejected false positive. +- [x] A fixture scan demonstrates one accepted static-corroborated finding and one rejected false positive. ## Phase 5: Harness Runtime (`usable_harness_mvp` prerequisite) @@ -117,12 +117,12 @@ Do not report OpenUltraSAST as a working security harness until `usable_harness_ ## Phase 6: Verification (`usable_harness_mvp` prerequisite) -- [ ] Define evidence ladder state machine. -- [ ] Implement independent verifier prompt and schema. -- [ ] Ensure verifier context excludes hunter reasoning. -- [ ] Implement pro-case, counter-case, tie-breaker, and required-next-step fields. -- [ ] Enforce that verified report status requires at least `static_corroboration`. -- [ ] Add tests for invalid evidence transitions. +- [x] Define evidence ladder state machine. +- [x] Implement independent verifier prompt and schema. +- [x] Ensure verifier context excludes hunter reasoning. +- [x] Implement pro-case, counter-case, tie-breaker, and required-next-step fields. +- [x] Enforce that verified report status requires at least `static_corroboration`. +- [x] Add tests for invalid evidence transitions. ## Gate: `standard_security_harness` diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py index 2ebe3bc..6271fe6 100644 --- a/src/openultrasast/cli.py +++ b/src/openultrasast/cli.py @@ -13,6 +13,7 @@ from .preprocess import preprocess_repository, write_preprocess_artifact from .rank import rank_targets, write_rankings from .reports import write_markdown_report from .run import create_scan_run +from .verification import verify_findings, write_verification_results def main(argv: list[str] | None = None) -> int: @@ -66,6 +67,8 @@ def _scan(path: Path, config_path: Path, mode: str) -> int: write_rankings(rankings, run.root / "rank" / "ranking.json") findings = runtime.run_stage("quick_findings", lambda: quick_scan_findings(run.target, targets, rankings)) write_findings(findings, run.root / "findings.json") + verifications = runtime.run_stage("verify", lambda: verify_findings(findings)) + write_verification_results(verifications, run.root / "verification.json") runtime.run_stage("report", lambda: write_markdown_report(findings, run.root / "report.md")) runtime.finish(status="succeeded") print(f"scan_id={run.scan_id}") diff --git a/src/openultrasast/verification.py b/src/openultrasast/verification.py new file mode 100644 index 0000000..dd12179 --- /dev/null +++ b/src/openultrasast/verification.py @@ -0,0 +1,131 @@ +from __future__ import annotations + +import json +from dataclasses import asdict, dataclass +from enum import StrEnum +from pathlib import Path + +from .findings import StaticFinding + + +class EvidenceLevel(StrEnum): + SUSPICION = "suspicion" + STATIC_CORROBORATION = "static_corroboration" + CRASH_REPRODUCED = "crash_reproduced" + ROOT_CAUSE_EXPLAINED = "root_cause_explained" + EXPLOIT_DEMONSTRATED = "exploit_demonstrated" + PATCH_VALIDATED = "patch_validated" + + +class VerificationStatus(StrEnum): + ACCEPTED = "accepted" + REJECTED = "rejected" + NEEDS_EVIDENCE = "needs_evidence" + + +@dataclass(frozen=True) +class VerificationResult: + finding_id: str + status: VerificationStatus + evidence_level: EvidenceLevel + verified: bool + pro_case: str + counter_case: str + tie_breaker: str + required_next_step: str + context_sources: list[str] + + +def validate_evidence_transition(current: EvidenceLevel | str, target: EvidenceLevel | str) -> EvidenceLevel: + current_level = EvidenceLevel(current) + target_level = EvidenceLevel(target) + if _EVIDENCE_ORDER[target_level] < _EVIDENCE_ORDER[current_level]: + raise ValueError(f"cannot move evidence level backward: {current_level} -> {target_level}") + if _EVIDENCE_ORDER[target_level] - _EVIDENCE_ORDER[current_level] > 1: + raise ValueError(f"cannot skip evidence levels: {current_level} -> {target_level}") + return target_level + + +def is_report_verified(evidence_level: EvidenceLevel | str, status: VerificationStatus | str) -> bool: + level = EvidenceLevel(evidence_level) + verification_status = VerificationStatus(status) + return verification_status == VerificationStatus.ACCEPTED and _EVIDENCE_ORDER[level] >= _EVIDENCE_ORDER[EvidenceLevel.STATIC_CORROBORATION] + + +def build_independent_verifier_context(finding: StaticFinding) -> dict[str, object]: + return { + "finding_id": finding.finding_id, + "path": finding.path, + "line": finding.line, + "function_name": finding.function_name, + "title": finding.title, + "severity": finding.severity, + "evidence_level": finding.evidence_level, + "reachability_status": finding.reachability_status, + "reachability_evidence": finding.reachability_evidence, + "reachability_conditions": finding.reachability_conditions, + "tags": finding.tags, + } + + +def verify_finding(finding: StaticFinding) -> VerificationResult: + level = EvidenceLevel(finding.evidence_level) + context = build_independent_verifier_context(finding) + context_sources = sorted(context.keys()) + + if _EVIDENCE_ORDER[level] < _EVIDENCE_ORDER[EvidenceLevel.STATIC_CORROBORATION]: + return VerificationResult( + finding_id=finding.finding_id, + status=VerificationStatus.NEEDS_EVIDENCE, + evidence_level=level, + verified=False, + pro_case="A candidate issue exists, but it has not reached static corroboration.", + counter_case="The current evidence can still be a model or heuristic suspicion.", + tie_breaker="Add static analyzer, source, or runtime evidence that corroborates the claim.", + required_next_step="Collect static corroboration before reporting as verified.", + context_sources=context_sources, + ) + + if finding.reachability_status == "unknown": + return VerificationResult( + finding_id=finding.finding_id, + status=VerificationStatus.REJECTED, + evidence_level=level, + verified=False, + pro_case="A static pattern matched the source line.", + counter_case="No function-level attacker-reachable entry point currently reaches this line.", + tie_breaker="Provide call-graph, route, CLI, parser, or dynamic evidence linking attacker input to this function.", + required_next_step="Add reachability evidence before prioritizing a fix.", + context_sources=context_sources, + ) + + return VerificationResult( + finding_id=finding.finding_id, + status=VerificationStatus.ACCEPTED, + evidence_level=level, + verified=is_report_verified(level, VerificationStatus.ACCEPTED), + pro_case="Static corroboration exists and function-level reachability evidence is attached.", + counter_case="Impact and exploitability still require deeper verifier or dynamic evidence.", + tie_breaker="Confirm attacker control, sanitizer absence, and impact in the next verification depth.", + required_next_step="Proceed to standard verifier or dynamic reproduction if severity warrants it.", + context_sources=context_sources, + ) + + +def verify_findings(findings: list[StaticFinding]) -> list[VerificationResult]: + return [verify_finding(finding) for finding in findings] + + +def write_verification_results(results: list[VerificationResult], path: Path) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps({"verifications": [asdict(result) for result in results]}, indent=2, sort_keys=True) + "\n") + + +_EVIDENCE_ORDER = { + EvidenceLevel.SUSPICION: 0, + EvidenceLevel.STATIC_CORROBORATION: 1, + EvidenceLevel.CRASH_REPRODUCED: 2, + EvidenceLevel.ROOT_CAUSE_EXPLAINED: 3, + EvidenceLevel.EXPLOIT_DEMONSTRATED: 4, + EvidenceLevel.PATCH_VALIDATED: 5, +} diff --git a/tests/test_verification.py b/tests/test_verification.py new file mode 100644 index 0000000..3d6c6c1 --- /dev/null +++ b/tests/test_verification.py @@ -0,0 +1,141 @@ +import json +from pathlib import Path + +import pytest + +from openultrasast.cli import main +from openultrasast.findings import StaticFinding +from openultrasast.verification import ( + EvidenceLevel, + VerificationStatus, + build_independent_verifier_context, + is_report_verified, + validate_evidence_transition, + verify_finding, + write_verification_results, +) + + +def test_evidence_transition_rejects_backward_and_skipped_levels() -> None: + assert validate_evidence_transition("suspicion", "static_corroboration") == EvidenceLevel.STATIC_CORROBORATION + + with pytest.raises(ValueError, match="backward"): + validate_evidence_transition("root_cause_explained", "static_corroboration") + + with pytest.raises(ValueError, match="skip"): + validate_evidence_transition("suspicion", "root_cause_explained") + + +def test_verified_status_requires_static_corroboration_and_acceptance() -> None: + assert is_report_verified(EvidenceLevel.STATIC_CORROBORATION, VerificationStatus.ACCEPTED) + assert not is_report_verified(EvidenceLevel.SUSPICION, VerificationStatus.ACCEPTED) + assert not is_report_verified(EvidenceLevel.STATIC_CORROBORATION, VerificationStatus.REJECTED) + + +def test_independent_verifier_context_excludes_hunter_reasoning() -> None: + finding = _finding(evidence_level="static_corroboration", reachability_status="reachable") + + context = build_independent_verifier_context(finding) + + assert "hunter_reasoning" not in context + assert "chain_of_thought" not in context + assert context["finding_id"] == finding.finding_id + assert context["reachability_status"] == "reachable" + + +def test_verifier_accepts_reachable_static_corroborated_finding() -> None: + finding = _finding(evidence_level="static_corroboration", reachability_status="reachable") + + result = verify_finding(finding) + + assert result.status == VerificationStatus.ACCEPTED + assert result.verified + assert result.evidence_level == EvidenceLevel.STATIC_CORROBORATION + assert result.pro_case + assert result.counter_case + assert result.tie_breaker + assert result.required_next_step + + +def test_verifier_rejects_unknown_reachability_static_finding() -> None: + finding = _finding(evidence_level="static_corroboration", reachability_status="unknown") + + result = verify_finding(finding) + + assert result.status == VerificationStatus.REJECTED + assert not result.verified + assert "reach" in result.counter_case.lower() + assert "call-graph" in result.tie_breaker + + +def test_verifier_does_not_verify_suspicion_only_finding() -> None: + finding = _finding(evidence_level="suspicion", reachability_status="reachable") + + result = verify_finding(finding) + + assert result.status == VerificationStatus.NEEDS_EVIDENCE + assert not result.verified + assert result.required_next_step == "Collect static corroboration before reporting as verified." + + +def test_write_verification_results_artifact(tmp_path: Path) -> None: + result = verify_finding(_finding(evidence_level="static_corroboration", reachability_status="reachable")) + output = tmp_path / "verification.json" + + write_verification_results([result], output) + + payload = json.loads(output.read_text()) + assert payload["verifications"][0]["finding_id"] == "python-unsafe-eval:app.py:3" + assert payload["verifications"][0]["status"] == "accepted" + + +def test_cli_scan_writes_accepted_and_rejected_verifications(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + repo = tmp_path / "repo" + repo.mkdir() + (repo / "app.py").write_text( + "@app.route('/admin')\n" + "def admin():\n" + " return eval(request.data)\n" + "\n" + "def helper(value):\n" + " return exec(value)\n" + ) + monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs") + + assert main(["scan", str(repo), "--mode", "quick"]) == 0 + + run_dirs = sorted((repo / ".runs").iterdir()) + verification_path = run_dirs[-1] / "verification.json" + payload = json.loads(verification_path.read_text()) + statuses = {item["status"] for item in payload["verifications"]} + + assert statuses == {"accepted", "rejected"} + + +def _finding(*, evidence_level: str, reachability_status: str) -> StaticFinding: + evidence = [ + { + "kind": "route", + "access_level": "public", + "line": 1, + "end_line": 3, + "function_name": "admin", + "conditions": [], + } + ] if reachability_status == "reachable" else [] + return StaticFinding( + finding_id="python-unsafe-eval:app.py:3", + path="app.py", + title="Dynamic Python execution needs review", + severity="high", + confidence="medium", + evidence_level=evidence_level, + rationale="Static pattern matched.", + line=3, + function_name="admin" if reachability_status == "reachable" else None, + reachability_status=reachability_status, + reachability_evidence=evidence, + reachability_conditions=[], + tags=["syscall_entry"], + ranking_priority=3.0, + )