diff --git a/src/openultrasast/hunter_tools.py b/src/openultrasast/hunter_tools.py index 183f96c..4dd4ded 100644 --- a/src/openultrasast/hunter_tools.py +++ b/src/openultrasast/hunter_tools.py @@ -5,7 +5,7 @@ import re from collections.abc import Iterable, Iterator, Mapping from pathlib import Path -from .preprocess import enumerate_source_files +from .preprocess import FileTarget, enumerate_source_files class PathEscapesRepo(ValueError): @@ -128,3 +128,172 @@ def _as_line(value: object) -> int | None: if isinstance(value, int): return value return None + + +# --- curated structural tools (learning-harness, Req 6.4, 6.5) -------------------------------------- +# A detector reads code with `read_file` and `read_definition`; the three list tools below answer with +# identifiers and line spans only, so they can never become a second way to dump the repository. Every +# path a caller supplies is clamped to the scan root first. No caching: a detector calls these a handful +# of times per region, and a stale view of a tree that a round may have rewritten is worse than a reparse. + + +def read_definition(root: Path, symbol: str, *, max_chars: int, path: str | None = None) -> dict[str, object] | None: + """The definition of one named function: where it is, how far it runs, and its text. None when absent.""" + resolved_root = root.resolve() + scope = clamp_repo_path(resolved_root, path) if path is not None else None + for relative, target in _iter_clamped_source_files(resolved_root): + if scope is not None and target != scope: + continue + text = target.read_text(errors="ignore") + span = _named_span(relative, text, symbol) + if span is None: + continue + lines = text.splitlines() + start, end = span + while start > 1 and lines[start - 2].lstrip().startswith("@"): + start -= 1 # a handler's decorators are part of what the detector must see + body = "\n".join(lines[start - 1 : end]) + return {"path": relative, "name": symbol, "start_line": start, "end_line": end, "text": body[: max(max_chars, 0)]} + return None + + +def entry_points(root: Path, *, path: str | None = None) -> list[dict[str, object]]: + """Handlers the mapper names, with their access classification and evidence. No source text.""" + from .mapping import analyze_entry_points + + resolved_root = root.resolve() + relative = _relative_or_none(resolved_root, path) + targets = _repo_targets(resolved_root) + rows: list[dict[str, object]] = [] + for record in analyze_entry_points(resolved_root, targets): + if relative is not None and record.path != relative: + continue + rows.append( + { + "path": record.path, + "function": record.function_name, + "name": record.name, + "kind": record.kind, + "line": record.line, + "end_line": record.end_line, + "access_level": record.access_level, + "evidence": list(record.access_evidence), + } + ) + return rows + + +def flows(root: Path, *, path: str, function: str) -> list[dict[str, object]]: + """Adjudicated source-to-sink records inside one function: which source, which sink, what the overlay said.""" + from .findings import quick_scan_findings + from .rank import rank_targets + from .semantic import adjudicate + + resolved_root = root.resolve() + relative = _relative_or_none(resolved_root, path) + targets = _repo_targets(resolved_root) + span = _span_in(resolved_root, relative, function) + if span is None: + return [] + findings = [item for item in quick_scan_findings(resolved_root, targets, rank_targets(targets), None) if item.status != "shadow"] + rows: list[dict[str, object]] = [] + for record in adjudicate(root=resolved_root, targets=targets, findings=findings): + if record.path != relative or record.line is None or not span[0] <= record.line <= span[1]: + continue + rows.append( + { + "path": record.path, + "function": function, + "line": record.line, + "sources": list(record.sources), + "sinks": list(record.sinks), + "disposition": record.disposition, + "cwe": record.cwe, + } + ) + return rows + + +def obligations(root: Path, *, path: str, function: str) -> list[dict[str, object]]: + """What one function owes and has not discharged: the operation, the missing discharger, why it is owed.""" + from .config import ObligationsConfig + from .semantic.facts import FactLoadError, load_facts + from .semantic.ir import parse_file + from .semantic.obligations import check_obligations, load_obligation_facts + from .semantic.obligations.dominance import OrderDominance + + resolved_root = root.resolve() + relative = _relative_or_none(resolved_root, path) + targets = _repo_targets(resolved_root) + try: + flow_facts = load_facts() + except FactLoadError: + return [] + irs: dict[str, tuple[object, str]] = {} + texts: dict[str, str] = {} + for target in targets: + try: + text = (resolved_root / target.path).read_text(errors="ignore") + except OSError: + continue + texts[target.path] = text + irs[target.path] = (parse_file(target.path, text, target.language), text) + from .mapping import analyze_entry_points + + result = check_obligations( + irs=irs, # type: ignore[arg-type] + entries=analyze_entry_points(resolved_root, targets), + facts=load_obligation_facts(), + flow_facts=flow_facts, + policy=None, + paths=(), + dominance=OrderDominance(texts=texts), + store_shapes=(), + min_siblings=ObligationsConfig().min_siblings, + ) + return [ + { + "path": finding.operation.path, + "function": finding.operation.function, + "line": finding.operation.line, + "operation": finding.operation.kind, + "resource": finding.operation.resource, + "missing": finding.missing, + "label": finding.label, + "evidence": list(finding.evidence), + } + for finding in result.findings + if (relative is None or finding.operation.path == relative) and finding.operation.function == function + ] + + +def _repo_targets(root: Path) -> list[FileTarget]: + from .preprocess import preprocess_repository + + _, targets = preprocess_repository(root) + return list(targets) + + +def _relative_or_none(root: Path, path: str | None) -> str | None: + if path is None: + return None + return clamp_repo_path(root, path).relative_to(root).as_posix() + + +def _named_span(relative: str, text: str, symbol: str) -> tuple[int, int] | None: + from .preprocess import detect_language + from .semantic.functions import named_function_ranges, spans_named + + ranges = named_function_ranges(relative, text, detect_language(Path(relative))) + spans = spans_named(ranges or (), symbol) + return spans[0] if spans else None + + +def _span_in(root: Path, relative: str | None, function: str) -> tuple[int, int] | None: + if relative is None: + return None + try: + text = (root / relative).read_text(errors="ignore") + except OSError: + return None + return _named_span(relative, text, function) diff --git a/src/openultrasast/tool_hunter.py b/src/openultrasast/tool_hunter.py index c53ef10..413411a 100644 --- a/src/openultrasast/tool_hunter.py +++ b/src/openultrasast/tool_hunter.py @@ -12,7 +12,17 @@ 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 .hunter_tools import ( + PathEscapesRepo, + clamp_repo_path, + entry_points, + find_refs, + flows, + grep_repo, + obligations, + read_definition, + read_file, +) from .provider.openrouter import OpenRouterChatClient, OpenRouterError _DEFAULT_MAX_CHARS = 4000 @@ -20,7 +30,8 @@ _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"}) +DEFAULT_TOOL_NAMES = ("read_file", "grep_repo", "find_refs") +_HUNTER_TOOL_NAMES = frozenset(DEFAULT_TOOL_NAMES) | {"read_definition", "entry_points", "flows", "obligations"} _SEVERITIES = frozenset({"critical", "high", "medium", "low", "info"}) _SCRIPTED_FLAGS = frozenset({"scripted", "script"}) _DUMP_FLAGS = frozenset({"dump", "dump-only"}) @@ -80,6 +91,50 @@ HUNTER_TOOLS: list[dict[str, object]] = [ }, }, }, + { + "type": "function", + "function": { + "name": "read_definition", + "description": "Resolve one named function to its file, line span and body. Use it instead of guessing what a callee does.", + "parameters": { + "type": "object", + "properties": {"symbol": {"type": "string"}, "path": {"type": "string"}, "max_chars": {"type": "integer"}}, + "required": ["symbol"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "entry_points", + "description": "List the handlers this repository exposes, with their access classification. Identifiers and line spans only.", + "parameters": {"type": "object", "properties": {"path": {"type": "string"}}, "required": []}, + }, + }, + { + "type": "function", + "function": { + "name": "flows", + "description": "List adjudicated source-to-sink records inside one function. Identifiers and line numbers only.", + "parameters": { + "type": "object", + "properties": {"path": {"type": "string"}, "function": {"type": "string"}}, + "required": ["path", "function"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "obligations", + "description": "List what one function owes and has not discharged: the operation, the missing discharger and why it is owed.", + "parameters": { + "type": "object", + "properties": {"path": {"type": "string"}, "function": {"type": "string"}}, + "required": ["path", "function"], + }, + }, + }, ] _SYSTEM_PROMPT = ( @@ -252,32 +307,59 @@ def run_tool_hunter( client: ChatClient, model: str, max_steps: int, + system_prompt: str = _SYSTEM_PROMPT, + user_prompt: str | None = None, + context_files: Sequence[str] = (), + tools: Sequence[str] | None = None, + max_chars: int = _DEFAULT_MAX_CHARS, + tags: Sequence[str] = ("tool-hunter",), ) -> list[StaticFinding]: + """One bounded hunt. Every keyword defaults to today's behaviour; a per-family detector overrides the + prompt, the context it starts from, the tools it may call and the tags its findings carry.""" if max_steps <= 0: return [] + schemas = _tool_schemas(tools) messages: list[dict[str, object]] = [ - {"role": "system", "content": _SYSTEM_PROMPT}, - {"role": "user", "content": _hotspot_prompt(hotspots)}, + {"role": "system", "content": system_prompt}, + {"role": "user", "content": _user_message(root, hotspots, user_prompt, context_files, max_chars)}, ] used_tools = False for _step in range(max_steps): - response = client.complete(model=model, messages=messages, tools=HUNTER_TOOLS) + response = client.complete(model=model, messages=messages, tools=schemas) tool_calls = tuple(response.tool_calls or ()) if tool_calls: - messages.append(_assistant_tool_message(response.content, tool_calls)) + messages.append(_assistant_tool_message(response.content, tool_calls, response.reasoning)) for call in tool_calls: if call.name in _HUNTER_TOOL_NAMES: used_tools = True - messages.append(_tool_result_message(root, call)) + messages.append(_tool_result_message(root, call, max_chars)) continue content = response.content or "" if not content.strip() or not used_tools: # A findings dump is not evidence unless the loop invoked a repo tool. return [] - return _findings_from_content(root, hotspots, content) + return _findings_from_content(root, hotspots, content, tags) return [] +def _tool_schemas(names: Sequence[str] | None) -> list[dict[str, object]]: + wanted = tuple(names) if names is not None else DEFAULT_TOOL_NAMES + by_name = {str(schema["function"]["name"]): schema for schema in HUNTER_TOOLS} # type: ignore[index] + return [by_name[name] for name in wanted if name in by_name] + + +def _user_message(root: Path, hotspots: Sequence[Hotspot], user_prompt: str | None, context_files: Sequence[str], max_chars: int) -> str: + body = user_prompt if user_prompt is not None else _hotspot_prompt(hotspots) + parts = [body] + for relative in context_files: + try: + text = read_file(root, relative, max_chars=max_chars) + except (PathEscapesRepo, OSError): + continue + parts.append(f"--- {relative}\n{text}") + return "\n\n".join(parts) + + def _hotspot_prompt(hotspots: Sequence[Hotspot]) -> str: payload = [ { @@ -296,8 +378,8 @@ def _hotspot_prompt(hotspots: Sequence[Hotspot]) -> str: ) -def _assistant_tool_message(content: str | None, tool_calls: Sequence[ToolCall]) -> dict[str, object]: - return { +def _assistant_tool_message(content: str | None, tool_calls: Sequence[ToolCall], reasoning: str | None = None) -> dict[str, object]: + message: dict[str, object] = { "role": "assistant", "content": content, "tool_calls": [ @@ -312,30 +394,51 @@ def _assistant_tool_message(content: str | None, tool_calls: Sequence[ToolCall]) for call in tool_calls ], } + if reasoning: + # The provider drops its own chain of thought unless it comes back on the next turn (DeepSeek tool calls). + message["reasoning_content"] = reasoning + return message -def _tool_result_message(root: Path, call: ToolCall) -> dict[str, object]: +def _tool_result_message(root: Path, call: ToolCall, max_chars: int = _DEFAULT_MAX_CHARS) -> dict[str, object]: return { "role": "tool", "tool_call_id": call.id, "name": call.name, - "content": _run_tool(root, call), + "content": _run_tool(root, call, max_chars), } -def _run_tool(root: Path, call: ToolCall) -> str: +def _run_tool(root: Path, call: ToolCall, max_chars: int = _DEFAULT_MAX_CHARS) -> str: try: - payload = _dispatch_tool(root, call.name, _arguments(call.arguments)) + payload = _dispatch_tool(root, call.name, _arguments(call.arguments), max_chars) except Exception as exc: # noqa: BLE001 — tool failures stay in-band so the scan continues payload = {"error": f"{type(exc).__name__}: {exc}"} return json.dumps(payload, sort_keys=True) -def _dispatch_tool(root: Path, name: str, arguments: Mapping[str, object]) -> object: +def _dispatch_tool(root: Path, name: str, arguments: Mapping[str, object], default_max_chars: int = _DEFAULT_MAX_CHARS) -> object: if name == "read_file": path = _require_str(arguments, "path") - max_chars = _optional_int(arguments, "max_chars", _DEFAULT_MAX_CHARS) + max_chars = _optional_int(arguments, "max_chars", default_max_chars) return {"path": path, "content": read_file(root, path, max_chars=max_chars)} + if name == "read_definition": + symbol = _require_str(arguments, "symbol") + scope = arguments.get("path") + found = read_definition( + root, + symbol, + max_chars=_optional_int(arguments, "max_chars", default_max_chars), + path=str(scope) if isinstance(scope, str) and scope else None, + ) + return found if found is not None else {"symbol": symbol, "found": False} + if name == "entry_points": + scope = arguments.get("path") + return entry_points(root, path=str(scope) if isinstance(scope, str) and scope else None) + if name == "flows": + return flows(root, path=_require_str(arguments, "path"), function=_require_str(arguments, "function")) + if name == "obligations": + return obligations(root, path=_require_str(arguments, "path"), function=_require_str(arguments, "function")) if name == "grep_repo": pattern = _require_str(arguments, "pattern") max_matches = _optional_int(arguments, "max_matches", _DEFAULT_MAX_MATCHES) @@ -371,7 +474,9 @@ def _optional_int(arguments: Mapping[str, object], key: str, default: int) -> in return value -def _findings_from_content(root: Path, hotspots: Sequence[Hotspot], content: str) -> list[StaticFinding]: +def _findings_from_content( + root: Path, hotspots: Sequence[Hotspot], content: str, tags: Sequence[str] = ("tool-hunter",) +) -> list[StaticFinding]: payload = _load_json(content) if isinstance(payload, list): items = payload @@ -384,7 +489,7 @@ def _findings_from_content(root: Path, hotspots: Sequence[Hotspot], content: str for item in items: if not isinstance(item, dict): continue - finding = _finding_from_item(root, item, scores) + finding = _finding_from_item(root, item, scores, tags) if finding is not None: findings.append(finding) return sorted(findings, key=lambda item: item.finding_id) @@ -402,7 +507,9 @@ def _load_json(content: str) -> object | None: return parsed -def _finding_from_item(root: Path, item: Mapping[str, object], scores: Mapping[str, float]) -> StaticFinding | None: +def _finding_from_item( + root: Path, item: Mapping[str, object], scores: Mapping[str, float], tags: Sequence[str] = ("tool-hunter",) +) -> StaticFinding | None: raw_path = item.get("path") if not isinstance(raw_path, str) or not raw_path: return None @@ -431,7 +538,7 @@ def _finding_from_item(root: Path, item: Mapping[str, object], scores: Mapping[s reachability_status="unknown", reachability_evidence=[], reachability_conditions=[], - tags=["tool-hunter"], + tags=list(tags), ranking_priority=float(scores.get(relative, 0.0)), proposed_snippet=proposed_snippet, ) diff --git a/tests/test_hunter_curated_tools.py b/tests/test_hunter_curated_tools.py new file mode 100644 index 0000000..1263fdd --- /dev/null +++ b/tests/test_hunter_curated_tools.py @@ -0,0 +1,151 @@ +"""learning-harness task 1.5: curated repository tools and hunter-loop parameters (offline).""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from openultrasast.hunter_tools import PathEscapesRepo +from openultrasast.tool_hunter import ChatResponse, ScriptedChatClient, ToolCall, run_tool_hunter + +APP = ( + "from flask import request\n\n\n" + "@app.route('/books/')\n@login_required\ndef guarded(title):\n" + " return Book.query.filter_by(book_title=title).first()\n\n\n" + "@app.route('/books/any/<title>')\ndef leaky(title):\n" + " owner = request.args.get('owner')\n" + " return Book.query.filter_by(user_id=owner, book_title=title).first()\n\n\n" + "def helper(value):\n" + " return value.strip()\n" +) + + +@pytest.fixture +def repo(tmp_path: Path) -> Path: + (tmp_path / "app.py").write_text(APP) + return tmp_path + + +def test_read_definition_returns_one_span_and_refuses_to_leave_the_repository(repo: Path) -> None: + from openultrasast.hunter_tools import read_definition + + found = read_definition(repo, "leaky", max_chars=4000) + assert found is not None + assert found["path"] == "app.py" and found["name"] == "leaky" + assert found["start_line"] == 10 and found["end_line"] == 13 + assert str(found["text"]).startswith("@app.route('/books/any/<title>')") # a handler's decorators come with it + assert "def leaky" in str(found["text"]) and "def guarded" not in str(found["text"]) + assert read_definition(repo, "not_a_function", max_chars=4000) is None + with pytest.raises(PathEscapesRepo): + read_definition(repo, "leaky", max_chars=4000, path="../outside.py") + + +def test_entry_points_lists_handlers_with_their_access_and_never_source_text(repo: Path) -> None: + from openultrasast.hunter_tools import entry_points + + rows = entry_points(repo) + by_function = {str(row["function"]): row for row in rows if row.get("function")} + assert {"guarded", "leaky"} <= set(by_function) + assert by_function["guarded"]["access_level"] == "authenticated" and by_function["leaky"]["access_level"] == "public" + assert by_function["leaky"]["path"] == "app.py" and isinstance(by_function["leaky"]["line"], int) + assert all("text" not in row and "content" not in row for row in rows) # identifiers and spans only + assert entry_points(repo, path="app.py") == rows + with pytest.raises(PathEscapesRepo): + entry_points(repo, path="../elsewhere.py") + + +def test_obligations_lists_what_a_function_owes_without_dumping_code(repo: Path) -> None: + from openultrasast.hunter_tools import obligations + + rows = obligations(repo, path="app.py", function="leaky") + assert rows and all(row["function"] == "leaky" for row in rows) + assert {"protected_read"} == {str(row["operation"]) for row in rows} + assert "identity_constraint" in {str(row["missing"]) for row in rows} + assert all("text" not in row and "content" not in row for row in rows) + assert obligations(repo, path="app.py", function="helper") == [] + + +def test_flows_lists_adjudicated_source_to_sink_records_for_one_function(repo: Path) -> None: + from openultrasast.hunter_tools import flows + + (repo / "flow.py").write_text("from flask import request\nimport os\n\n\ndef run():\n os.system(request.args.get('cmd'))\n") + rows = flows(repo, path="flow.py", function="run") + assert all(row["function"] == "run" and row["path"] == "flow.py" for row in rows) + assert all(set(row) <= {"path", "function", "line", "sources", "sinks", "disposition", "cwe"} for row in rows) + assert flows(repo, path="app.py", function="helper") == [] + with pytest.raises(PathEscapesRepo): + flows(repo, path="/etc/passwd", function="run") + + +def test_the_loop_keeps_todays_behaviour_by_default(repo: Path) -> None: + from openultrasast.complexity.map import Hotspot + from openultrasast.tool_hunter import DEFAULT_TOOL_NAMES + + hotspot = Hotspot( + path="app.py", + function_name=None, + score=1.0, + band="high", + signals={}, + rationale="fixture", + test_hint=None, + inventory_finding_ids=(), + ) + client = ScriptedChatClient( + [ + ChatResponse(tool_calls=(ToolCall(id="1", name="read_file", arguments={"path": "app.py"}),)), + ChatResponse(content='[{"path": "app.py", "line": 12, "title": "t", "rationale": "r"}]'), + ] + ) + findings = run_tool_hunter(repo, [hotspot], client=client, model="m", max_steps=4) + assert [finding.finding_id for finding in findings] == ["tool-hunter:app.py:12"] + assert findings[0].tags == ["tool-hunter"] and findings[0].evidence_level == "suspicion" + assert tuple(tool["function"]["name"] for tool in client.calls[0]["tools"]) == DEFAULT_TOOL_NAMES + + +def test_the_loop_takes_a_prompt_context_tools_and_tags_and_replays_reasoning(repo: Path) -> None: + from openultrasast.complexity.map import Hotspot + + hotspot = Hotspot( + path="app.py", + function_name="leaky", + score=1.0, + band="high", + signals={}, + rationale="fixture", + test_hint=None, + inventory_finding_ids=(), + ) + client = ScriptedChatClient( + [ + ChatResponse( + content="looking", + reasoning="first I check the route", + tool_calls=(ToolCall(id="1", name="obligations", arguments={"path": "app.py", "function": "leaky"}),), + ), + ChatResponse(content='[{"path": "app.py", "line": 12, "title": "t", "rationale": "r"}]'), + ] + ) + findings = run_tool_hunter( + repo, + [hotspot], + client=client, + model="m", + max_steps=4, + system_prompt="You hunt access control only.", + user_prompt="Look at leaky.", + context_files=("app.py",), + tools=("read_definition", "obligations"), + tags=("family:access_control", "detector:access_control@1"), + ) + first = client.calls[0] + assert first["messages"][0]["content"] == "You hunt access control only." + assert "Look at leaky." in str(first["messages"][1]["content"]) and "def leaky" in str(first["messages"][1]["content"]) + assert [tool["function"]["name"] for tool in first["tools"]] == ["read_definition", "obligations"] + replayed = [message for message in client.calls[1]["messages"] if message.get("role") == "assistant"] + assert replayed and replayed[0]["reasoning_content"] == "first I check the route" + tool_reply = json.loads(str(client.calls[1]["messages"][-1]["content"])) + assert tool_reply and tool_reply[0]["missing"] == "identity_constraint" + assert findings[0].tags == ["family:access_control", "detector:access_control@1"]