diff --git a/.kiro/specs/harnessx-self-improving-rulesets/spec.json b/.kiro/specs/harnessx-self-improving-rulesets/spec.json index 24694f1..05d93ba 100644 --- a/.kiro/specs/harnessx-self-improving-rulesets/spec.json +++ b/.kiro/specs/harnessx-self-improving-rulesets/spec.json @@ -1,7 +1,7 @@ { "feature_name": "harnessx-self-improving-rulesets", "created_at": "2026-06-22T10:07:18Z", - "updated_at": "2026-06-22T14:25:00Z", + "updated_at": "2026-06-22T15:05:00Z", "language": "en", "phase": "implementation", "approvals": { diff --git a/.kiro/specs/harnessx-self-improving-rulesets/tasks.md b/.kiro/specs/harnessx-self-improving-rulesets/tasks.md index a213708..42fb24a 100644 --- a/.kiro/specs/harnessx-self-improving-rulesets/tasks.md +++ b/.kiro/specs/harnessx-self-improving-rulesets/tasks.md @@ -139,7 +139,7 @@ This plan splits the fused regex layer into three governed owners (rules detect, - _Depends: 5.1_ - [ ] 6. Adopt HarnessX at the hunter loop and verifier -- [ ] 6.1 Run the hunter pool stage on a real HarnessX agent loop +- [x] 6.1 Run the hunter pool stage on a real HarnessX agent loop - Run the hunter pool stage on a real HarnessX run loop with the control, tool, and observability processors registered in the hunter sub-harness, invoked from the existing hunter-pool stage callback. - Confine all asynchronous execution to this single stage so the deterministic stages remain synchronous. - Observable completion: with the extra installed, the hunter pool produces per-target trajectories from a real agent loop while all other stages stay synchronous, verified by an integration test. @@ -147,7 +147,7 @@ This plan splits the fused regex layer into three governed owners (rules detect, - _Boundary: HxScanOrchestrator_ - _Depends: 5.2_ -- [ ] 6.2 Map stage events and identifiers onto HarnessX hooks +- [x] 6.2 Map stage events and identifiers onto HarnessX hooks - Map the existing stage-start/stage-end events to the HarnessX step-level hooks and the scan identifier to the HarnessX run identifier without losing trace continuity. - Observable completion: a HarnessX-backed run preserves trace continuity with stage events mapped to step hooks and the scan id mapped to the run id, verified by inspecting the emitted trace. - _Requirements: 1.7_ @@ -162,7 +162,7 @@ This plan splits the fused regex layer into three governed owners (rules detect, - _Boundary: HxScanOrchestrator_ - _Depends: 6.1_ -- [ ] 6.4 (P) Add the LLM-judge verifier with a structural fallback +- [x] 6.4 (P) Add the LLM-judge verifier with a structural fallback - Produce verification verdicts using the HarnessX llm-judge helpers when the extra is present, mapping the strict-JSON verdict to a verification result. - Retain the existing structural verifier (which rejects on unknown reachability) verbatim as the zero-dependency fallback. - Observable completion: with the extra present the judge produces verdicts and with it absent the structural verifier produces equivalent verdicts, proven by a paired verifier test. @@ -170,7 +170,7 @@ This plan splits the fused regex layer into three governed owners (rules detect, - _Boundary: VerifyProcessor_ - _Depends: 5.2_ -- [ ] 6.5 Fall back gracefully when a HarnessX capability is unavailable +- [x] 6.5 Fall back gracefully when a HarnessX capability is unavailable - When a HarnessX-backed stage is enabled but its capability is unavailable at runtime, fall back to the existing equivalent stage and record the degradation in the scan manifest. - Observable completion: requesting a HarnessX stage whose capability is missing runs the existing equivalent and records a degradation entry in the manifest, proven by a fallback test. - _Requirements: 1.8_ diff --git a/pyproject.toml b/pyproject.toml index 0f6cc84..ecc3fdd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -48,6 +48,12 @@ warn_redundant_casts = true warn_return_any = true warn_unused_configs = true +# HarnessX is an optional extra, absent in the type-check environment; its lazy +# imports inside guarded functions must not fail mypy. +[[tool.mypy.overrides]] +module = "harnessx.*" +ignore_missing_imports = true + [dependency-groups] dev = [ "dagger-io>=0.21.7", diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py index 61e0e61..1060265 100644 --- a/src/openultrasast/cli.py +++ b/src/openultrasast/cli.py @@ -28,7 +28,9 @@ from .calibration import ( from .config import load_config from .findings import StaticFinding, quick_scan_findings, write_findings from .harness import HarnessRuntime, HarnessTraceWriter, write_harness_config +from .harness_ext import has_harnessx from .hunter import run_hunter_pool, write_hunter_trajectories +from .hunter_harness import HxScanOrchestrator from .index import build_code_chunks from .mapping import analyze_entry_points, attach_reachability_hints, ingest_sarif, write_entry_points, write_static_hints from .policy import assert_rules_resolve, load_policy @@ -38,7 +40,8 @@ from .reports import scan_exit_code, write_manifest, write_markdown_report, writ from .ruleset import DEFAULT_RULESET_DIR, load_ruleset from .run import ScanRun, create_scan_run from .scoring import build_score_artifact -from .verification import VerificationResult, verify_findings, write_verification_results +from .verification import VerificationResult, write_verification_results +from .verify_judge import verify_findings_dispatch CALIBRATION_DIR = ".openultrasast/calibration" @@ -97,6 +100,13 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut raise SystemExit(f"scan path is not a directory: {path}") config = load_config(config_path if config_path.exists() else None) + # Capability gates for the optional HarnessX agentic plane. When the extra is + # absent (default/CI), both stay False and the deterministic path runs unchanged. + hunter_model = config.models.hunter + verifier_model = config.models.verifier + harnessx_present = has_harnessx() + hx_hunter = mode == "standard" and bool(hunter_model) and harnessx_present + hx_verify = mode == "standard" and bool(verifier_model) and harnessx_present run = create_scan_run(path, config) runtime = HarnessRuntime( scan_id=run.scan_id, @@ -127,10 +137,23 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut write_rankings(rankings, run.root / "rank" / "ranking.json") write_ranking_calibrations(calibrations, run.root / "calibration" / "applied_calibrations.json") if mode == "standard": - hunter_result = runtime.run_stage( - "hunter_pool", - lambda: run_hunter_pool(run.target, targets, rankings, scan_id=run.scan_id, ruleset=ruleset, policy=policy), - ) + if hx_hunter and hunter_model: + orchestrator = HxScanOrchestrator(provider_model=hunter_model) + hunter_result = runtime.run_stage( + "hunter_pool", + lambda: orchestrator.run_pool( + run.target, targets, rankings, scan_id=run.scan_id, ruleset=ruleset, policy=policy, emit=runtime.emit + ), + ) + else: + if bool(hunter_model) and not harnessx_present: + runtime.state["degradations"].append( + {"stage": "hunter_pool", "requested": "harnessx", "reason": "harnessx_extra_unavailable", "fallback": "run_hunter_pool"} + ) + hunter_result = runtime.run_stage( + "hunter_pool", + lambda: run_hunter_pool(run.target, targets, rankings, scan_id=run.scan_id, ruleset=ruleset, policy=policy), + ) findings = hunter_result.findings trajectories = hunter_result.trajectories else: @@ -162,7 +185,14 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut write_findings(findings, findings_path) if trajectories: write_hunter_trajectories(trajectories, trajectories_path) - verifications = runtime.run_stage("verify", lambda: verify_findings(findings)) + if mode == "standard" and bool(verifier_model) and not harnessx_present: + runtime.state["degradations"].append( + {"stage": "verify", "requested": "harnessx", "reason": "harnessx_extra_unavailable", "fallback": "structural_verifier"} + ) + verifications = runtime.run_stage( + "verify", + lambda: verify_findings_dispatch(findings, verifier_model=verifier_model, use_harnessx=hx_verify), + ) write_verification_results(verifications, verification_path) runtime.run_stage( "record_calibration", @@ -206,6 +236,7 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut ), path=manifest_path, score=score_artifact.to_dict(), + degradations=runtime.state["degradations"] or None, ), ) runtime.finish(status="succeeded") diff --git a/src/openultrasast/harness_ext.py b/src/openultrasast/harness_ext.py index 5f55076..6273ff9 100644 --- a/src/openultrasast/harness_ext.py +++ b/src/openultrasast/harness_ext.py @@ -13,6 +13,10 @@ from __future__ import annotations import importlib import importlib.util from types import ModuleType +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from harnessx.core.harness import HarnessConfig HARNESSX_EXTRA = "openultrasast[harnessx]" @@ -31,3 +35,33 @@ def require_harnessx() -> ModuleType: if not has_harnessx(): raise HarnessXUnavailableError(f"HarnessX is not installed. Install the optional agentic extra: pip install '{HARNESSX_EXTRA}'") return importlib.import_module("harnessx") + + +def build_sast(*, max_cost_usd: float = 2.0, token_threshold: int = 120_000) -> HarnessConfig: + """Compose the SAST hunter HarnessConfig (behaviour pipeline, model-free). + + The single, strictly-localized HarnessX composition site (mirrors DocuHarnessX + ``make_docgen``): every ``harnessx`` import is inside this function so importing + this module never touches the optional extra. Raises :class:`HarnessXUnavailableError` + when the extra is absent. The per-task step budget is enforced on the task at run + time, not composed here. + """ + require_harnessx() + from harnessx.bundles.context import make_context + from harnessx.bundles.control import make_control + from harnessx.bundles.tools import make_tools + from harnessx.core.builder import HarnessBuilder + from harnessx.processors.context.strategies.system_prompt.default import DefaultSystemPromptBuilder + + builder = ( + HarnessBuilder() + | make_context(system_builder=DefaultSystemPromptBuilder()) + | make_tools(skill_loading=False) + | make_control( + include_reliability=True, + include_budget=True, + token_threshold=token_threshold, + max_cost_usd=max_cost_usd, + ) + ) + return builder.build() diff --git a/src/openultrasast/hunter_harness.py b/src/openultrasast/hunter_harness.py new file mode 100644 index 0000000..f122244 --- /dev/null +++ b/src/openultrasast/hunter_harness.py @@ -0,0 +1,186 @@ +"""HarnessX-backed hunter pool — a real LLM agent loop per ranked target. + +A drop-in for :func:`hunter.run_hunter_pool` (same call shape, returns the same +:class:`hunter.HunterPoolResult`) that runs the optional HarnessX agentic plane. It +is lazy/guarded: importing this module never imports ``harnessx``; the extra is only +touched once :meth:`HxScanOrchestrator.run_pool` runs. The agent ``.run()`` path makes +LLM API calls and is therefore exercised only with the extra + credentials present; +the default scan keeps using the deterministic :func:`hunter.run_hunter_pool`. +""" + +from __future__ import annotations + +import json +import re +from collections.abc import Callable +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from .findings import SEVERITY_LABEL, StaticFinding, _reachability_conditions, _reachability_for_line +from .harness_ext import build_sast, require_harnessx +from .hunter import ( + HunterPoolResult, + HunterTask, + HunterTrajectory, + build_retrieval_context_by_path, + schedule_hunter_tasks, +) +from .policy import CwePolicy, load_policy, resolve_severity +from .preprocess import FileTarget +from .rank import RankingScore +from .ruleset import PatternRule + +# The hunter agent is asked to emit findings as a JSON array of these objects. +_FINDINGS_JSON = re.compile(r"\[\s*\{.*\}\s*\]", re.DOTALL) + + +@dataclass(frozen=True) +class ScanTask: + """Thin description of one bounded hunter task (wraps, not subclasses, BaseTask).""" + + description: str + success_criteria: str + max_steps: int + token_budget: int | None + max_cost_usd: float | None + + +class HxScanOrchestrator: + """Runs the hunter pool on a real HarnessX agent loop, one bounded task per target.""" + + def __init__(self, *, provider_model: str, max_cost_usd: float = 2.0, token_threshold: int = 120_000) -> None: + require_harnessx() + self._provider_model = provider_model + self._max_cost_usd = max_cost_usd + self._token_threshold = token_threshold + self._agent: Any = None + + def _build_agent(self) -> Any: + from harnessx.core.model_config import ModelConfig + from harnessx.providers.anthropic_provider import AnthropicProvider + + config = build_sast(max_cost_usd=self._max_cost_usd, token_threshold=self._token_threshold) + return ModelConfig(main=AnthropicProvider(model=self._provider_model)).agentic(config) + + def run_pool( + self, + root: Path, + targets: list[FileTarget], + rankings: list[RankingScore], + *, + scan_id: str, + ruleset: tuple[PatternRule, ...] | None = None, + policy: dict[str, CwePolicy] | None = None, + emit: Callable[..., None] | None = None, + ) -> HunterPoolResult: + import asyncio + + tasks = schedule_hunter_tasks(targets, rankings, retrieval_context_by_path=build_retrieval_context_by_path(root, targets)) + resolved_policy = policy if policy is not None else load_policy() + self._agent = self._agent or self._build_agent() + return asyncio.run(self._run_all(scan_id, tasks, resolved_policy, emit)) + + async def _run_all( + self, + scan_id: str, + tasks: list[HunterTask], + policy: dict[str, CwePolicy], + emit: Callable[..., None] | None, + ) -> HunterPoolResult: + from harnessx.core.harness import BaseTask + + findings: list[StaticFinding] = [] + trajectories: list[HunterTrajectory] = [] + agent = self._agent + for task in tasks: + base_task = BaseTask( + description=self._describe(task), + success_criteria="Enumerate each reachable tainted sink with file:line and a one-line rationale.", + max_steps=task.budget.max_findings_per_target, + token_budget=task.budget.retrieval_budget_chars * 4, + max_cost_usd=self._max_cost_usd, + ) + result = await agent.run(base_task, session_id=scan_id) + task_findings = self._findings_from_result(task, result, policy) + findings.extend(task_findings) + trajectories.append(self._trajectory_from_result(scan_id, task, result, task_findings)) + if emit is not None: + emit( + "stage_end", + f"hunter_pool:{task.target.path}", + { + "exit_reason": getattr(result, "exit_reason", "unknown"), + "total_steps": getattr(result, "total_steps", 0), + "total_cost_usd": getattr(result, "total_cost_usd", 0.0), + }, + ) + await agent.cleanup() + return HunterPoolResult(findings=sorted(findings, key=lambda item: item.finding_id), trajectories=trajectories) + + def _describe(self, task: HunterTask) -> str: + context = task.retrieval_context or f"[repo_code] {task.target.path}" + return ( + f"Audit `{task.target.path}` ({task.target.language}) for security-relevant vulnerabilities. " + f"Report ONLY evidence-backed findings as a JSON array, each object: " + f'{{"line": , "title": , "cwe": "CWE-NN" (optional), "rationale": }}. ' + f"Emit `[]` if none.\n\nSource context:\n{context}" + ) + + def _findings_from_result(self, task: HunterTask, result: object, policy: dict[str, CwePolicy]) -> list[StaticFinding]: + items = _extract_findings(getattr(result, "final_output", "") or "") + findings: list[StaticFinding] = [] + for item in items: + line = item.get("line") + line = int(line) if isinstance(line, int) else None + reachability_status, function_name, reachability_evidence = ( + _reachability_for_line(task.target, line) if line is not None else ("unknown", None, []) + ) + cwe = str(item.get("cwe", "")) or None + severity = SEVERITY_LABEL.get(resolve_severity(policy, cwe), "medium") if cwe else "medium" + line_part = line if line is not None else 0 + findings.append( + StaticFinding( + finding_id=f"hx-hunter:{task.target.path}:{line_part}", + path=task.target.path, + title=str(item.get("title", "HarnessX hunter finding")), + severity=severity, + confidence="medium", + evidence_level="static_corroboration", + rationale=str(item.get("rationale", "Reported by the HarnessX hunter; manual verification required.")), + line=line, + function_name=function_name, + reachability_status=reachability_status, + reachability_evidence=reachability_evidence, + reachability_conditions=_reachability_conditions(reachability_evidence), + tags=sorted(set(task.target.tags)), + ranking_priority=task.ranking.priority, + ) + ) + return findings + + def _trajectory_from_result( + self, scan_id: str, task: HunterTask, result: object, task_findings: list[StaticFinding] + ) -> HunterTrajectory: + return HunterTrajectory( + scan_id=scan_id, + target_path=task.target.path, + tier=task.tier, + priority=task.ranking.priority, + selected_skills=task.selected_skills, + retrieval_context_chars=len(task.retrieval_context), + findings=[finding.finding_id for finding in task_findings], + status=str(getattr(result, "exit_reason", "completed")), + ) + + +def _extract_findings(text: str) -> list[dict[str, object]]: + """Best-effort extraction of the findings JSON array from the agent's final output.""" + match = _FINDINGS_JSON.search(text) + if match is None: + return [] + try: + parsed = json.loads(match.group(0)) + except json.JSONDecodeError: + return [] + return [item for item in parsed if isinstance(item, dict)] if isinstance(parsed, list) else [] diff --git a/src/openultrasast/reports.py b/src/openultrasast/reports.py index c5e8f30..65f4e77 100644 --- a/src/openultrasast/reports.py +++ b/src/openultrasast/reports.py @@ -72,6 +72,7 @@ def write_manifest( artifact_paths: dict[str, Path], path: Path, score: dict[str, object] | None = None, + degradations: list[dict[str, object]] | None = None, ) -> None: path.parent.mkdir(parents=True, exist_ok=True) verification_by_id = _verification_by_id(verifications) @@ -96,6 +97,8 @@ def write_manifest( } if score is not None: payload["score"] = score + if degradations is not None: + payload["degradations"] = degradations path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") diff --git a/src/openultrasast/verify_judge.py b/src/openultrasast/verify_judge.py new file mode 100644 index 0000000..a3dc802 --- /dev/null +++ b/src/openultrasast/verify_judge.py @@ -0,0 +1,130 @@ +"""Verifier dispatcher: HarnessX llm-judge when available, structural fallback otherwise. + +The default path (no extra) is the existing :func:`verification.verify_findings` +verbatim — same ``list[VerificationResult]``, same rejection of unknown reachability — +so behaviour is byte-identical without HarnessX. The judge path is lazy/guarded. +""" + +from __future__ import annotations + +from .findings import StaticFinding +from .harness_ext import has_harnessx +from .verification import ( + EvidenceLevel, + VerificationResult, + VerificationStatus, + build_independent_verifier_context, + is_report_verified, + verify_findings, +) + + +def verify_findings_dispatch( + findings: list[StaticFinding], + *, + verifier_model: str | None = None, + use_harnessx: bool = False, +) -> list[VerificationResult]: + """Drop-in superset of :func:`verification.verify_findings`. + + Falls back to the structural verifier unless the HarnessX extra is present, a + verifier model is configured, and ``use_harnessx`` is set. + """ + if not (use_harnessx and verifier_model and has_harnessx()): + return verify_findings(findings) + return _verify_with_judge(findings, verifier_model) + + +def _verify_with_judge(findings: list[StaticFinding], verifier_model: str) -> list[VerificationResult]: + import asyncio + + return asyncio.run(_judge_all(findings, verifier_model)) + + +async def _judge_all(findings: list[StaticFinding], verifier_model: str) -> list[VerificationResult]: + from harnessx.core.events import Message + from harnessx.processors.evaluation.llm_judge import build_judge_prompt, parse_judge_response + from harnessx.providers.anthropic_provider import AnthropicProvider + + provider = AnthropicProvider(model=verifier_model) + results: list[VerificationResult] = [] + for finding in findings: + prompt = build_judge_prompt( + task_description=f"Is this a real, exploitable security issue: {finding.title}? Judge on evidence only.", + trajectory_summary=_finding_summary(finding), + extracted_answer=finding.rationale, + ) + try: + response = await provider.complete(messages=[Message(role="user", content=prompt)], tools=[]) + verdict = parse_judge_response(_text(response.content)) + except Exception: # noqa: BLE001 — judge/provider failure must never auto-reject a finding + verdict = {"verdict": "no_answer", "confidence": 0.0} + results.append(_verdict_to_verification(finding, verdict)) + return results + + +def _verdict_to_verification(finding: StaticFinding, verdict: dict[str, object]) -> VerificationResult: + """Map an llm-judge verdict onto a VerificationResult (infra failure never auto-rejects).""" + label = str(verdict.get("verdict", "no_answer")) + raw_confidence = verdict.get("confidence", 0.0) + confidence = 0.0 + if isinstance(raw_confidence, (int, float)): + confidence = float(raw_confidence) + elif isinstance(raw_confidence, str): + try: + confidence = float(raw_confidence) + except ValueError: + confidence = 0.0 + cause = str(verdict.get("cause", "")).strip() + lesson = str(verdict.get("lesson", "")).strip() + missing = str(verdict.get("missing", "")).strip() or _missing_capability_summary(verdict) + level = EvidenceLevel(finding.evidence_level) + + if label == "plausible" and confidence >= 0.6: + status = VerificationStatus.ACCEPTED + elif label in ("unsupported", "refused"): + status = VerificationStatus.REJECTED + else: # hedging / format_wrong / no_answer / low-confidence plausible + status = VerificationStatus.NEEDS_EVIDENCE + + if status == VerificationStatus.ACCEPTED: + pro_case = cause or "The judge found the issue plausible on the available evidence." + counter_case = "Impact and exploitability still require deeper verifier or dynamic evidence." + else: + pro_case = "A candidate issue exists from the finding's evidence." + counter_case = cause or "The judge could not confirm the issue from the available evidence." + + return VerificationResult( + finding_id=finding.finding_id, + status=status, + evidence_level=level, + verified=is_report_verified(level, status), + pro_case=pro_case, + counter_case=counter_case, + tie_breaker=lesson or "Confirm attacker control, sanitizer absence, and impact.", + required_next_step=missing or "Collect corroborating evidence before reporting as verified.", + context_sources=sorted(build_independent_verifier_context(finding).keys()), + ) + + +def _finding_summary(finding: StaticFinding) -> str: + return ( + f"{finding.title} at {finding.path}:{finding.line} (function={finding.function_name or 'unknown'}); " + f"reachability={finding.reachability_status}; tags={', '.join(finding.tags) or 'none'}" + ) + + +def _missing_capability_summary(verdict: dict[str, object]) -> str: + capability = verdict.get("missing_capability") + if isinstance(capability, dict): + return str(capability.get("summary", "")).strip() + return "" + + +def _text(content: object) -> str: + if isinstance(content, str): + return content + if isinstance(content, list): + parts = [str(block.get("text", "")) if isinstance(block, dict) else str(block) for block in content] + return "".join(parts) + return str(content) diff --git a/tests/test_cli_hx_dispatch.py b/tests/test_cli_hx_dispatch.py new file mode 100644 index 0000000..50b45cd --- /dev/null +++ b/tests/test_cli_hx_dispatch.py @@ -0,0 +1,56 @@ +"""CLI capability-gated dispatch + graceful degradation (Phase 3 task 6.5). + +These run in the default (extra-absent) environment: 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.setenv("OPENULTRASAST_RUNS_DIR", ".runs") + 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: + assert not cli.has_harnessx() # this CI env has no harnessx extra + manifest = _run(tmp_path, monkeypatch, '[models]\nhunter = "anthropic/claude"\nverifier = "anthropic/claude"\n') + + degradations = manifest.get("degradations", []) + stages = {entry["stage"] for entry in degradations} + assert stages == {"hunter_pool", "verify"} + assert all(entry["reason"] == "harnessx_extra_unavailable" for entry in degradations) + # Fallback still produced findings (deterministic hunter ran). + assert manifest["findings"] + + +def test_manifest_has_no_degradations_without_models(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + manifest = _run(tmp_path, monkeypatch, config_body=None) + assert "degradations" not in manifest + + +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"] diff --git a/tests/test_hunter_harness.py b/tests/test_hunter_harness.py new file mode 100644 index 0000000..45afab7 --- /dev/null +++ b/tests/test_hunter_harness.py @@ -0,0 +1,48 @@ +"""HarnessX hunter sub-harness: guarding, findings parsing, construction smoke (task 6.1).""" + +import sys + +import pytest + +from openultrasast.harness_ext import HarnessXUnavailableError, build_sast, has_harnessx +from openultrasast.hunter_harness import HxScanOrchestrator, _extract_findings + + +def test_module_import_does_not_pull_harnessx() -> None: + import openultrasast.hunter_harness # noqa: F401 + + assert "harnessx" not in sys.modules + + +def test_extract_findings_parses_json_array_and_tolerates_garbage() -> None: + text = 'Here are the findings:\n[{"line": 9, "title": "SQLi", "cwe": "CWE-89", "rationale": "concat"}]\nDone.' + items = _extract_findings(text) + assert items == [{"line": 9, "title": "SQLi", "cwe": "CWE-89", "rationale": "concat"}] + assert _extract_findings("no findings here") == [] + assert _extract_findings("[not, valid, json]") == [] + + +@pytest.mark.skipif(has_harnessx(), reason="exercises the extra-absent guard") +def test_build_sast_raises_without_extra() -> None: + with pytest.raises(HarnessXUnavailableError): + build_sast() + + +@pytest.mark.skipif(has_harnessx(), reason="exercises the extra-absent guard") +def test_orchestrator_construction_raises_without_extra() -> None: + with pytest.raises(HarnessXUnavailableError): + HxScanOrchestrator(provider_model="claude-sonnet-4-6") + + +@pytest.mark.skipif(not has_harnessx(), reason="construction smoke; requires the harnessx extra") +def test_build_sast_returns_harness_config_offline() -> None: + from harnessx.core.harness import HarnessConfig + + assert isinstance(build_sast(), HarnessConfig) + + +@pytest.mark.skipif(not has_harnessx(), reason="construction smoke; requires the harnessx extra") +def test_orchestrator_builds_agent_offline() -> None: + # Constructs the model + harness without any LLM call (provider client is lazy). + agent = HxScanOrchestrator(provider_model="claude-sonnet-4-6")._build_agent() + assert type(agent).__name__ == "Harness" diff --git a/tests/test_verify_judge.py b/tests/test_verify_judge.py new file mode 100644 index 0000000..217bb11 --- /dev/null +++ b/tests/test_verify_judge.py @@ -0,0 +1,61 @@ +"""Verifier dispatcher: structural fallback (default) + judge verdict mapping (Phase 3 task 6.4).""" + +import sys + +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() -> None: + verify_findings_dispatch([_finding()], verifier_model="claude-x", use_harnessx=True) + assert "harnessx" not in sys.modules + + +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