mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 03:49:26 +00:00
refactor(harnessx-removal): retire the HarnessX scan paths (task 5)
The standard scan now runs run_hunter_pool, verify_findings and the deterministic fuse_findings directly; the HarnessX hunter orchestrator, the LLM judge dispatcher (verify_judge.py, deleted with its test) and the LLM fusion panels are gone, and so are their extra-absent degradations. host_under_harnessx is dropped from stage_processors. config: [models] verifier and [fusion] panel_model/decider_model raise RetiredConfigError naming the plane replacement (the CLI prints it and exits 2); a [harnessx] section loads with one warning and is otherwise ignored. harness.json keeps "verifier": null. Tests that asserted the fallback are renamed to the behaviour they assert; the module-audit manifest drops verify_judge. Nothing outside the two modules deleted in task 6 imports harness_ext or hunter_harness. Co-Authored-By: Claude Opus 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo
This commit is contained in:
Marius Ciepluch
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
b68fe27c74
commit
cbadc38d5e
@@ -1,65 +0,0 @@
|
||||
"""CLI capability-gated dispatch + graceful degradation (Phase 3 task 6.5).
|
||||
|
||||
Simulates the extra-absent environment (via a monkeypatched capability probe so the
|
||||
behaviour is deterministic whether or not the optional extra is installed): standard
|
||||
mode with models configured must fall back to the deterministic hunter/structural
|
||||
verifier and record a degradation; with no models configured the manifest stays
|
||||
byte-identical.
|
||||
"""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from openultrasast import cli
|
||||
from openultrasast.cli import main
|
||||
|
||||
_VULN = "@app.route('/admin')\ndef admin():\n return eval(request.data)\n"
|
||||
|
||||
|
||||
def _run(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, config_body: str | None) -> dict:
|
||||
repo = tmp_path / "repo"
|
||||
repo.mkdir()
|
||||
(repo / "app.py").write_text(_VULN)
|
||||
monkeypatch.setattr(cli, "has_harnessx", lambda: False) # simulate the extra being absent
|
||||
monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs")
|
||||
monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
|
||||
monkeypatch.delenv("OPENULTRASAST_HUNTER_CLIENT", raising=False)
|
||||
argv = ["scan", str(repo), "--mode", "standard"]
|
||||
if config_body is not None:
|
||||
cfg = tmp_path / "openultrasast.toml"
|
||||
cfg.write_text(config_body)
|
||||
argv += ["--config", str(cfg)]
|
||||
assert main(argv) == 0
|
||||
run_dir = sorted((repo / ".runs").iterdir())[-1]
|
||||
return json.loads((run_dir / "manifest.json").read_text())
|
||||
|
||||
|
||||
def test_degradation_recorded_when_models_set_but_extra_absent(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
manifest = _run(tmp_path, monkeypatch, '[models]\nhunter = "anthropic/claude"\nverifier = "anthropic/claude"\n')
|
||||
|
||||
degradations = manifest.get("degradations", [])
|
||||
stages = {entry["stage"] for entry in degradations}
|
||||
# `model` is present because the model layer records that no CPG engine was available rather than being
|
||||
# silently absent (contributor-scan Req 1.3). This test is about the HarnessX extra, so it asserts over
|
||||
# the HarnessX stages specifically instead of the whole set.
|
||||
assert stages == {"hunter_pool", "verify", "model"}
|
||||
assert all(entry["reason"] == "harnessx_extra_unavailable" for entry in degradations if entry["stage"] != "model")
|
||||
# Fallback still produced findings (deterministic hunter ran).
|
||||
assert manifest["findings"]
|
||||
|
||||
|
||||
def test_manifest_has_no_harnessx_degradations_without_models(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
manifest = _run(tmp_path, monkeypatch, config_body=None)
|
||||
degradations = manifest.get("degradations", [])
|
||||
assert all(entry.get("reason") != "harnessx_extra_unavailable" for entry in degradations)
|
||||
assert any(entry.get("reason") == "hunter_model_unavailable" for entry in degradations)
|
||||
|
||||
|
||||
def test_standard_scan_falls_back_to_run_hunter_pool(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
calls: list[str] = []
|
||||
real = cli.run_hunter_pool
|
||||
monkeypatch.setattr(cli, "run_hunter_pool", lambda *a, **k: calls.append("fallback") or real(*a, **k))
|
||||
_run(tmp_path, monkeypatch, '[models]\nhunter = "anthropic/claude"\n')
|
||||
assert calls == ["fallback"]
|
||||
+47
-27
@@ -1,7 +1,11 @@
|
||||
import logging
|
||||
import os
|
||||
from dataclasses import replace
|
||||
from pathlib import Path
|
||||
|
||||
from openultrasast.config import load_config, load_dotenv, write_resolved_config
|
||||
import pytest
|
||||
|
||||
from openultrasast.config import RetiredConfigError, load_config, load_dotenv, write_resolved_config
|
||||
|
||||
|
||||
def test_load_config_reads_toml(tmp_path: Path) -> None:
|
||||
@@ -29,48 +33,64 @@ def test_load_config_reads_toml(tmp_path: Path) -> None:
|
||||
assert config.dynamic.network_scope == ("127.0.0.1:8080",)
|
||||
|
||||
|
||||
def test_harnessx_defaults_when_section_absent() -> None:
|
||||
config = load_config(None)
|
||||
assert config.harnessx.provider == "anthropic"
|
||||
assert config.harnessx.max_cost_usd == 2.0
|
||||
assert config.harnessx.token_threshold == 120_000
|
||||
|
||||
|
||||
def test_load_config_reads_harnessx_section(tmp_path: Path) -> None:
|
||||
def test_retired_agentic_section_loads_with_one_warning(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None:
|
||||
# retired 2026-09-30: a budget-only [harnessx] section loads with one warning naming the plane (Req 3.3)
|
||||
config_path = tmp_path / "openultrasast.toml"
|
||||
config_path.write_text(
|
||||
"\n".join(
|
||||
[
|
||||
"[harnessx]",
|
||||
'provider = "openai"',
|
||||
"max_cost_usd = 5.0",
|
||||
"token_threshold = 200000",
|
||||
]
|
||||
)
|
||||
)
|
||||
config_path.write_text('[harnessx]\nprovider = "openai"\nmax_cost_usd = 5.0\n[score]\nmin_score = 70\n') # retired 2026-09-30
|
||||
|
||||
config = load_config(config_path)
|
||||
with caplog.at_level(logging.WARNING, logger="openultrasast.config"):
|
||||
config = load_config(config_path)
|
||||
|
||||
assert config.harnessx.provider == "openai"
|
||||
assert config.harnessx.max_cost_usd == 5.0
|
||||
assert config.harnessx.token_threshold == 200_000
|
||||
records = [record for record in caplog.records if record.name == "openultrasast.config"]
|
||||
assert len(records) == 1
|
||||
assert "ax plane" in records[0].getMessage() and "ops/ax/README.md" in records[0].getMessage()
|
||||
assert config.score.min_score == 70 # the rest of the file still loads
|
||||
assert config == replace(load_config(None), score=config.score) # and nothing else is read from the section
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("body", "key", "replacement"),
|
||||
[
|
||||
('[models]\nverifier = "gpt-4o"\n', "`[models] verifier`", "`verify` + `agree`"),
|
||||
('[fusion]\npanel_model = "gpt-4o"\n', "`[fusion] panel_model`", "`agree` task"),
|
||||
('[fusion]\ndecider_model = "gpt-4o"\n', "`[fusion] decider_model`", "`agree` task"),
|
||||
],
|
||||
)
|
||||
def test_retired_llm_keys_fail_naming_the_replacement(tmp_path: Path, body: str, key: str, replacement: str) -> None:
|
||||
config_path = tmp_path / "openultrasast.toml"
|
||||
config_path.write_text(body)
|
||||
|
||||
with pytest.raises(RetiredConfigError) as raised:
|
||||
load_config(config_path)
|
||||
|
||||
message = str(raised.value)
|
||||
assert key in message and replacement in message and "ousast plane run" in message and "Remove the key" in message
|
||||
|
||||
|
||||
def test_the_cli_exits_2_on_a_retired_key(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
from openultrasast.cli import main
|
||||
|
||||
(tmp_path / "repo").mkdir()
|
||||
config_path = tmp_path / "openultrasast.toml"
|
||||
config_path.write_text('[models]\nverifier = "gpt-4o"\n')
|
||||
|
||||
assert main(["scan", str(tmp_path / "repo"), "--config", str(config_path)]) == 2
|
||||
assert "`[models] verifier`" in capsys.readouterr().err
|
||||
|
||||
|
||||
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_path.write_text("[fusion]\nenabled = false\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.enabled is False
|
||||
assert config.fusion.high_assurance is True
|
||||
|
||||
|
||||
|
||||
+2
-11
@@ -9,11 +9,10 @@ from openultrasast.findings import StaticFinding
|
||||
from openultrasast.fusion import (
|
||||
FusionDisposition,
|
||||
fuse_finding,
|
||||
fuse_findings_dispatch,
|
||||
fuse_findings,
|
||||
fusion_triggers,
|
||||
should_fuse,
|
||||
)
|
||||
from openultrasast.harness_ext import has_harnessx
|
||||
from openultrasast.verification import EvidenceLevel, VerificationResult, VerificationStatus
|
||||
|
||||
|
||||
@@ -135,18 +134,10 @@ def test_dispatch_only_fuses_triggered_findings() -> None:
|
||||
_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, [])
|
||||
decisions = fuse_findings(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()))
|
||||
|
||||
@@ -141,7 +141,6 @@ def test_deep_scan_unsafe_hunter_snippet_is_safety_rejected_without_docker_job(t
|
||||
monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs")
|
||||
monkeypatch.setenv("OPENULTRASAST_SANDBOX_PROBE", "1")
|
||||
monkeypatch.setenv("OPENULTRASAST_HUNTER_CLIENT", "unsafe-snippet")
|
||||
monkeypatch.setattr(cli, "has_harnessx", lambda: False)
|
||||
monkeypatch.setattr(cli, "resolve_sandbox_runner", lambda: fake)
|
||||
docker_argv = _guard_docker(monkeypatch)
|
||||
|
||||
|
||||
@@ -1,12 +1,10 @@
|
||||
"""Scan-pipeline integrity & serialization (Phase 4 task 9).
|
||||
|
||||
9.1 — with the extra absent, quick and standard scans run through the existing sync
|
||||
driver with behaviour unchanged from the deterministic baseline, and a
|
||||
requested-but-unavailable HarnessX stage records a manifest degradation.
|
||||
9.1 — quick and standard scans run through the sync driver with the deterministic
|
||||
hunter, verifier and fusion; a configured hunter model never changes the
|
||||
deterministic findings.
|
||||
9.2 — every persisted scan-state slot serializes and restores without loss (guards
|
||||
against a non-serializable slot silently restoring as empty). This covers the
|
||||
current artifact-persistence model; the HarnessX processor slot-contract variant
|
||||
(spec task 9.2 over task 6.3) remains deferred with 6.3.
|
||||
against a non-serializable slot silently restoring as empty).
|
||||
"""
|
||||
|
||||
import json
|
||||
@@ -15,7 +13,6 @@ from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from openultrasast import cli
|
||||
from openultrasast.benchmark import load_findings
|
||||
from openultrasast.cli import main
|
||||
from openultrasast.findings import StaticFinding, write_findings
|
||||
@@ -27,14 +24,13 @@ from openultrasast.verification import EvidenceLevel, VerificationResult, Verifi
|
||||
_VULN = "@app.route('/admin')\ndef admin():\n return eval(request.data)\n"
|
||||
|
||||
|
||||
# ---- 9.1 fallback parity + degradation --------------------------------------
|
||||
# ---- 9.1 deterministic parity + degradation --------------------------------------
|
||||
|
||||
|
||||
def _scan(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, *, mode: str, config_body: str | None, harnessx: bool) -> tuple[list, dict]:
|
||||
def _scan(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, *, mode: str, config_body: str | None) -> tuple[list, dict]:
|
||||
repo = tmp_path / "repo"
|
||||
repo.mkdir(parents=True)
|
||||
(repo / "app.py").write_text(_VULN)
|
||||
monkeypatch.setattr(cli, "has_harnessx", lambda: harnessx)
|
||||
monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs")
|
||||
monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
|
||||
monkeypatch.delenv("OPENULTRASAST_HUNTER_CLIENT", raising=False)
|
||||
@@ -50,40 +46,27 @@ def _scan(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, *, mode: str, config_
|
||||
return findings, manifest
|
||||
|
||||
|
||||
def test_standard_scan_extra_absent_matches_deterministic_baseline(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
# Requested-but-unavailable HarnessX (models set, extra absent) must produce the
|
||||
# same findings as the plain deterministic path (no models), differing only by
|
||||
# the recorded degradation.
|
||||
requested, requested_manifest = _scan(
|
||||
tmp_path / "a",
|
||||
monkeypatch,
|
||||
mode="standard",
|
||||
config_body='[models]\nhunter = "openai/gpt-4o"\nverifier = "openai/gpt-4o"\n',
|
||||
harnessx=False,
|
||||
)
|
||||
baseline, baseline_manifest = _scan(tmp_path / "b", monkeypatch, mode="standard", config_body=None, harnessx=False)
|
||||
def test_standard_scan_with_a_hunter_model_keeps_deterministic_findings(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
# A configured hunter model (no client reachable here) must leave the deterministic findings byte-identical.
|
||||
requested, requested_manifest = _scan(tmp_path / "a", monkeypatch, mode="standard", config_body='[models]\nhunter = "openai/gpt-4o"\n')
|
||||
baseline, baseline_manifest = _scan(tmp_path / "b", monkeypatch, mode="standard", config_body=None)
|
||||
|
||||
assert requested == baseline # byte-identical findings -> sync driver behaviour unchanged
|
||||
stages = {entry["stage"] for entry in requested_manifest.get("degradations", [])}
|
||||
# `model` joins the set when the model layer runs without a CPG engine: findings are byte-identical
|
||||
# (asserted above) and the only difference is that the scan now SAYS the arbiter was unavailable
|
||||
# (contributor-scan Req 1.3) rather than being silently absent.
|
||||
assert stages == {"hunter_pool", "verify", "model"}
|
||||
assert {e["reason"] for e in requested_manifest["degradations"] if e["stage"] == "model"} == {"cpg_unavailable"}
|
||||
assert all(entry["reason"] == "harnessx_extra_unavailable" for entry in requested_manifest["degradations"] if entry["stage"] != "model")
|
||||
assert all(entry["reason"] != "hunter_model_unavailable" for entry in requested_manifest["degradations"])
|
||||
assert [e["reason"] for e in baseline_manifest.get("degradations", []) if e["stage"] != "model"] == ["hunter_model_unavailable"]
|
||||
assert requested == baseline
|
||||
# The hunter model runs the tool hunter (its client is absent here, so it contributes nothing) and removes the
|
||||
# `map` skip; `model` records that no CPG engine was available (contributor-scan Req 1.3).
|
||||
assert requested_manifest["degradations"] == [{"reason": "cpg_unavailable", "stage": "model"}]
|
||||
assert {(e["stage"], e["reason"]) for e in baseline_manifest["degradations"]} == {
|
||||
("model", "cpg_unavailable"),
|
||||
("map", "hunter_model_unavailable"),
|
||||
}
|
||||
|
||||
|
||||
def test_quick_scan_is_unaffected_by_the_extra(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
# Quick mode never touches the agentic plane: identical findings whether the
|
||||
# extra is present or absent, and never a degradation.
|
||||
present, present_manifest = _scan(tmp_path / "a", monkeypatch, mode="quick", config_body=None, harnessx=True)
|
||||
absent, _ = _scan(tmp_path / "b", monkeypatch, mode="quick", config_body=None, harnessx=False)
|
||||
def test_quick_scan_records_no_degradation(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
# Quick mode never touches a model: findings are produced and no degradation is recorded.
|
||||
findings, manifest = _scan(tmp_path, monkeypatch, mode="quick", config_body=None)
|
||||
|
||||
assert present == absent
|
||||
assert present # findings were produced
|
||||
assert "degradations" not in present_manifest
|
||||
assert findings
|
||||
assert "degradations" not in manifest
|
||||
|
||||
|
||||
# ---- 9.2 per-slot serialization round-trip ----------------------------------
|
||||
|
||||
@@ -2,8 +2,7 @@
|
||||
|
||||
Proves: deterministic stages carry work in declared slots, a slot-contract violation
|
||||
is raised on undeclared slot access, the pipeline is model-disabled, the slot pipeline
|
||||
reproduces the deterministic quick-scan path byte-for-byte, lifecycle hooks fire, and
|
||||
(when the extra is present) a stage can run under HarnessX with the model disabled.
|
||||
reproduces the deterministic quick-scan path byte-for-byte, and lifecycle hooks fire.
|
||||
"""
|
||||
|
||||
from collections.abc import Mapping
|
||||
@@ -13,13 +12,10 @@ from typing import Any
|
||||
import pytest
|
||||
|
||||
from openultrasast.gate import _quick_scan
|
||||
from openultrasast.harness_ext import has_harnessx
|
||||
from openultrasast.slot_contract import SlotContractError, SlotContractMixin, SlotPipeline
|
||||
from openultrasast.stage_processors import (
|
||||
PreprocessProcessor,
|
||||
build_deterministic_pipeline,
|
||||
deterministic_processors,
|
||||
host_under_harnessx,
|
||||
run_quick_pipeline,
|
||||
)
|
||||
|
||||
@@ -110,24 +106,3 @@ def test_pipeline_emits_lifecycle_hooks_per_stage() -> None:
|
||||
for name in names:
|
||||
assert ("stage_start", name) in recorder.events
|
||||
assert ("stage_end", name) in recorder.events
|
||||
|
||||
|
||||
# ---- optional: run under HarnessX with the model disabled (1.6) --------------
|
||||
|
||||
|
||||
@pytest.mark.skipif(not has_harnessx(), reason="requires the harnessx extra")
|
||||
def test_host_under_harnessx_disables_the_model() -> None: # pragma: no cover - needs the extra
|
||||
import asyncio
|
||||
|
||||
from harnessx.core.events import BeforeModelEvent
|
||||
from harnessx.core.processor import MultiHookProcessor
|
||||
|
||||
hosted = host_under_harnessx(PreprocessProcessor())
|
||||
assert isinstance(hosted, MultiHookProcessor)
|
||||
|
||||
async def drive() -> list[Any]:
|
||||
event = BeforeModelEvent(run_id="test", step_id=0)
|
||||
return [out async for out in hosted.on_before_model(event)]
|
||||
|
||||
emitted = asyncio.run(drive())
|
||||
assert emitted and emitted[-1].skip_model is True
|
||||
|
||||
@@ -219,14 +219,13 @@ def _guard_docker(monkeypatch: pytest.MonkeyPatch) -> list[list[str]]:
|
||||
|
||||
|
||||
def test_standard_scripted_hunter_emits_tool_hunter_suspicion(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
from openultrasast import cli, tool_hunter
|
||||
from openultrasast import tool_hunter
|
||||
from openultrasast.cli import main
|
||||
|
||||
repo = _standard_repo(tmp_path)
|
||||
config = _write_hunter_config(tmp_path)
|
||||
monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs")
|
||||
monkeypatch.setenv("OPENULTRASAST_HUNTER_CLIENT", "scripted")
|
||||
monkeypatch.setattr(cli, "has_harnessx", lambda: False)
|
||||
docker_argv = _guard_docker(monkeypatch)
|
||||
hunter_calls: list[int] = []
|
||||
real_hunter = tool_hunter.run_tool_hunter
|
||||
@@ -256,14 +255,13 @@ def test_standard_scripted_hunter_emits_tool_hunter_suspicion(tmp_path: Path, mo
|
||||
|
||||
|
||||
def test_dump_only_hunter_client_emits_no_hunter_findings(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
from openultrasast import cli, tool_hunter
|
||||
from openultrasast import tool_hunter
|
||||
from openultrasast.cli import main
|
||||
|
||||
repo = _standard_repo(tmp_path)
|
||||
config = _write_hunter_config(tmp_path)
|
||||
monkeypatch.setenv("OPENULTRASAST_RUNS_DIR", ".runs")
|
||||
monkeypatch.setenv("OPENULTRASAST_HUNTER_CLIENT", "dump-only")
|
||||
monkeypatch.setattr(cli, "has_harnessx", lambda: False)
|
||||
docker_argv = _guard_docker(monkeypatch)
|
||||
hunter_calls: list[int] = []
|
||||
real_hunter = tool_hunter.run_tool_hunter
|
||||
|
||||
@@ -1,70 +0,0 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user