diff --git a/.kiro/specs/three-stage-scan/tasks.md b/.kiro/specs/three-stage-scan/tasks.md index 8570030..891942f 100644 --- a/.kiro/specs/three-stage-scan/tasks.md +++ b/.kiro/specs/three-stage-scan/tasks.md @@ -150,7 +150,7 @@ This plan ships the three-stage scan without rebuilding CWE policy or the regex - _Requirements: 1.3, 1.5, 5.1, 9.3, 9.6_ - _Depends: 5.3, 6.2_ -- [ ] 6.4 Attach the tool hunter to map hotspots when a model is configured +- [x] 6.4 Attach the tool hunter to map hotspots when a model is configured - Hunter findings are suspicion-only and may propose a snippet that still must pass the safety check and run only in the sandbox. - Observable completion: with a scripted client, standard+model produces tool-hunter suspicions; a proposed snippet is executed only on deep after the safety check. - _Requirements: 6.1, 6.3, 6.5, 6.6_ diff --git a/src/openultrasast/cli.py b/src/openultrasast/cli.py index bc82767..84a6af1 100644 --- a/src/openultrasast/cli.py +++ b/src/openultrasast/cli.py @@ -3,10 +3,11 @@ from __future__ import annotations import argparse import json import tempfile -from dataclasses import dataclass +from dataclasses import dataclass, replace from pathlib import Path from time import perf_counter +from . import tool_hunter from .benchmark import ( create_benchmark_run, evaluate_benchmark, @@ -48,7 +49,7 @@ from .run import ScanRun, create_scan_run from .sandbox import resolve_sandbox_probe, resolve_sandbox_runner from .scoring import build_score_artifact from .stages import Stage, plan_for_mode, record_completed, record_skip, skip_as_degradation, stages_payload -from .verification import VerificationResult, write_verification_results +from .verification import VerificationResult, verify_findings, write_verification_results from .verify_judge import verify_findings_dispatch CALIBRATION_DIR = ".openultrasast/calibration" @@ -320,6 +321,20 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut plan = record_completed(plan, Stage.MAP) if not hunter_model: runtime.state["degradations"].append(skip_as_degradation(Stage.MAP, "hunter_model_unavailable")) + else: + built_map, findings, verifications = _attach_tool_hunter( + root=run.target, + built_map=built_map, + findings=findings, + verifications=verifications, + hunter_model=hunter_model, + max_hotspots=config.complexity.max_hunter_hotspots, + max_findings=max_findings, + findings_path=findings_path, + verification_path=verification_path, + complexity_map_path=complexity_map_path, + runtime=runtime, + ) complexity_payload = { "hotspot_count": len(built_map.hotspots), "heuristic_only": built_map.heuristic_only, @@ -396,6 +411,55 @@ def _run_scan(path: Path, config_path: Path, mode: str, fail_on: str) -> ScanOut ) +def _attach_tool_hunter( + *, + root: Path, + built_map: complexity_map.ComplexityMap, + findings: list[StaticFinding], + verifications: list[VerificationResult], + hunter_model: str, + max_hotspots: int, + max_findings: int, + findings_path: Path, + verification_path: Path, + complexity_map_path: Path, + runtime: HarnessRuntime, +) -> tuple[complexity_map.ComplexityMap, list[StaticFinding], list[VerificationResult]]: + client = tool_hunter.resolve_hunter_client() + if client is None: + return built_map, findings, verifications + hunter_hotspots = list(built_map.hotspots[: max(max_hotspots, 0)]) + hunter_findings = runtime.run_stage( + "tool_hunter", + lambda: tool_hunter.run_tool_hunter( + root, + hunter_hotspots, + client=client, + model=hunter_model, + max_steps=tool_hunter.DEFAULT_MAX_STEPS, + ), + ) + seen = {finding.finding_id for finding in findings} + extra = [finding for finding in hunter_findings if finding.finding_id not in seen] + if extra: + findings = findings + extra + if max_findings and len(findings) > max_findings: + if not any(entry.get("reason") == "max_findings_exceeded" for entry in runtime.state["degradations"]): + runtime.state["degradations"].append( + {"stage": "budget", "reason": "max_findings_exceeded", "requested": max_findings, "actual": len(findings)} + ) + findings = findings[:max_findings] + kept = {finding.finding_id for finding in findings} + extra = [finding for finding in extra if finding.finding_id in kept] + if extra: + verifications = verifications + verify_findings(extra) + write_verification_results(verifications, verification_path) + write_findings(findings, findings_path) + built_map = replace(built_map, heuristic_only=False) + complexity_map.write_complexity_map(built_map, complexity_map_path) + return built_map, findings, verifications + + def _persist_calibration_feedback( run: ScanRun, ledger_path: Path, diff --git a/src/openultrasast/findings.py b/src/openultrasast/findings.py index 20fece1..16ec29b 100644 --- a/src/openultrasast/findings.py +++ b/src/openultrasast/findings.py @@ -28,6 +28,7 @@ class StaticFinding: tags: list[str] ranking_priority: float status: str = "enabled" + proposed_snippet: str | None = None SEVERITY_LABEL = {5: "critical", 4: "high", 3: "medium", 2: "low", 1: "low", 0: "info"} diff --git a/src/openultrasast/provider/openrouter.py b/src/openultrasast/provider/openrouter.py index c2b9c40..a2cdf2f 100644 --- a/src/openultrasast/provider/openrouter.py +++ b/src/openultrasast/provider/openrouter.py @@ -6,7 +6,7 @@ import re import time import urllib.error import urllib.request -from collections.abc import Callable +from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass from typing import TypeVar, cast @@ -61,8 +61,18 @@ class OpenRouterChatClient: raise OpenRouterError("OPENROUTER_API_KEY is required for OpenRouter chat calls") return cls(api_key=api_key, base_url=os.environ.get("OPENROUTER_BASE_URL", cls.base_url)) - def complete_json(self, *, model: str, messages: list[dict[str, str]], timeout_seconds: int = 60) -> object: - payload = json.dumps({"model": model, "messages": messages, "temperature": 0}).encode() + def complete_chat( + self, + *, + model: str, + messages: Sequence[Mapping[str, object]], + tools: Sequence[Mapping[str, object]] | None = None, + timeout_seconds: int = 60, + ) -> dict[str, object]: + body: dict[str, object] = {"model": model, "messages": list(messages), "temperature": 0} + if tools: + body["tools"] = list(tools) + payload = json.dumps(body).encode() request = urllib.request.Request( f"{self.base_url.rstrip('/')}/chat/completions", data=payload, @@ -81,8 +91,13 @@ class OpenRouterChatClient: response_payload = call_with_retry(_do, attempts=self.max_attempts, base_delay=self.retry_base_delay) except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc: raise OpenRouterError(f"OpenRouter chat request failed: {exc}") from exc + return _extract_message(response_payload) - content = _extract_message_content(response_payload) + def complete_json(self, *, model: str, messages: list[dict[str, str]], timeout_seconds: int = 60) -> object: + message = self.complete_chat(model=model, messages=messages, timeout_seconds=timeout_seconds) + content = message.get("content") + if not isinstance(content, str) or not content.strip(): + raise OpenRouterError("OpenRouter message content was empty") return parse_json_content(content) @@ -156,7 +171,7 @@ def parse_json_content(content: str) -> object: raise OpenRouterError(f"model response was not valid JSON: {exc}") from exc -def _extract_message_content(payload: object) -> str: +def _extract_message(payload: object) -> dict[str, object]: if not isinstance(payload, dict): raise OpenRouterError("OpenRouter response must be a JSON object") choices = payload.get("choices") @@ -168,7 +183,11 @@ def _extract_message_content(payload: object) -> str: message = first.get("message") if not isinstance(message, dict): raise OpenRouterError("OpenRouter choice did not include a message") - content = message.get("content") + return message + + +def _extract_message_content(payload: object) -> str: + content = _extract_message(payload).get("content") if not isinstance(content, str) or not content.strip(): raise OpenRouterError("OpenRouter message content was empty") return content diff --git a/src/openultrasast/regress/candidate.py b/src/openultrasast/regress/candidate.py index 1b7c7ee..0fdf55a 100644 --- a/src/openultrasast/regress/candidate.py +++ b/src/openultrasast/regress/candidate.py @@ -196,7 +196,7 @@ def run_regression( records: list[CandidateVerdict] = [] for hotspot in selected: language = _language_for(hotspot.path, languages_by_path) - snippet = snippet_for(language, hotspot.path, hotspot.function_name) + snippet = _proposed_snippet(hotspot, findings) or snippet_for(language, hotspot.path, hotspot.function_name) image = images.get(language) or DEFAULT_IMAGES.get(language, "ousast-missing-image") mapped = runner.run_recipe(language, snippet, image, sandbox_limits, repo_root=repo_root) reachability = _reachability_for(hotspot, findings_by_id) @@ -220,6 +220,16 @@ def write_verdicts(verdicts: Sequence[CandidateVerdict], path: Path) -> None: path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") +def _proposed_snippet(hotspot: Hotspot, findings: Sequence[StaticFinding]) -> str | None: + for finding in findings: + snippet = finding.proposed_snippet + if not snippet: + continue + if finding.path == hotspot.path or finding.finding_id in hotspot.inventory_finding_ids: + return snippet + return None + + def _language_for(path: str, languages_by_path: Mapping[str, str]) -> str: if path in languages_by_path: return languages_by_path[path] diff --git a/src/openultrasast/tool_hunter.py b/src/openultrasast/tool_hunter.py index 338ace7..e701366 100644 --- a/src/openultrasast/tool_hunter.py +++ b/src/openultrasast/tool_hunter.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import os import re from collections.abc import Mapping, Sequence from dataclasses import dataclass @@ -12,12 +13,29 @@ from typing import Protocol from .complexity.map import Hotspot from .findings import StaticFinding from .hunter_tools import PathEscapesRepo, clamp_repo_path, find_refs, grep_repo, read_file +from .provider.openrouter import OpenRouterChatClient, OpenRouterError _DEFAULT_MAX_CHARS = 4000 _DEFAULT_MAX_MATCHES = 50 +DEFAULT_MAX_STEPS = 4 +CLIENT_ENV = "OPENULTRASAST_HUNTER_CLIENT" _FINDING_ID_PREFIX = "tool-hunter:" _HUNTER_TOOL_NAMES = frozenset({"read_file", "grep_repo", "find_refs"}) _SEVERITIES = frozenset({"critical", "high", "medium", "low", "info"}) +_SCRIPTED_FLAGS = frozenset({"scripted", "script"}) +_DUMP_FLAGS = frozenset({"dump", "dump-only"}) +_UNSAFE_FLAGS = frozenset({"unsafe", "unsafe-snippet"}) +_OPENROUTER_FLAGS = frozenset({"openrouter", "live"}) +_UNSAFE_SNIPPET = 'client = docker.DockerClient(base_url="unix://var/run/docker.sock")\n' +_SCRIPTED_FINDINGS = [ + { + "path": "app.py", + "line": 3, + "title": "Dynamic execution of attacker-controlled data", + "rationale": "eval/execute follows request input across the hotspot.", + "evidence_level": "static_corroboration", + } +] HUNTER_TOOLS: list[dict[str, object]] = [ { @@ -97,6 +115,117 @@ class ChatClient(Protocol): raise NotImplementedError +class ScriptedChatClient: + """Deterministic ChatClient for tests and OPENULTRASAST_HUNTER_CLIENT injection.""" + + def __init__(self, turns: list[ChatResponse]) -> None: + self._turns = turns + self.calls: list[dict[str, object]] = [] + + def complete( + self, + *, + model: str, + messages: list[dict[str, object]], + tools: list[dict[str, object]], + timeout_seconds: int = 60, + ) -> ChatResponse: + self.calls.append({"model": model, "messages": list(messages), "tools": tools, "timeout_seconds": timeout_seconds}) + index = len(self.calls) - 1 + if index >= len(self._turns): + return ChatResponse(content="") + return self._turns[index] + + +class OpenRouterHunterClient: + """Adapt OpenRouterChatClient to the tool-hunter ChatClient protocol.""" + + def __init__(self, client: OpenRouterChatClient) -> None: + self._client = client + + @classmethod + def from_env(cls) -> OpenRouterHunterClient: + return cls(OpenRouterChatClient.from_env()) + + def complete( + self, + *, + model: str, + messages: list[dict[str, object]], + tools: list[dict[str, object]], + timeout_seconds: int = 60, + ) -> ChatResponse: + message = self._client.complete_chat(model=model, messages=messages, tools=tools, timeout_seconds=timeout_seconds) + return chat_response_from_message(message) + + +def scripted_hunter_client(mode: str = "scripted") -> ScriptedChatClient: + grep_turn = ChatResponse( + tool_calls=(ToolCall(id="call_grep", name="grep_repo", arguments={"pattern": "eval|execute|query", "max_matches": 10}),) + ) + if mode in _DUMP_FLAGS: + return ScriptedChatClient([ChatResponse(content=json.dumps(_SCRIPTED_FINDINGS))]) + if mode in _UNSAFE_FLAGS: + payload = [{**_SCRIPTED_FINDINGS[0], "snippet": _UNSAFE_SNIPPET}] + return ScriptedChatClient([grep_turn, ChatResponse(content=json.dumps(payload))]) + return ScriptedChatClient([grep_turn, ChatResponse(content=json.dumps(_SCRIPTED_FINDINGS))]) + + +def resolve_hunter_client() -> ChatClient | None: + """Return a hunter client from env, or OpenRouter when an API key is present.""" + flag = os.environ.get(CLIENT_ENV, "").strip().lower() + if flag in _SCRIPTED_FLAGS: + return scripted_hunter_client("scripted") + if flag in _DUMP_FLAGS: + return scripted_hunter_client("dump-only") + if flag in _UNSAFE_FLAGS: + return scripted_hunter_client("unsafe-snippet") + if flag in _OPENROUTER_FLAGS or (not flag and os.environ.get("OPENROUTER_API_KEY")): + try: + return OpenRouterHunterClient.from_env() + except OpenRouterError: + return None + return None + + +def chat_response_from_message(message: Mapping[str, object]) -> ChatResponse: + content = message.get("content") + text = content if isinstance(content, str) else None + raw_calls = message.get("tool_calls") or () + calls: list[ToolCall] = [] + if isinstance(raw_calls, Sequence) and not isinstance(raw_calls, str | bytes): + for index, item in enumerate(raw_calls): + parsed = _tool_call_from_openrouter(item, index) + if parsed is not None: + calls.append(parsed) + return ChatResponse(content=text, tool_calls=tuple(calls)) + + +def _tool_call_from_openrouter(item: object, index: int) -> ToolCall | None: + if not isinstance(item, Mapping): + return None + function = item.get("function") + if not isinstance(function, Mapping): + return None + name = function.get("name") + if not isinstance(name, str) or not name: + return None + call_id = item.get("id") + identifier = call_id if isinstance(call_id, str) and call_id else f"call_{index}" + arguments = function.get("arguments", {}) + if isinstance(arguments, str): + try: + loaded = json.loads(arguments) + except json.JSONDecodeError: + loaded = {} + parsed_args = loaded if isinstance(loaded, dict) else {} + elif isinstance(arguments, Mapping): + parsed_args = dict(arguments) + else: + parsed_args = {} + return ToolCall(id=identifier, name=name, arguments=parsed_args) + + def run_tool_hunter( root: Path, hotspots: list[Hotspot], @@ -268,6 +397,8 @@ def _finding_from_item(root: Path, item: Mapping[str, object], scores: Mapping[s rationale = item.get("rationale") function_name = item.get("function_name") severity = item.get("severity") + snippet = item.get("snippet") + proposed_snippet = snippet if isinstance(snippet, str) and snippet else None return StaticFinding( finding_id=f"{_FINDING_ID_PREFIX}{relative}:{line_no if line_no is not None else 0}", path=relative, @@ -283,6 +414,7 @@ def _finding_from_item(root: Path, item: Mapping[str, object], scores: Mapping[s reachability_conditions=[], tags=["tool-hunter"], ranking_priority=float(scores.get(relative, 0.0)), + proposed_snippet=proposed_snippet, ) diff --git a/tests/test_cli_hx_dispatch.py b/tests/test_cli_hx_dispatch.py index 15286e0..26725b2 100644 --- a/tests/test_cli_hx_dispatch.py +++ b/tests/test_cli_hx_dispatch.py @@ -24,6 +24,8 @@ def _run(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, config_body: str | Non (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" diff --git a/tests/test_scan_deep.py b/tests/test_scan_deep.py index 6125966..b2f2405 100644 --- a/tests/test_scan_deep.py +++ b/tests/test_scan_deep.py @@ -99,3 +99,36 @@ def test_deep_scan_with_fake_runner_writes_verdicts(tmp_path: Path, monkeypatch: assert "path" in item assert manifest["artifacts"]["verdicts"] == "verdicts.json" assert manifest["artifacts"]["complexity_map"] == "complexity_map.json" + + +def test_deep_scan_unsafe_hunter_snippet_is_safety_rejected_without_docker_job(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + from openultrasast import cli + from openultrasast.sandbox import FakeSandboxRunner + + repo = _repo(tmp_path) + config = tmp_path / "openultrasast.toml" + config.write_text('[models]\nhunter = "test-hunter"\n') + fake = FakeSandboxRunner() + 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) + + assert main(["scan", str(repo), "--mode", "deep", "--config", str(config)]) == 0 + + run_dir = _latest_run(repo) + payload = json.loads((run_dir / "verdicts.json").read_text()) + findings = json.loads((run_dir / "findings.json").read_text())["findings"] + hunter_findings = [item for item in findings if str(item["finding_id"]).startswith("tool-hunter:")] + + assert docker_argv == [] + assert fake.jobs == [] + assert hunter_findings + assert all(item["evidence_level"] == "suspicion" for item in hunter_findings) + assert payload["verdicts"] + assert any(item["reason"] == "safety_rejected" for item in payload["verdicts"]) + for item in payload["verdicts"]: + if item["reason"] == "safety_rejected": + assert item["verdict"] == "inconclusive" diff --git a/tests/test_scan_pipeline_integrity.py b/tests/test_scan_pipeline_integrity.py index 8a4d370..2a3ade5 100644 --- a/tests/test_scan_pipeline_integrity.py +++ b/tests/test_scan_pipeline_integrity.py @@ -36,6 +36,8 @@ def _scan(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, *, mode: str, config_ (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) argv = ["scan", str(repo), "--mode", mode] if config_body is not None: cfg = tmp_path / "openultrasast.toml" diff --git a/tests/test_tool_hunter.py b/tests/test_tool_hunter.py index 6840bd4..826da01 100644 --- a/tests/test_tool_hunter.py +++ b/tests/test_tool_hunter.py @@ -1,8 +1,10 @@ import json from pathlib import Path +import pytest + from openultrasast.complexity.map import Hotspot -from openultrasast.tool_hunter import ChatResponse, ToolCall, run_tool_hunter +from openultrasast.tool_hunter import ChatResponse, OpenRouterHunterClient, ScriptedChatClient, ToolCall, run_tool_hunter SPLIT_SINK = """\ term = request.args["q"] @@ -21,26 +23,6 @@ _DUMP_FINDINGS = [ ] -class ScriptedChatClient: - def __init__(self, turns: list[ChatResponse]) -> None: - self._turns = turns - self.calls: list[dict[str, object]] = [] - - def complete( - self, - *, - model: str, - messages: list[dict[str, object]], - tools: list[dict[str, object]], - timeout_seconds: int = 60, - ) -> ChatResponse: - self.calls.append({"model": model, "messages": list(messages), "tools": tools, "timeout_seconds": timeout_seconds}) - index = len(self.calls) - 1 - if index >= len(self._turns): - return ChatResponse(content="") - return self._turns[index] - - def _repo_with_split_sink(tmp_path: Path) -> Path: root = tmp_path / "repo" root.mkdir() @@ -164,3 +146,182 @@ def test_max_steps_stops_before_ungrounded_json(tmp_path: Path) -> None: assert findings == [] assert len(client.calls) == 1 + + +_UNSAFE_SNIPPET = 'client = docker.DockerClient(base_url="unix://var/run/docker.sock")\n' +_VULN = "@app.route('/admin')\ndef admin():\n return eval(request.data)\n" + + +def test_hunter_json_snippet_stays_suspicion_and_is_not_static_corroboration(tmp_path: Path) -> None: + root = _repo_with_split_sink(tmp_path) + payload = [{**_DUMP_FINDINGS[0], "snippet": _UNSAFE_SNIPPET, "evidence_level": "static_corroboration"}] + client = ScriptedChatClient( + [ + ChatResponse(tool_calls=(ToolCall(id="call_grep", name="grep_repo", arguments={"pattern": "query"}),)), + ChatResponse(content=json.dumps(payload)), + ] + ) + + findings = run_tool_hunter(root, [_hotspot()], client=client, model="test-hunter", max_steps=4) + + assert len(findings) == 1 + assert findings[0].evidence_level == "suspicion" + assert findings[0].finding_id.startswith("tool-hunter:") + assert findings[0].proposed_snippet == _UNSAFE_SNIPPET + + +def _write_hunter_config(tmp_path: Path, body: str = '[models]\nhunter = "test-hunter"\n') -> Path: + config = tmp_path / "openultrasast.toml" + config.write_text(body) + return config + + +def _standard_repo(tmp_path: Path) -> Path: + repo = tmp_path / "repo" + repo.mkdir() + (repo / "app.py").write_text(_VULN) + return repo + + +def _latest_run(repo: Path) -> Path: + return sorted((repo / ".runs").iterdir())[-1] + + +def _guard_docker(monkeypatch: pytest.MonkeyPatch) -> list[list[str]]: + import subprocess + + from openultrasast.sandbox import runner as sandbox_runner + + docker_argv: list[list[str]] = [] + original_run = subprocess.run + + def guarded_run(command: object, *args: object, **kwargs: object) -> object: + argv = list(command) if isinstance(command, list | tuple) else [command] + if argv and Path(str(argv[0])).name == "docker": + docker_argv.append([str(item) for item in argv]) + raise AssertionError(f"hunter scan test must not start docker: {argv}") + return original_run(command, *args, **kwargs) + + def fail(*args: object, **kwargs: object) -> object: + raise AssertionError("hunter scan test must not use DockerCliRunner") + + monkeypatch.setattr(subprocess, "run", guarded_run) + monkeypatch.setattr(sandbox_runner.DockerCliRunner, "run", fail) + return docker_argv + + +def test_standard_scripted_hunter_emits_tool_hunter_suspicion(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + from openultrasast import cli, 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 + + def wrapped(*args: object, **kwargs: object) -> object: + hunter_calls.append(1) + return real_hunter(*args, **kwargs) + + monkeypatch.setattr(tool_hunter, "run_tool_hunter", wrapped) + + assert main(["scan", str(repo), "--mode", "standard", "--config", str(config)]) == 0 + + run_dir = _latest_run(repo) + findings = json.loads((run_dir / "findings.json").read_text())["findings"] + complexity_map = json.loads((run_dir / "complexity_map.json").read_text()) + manifest = json.loads((run_dir / "manifest.json").read_text()) + hunter_findings = [item for item in findings if str(item["finding_id"]).startswith("tool-hunter:")] + + assert hunter_calls + assert docker_argv == [] + assert hunter_findings + assert all(item["evidence_level"] == "suspicion" for item in hunter_findings) + assert all(item["evidence_level"] != "static_corroboration" for item in hunter_findings) + assert complexity_map["heuristic_only"] is False + assert manifest["complexity"]["heuristic_only"] is False + assert not any(entry.get("reason") == "hunter_model_unavailable" for entry in manifest.get("degradations", [])) + + +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.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 + + def wrapped(*args: object, **kwargs: object) -> object: + hunter_calls.append(1) + return real_hunter(*args, **kwargs) + + monkeypatch.setattr(tool_hunter, "run_tool_hunter", wrapped) + + assert main(["scan", str(repo), "--mode", "standard", "--config", str(config)]) == 0 + + run_dir = _latest_run(repo) + findings = json.loads((run_dir / "findings.json").read_text())["findings"] + complexity_map = json.loads((run_dir / "complexity_map.json").read_text()) + hunter_findings = [item for item in findings if str(item["finding_id"]).startswith("tool-hunter:")] + + assert hunter_calls, "dump-only still runs the tool hunter loop" + assert docker_argv == [] + assert hunter_findings == [] + assert complexity_map["heuristic_only"] is False + + +def test_resolve_hunter_client_scripted_and_dump_only_env(monkeypatch: pytest.MonkeyPatch) -> None: + from openultrasast import tool_hunter + + monkeypatch.delenv("OPENROUTER_API_KEY", raising=False) + monkeypatch.setenv("OPENULTRASAST_HUNTER_CLIENT", "scripted") + scripted = tool_hunter.resolve_hunter_client() + assert scripted is not None + assert type(scripted).__name__ == "ScriptedChatClient" + + monkeypatch.setenv("OPENULTRASAST_HUNTER_CLIENT", "dump-only") + dump_only = tool_hunter.resolve_hunter_client() + assert dump_only is not None + assert type(dump_only).__name__ == "ScriptedChatClient" + assert dump_only is not scripted + + monkeypatch.delenv("OPENULTRASAST_HUNTER_CLIENT", raising=False) + assert tool_hunter.resolve_hunter_client() is None + + monkeypatch.setenv("OPENROUTER_API_KEY", "test-key") + live = tool_hunter.resolve_hunter_client() + assert isinstance(live, OpenRouterHunterClient) + + +def test_openrouter_hunter_client_maps_tool_calls_without_network() -> None: + class _Inner: + def complete_chat(self, **kwargs: object) -> dict[str, object]: + assert kwargs["model"] == "test-hunter" + assert kwargs["tools"] + return { + "content": None, + "tool_calls": [ + { + "id": "call_grep", + "type": "function", + "function": {"name": "grep_repo", "arguments": '{"pattern": "eval", "max_matches": 4}'}, + } + ], + } + + client = OpenRouterHunterClient(_Inner()) # type: ignore[arg-type] + response = client.complete(model="test-hunter", messages=[], tools=[{"type": "function"}], timeout_seconds=5) + + assert response.content is None + assert len(response.tool_calls) == 1 + assert response.tool_calls[0].name == "grep_repo" + assert response.tool_calls[0].arguments == {"pattern": "eval", "max_matches": 4}