mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 03:49:26 +00:00
Add a [harnessx] config block so the agentic plane can be steered from
openultrasast.toml, and make the LLM provider selectable (the hunter and verifier
previously hardcoded AnthropicProvider).
- config.HarnessxConfig: provider ("anthropic"|"openai"|"litellm", default
anthropic), max_cost_usd (2.0), token_threshold (120000); parsed from [harnessx].
- harness_ext.build_provider: single lazy, capability-guarded provider factory.
The provider name is validated before any import, so an unknown choice fails with
a clear ValueError even when the extra is absent; each provider reads its own
standard key env var (openai -> OPENAI_API_KEY, anthropic -> ANTHROPIC_API_KEY).
- HxScanOrchestrator and verify_findings_dispatch take a provider and route through
build_provider; cli threads config.harnessx.{provider,max_cost_usd,token_threshold}
into both the hunter and the verifier.
- README: "Configuring the HarnessX agentic plane" section (install, models,
provider, budgets, fallback/degradation).
Test isolation: convert the "harnessx not in sys.modules" import-guards to a
hermetic subprocess fixture (tests/conftest.py assert_cold_of_harnessx) and
simulate extra-absent in the CLI dispatch tests, so the suite is deterministic
whether or not the optional extra is installed (it now is, locally). Verified the
OpenAI path constructs end-to-end (build_provider -> OpenAIProvider; orchestrator
-> Harness) with the budget threaded through.
137 passed, 3 skipped; ruff + mypy clean; detection gate PASS.
Co-Authored-By: Claude Opus 4.8 <[email protected]>
71 lines
3.7 KiB
Python
71 lines
3.7 KiB
Python
"""Verifier dispatcher: structural fallback (default) + judge verdict mapping (Phase 3 task 6.4)."""
|
|
|
|
from openultrasast.findings import StaticFinding
|
|
from openultrasast.verification import VerificationStatus, verify_findings
|
|
from openultrasast.verify_judge import _verdict_to_verification, verify_findings_dispatch
|
|
|
|
|
|
def _finding(reachability: str = "reachable") -> StaticFinding:
|
|
return StaticFinding(
|
|
finding_id="python-unsafe-eval:app.py:3",
|
|
path="app.py",
|
|
title="Dynamic Python execution needs review",
|
|
severity="critical",
|
|
confidence="medium",
|
|
evidence_level="static_corroboration",
|
|
rationale="Static pattern matched.",
|
|
line=3,
|
|
function_name="admin" if reachability == "reachable" else None,
|
|
reachability_status=reachability,
|
|
reachability_evidence=[{"kind": "route", "access_level": "public", "line": 1, "end_line": 3}]
|
|
if reachability == "reachable"
|
|
else [],
|
|
reachability_conditions=[],
|
|
tags=["syscall_entry"],
|
|
ranking_priority=3.0,
|
|
)
|
|
|
|
|
|
def test_dispatch_falls_back_to_structural_by_default() -> None:
|
|
findings = [_finding("reachable"), _finding("unknown")]
|
|
assert verify_findings_dispatch(findings) == verify_findings(findings)
|
|
|
|
|
|
def test_dispatch_stays_structural_when_model_unset_or_extra_absent() -> None:
|
|
findings = [_finding("reachable")]
|
|
# use_harnessx requested but no model -> structural; and no model -> structural.
|
|
assert verify_findings_dispatch(findings, verifier_model=None, use_harnessx=True) == verify_findings(findings)
|
|
assert verify_findings_dispatch(findings, verifier_model="claude-x", use_harnessx=False) == verify_findings(findings)
|
|
|
|
|
|
def test_dispatch_does_not_import_harnessx_on_fallback(assert_cold_of_harnessx) -> None: # type: ignore[no-untyped-def]
|
|
# Even with a model and use_harnessx=True, when the capability is absent the
|
|
# dispatch must fall back without importing the extra. Simulate absence so the
|
|
# guarantee holds whether or not the extra is installed in this environment.
|
|
assert_cold_of_harnessx(
|
|
"import openultrasast.verify_judge as vj\n"
|
|
"vj.has_harnessx = lambda: False\n"
|
|
"from openultrasast.findings import StaticFinding\n"
|
|
"f = StaticFinding(finding_id='x:app.py:1', path='app.py', title='t', severity='low',\n"
|
|
" confidence='low', evidence_level='static_corroboration', rationale='r', line=1,\n"
|
|
" function_name=None, reachability_status='unknown', reachability_evidence=[],\n"
|
|
" reachability_conditions=[], tags=[], ranking_priority=1.0)\n"
|
|
"vj.verify_findings_dispatch([f], verifier_model='claude-x', use_harnessx=True)\n"
|
|
)
|
|
|
|
|
|
def test_verdict_mapping_is_evidence_gated() -> None:
|
|
finding = _finding("reachable")
|
|
accepted = _verdict_to_verification(finding, {"verdict": "plausible", "confidence": 0.9, "cause": "tainted sink reached"})
|
|
low_conf = _verdict_to_verification(finding, {"verdict": "plausible", "confidence": 0.3})
|
|
rejected = _verdict_to_verification(finding, {"verdict": "unsupported", "cause": "input is constant"})
|
|
hedging = _verdict_to_verification(finding, {"verdict": "hedging"})
|
|
infra = _verdict_to_verification(finding, {"verdict": "no_answer"})
|
|
|
|
assert accepted.status == VerificationStatus.ACCEPTED and accepted.verified
|
|
assert accepted.pro_case == "tainted sink reached"
|
|
assert low_conf.status == VerificationStatus.NEEDS_EVIDENCE
|
|
assert rejected.status == VerificationStatus.REJECTED and rejected.counter_case == "input is constant"
|
|
assert hedging.status == VerificationStatus.NEEDS_EVIDENCE
|
|
assert infra.status == VerificationStatus.NEEDS_EVIDENCE # infra failure never auto-rejects
|