scriptoriumadmin/strix
1
0
Fork 0
mirror of https://github.com/usestrix/strix.git synced 2026-11-11 08:37:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

feat(report): opt-in verify-before-emit pass for findings

Browse source 
Adds `STRIX_VERIFY`: a per-finding adjudication step that runs just before a
report is persisted — a sibling of the existing dedupe reject in
`_do_create`. A (typically cheaper) model re-derives whether the candidate is
REAL or a FALSE_POSITIVE, and a confident false positive is dropped before it
reaches the report state.

The pass is off by default and deliberately fail-open + asymmetric, so it can
never lose a real finding:

  - only a FALSE_POSITIVE verdict at/above STRIX_VERIFY_MIN_CONFIDENCE (default
    0.8) suppresses; REAL, uncertain, below-severity, unparseable, no-model, or
    any error all EMIT
  - only findings at/above STRIX_VERIFY_MIN_SEVERITY (default high) are verified
  - raising min_confidence only makes it more conservative (emits more)

Mirrors the dedupe module's shape (StrixProvider.get_model + get_response, JSON
verdict, usage recorded via record_sdk_usage) and reuses its settings pattern
(dedicated STRIX_VERIFY_MODEL with its own creds/headers, falling back to the
main model). No new dependencies.

17 unit tests cover the severity gate, verdict parsing, the confidence
threshold in both directions, and the fail-open paths; existing reporting and
dedupe suites still pass.
This commit is contained in:
Sean Turner 2026-08-06 23:02:19 +01:00
parent b08662449d
commit 312c774802
No known key found for this signature in database
4 changed files with 472 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

34
strix/config/settings.py
View file

@ -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)

229
strix/report/verify.py Normal file
View file

@ -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": "<one sentence>"}
"""
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]}

22
strix/tools/reporting/tool.py
View file

@ -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,

188
tests/test_verify.py Normal file
View file

@ -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"