diff --git a/strix/config/settings.py b/strix/config/settings.py index 42a2c97ea..fae6cdb5e 100644 --- a/strix/config/settings.py +++ b/strix/config/settings.py @@ -82,6 +82,39 @@ class DedupeSettings(BaseSettings): ) +class VerifySettings(BaseSettings): + """Opt-in verify-before-emit pass. + + When enabled, each candidate finding at or above ``min_severity`` is + re-adjudicated by a (typically cheaper) model just before it is persisted — + a sibling of the existing dedupe reject. The pass is deliberately + asymmetric and fail-open: only a high-confidence FALSE_POSITIVE verdict + suppresses a report; REAL, uncertain, unparseable, below-threshold, or any + error all emit, so a verifier miss can never drop a real finding. + """ + + model_config = _BASE_CONFIG + + enabled: bool = Field(default=False, alias="STRIX_VERIFY") + model: str | None = Field(default=None, alias="STRIX_VERIFY_MODEL") + reasoning_effort: ReasoningEffort | None = Field( + default=None, + alias="STRIX_VERIFY_REASONING_EFFORT", + ) + # Only findings at/above this severity are verified — low/info findings are + # cheap to triage by hand and rarely worth an extra model round-trip. + min_severity: str = Field(default="high", alias="STRIX_VERIFY_MIN_SEVERITY") + # A FALSE_POSITIVE verdict must clear this confidence to actually suppress. + min_confidence: float = Field(default=0.8, ge=0.0, le=1.0, alias="STRIX_VERIFY_MIN_CONFIDENCE") + api_key: str | None = Field(default=None, alias="VERIFY_LLM_API_KEY", repr=False) + api_base: str | None = Field(default=None, alias="VERIFY_LLM_API_BASE") + extra_headers: dict[str, str] | None = Field( + default=None, + alias="VERIFY_LLM_EXTRA_HEADERS", + repr=False, + ) + + class ContextSettings(BaseSettings): """Context-window management: per-tool-output caps and history compaction.""" @@ -149,6 +182,7 @@ class Settings(BaseSettings): llm: LlmSettings = Field(default_factory=LlmSettings) dedupe: DedupeSettings = Field(default_factory=DedupeSettings) + verify: VerifySettings = Field(default_factory=VerifySettings) runtime: RuntimeSettings = Field(default_factory=RuntimeSettings) context: ContextSettings = Field(default_factory=ContextSettings) telemetry: TelemetrySettings = Field(default_factory=TelemetrySettings) diff --git a/strix/report/verify.py b/strix/report/verify.py new file mode 100644 index 000000000..d8e2f0b91 --- /dev/null +++ b/strix/report/verify.py @@ -0,0 +1,229 @@ +"""Opt-in verify-before-emit pass for vulnerability findings. + +A sibling of the dedupe reject in ``strix.tools.reporting.tool``: just before a +candidate finding is persisted, a (typically cheaper) model re-adjudicates it +and may reject a confident false positive. The pass is off by default and +**fail-open + asymmetric** — only a FALSE_POSITIVE verdict at or above +``min_confidence`` suppresses the report. REAL, uncertain, below-threshold, +unparseable, no-model, or any error all EMIT, so a verifier miss can never drop +a real finding. +""" + +from __future__ import annotations + +import json +import logging +from typing import TYPE_CHECKING, Any + +from agents.model_settings import ModelSettings +from agents.models.interface import ModelTracing +from openai.types.responses import ResponseOutputMessage + +from strix.config import load_settings +from strix.config.models import StrixProvider, configure_sdk_model_defaults +from strix.core.inputs import make_model_settings +from strix.report.state import get_global_report_state + + +if TYPE_CHECKING: + from agents.items import ModelResponse + + from strix.config.settings import VerifySettings + +logger = logging.getLogger(__name__) + +# Severity floor ranking (higher = more severe). A finding is verified only when +# its severity rank is >= the configured min_severity's rank. +_SEVERITY_RANK = { + "info": 0, + "informational": 0, + "low": 1, + "medium": 2, + "high": 3, + "critical": 4, +} + +VERIFY_SYSTEM_PROMPT = """\ +You are a security finding verifier. You are given a single vulnerability \ +finding that another agent has proposed reporting. Decide whether it is a REAL \ +vulnerability or a FALSE_POSITIVE, re-deriving the conclusion from the evidence \ +provided — do not assume the proposing agent was correct. + +Rules: +- A PARTIAL or INCOMPLETE fix is still REAL. If a control only covers some of \ +the dangerous inputs (e.g. a sanitizer that misses an encoding, a path guard \ +bypassable via symlink, a blocklist missing an alias), the finding is REAL. +- The absence of a proof-of-concept is NOT grounds for FALSE_POSITIVE. Judge \ +the code path, not whether an exploit script was attached. +- "A scanner flagged it but it may not be reachable" is NOT, on its own, \ +grounds for FALSE_POSITIVE — say REAL unless you can show the sink is \ +unreachable or already neutralised on every path. +- Be asymmetric: only answer FALSE_POSITIVE with confidence >= 0.8 when the \ +refutation is airtight. On any genuine doubt, answer REAL. + +Respond with ONLY a JSON object: +{"verdict": "REAL" | "FALSE_POSITIVE", "confidence": <0.0-1.0>, "reason": ""} +""" + + +def _verify_extra_args(verify: VerifySettings) -> dict[str, str]: + extra: dict[str, str] = {} + if verify.api_key and verify.api_key.strip(): + extra["api_key"] = verify.api_key.strip() + if verify.api_base and verify.api_base.strip(): + extra["api_base"] = verify.api_base.strip() + return extra + + +def _verify_model_settings( + verify: VerifySettings, model_name: str, request_timeout: float | None +) -> ModelSettings: + llm = load_settings().llm + settings = make_model_settings( + verify.reasoning_effort, + model_name=model_name, + force_required_tool_choice=False, + request_timeout=request_timeout, + # Only forward the main endpoint's headers when the verifier falls back + # to the main model; a dedicated verify model may route elsewhere and + # must not receive the main endpoint's credentials. + extra_headers=verify.extra_headers if verify.model else llm.extra_headers, + ) + extra = _verify_extra_args(verify) + if extra: + settings = settings.resolve(ModelSettings(extra_args=extra)) + return settings + + +def _extract_text(response: ModelResponse) -> str: + parts: list[str] = [] + for item in response.output: + if not isinstance(item, ResponseOutputMessage): + continue + for chunk in item.content: + text = getattr(chunk, "text", None) + if text: + parts.append(text) + return "".join(parts) + + +def _parse_verdict(content: str) -> dict[str, Any]: + text = content.strip() + if text.startswith("```"): + text = text.strip("`") + if text.lower().startswith("json"): + text = text[4:] + text = text.strip() + start = text.find("{") + end = text.rfind("}") + if start == -1 or end == -1 or end <= start: + raise ValueError(f"No JSON object found in verify response: {content[:500]}") + parsed = json.loads(text[start : end + 1]) + + verdict = str(parsed.get("verdict") or "").strip().upper() + reason = str(parsed.get("reason") or "")[:500] + try: + confidence = float(parsed.get("confidence", 0.0)) + except (TypeError, ValueError): + confidence = 0.0 + return {"verdict": verdict, "confidence": confidence, "reason": reason} + + +def _meets_min_severity(severity: str, min_severity: str) -> bool: + have = _SEVERITY_RANK.get((severity or "").strip().lower(), 0) + need = _SEVERITY_RANK.get((min_severity or "high").strip().lower(), 3) + return have >= need + + +async def verify_finding(candidate: dict[str, Any], severity: str) -> dict[str, Any]: + """Adjudicate a candidate finding. + + Returns a dict with ``reject`` (bool), ``verdict``, ``confidence`` and + ``reason``. ``reject`` is True only for a high-confidence FALSE_POSITIVE; + every other outcome — including errors — returns ``reject=False`` so the + finding is emitted (fail-open, FN=0 posture). + """ + settings = load_settings() + verify = settings.verify + + if not verify.enabled: + return {"reject": False, "verdict": "SKIPPED", "confidence": 0.0, "reason": "disabled"} + + if not _meets_min_severity(severity, verify.min_severity): + return { + "reject": False, + "verdict": "SKIPPED", + "confidence": 0.0, + "reason": f"severity {severity!r} below min_severity {verify.min_severity!r}", + } + + model_name = (verify.model or "").strip() or settings.llm.model + if not model_name: + return { + "reject": False, + "verdict": "SKIPPED", + "confidence": 0.0, + "reason": "no LLM model configured; emitting without verification", + } + + try: + user_msg = ( + "Verify this candidate vulnerability finding:\n\n" + f"{json.dumps(candidate, indent=2)}\n\n" + "Respond with ONLY the JSON object described in the system prompt." + ) + + configure_sdk_model_defaults(settings) + resolved_model = model_name.strip() + model = StrixProvider().get_model(resolved_model) + response = await model.get_response( + system_instructions=VERIFY_SYSTEM_PROMPT, + input=user_msg, + model_settings=_verify_model_settings(verify, resolved_model, settings.llm.timeout), + tools=[], + output_schema=None, + handoffs=[], + tracing=ModelTracing.DISABLED, + previous_response_id=None, + conversation_id=None, + prompt=None, + ) + + report_state = get_global_report_state() + if report_state is not None: + report_state.record_sdk_usage( + agent_id="verify", + agent_name="verify", + model=resolved_model, + usage=response.usage, + ) + + content = _extract_text(response) + if not content: + return { + "reject": False, + "verdict": "ERROR", + "confidence": 0.0, + "reason": "empty verifier response; emitting", + } + + parsed = _parse_verdict(content) + reject = ( + parsed["verdict"] == "FALSE_POSITIVE" and parsed["confidence"] >= verify.min_confidence + ) + logger.info( + "verify: verdict=%s confidence=%.2f reject=%s title=%s", + parsed["verdict"], + parsed["confidence"], + reject, + candidate.get("title", ""), + ) + return { + "reject": reject, + "verdict": parsed["verdict"], + "confidence": parsed["confidence"], + "reason": parsed["reason"], + } + except Exception as exc: # noqa: BLE001 — verifier must never drop a finding + logger.warning("verify: adjudication failed (%s); emitting without verification", exc) + return {"reject": False, "verdict": "ERROR", "confidence": 0.0, "reason": str(exc)[:200]} diff --git a/strix/tools/reporting/tool.py b/strix/tools/reporting/tool.py index 12dd20463..f7400f805 100644 --- a/strix/tools/reporting/tool.py +++ b/strix/tools/reporting/tool.py @@ -163,7 +163,7 @@ _REQUIRED_FIELDS = { _VALID_FIX_EFFORT = frozenset({"trivial", "low", "medium", "high"}) -async def _do_create( # noqa: PLR0912 +async def _do_create( # noqa: PLR0912, PLR0911, PLR0915 *, title: str, description: str, @@ -285,6 +285,26 @@ async def _do_create( # noqa: PLR0912 "reason": dedupe.get("reason", ""), } + # Opt-in verify-before-emit (STRIX_VERIFY). A sibling of the dedupe + # reject: re-adjudicate the candidate and drop only a high-confidence + # false positive. Fail-open — anything short of a confident + # FALSE_POSITIVE (including errors) emits, so it can't lose a real find. + from strix.report.verify import verify_finding + + verdict = await verify_finding(candidate, severity) + if verdict.get("reject"): + return { + "success": False, + "error": ( + "Rejected by verify-before-emit as a likely false positive " + f"(confidence={verdict.get('confidence', 0.0):.2f}): " + f"{verdict.get('reason', '')}" + ), + "verify_verdict": verdict.get("verdict", ""), + "confidence": verdict.get("confidence", 0.0), + "reason": verdict.get("reason", ""), + } + report_id = report_state.add_vulnerability_report( title=title, description=description, diff --git a/tests/test_verify.py b/tests/test_verify.py new file mode 100644 index 000000000..5d7289757 --- /dev/null +++ b/tests/test_verify.py @@ -0,0 +1,188 @@ +"""Unit tests for the opt-in verify-before-emit pass (strix.report.verify).""" + +from __future__ import annotations + +import pytest +from pydantic import ValidationError + +from strix.config import loader +from strix.report import verify + + +def _reset_cache() -> None: + loader._cached = None + loader._override = None + + +@pytest.fixture(autouse=True) +def clean_settings(monkeypatch: pytest.MonkeyPatch): + for key in ( + "STRIX_VERIFY", + "STRIX_VERIFY_MODEL", + "STRIX_VERIFY_MIN_SEVERITY", + "STRIX_VERIFY_MIN_CONFIDENCE", + "STRIX_LLM", + ): + monkeypatch.delenv(key, raising=False) + _reset_cache() + yield + _reset_cache() + + +def test_min_confidence_defaults_to_high(): + _reset_cache() + assert loader.load_settings().verify.min_confidence == 0.8 + + +def test_min_confidence_is_configurable(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setenv("STRIX_VERIFY_MIN_CONFIDENCE", "0.95") + _reset_cache() + assert loader.load_settings().verify.min_confidence == 0.95 + + +def test_min_confidence_rejects_out_of_range(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setenv("STRIX_VERIFY_MIN_CONFIDENCE", "1.5") + _reset_cache() + with pytest.raises(ValidationError): + loader.load_settings() + + +def test_verify_disabled_by_default(): + _reset_cache() + assert loader.load_settings().verify.enabled is False + + +def test_severity_gate(): + assert verify._meets_min_severity("critical", "high") + assert verify._meets_min_severity("high", "high") + assert not verify._meets_min_severity("medium", "high") + assert not verify._meets_min_severity("low", "high") + assert not verify._meets_min_severity("info", "high") + # a lower floor lets more through + assert verify._meets_min_severity("medium", "medium") + assert verify._meets_min_severity("low", "info") + + +def test_parse_verdict_plain(): + v = verify._parse_verdict( + '{"verdict": "REAL", "confidence": 0.9, "reason": "guard bypassable"}' + ) + assert v["verdict"] == "REAL" + assert v["confidence"] == 0.9 + assert v["reason"] == "guard bypassable" + + +def test_parse_verdict_fenced(): + v = verify._parse_verdict('```json\n{"verdict": "FALSE_POSITIVE", "confidence": 0.85}\n```') + assert v["verdict"] == "FALSE_POSITIVE" + assert v["confidence"] == 0.85 + + +def test_parse_verdict_bad_confidence_defaults_zero(): + v = verify._parse_verdict('{"verdict": "REAL", "confidence": "high"}') + assert v["confidence"] == 0.0 + + +def test_parse_verdict_no_json_raises(): + with pytest.raises(ValueError): + verify._parse_verdict("I think this is real, sorry no JSON") + + +async def test_disabled_emits(): + # enabled defaults to False → never rejects, no model call. + _reset_cache() + out = await verify.verify_finding({"title": "x"}, "critical") + assert out["reject"] is False + assert out["verdict"] == "SKIPPED" + + +async def test_below_min_severity_skips(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setenv("STRIX_VERIFY", "true") + monkeypatch.setenv("STRIX_LLM", "openai/gpt-x") + _reset_cache() + out = await verify.verify_finding({"title": "x"}, "low") + assert out["reject"] is False + assert out["verdict"] == "SKIPPED" + assert "below min_severity" in out["reason"] + + +async def test_missing_or_unreachable_model_emits(monkeypatch: pytest.MonkeyPatch): + # Enabled + a model name that can't actually be reached (no creds) must never + # reject — the fail-open posture turns any call failure into an EMIT. Whether + # the guard skips (no model resolved) or the call errors, reject stays False. + monkeypatch.setenv("STRIX_VERIFY", "true") + monkeypatch.setenv("STRIX_VERIFY_MODEL", "openai/definitely-not-reachable") + _reset_cache() + out = await verify.verify_finding({"title": "x"}, "critical") + assert out["reject"] is False + assert out["verdict"] in {"SKIPPED", "ERROR"} + + +async def _run_with_stubbed_model( + monkeypatch: pytest.MonkeyPatch, response_text: str, severity: str, min_conf: str | None = None +) -> dict: + """Drive verify_finding with the model call fully stubbed (no network). + + _extract_text is patched to return the canned response, so the fake model's + get_response only needs to be awaitable — its return value is never read. + """ + monkeypatch.setenv("STRIX_VERIFY", "true") + monkeypatch.setenv("STRIX_LLM", "openai/gpt-x") + if min_conf is not None: + monkeypatch.setenv("STRIX_VERIFY_MIN_CONFIDENCE", min_conf) + _reset_cache() + + class _FakeModel: + async def get_response(self, *args: object, **kwargs: object) -> object: # noqa: ARG002 + return object() + + monkeypatch.setattr(verify.StrixProvider, "get_model", lambda self, m: _FakeModel()) # noqa: ARG005 + monkeypatch.setattr(verify, "configure_sdk_model_defaults", lambda s: None) # noqa: ARG005 + monkeypatch.setattr(verify, "_extract_text", lambda r: response_text) # noqa: ARG005 + monkeypatch.setattr(verify, "get_global_report_state", lambda: None) + return await verify.verify_finding({"title": "candidate"}, severity) + + +async def test_confident_false_positive_rejects(monkeypatch: pytest.MonkeyPatch): + out = await _run_with_stubbed_model( + monkeypatch, + '{"verdict": "FALSE_POSITIVE", "confidence": 0.9, "reason": "already patched"}', + "high", + ) + assert out["reject"] is True + assert out["verdict"] == "FALSE_POSITIVE" + + +async def test_low_confidence_false_positive_emits(monkeypatch: pytest.MonkeyPatch): + # Below the 0.8 default → NOT suppressed (FN=0 posture). + out = await _run_with_stubbed_model( + monkeypatch, '{"verdict": "FALSE_POSITIVE", "confidence": 0.55, "reason": "maybe"}', "high" + ) + assert out["reject"] is False + + +async def test_real_verdict_emits(monkeypatch: pytest.MonkeyPatch): + out = await _run_with_stubbed_model( + monkeypatch, + '{"verdict": "REAL", "confidence": 0.99, "reason": "sink reachable"}', + "critical", + ) + assert out["reject"] is False + assert out["verdict"] == "REAL" + + +async def test_raised_min_confidence_makes_it_more_conservative(monkeypatch: pytest.MonkeyPatch): + # Same 0.9 FP verdict that rejects at default 0.8 must EMIT when the floor is 0.95. + out = await _run_with_stubbed_model( + monkeypatch, + '{"verdict": "FALSE_POSITIVE", "confidence": 0.9, "reason": "patched"}', + "high", + min_conf="0.95", + ) + assert out["reject"] is False + + +async def test_unparseable_response_emits(monkeypatch: pytest.MonkeyPatch): + out = await _run_with_stubbed_model(monkeypatch, "no json here", "critical") + assert out["reject"] is False + assert out["verdict"] == "ERROR"