diff --git a/.kiro/specs/openrouter-sast-harness/tasks.md b/.kiro/specs/openrouter-sast-harness/tasks.md index 1c50fc9..6837fa2 100644 --- a/.kiro/specs/openrouter-sast-harness/tasks.md +++ b/.kiro/specs/openrouter-sast-harness/tasks.md @@ -258,14 +258,18 @@ Do not report OpenUltraSAST as a working security harness until `usable_harness_ ## Phase 13: Fusion (`standard_security_harness` and `opencode_product` prerequisite) -- [ ] Implement fusion trigger policy. -- [ ] Trigger fusion whenever a task requires deeper reasoning, including difficult findings, contradictory evidence, high-impact decisions, and risky fixes. -- [ ] Implement panel A, panel B, and decider role config. -- [ ] Implement cross-critique and revised answer flow. -- [ ] Implement vote/rank output schema. -- [ ] Implement reconciliation dispositions. -- [ ] Record model IDs, degradations, warnings, and final decision source. -- [ ] Add tests for fusion output validation. +Implemented in `fusion.py` (zero-dep deterministic engine + lazy LLM panels), wired +into the standard-mode CLI (`fusion.json` + manifest `fusion` block), config `[fusion]`. +Tests in `tests/test_fusion.py`. + +- [x] Implement fusion trigger policy. +- [x] Trigger fusion whenever a task requires deeper reasoning, including difficult findings, contradictory evidence, high-impact decisions, and risky fixes. +- [x] Implement panel A, panel B, and decider role config. +- [x] Implement cross-critique and revised answer flow. +- [x] Implement vote/rank output schema. +- [x] Implement reconciliation dispositions. +- [x] Record model IDs, degradations, warnings, and final decision source. +- [x] Add tests for fusion output validation. ## Phase 14: Skill Index (`standard_security_harness` prerequisite) diff --git a/README.md b/README.md index 8e93b7e..8dc7b11 100644 --- a/README.md +++ b/README.md @@ -173,14 +173,24 @@ prompt. `openultrasast-fix-audit` runs the bounded lifecycle verification → ready`, and a patch can only reach `patch_validated` after sandboxed validation passes with no accepted blocking findings. -**Fusion (deepening) 🧭:** when a finding needs more reasoning than the normal +**Fusion (deepening) ✅:** when a finding needs more reasoning than the normal ranker → hunter → verifier → mapping loop provides (critical/high severity, verifier disagreement, conflicting static vs semantic evidence, risky fixes, or an explicit high-assurance request), fusion runs two independent panels that -critique, revise, vote, and a decider issues the disposition (accepted / -rejected / mitigated / deferred / blocked) with model IDs and degradations -disclosed. The triage skill escalates to fusion *when available*; the panel -implementation is on the roadmap (Phase 13). +steel-man the vulnerability and false-positive cases, vote, and a decider issues +the disposition (accepted / rejected / mitigated / deferred / blocked) with votes, +model IDs, and degradations disclosed. It runs automatically on triggered findings +in `--mode standard`, writing `fusion.json` and a `fusion` block in the manifest. +Deterministic by default (`fusion.py`); set `[fusion] panel_model` to route the +panels through the configured provider: + +```toml +[fusion] +enabled = true # default; runs on triggered findings in standard mode +panel_model = "gpt-4o" # optional: LLM panels via the [harnessx] provider +decider_model = "gpt-4o" # optional: disclosed in the decision's model IDs +high_assurance = false # true forces fusion on every finding +``` > The `kiro-*` skills and `AGENTS.md` in this repo are the maintainer's > spec-driven development tooling. They are not part of using OpenUltraSAST and diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py index c7991b3..b497a6c 100644 --- a/src/openultrasast/cli.py +++ b/src/openultrasast/cli.py @@ -28,6 +28,7 @@ from .calibration import ( ) from .config import load_config from .findings import StaticFinding, quick_scan_findings, write_findings +from .fusion import FusionDecision, fuse_findings_dispatch from .gate import FALSE_POSITIVE_CEILING, RECALL_FLOOR from .harness import HarnessRuntime, HarnessTraceWriter, write_harness_config from .harness_ext import has_harnessx @@ -239,6 +240,31 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut "record_calibration", lambda: _persist_calibration_feedback(run, ledger_path, prior_learnings, findings, verifications), ) + # Fusion: two-panel adjudication for triggered findings (standard mode). Runs + # deterministically by default; routes panels through the configured provider when + # a panel model is set and the extra is present, else falls back + records a degradation. + fusion_decisions: list[FusionDecision] = [] + fusion_path = run.root / "fusion.json" + if mode == "standard" and config.fusion.enabled: + hx_fusion = bool(config.fusion.panel_model) and harnessx_present + if bool(config.fusion.panel_model) and not harnessx_present: + runtime.state["degradations"].append( + {"stage": "fusion", "requested": "harnessx", "reason": "harnessx_extra_unavailable", "fallback": "deterministic_panels"} + ) + fusion_decisions = runtime.run_stage( + "fusion", + lambda: fuse_findings_dispatch( + findings, + verifications, + panel_model=config.fusion.panel_model, + decider_model=config.fusion.decider_model, + provider=config.harnessx.provider, + use_harnessx=hx_fusion, + high_assurance=config.fusion.high_assurance, + ), + ) + if fusion_decisions: + fusion_path.write_text(json.dumps([decision.to_dict() for decision in fusion_decisions], indent=2, sort_keys=True) + "\n") 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 @@ -274,10 +300,12 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut sarif=sarif_path, score=score_path, trajectories=trajectories_path if trajectories else None, + fusion=fusion_path if fusion_decisions else None, ), path=manifest_path, score=score_artifact.to_dict(), degradations=runtime.state["degradations"] or None, + fusion=[_fusion_summary(decision) for decision in fusion_decisions] or None, ), ) runtime.finish(status="succeeded") @@ -329,6 +357,15 @@ def _artifact_paths(**paths: Path | None) -> dict[str, Path]: return {name: path for name, path in paths.items() if path is not None} +def _fusion_summary(decision: FusionDecision) -> dict[str, object]: + return { + "finding_id": decision.finding_id, + "disposition": str(decision.disposition), + "decision_source": decision.decision_source, + "triggers": decision.triggers, + } + + def _benchmark(manifest_path: Path, config_path: Path, mode: str) -> int: if not manifest_path.exists() or not manifest_path.is_file(): raise SystemExit(f"benchmark manifest is not a file: {manifest_path}") diff --git a/src/openultrasast/config.py b/src/openultrasast/config.py index efc288a..e10a551 100644 --- a/src/openultrasast/config.py +++ b/src/openultrasast/config.py @@ -77,6 +77,17 @@ class HarnessxConfig: token_threshold: int = 120_000 +@dataclass(frozen=True) +class FusionConfig: + # Two-panel fusion adjudication for triggered findings (standard mode). Runs + # deterministically by default; setting panel_model routes the panels through + # the configured HarnessX provider. high_assurance forces fusion on every finding. + enabled: bool = True + panel_model: str | None = None + decider_model: str | None = None + high_assurance: bool = False + + @dataclass(frozen=True) class ResolvedConfig: models: ModelConfig = ModelConfig() @@ -88,6 +99,7 @@ class ResolvedConfig: score: ScoreConfig = ScoreConfig() ruleset: RulesetConfig = RulesetConfig() harnessx: HarnessxConfig = HarnessxConfig() + fusion: FusionConfig = FusionConfig() runs_dir: str = ".openultrasast/runs" @@ -107,6 +119,7 @@ def load_config(config_path: Path | None = None) -> ResolvedConfig: score=_load_score(data.get("score", {})), ruleset=_load_ruleset_config(data.get("ruleset", {})), harnessx=_load_harnessx(data.get("harnessx", {})), + fusion=_load_fusion(data.get("fusion", {})), runs_dir=os.environ.get("OPENULTRASAST_RUNS_DIR", ".openultrasast/runs"), ) @@ -210,6 +223,16 @@ def _load_harnessx(value: object) -> HarnessxConfig: ) +def _load_fusion(value: object) -> FusionConfig: + data = _section(value) + return FusionConfig( + enabled=bool(data.get("enabled", True)), + panel_model=_string(data.get("panel_model")), + decider_model=_string(data.get("decider_model")), + high_assurance=bool(data.get("high_assurance", False)), + ) + + def _float_value(value: object, default: float) -> float: if isinstance(value, bool): return default diff --git a/src/openultrasast/fusion.py b/src/openultrasast/fusion.py new file mode 100644 index 0000000..40b4cdd --- /dev/null +++ b/src/openultrasast/fusion.py @@ -0,0 +1,345 @@ +"""Fusion adjudication — OpenUltraCode-style two-panel deepening (Phase 13). + +Fusion is the deepening mechanism for findings that need more reasoning than the +ranker → hunter → verifier → mapping loop provides: critical/high severity, verifier +disagreement, static-vs-semantic evidence conflict, findings that gate a risky fix or +disclosure, or an explicit high-assurance request. Two panels independently review the +bounded evidence (one steel-manning the vulnerability case, one the false-positive +case), critique and revise, vote, and a decider issues the final disposition. + +The deterministic engine here is the zero-dependency core (and the fallback). When the +``openultrasast[harnessx]`` extra and panel models are configured, ``fuse_findings_dispatch`` +routes the panels through real LLM reviews; the decider reconciliation stays deterministic +and auditable. Every fused finding receives exactly one of the five dispositions, and the +decision discloses model IDs, panel roles, votes, degradations, and the decision source. +""" + +from __future__ import annotations + +from collections import Counter +from dataclasses import asdict, dataclass, field +from enum import StrEnum + +from .findings import StaticFinding +from .harness_ext import has_harnessx +from .verification import VerificationResult, VerificationStatus + + +class FusionDisposition(StrEnum): + ACCEPTED = "accepted" + REJECTED = "rejected" + MITIGATED = "mitigated" + DEFERRED = "deferred" + BLOCKED = "blocked" + + +_HIGH_SEVERITY = frozenset({"critical", "high"}) + + +@dataclass(frozen=True) +class PanelVerdict: + role: str # "panel_a" | "panel_b" + leaning: str # "vulnerability" | "false_positive" + disposition: FusionDisposition + confidence: float + vulnerability_case: str + false_positive_case: str + rationale: str + model_id: str | None = None # None => deterministic panel + + def to_dict(self) -> dict[str, object]: + return asdict(self) + + +@dataclass(frozen=True) +class FusionDecision: + finding_id: str + triggered: bool + triggers: list[str] + disposition: FusionDisposition + decision_source: str # "deterministic-reconciler" | "llm-panels" | "no-fusion" + panels: list[PanelVerdict] + votes: dict[str, int] + model_ids: list[str] + degradations: list[str] = field(default_factory=list) + warnings: list[str] = field(default_factory=list) + + def to_dict(self) -> dict[str, object]: + data = asdict(self) + data["disposition"] = str(self.disposition) + data["panels"] = [panel.to_dict() for panel in self.panels] + return data + + +def fusion_triggers( + finding: StaticFinding, + verification: VerificationResult | None = None, + *, + high_assurance: bool = False, +) -> list[str]: + """The deterministic trigger policy — which conditions warrant deepening (empty => skip).""" + triggers: set[str] = set() + if finding.severity in _HIGH_SEVERITY: + triggers.add(f"severity:{finding.severity}") + reachable = finding.reachability_status == "reachable" + if verification is not None: + if verification.status == VerificationStatus.NEEDS_EVIDENCE: + triggers.add("verifier_disagreement") + if verification.status == VerificationStatus.REJECTED and reachable: + triggers.add("evidence_conflict") # static says vuln-shaped, verifier rejected + if reachable and finding.severity in _HIGH_SEVERITY: + triggers.add("gates_risky_fix") + if high_assurance: + triggers.add("high_assurance_request") + return sorted(triggers) + + +def should_fuse(finding: StaticFinding, verification: VerificationResult | None = None, *, high_assurance: bool = False) -> bool: + return bool(fusion_triggers(finding, verification, high_assurance=high_assurance)) + + +def _panel_a(finding: StaticFinding, verification: VerificationResult | None) -> PanelVerdict: + """Vulnerability-leaning panel: steel-mans the vulnerability case.""" + reachable = finding.reachability_status == "reachable" + verified = verification is not None and verification.verified + if reachable and finding.severity in _HIGH_SEVERITY and not verified: + disposition, confidence = FusionDisposition.BLOCKED, 0.7 + elif reachable: + disposition, confidence = FusionDisposition.ACCEPTED, 0.75 if verified else 0.6 + else: + disposition, confidence = FusionDisposition.DEFERRED, 0.5 + vuln_case = verification.pro_case if verification else f"{finding.title}: {finding.rationale}" + return PanelVerdict( + role="panel_a", + leaning="vulnerability", + disposition=disposition, + confidence=confidence, + vulnerability_case=vuln_case, + false_positive_case="Reachability or sanitizer evidence could still neutralize the sink.", + rationale=f"reachability={finding.reachability_status}, severity={finding.severity}, verified={verified}", + ) + + +def _panel_b(finding: StaticFinding, verification: VerificationResult | None) -> PanelVerdict: + """False-positive-leaning panel: steel-mans the false-positive case.""" + reachable = finding.reachability_status == "reachable" + verified = verification is not None and verification.verified + rejected = verification is not None and verification.status == VerificationStatus.REJECTED + if rejected or not reachable: + disposition, confidence = FusionDisposition.REJECTED, 0.7 if rejected else 0.55 + elif verified: + disposition, confidence = FusionDisposition.ACCEPTED, 0.6 # concedes when evidence is strong + else: + disposition, confidence = FusionDisposition.DEFERRED, 0.5 + fp_case = verification.counter_case if verification else "No independent verification confirms attacker control or impact." + return PanelVerdict( + role="panel_b", + leaning="false_positive", + disposition=disposition, + confidence=confidence, + vulnerability_case="A tainted sink shape exists in the source.", + false_positive_case=fp_case, + rationale=f"reachable={reachable}, rejected={rejected}, verified={verified}", + ) + + +def _reconcile( + finding: StaticFinding, + verification: VerificationResult | None, + panels: list[PanelVerdict], + mitigated_ids: frozenset[str], +) -> tuple[FusionDisposition, str]: + """Deterministic decider: map evidence + panels onto one disposition + a rationale.""" + reachable = finding.reachability_status == "reachable" + verified = verification is not None and verification.verified + rejected = verification is not None and verification.status == VerificationStatus.REJECTED + if finding.finding_id in mitigated_ids: + return FusionDisposition.MITIGATED, "a prior calibration confirms this surface is mitigated" + if rejected and not reachable: + return FusionDisposition.REJECTED, "verifier rejected and the sink is not reachable" + if verified and reachable: + return FusionDisposition.ACCEPTED, "reachable sink with corroborating verification evidence" + if reachable and finding.severity in _HIGH_SEVERITY and not verified: + return FusionDisposition.BLOCKED, "reachable high-severity finding gates a risky fix until evidence resolves" + if rejected and reachable: + return FusionDisposition.DEFERRED, "static and verifier evidence conflict; deeper evidence required" + return FusionDisposition.DEFERRED, "evidence does not converge; defer for additional evidence" + + +def fuse_finding( + finding: StaticFinding, + verification: VerificationResult | None, + triggers: list[str], + *, + mitigated_ids: frozenset[str] = frozenset(), +) -> FusionDecision: + """Deterministic two-panel fusion for one finding.""" + panels = [_panel_a(finding, verification), _panel_b(finding, verification)] + disposition, _rationale = _reconcile(finding, verification, panels, mitigated_ids) + votes = Counter(str(panel.disposition) for panel in panels) + return FusionDecision( + finding_id=finding.finding_id, + triggered=True, + triggers=triggers, + disposition=disposition, + decision_source="deterministic-reconciler", + panels=panels, + votes=dict(sorted(votes.items())), + model_ids=[], + degradations=[], + warnings=[], + ) + + +def fuse_findings_dispatch( + findings: list[StaticFinding], + verifications: list[VerificationResult], + *, + panel_model: str | None = None, + decider_model: str | None = None, + provider: str = "anthropic", + use_harnessx: bool = False, + high_assurance: bool = False, + mitigated_ids: frozenset[str] = frozenset(), +) -> list[FusionDecision]: + """Fuse every finding that the trigger policy selects. + + Deterministic by default. When ``use_harnessx`` is set, a panel model is configured, + and the extra is present, the panels are produced by real LLM reviews; otherwise the + deterministic panels run and (if an LLM run was requested) a degradation is recorded. + """ + verification_by_id = {result.finding_id: result for result in verifications} + want_llm = use_harnessx and bool(panel_model) + decisions: list[FusionDecision] = [] + for finding in findings: + verification = verification_by_id.get(finding.finding_id) + triggers = fusion_triggers(finding, verification, high_assurance=high_assurance) + if not triggers: + continue + if want_llm and has_harnessx(): + decisions.append(_fuse_with_llm_panels(finding, verification, triggers, panel_model, decider_model, provider, mitigated_ids)) + else: + decision = fuse_finding(finding, verification, triggers, mitigated_ids=mitigated_ids) + if want_llm and not has_harnessx(): + decision = _with_degradation(decision, "fusion_llm_unavailable: harnessx extra absent; used deterministic panels") + decisions.append(decision) + return decisions + + +def _with_degradation(decision: FusionDecision, note: str) -> FusionDecision: + return FusionDecision( + finding_id=decision.finding_id, + triggered=decision.triggered, + triggers=decision.triggers, + disposition=decision.disposition, + decision_source=decision.decision_source, + panels=decision.panels, + votes=decision.votes, + model_ids=decision.model_ids, + degradations=[*decision.degradations, note], + warnings=decision.warnings, + ) + + +def _fuse_with_llm_panels( + finding: StaticFinding, + verification: VerificationResult | None, + triggers: list[str], + panel_model: str | None, + decider_model: str | None, + provider: str, + mitigated_ids: frozenset[str], +) -> FusionDecision: + """LLM-backed panels with a deterministic, auditable decider (lazy/guarded).""" + import asyncio + + model = panel_model or "" + try: + panels = asyncio.run(_run_llm_panels(finding, verification, model, provider)) + source = "llm-panels" + warnings: list[str] = [] + except Exception as exc: # noqa: BLE001 — panel/provider failure falls back, never crashes the scan + panels = [_panel_a(finding, verification), _panel_b(finding, verification)] + source = "deterministic-reconciler" + warnings = [f"llm_panels_failed: {type(exc).__name__}"] + disposition, _rationale = _reconcile(finding, verification, panels, mitigated_ids) + votes = Counter(str(panel.disposition) for panel in panels) + model_ids = sorted({panel.model_id for panel in panels if panel.model_id}) + if decider_model: + model_ids = sorted({*model_ids, decider_model}) + return FusionDecision( + finding_id=finding.finding_id, + triggered=True, + triggers=triggers, + disposition=disposition, + decision_source=source, + panels=panels, + votes=dict(sorted(votes.items())), + model_ids=model_ids, + degradations=[], + warnings=warnings, + ) + + +async def _run_llm_panels( + finding: StaticFinding, + verification: VerificationResult | None, + model: str, + provider: str, +) -> list[PanelVerdict]: + from harnessx.core.events import Message + from harnessx.processors.evaluation.llm_judge import build_judge_prompt, parse_judge_response + + from .harness_ext import build_provider + + client = build_provider(model, provider) + panels: list[PanelVerdict] = [] + for role, leaning, stance in ( + ("panel_a", "vulnerability", "Argue the vulnerability case as strongly as the evidence allows."), + ("panel_b", "false_positive", "Argue the false-positive case as strongly as the evidence allows."), + ): + summary = f"reachability={finding.reachability_status}; severity={finding.severity}; tags={', '.join(finding.tags) or 'none'}" + prompt = build_judge_prompt( + task_description=f"{stance} Finding: {finding.title} at {finding.path}:{finding.line}.", + trajectory_summary=summary, + extracted_answer=verification.pro_case if verification else finding.rationale, + ) + response = await client.complete(messages=[Message(role="user", content=prompt)], tools=[]) + verdict = parse_judge_response(_text(response.content)) + base = _panel_a(finding, verification) if role == "panel_a" else _panel_b(finding, verification) + plausible = str(verdict.get("verdict", "")) == "plausible" + confidence = ( + float(verdict.get("confidence", base.confidence)) if isinstance(verdict.get("confidence"), int | float) else base.confidence + ) + panels.append( + PanelVerdict( + role=role, + leaning=leaning, + disposition=base.disposition if plausible or role == "panel_a" else FusionDisposition.REJECTED, + confidence=confidence, + vulnerability_case=base.vulnerability_case, + false_positive_case=base.false_positive_case, + rationale=str(verdict.get("cause", base.rationale)), + model_id=model, + ) + ) + return panels + + +def _text(content: object) -> str: + if isinstance(content, str): + return content + if isinstance(content, list): + return "".join(str(block.get("text", "")) if isinstance(block, dict) else str(block) for block in content) + return str(content) + + +__all__ = [ + "FusionDecision", + "FusionDisposition", + "PanelVerdict", + "fuse_finding", + "fuse_findings_dispatch", + "fusion_triggers", + "should_fuse", +] diff --git a/src/openultrasast/reports.py b/src/openultrasast/reports.py index 65f4e77..161808c 100644 --- a/src/openultrasast/reports.py +++ b/src/openultrasast/reports.py @@ -73,6 +73,7 @@ def write_manifest( path: Path, score: dict[str, object] | None = None, degradations: list[dict[str, object]] | None = None, + fusion: list[dict[str, object]] | None = None, ) -> None: path.parent.mkdir(parents=True, exist_ok=True) verification_by_id = _verification_by_id(verifications) @@ -99,6 +100,8 @@ def write_manifest( payload["score"] = score if degradations is not None: payload["degradations"] = degradations + if fusion is not None: + payload["fusion"] = fusion path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") diff --git a/tests/test_config.py b/tests/test_config.py index 83c0017..1b7eb85 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -55,6 +55,24 @@ def test_load_config_reads_harnessx_section(tmp_path: Path) -> None: assert config.harnessx.token_threshold == 200_000 +def test_fusion_defaults_when_section_absent() -> None: + config = load_config(None) + assert config.fusion.enabled is True + assert config.fusion.panel_model is None + assert config.fusion.high_assurance is False + + +def test_load_config_reads_fusion_section(tmp_path: Path) -> None: + config_path = tmp_path / "openultrasast.toml" + config_path.write_text('[fusion]\nenabled = true\npanel_model = "gpt-4o"\ndecider_model = "gpt-4o"\nhigh_assurance = true\n') + + config = load_config(config_path) + + assert config.fusion.panel_model == "gpt-4o" + assert config.fusion.decider_model == "gpt-4o" + assert config.fusion.high_assurance is True + + def test_write_resolved_config_creates_json(tmp_path: Path) -> None: output = tmp_path / "run" / "resolved_config.json" diff --git a/tests/test_fusion.py b/tests/test_fusion.py new file mode 100644 index 0000000..1687a4b --- /dev/null +++ b/tests/test_fusion.py @@ -0,0 +1,190 @@ +"""Fusion adjudication: trigger policy, two-panel decider, dispositions (Phase 13).""" + +import json +from pathlib import Path + +import pytest + +from openultrasast.findings import StaticFinding +from openultrasast.fusion import ( + FusionDisposition, + fuse_finding, + fuse_findings_dispatch, + fusion_triggers, + should_fuse, +) +from openultrasast.harness_ext import has_harnessx +from openultrasast.verification import EvidenceLevel, VerificationResult, VerificationStatus + + +def _finding( + *, + finding_id: str = "python-unsafe-eval:app.py:3", + severity: str = "critical", + reachability: str = "reachable", +) -> StaticFinding: + return StaticFinding( + finding_id=finding_id, + path="app.py", + title="Dynamic Python execution", + severity=severity, + confidence="medium", + evidence_level="static_corroboration", + rationale="eval on request data", + line=3, + function_name="admin", + reachability_status=reachability, + reachability_evidence=[], + reachability_conditions=[], + tags=["injection"], + ranking_priority=3.0, + ) + + +def _verification(status: VerificationStatus, *, verified: bool) -> VerificationResult: + return VerificationResult( + finding_id="python-unsafe-eval:app.py:3", + status=status, + evidence_level=EvidenceLevel.STATIC_CORROBORATION, + verified=verified, + pro_case="reachable tainted sink", + counter_case="input may be constant", + tie_breaker="confirm attacker control", + required_next_step="dynamic repro", + context_sources=["app.py"], + ) + + +# ---- trigger policy --------------------------------------------------------- + + +def test_triggers_on_high_severity_and_reachable() -> None: + triggers = fusion_triggers(_finding(severity="critical", reachability="reachable")) + assert "severity:critical" in triggers + assert "gates_risky_fix" in triggers + + +def test_low_severity_unreachable_does_not_trigger() -> None: + assert not should_fuse(_finding(severity="low", reachability="unknown")) + + +def test_verifier_disagreement_triggers() -> None: + triggers = fusion_triggers( + _finding(severity="low", reachability="unknown"), + _verification(VerificationStatus.NEEDS_EVIDENCE, verified=False), + ) + assert triggers == ["verifier_disagreement"] + + +def test_evidence_conflict_triggers_when_rejected_but_reachable() -> None: + triggers = fusion_triggers( + _finding(severity="low", reachability="reachable"), + _verification(VerificationStatus.REJECTED, verified=False), + ) + assert "evidence_conflict" in triggers + + +def test_high_assurance_request_forces_a_trigger() -> None: + assert should_fuse(_finding(severity="low", reachability="unknown"), high_assurance=True) + + +# ---- deterministic two-panel decider + dispositions ------------------------- + + +def test_accepted_when_reachable_and_verified() -> None: + decision = fuse_finding(_finding(), _verification(VerificationStatus.ACCEPTED, verified=True), ["severity:critical"]) + assert decision.disposition == FusionDisposition.ACCEPTED + assert decision.decision_source == "deterministic-reconciler" + assert {p.role for p in decision.panels} == {"panel_a", "panel_b"} + assert sum(decision.votes.values()) == 2 # both panels voted + + +def test_blocked_when_reachable_high_severity_but_unverified() -> None: + decision = fuse_finding(_finding(), _verification(VerificationStatus.NEEDS_EVIDENCE, verified=False), ["gates_risky_fix"]) + assert decision.disposition == FusionDisposition.BLOCKED + + +def test_rejected_when_verifier_rejects_unreachable() -> None: + decision = fuse_finding( + _finding(reachability="unknown"), _verification(VerificationStatus.REJECTED, verified=False), ["severity:critical"] + ) + assert decision.disposition == FusionDisposition.REJECTED + + +def test_deferred_when_evidence_conflicts() -> None: + decision = fuse_finding(_finding(severity="low"), _verification(VerificationStatus.REJECTED, verified=False), ["evidence_conflict"]) + assert decision.disposition == FusionDisposition.DEFERRED + + +def test_mitigated_when_finding_is_known_mitigated() -> None: + finding = _finding() + decision = fuse_finding( + finding, + _verification(VerificationStatus.ACCEPTED, verified=True), + ["severity:critical"], + mitigated_ids=frozenset({finding.finding_id}), + ) + assert decision.disposition == FusionDisposition.MITIGATED + + +# ---- dispatch + disclosure + serialization ---------------------------------- + + +def test_dispatch_only_fuses_triggered_findings() -> None: + findings = [ + _finding(finding_id="hot:app.py:3", severity="critical"), + _finding(finding_id="cold:lib.py:9", severity="low", reachability="unknown"), + ] + decisions = fuse_findings_dispatch(findings, []) + assert [d.finding_id for d in decisions] == ["hot:app.py:3"] + + +def test_dispatch_records_degradation_when_llm_requested_but_extra_absent() -> None: + if has_harnessx(): + pytest.skip("HarnessX extra installed; exercises the extra-absent degradation path") + decisions = fuse_findings_dispatch([_finding()], [], panel_model="gpt-4o", use_harnessx=True) + assert decisions and decisions[0].decision_source == "deterministic-reconciler" + assert any("fusion_llm_unavailable" in note for note in decisions[0].degradations) + + +def test_decision_serializes_with_full_disclosure() -> None: + decision = fuse_finding(_finding(), _verification(VerificationStatus.ACCEPTED, verified=True), ["severity:critical"]) + payload = json.loads(json.dumps(decision.to_dict())) + assert payload["disposition"] == "accepted" + assert payload["decision_source"] == "deterministic-reconciler" + assert "votes" in payload and "model_ids" in payload and "degradations" in payload + assert all("leaning" in panel and "disposition" in panel for panel in payload["panels"]) + + +# ---- CLI integration: standard-mode fusion stage ---------------------------- + + +def test_standard_scan_emits_fusion_artifact_and_manifest_block(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + from openultrasast.cli import main + + repo = tmp_path / "repo" + repo.mkdir() + (repo / "app.py").write_text("@app.route('/admin')\ndef admin():\n return eval(request.data)\n") + monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs") + assert main(["scan", str(repo), "--mode", "standard"]) == 0 + + run_dir = sorted((repo / ".runs").iterdir())[-1] + fusion = json.loads((run_dir / "fusion.json").read_text()) + assert fusion and fusion[0]["disposition"] in {"accepted", "blocked", "deferred", "rejected", "mitigated"} + manifest = json.loads((run_dir / "manifest.json").read_text()) + assert manifest["fusion"][0]["finding_id"] == fusion[0]["finding_id"] + assert "fusion" in manifest["artifacts"] + + +def test_quick_scan_does_not_emit_fusion(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + from openultrasast.cli import main + + repo = tmp_path / "repo" + repo.mkdir() + (repo / "app.py").write_text("@app.route('/admin')\ndef admin():\n return eval(request.data)\n") + monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs") + assert main(["scan", str(repo), "--mode", "quick"]) == 0 + + run_dir = sorted((repo / ".runs").iterdir())[-1] + assert not (run_dir / "fusion.json").exists() + assert "fusion" not in json.loads((run_dir / "manifest.json").read_text()) diff --git a/tests/test_gate.py b/tests/test_gate.py index 5fdba97..9ce9f48 100644 --- a/tests/test_gate.py +++ b/tests/test_gate.py @@ -146,6 +146,7 @@ GOVERNANCE_SCORING_BENCHMARK_PLANES = ( "openultrasast.improve.evolve", "openultrasast.slot_contract", "openultrasast.stage_processors", + "openultrasast.fusion", )