diff --git a/strix/agents/factory.py b/strix/agents/factory.py index f559033ad..a86dbdecb 100644 --- a/strix/agents/factory.py +++ b/strix/agents/factory.py @@ -42,6 +42,11 @@ from strix.tools.proxy.tools import ( view_sitemap_entry, ) from strix.tools.reporting.tool import create_vulnerability_report +from strix.tools.test_log.tools import ( + query_tests, + record_test, + test_log_summary, +) from strix.tools.thinking.tool import think from strix.tools.todo.tools import ( create_todo, @@ -333,6 +338,9 @@ _BASE_TOOLS: tuple[Tool, ...] = ( get_note, update_note, delete_note, + record_test, + query_tests, + test_log_summary, web_search, create_vulnerability_report, list_requests, diff --git a/strix/agents/prompts/system_prompt.jinja b/strix/agents/prompts/system_prompt.jinja index 773d6a2af..7741443b7 100644 --- a/strix/agents/prompts/system_prompt.jinja +++ b/strix/agents/prompts/system_prompt.jinja @@ -283,6 +283,13 @@ ROOT AGENT ROLE: 8. **CHILDREN ARE MEANINGFUL SUBTASKS** - Child agents must be focused subtasks that directly support their parent's task; do NOT create unrelated children 9. **UNIQUENESS** - Do not create two agents with the same task; ensure clear, non-overlapping responsibilities for every agent +ENDPOINT TEST MEMORY (REQUIRED): +- Before testing any endpoint for a vulnerability class, call `query_tests(endpoint=..., vuln_class=...)` to see what prior agents already covered. If another agent confirmed `not_vulnerable` for the same endpoint+class with payloads you would have tried, do NOT re-test — pick a different endpoint or a different class. +- After each meaningful test attempt, call `record_test(endpoint, vuln_class, outcome, payloads=[short summaries], notes="")`. Use canonical `"METHOD /path/{id}"` form so entries de-duplicate across IDs. +- Outcomes: `vulnerable` (confirmed with PoC), `not_vulnerable` (tested, nothing found), `needs_more_testing` (partial signal, hand-off), `inconclusive` (couldn't get a clean read), `blocked` (auth/WAF/rate-limit prevented testing). +- The root agent should call `test_log_summary()` periodically to see coverage gaps and route new specialists toward uncovered surface. +- This log persists across `--resume`; on resume, every agent must query the log first to avoid re-running work the prior session already did. + WHEN TO CREATE NEW AGENTS: BLACK-BOX (domain/URL only): diff --git a/strix/core/execution.py b/strix/core/execution.py index 84fe72f84..464cdbeea 100644 --- a/strix/core/execution.py +++ b/strix/core/execution.py @@ -506,6 +506,7 @@ async def _start_child_runner( child_ctx: dict[str, Any] = dict(parent_ctx) child_ctx["agent_id"] = child_id + child_ctx["agent_name"] = name child_ctx["parent_id"] = parent_id child_ctx["task"] = task diff --git a/strix/core/runner.py b/strix/core/runner.py index 0e6d6583e..a764bd199 100644 --- a/strix/core/runner.py +++ b/strix/core/runner.py @@ -103,10 +103,12 @@ async def run_strix_scan( coordinator.set_snapshot_path(agents_path) from strix.tools.notes.tools import hydrate_notes_from_disk + from strix.tools.test_log.tools import hydrate_test_log_from_disk from strix.tools.todo.tools import hydrate_todos_from_disk hydrate_todos_from_disk(state_dir) hydrate_notes_from_disk(state_dir) + hydrate_test_log_from_disk(state_dir) root_id: str | None = None if is_resume: @@ -211,6 +213,7 @@ async def run_strix_scan( "sandbox_session": bundle["session"], "caido_client": bundle["caido_client"], "agent_id": root_id, + "agent_name": "strix", "parent_id": None, "interactive": interactive, "spawn_child_agent": spawn_child_agent, diff --git a/strix/interface/assets/tui_styles.tcss b/strix/interface/assets/tui_styles.tcss index d2984c8bc..74d427c6a 100644 --- a/strix/interface/assets/tui_styles.tcss +++ b/strix/interface/assets/tui_styles.tcss @@ -205,7 +205,7 @@ VulnerabilityDetailScreen { } #chat_area_container { - width: 80%; + width: 44%; background: transparent; } @@ -213,6 +213,66 @@ VulnerabilityDetailScreen { width: 100%; } +#activity_panel { + width: 18%; + background: transparent; + margin-right: 1; +} + +#activity_panel.-hidden { + display: none; +} + +#activity_scroll { + height: 1fr; + background: transparent; + border: round #333333; + border-title-color: #a8a29e; + border-title-style: bold; + padding: 0 1; + scrollbar-background: #000000; + scrollbar-color: #1a1a1a; + scrollbar-corner-color: #000000; + scrollbar-size: 1 1; +} + +#activity_display { + height: auto; + background: transparent; + padding: 0; + margin: 0; +} + +#plan_panel { + width: 18%; + background: transparent; + margin-left: 1; +} + +#plan_panel.-hidden { + display: none; +} + +#plan_scroll { + height: 1fr; + background: transparent; + border: round #333333; + border-title-color: #a8a29e; + border-title-style: bold; + padding: 0 1; + scrollbar-background: #000000; + scrollbar-color: #1a1a1a; + scrollbar-corner-color: #000000; + scrollbar-size: 1 1; +} + +#plan_display { + height: auto; + background: transparent; + padding: 0; + margin: 0; +} + #chat_history { height: 1fr; background: transparent; diff --git a/strix/interface/tui/app.py b/strix/interface/tui/app.py index 65a491cc8..5a4fece4a 100644 --- a/strix/interface/tui/app.py +++ b/strix/interface/tui/app.py @@ -846,9 +846,19 @@ class StrixTUIApp(App): # type: ignore[misc] vulnerabilities_panel = VulnerabilitiesPanel(id="vulnerabilities_panel") + activity_display = Static("", id="activity_display") + activity_scroll = VerticalScroll(activity_display, id="activity_scroll") + activity_panel = Vertical(activity_scroll, id="activity_panel") + + plan_display = Static("", id="plan_display") + plan_scroll = VerticalScroll(plan_display, id="plan_scroll") + plan_panel = Vertical(plan_scroll, id="plan_panel") + sidebar = Vertical(agents_tree, vulnerabilities_panel, stats_scroll, id="sidebar") + content_container.mount(activity_panel) content_container.mount(chat_area_container) + content_container.mount(plan_panel) content_container.mount(sidebar) chat_area_container.mount(chat_history) @@ -937,6 +947,10 @@ class StrixTUIApp(App): # type: ignore[misc] self._update_vulnerabilities_panel() + self._update_activity_panel() + + self._update_plan_panel() + def _sync_agent_graph(self) -> None: future = self._agent_graph_sync_future if future is not None: @@ -1277,6 +1291,157 @@ class StrixTUIApp(App): # type: ignore[misc] self._safe_widget_operation(vuln_panel.remove_class, "hidden") vuln_panel.update_vulnerabilities(enriched_vulns) + def _update_activity_panel(self) -> None: + try: + display = self.query_one("#activity_display", Static) + except (ValueError, Exception): + return + if not self._is_widget_safe(display): + return + + agent_id = self.selected_agent_id + if not agent_id or agent_id not in self.live_view.agents: + agent_id = next(iter(self.live_view.agents), None) + + text = Text() + agent_name = "" + if agent_id: + agent_name = self.live_view.agents.get(agent_id, {}).get("name", agent_id) + text.append("Activity", style="bold #a8a29e") + if agent_name: + text.append(" ") + text.append(agent_name, style="dim") + text.append("\n") + + events = self._gather_agent_events(agent_id) if agent_id else [] + tool_events = [e for e in events if e.get("type") == "tool"] + + if not tool_events: + text.append("\n") + text.append("no tool calls yet", style="dim") + self._safe_widget_operation(display.update, text) + return + + recent = tool_events[-200:] + status_color = { + "completed": "#22c55e", + "running": "#eab308", + "error": "#ef4444", + "failed": "#ef4444", + } + for event in recent: + data = event.get("data") or {} + tool_name = str(data.get("tool_name") or "?") + status = str(data.get("status") or "running").lower() + color = status_color.get(status, "#a8a29e") + timestamp = event.get("timestamp", "") + ts_suffix = timestamp[11:19] if len(timestamp) >= 19 else "" + + marker = "●" if status == "running" else "✓" if status == "completed" else "✗" + text.append(f"{marker} ", style=color) + text.append(tool_name, style="white") + args = data.get("args") + if isinstance(args, dict) and args: + summary = self._summarize_tool_args(tool_name, args) + if summary: + text.append(" ") + text.append(summary, style="dim") + if ts_suffix: + text.append(f" {ts_suffix}", style="#6b7280") + text.append("\n") + + self._safe_widget_operation(display.update, text) + + @staticmethod + def _summarize_tool_args(tool_name: str, args: dict[str, Any]) -> str: + priority_keys = ( + "endpoint", + "url", + "path", + "target", + "command", + "vuln_class", + "skill", + "title", + "name", + "query", + ) + for key in priority_keys: + value = args.get(key) + if isinstance(value, str) and value: + trimmed = value.strip().replace("\n", " ") + if len(trimmed) > 60: + trimmed = trimmed[:57] + "..." + return trimmed + for value in args.values(): + if isinstance(value, str) and value: + trimmed = value.strip().replace("\n", " ") + if len(trimmed) > 60: + trimmed = trimmed[:57] + "..." + return trimmed + return "" + + def _update_plan_panel(self) -> None: + try: + display = self.query_one("#plan_display", Static) + except (ValueError, Exception): + return + if not self._is_widget_safe(display): + return + + from strix.tools.todo.tools import _sorted_todos, _todos_storage + + agent_id = self.selected_agent_id + if not agent_id or agent_id not in self.live_view.agents: + agent_id = next(iter(self.live_view.agents), None) + + text = Text() + text.append("Plan / Todos", style="bold #a8a29e") + if agent_id: + agent_name = self.live_view.agents.get(agent_id, {}).get("name", agent_id) + text.append(" ") + text.append(agent_name, style="dim") + text.append("\n") + + if not agent_id: + text.append("\nno agent selected", style="dim") + self._safe_widget_operation(display.update, text) + return + + todos = _sorted_todos(agent_id) if agent_id in _todos_storage else [] + if not todos: + text.append("\nno todos yet", style="dim") + self._safe_widget_operation(display.update, text) + return + + status_marker = {"done": "[x]", "in_progress": "[~]", "pending": "[ ]"} + status_color = { + "done": "#22c55e", + "in_progress": "#eab308", + "pending": "#a8a29e", + } + priority_color = { + "critical": "#ef4444", + "high": "#f97316", + "normal": "#a8a29e", + "low": "#6b7280", + } + for todo in todos: + status = str(todo.get("status") or "pending") + priority = str(todo.get("priority") or "normal") + marker = status_marker.get(status, "[?]") + text.append(marker, style=status_color.get(status, "#a8a29e")) + text.append(" ") + if priority in ("critical", "high"): + text.append(f"({priority[0].upper()}) ", style=priority_color.get(priority)) + content = str(todo.get("content") or todo.get("title") or "") + if len(content) > 200: + content = content[:197] + "..." + text.append(content, style="white" if status != "done" else "dim") + text.append("\n") + + self._safe_widget_operation(display.update, text) + def _get_sweep_animation(self, color_palette: list[str]) -> Text: text = Text() num_squares = self._sweep_num_squares @@ -1377,6 +1542,8 @@ class StrixTUIApp(App): # type: ignore[misc] self._displayed_events.clear() self.call_later(self._update_chat_view) + self.call_later(self._update_activity_panel) + self.call_later(self._update_plan_panel) self._update_agent_status_display() def _start_scan_thread(self) -> None: diff --git a/strix/skills/coordination/root_agent.md b/strix/skills/coordination/root_agent.md index 981c3b973..b2f216fe4 100644 --- a/strix/skills/coordination/root_agent.md +++ b/strix/skills/coordination/root_agent.md @@ -90,3 +90,14 @@ When all agents report completion: 2. Assess overall security posture 3. Compile executive summary with prioritized recommendations 4. Invoke finish tool with final report + +## Test-Coverage Memory + +A persistent endpoint test log lives in `{state_dir}/test_log.json` and is exposed via three tools: `record_test`, `query_tests`, and `test_log_summary`. + +As the coordinator: + +- Before spawning a new specialist, run `test_log_summary()` and `query_tests(vuln_class="")` to see whether that surface is already covered. +- Brief each spawned specialist with the coverage already done so it can `query_tests(endpoint=...)` to avoid duplicate work. +- On `--resume`, your first action is `test_log_summary()` — it tells you what the prior session finished, what's partial (`needs_more_testing`), and what's untouched. +- The Vulnerability panel shows confirmed `findings`; the test log records `negative` outcomes too, which the vulnerability panel never will. diff --git a/strix/skills/scan_modes/deep.md b/strix/skills/scan_modes/deep.md index 657bd55c7..68731e413 100644 --- a/strix/skills/scan_modes/deep.md +++ b/strix/skills/scan_modes/deep.md @@ -165,3 +165,7 @@ Before spawning vulnerability agents, spawn a single `JS Analysis Agent` with `s Relentless. Creative. Patient. Thorough. Persistent. This is about finding what others miss. Test every parameter, every endpoint, every edge case. If one approach fails, try ten more. Understand how components interact to find systemic issues. + +## Test Memory + +Every endpoint test must be recorded via `record_test` and every endpoint must be checked via `query_tests` before testing — this is how deep mode avoids burning iterations on already-covered ground, especially after `--resume`. The root agent uses `test_log_summary` to spot coverage gaps. diff --git a/strix/skills/scan_modes/hunter.md b/strix/skills/scan_modes/hunter.md index f182d9f50..a4561b304 100644 --- a/strix/skills/scan_modes/hunter.md +++ b/strix/skills/scan_modes/hunter.md @@ -197,3 +197,4 @@ Before spawning vulnerability agents, spawn a single `JS Analysis Agent` with `s - Never test targets not explicitly authorized - Never stop mid-scan without a summary of findings - Zero hallucination: never write a response you didn't receive +- **Test memory is mandatory**: call `query_tests` before testing any endpoint and `record_test` after every attempt. On `--resume`, this lets agents skip ground already covered. diff --git a/strix/skills/scan_modes/quick.md b/strix/skills/scan_modes/quick.md index 6bf19225a..e8938832c 100644 --- a/strix/skills/scan_modes/quick.md +++ b/strix/skills/scan_modes/quick.md @@ -70,3 +70,7 @@ When a strong primitive is found (auth weakness, injection point, internal acces ## Mindset Think like a time-boxed bug bounty hunter going for quick wins. Prioritize breadth over depth on critical areas. If something looks exploitable, validate quickly and move on. Don't get stuck—if an attack vector isn't yielding results quickly, pivot. + +## Test Memory + +Even in quick mode, call `query_tests` before testing an endpoint and `record_test` after — this is the cheapest possible deduplication. On `--resume` it lets you skip directly to surface you haven't touched yet. diff --git a/strix/skills/scan_modes/standard.md b/strix/skills/scan_modes/standard.md index 14c26c776..edf80f853 100644 --- a/strix/skills/scan_modes/standard.md +++ b/strix/skills/scan_modes/standard.md @@ -101,3 +101,7 @@ When you discover a useful pivot (info leak, weak boundary, partial access), imm ## Mindset Methodical and systematic. Document as you go. Validate everything—no assumptions about exploitability. Think about business impact, not just technical severity. + +## Test Memory + +Before testing an endpoint, call `query_tests(endpoint=..., vuln_class=...)`. After every test attempt — including negatives — call `record_test(...)`. Other agents (and the resumed session) read this log to skip duplicate work. diff --git a/strix/tools/test_log/__init__.py b/strix/tools/test_log/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/strix/tools/test_log/tools.py b/strix/tools/test_log/tools.py new file mode 100644 index 000000000..a8c4366e1 --- /dev/null +++ b/strix/tools/test_log/tools.py @@ -0,0 +1,455 @@ +"""Per-run endpoint test memory — mirrored to {state_dir}/test_log.json. + +Tracks which endpoints have been tested for which vulnerability classes, +which payloads were tried, and what the outcome was. Persists across +``--resume`` so spawned agents don't re-test ground other agents already +covered. + +Shape on disk:: + + { + "entries": { + "": { + "endpoint": "GET /api/v2/orders/{id}", + "vuln_class": "idor", + "payloads": ["id=123 -> 200 victim data", ...], + "outcome": "vulnerable" | "not_vulnerable" | "needs_more_testing", + "agent_id": "", + "agent_name": "IDOR Agent", + "notes": "...", + "tags": ["auth", "production"], + "created_at": "...", + "updated_at": "..." + } + }, + "by_endpoint": { "": ["", ...] } + } +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import tempfile +import threading +import uuid +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +from agents import RunContextWrapper, function_tool + + +logger = logging.getLogger(__name__) + + +_VALID_OUTCOMES = { + "vulnerable", + "not_vulnerable", + "needs_more_testing", + "inconclusive", + "blocked", +} + +_VALID_VULN_CLASSES = { + "idor", + "ssrf", + "xss", + "sqli", + "rce", + "lfi", + "rfi", + "xxe", + "ssti", + "auth_bypass", + "broken_access_control", + "csrf", + "open_redirect", + "file_upload", + "race_condition", + "business_logic", + "mass_assignment", + "nosql_injection", + "header_injection", + "http_smuggling", + "cache_poison", + "subdomain_takeover", + "info_disclosure", + "secrets_exposure", + "graphql", + "oauth", + "saml", + "mfa_bypass", + "prototype_pollution", + "other", +} + + +_storage: dict[str, dict[str, Any]] = {"entries": {}, "by_endpoint": {}} +_lock = threading.RLock() +_path: Path | None = None + + +def hydrate_test_log_from_disk(state_dir: Path) -> None: + global _path # noqa: PLW0603 + _path = state_dir / "test_log.json" + with _lock: + _storage["entries"] = {} + _storage["by_endpoint"] = {} + if not _path.exists(): + return + try: + data = json.loads(_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + logger.exception( + "test_log.json at %s is unreadable; starting empty", + _path, + ) + return + if not isinstance(data, dict): + return + entries = data.get("entries") + if isinstance(entries, dict): + _storage["entries"] = { + eid: e + for eid, e in entries.items() + if isinstance(eid, str) and isinstance(e, dict) + } + by_endpoint = data.get("by_endpoint") + if isinstance(by_endpoint, dict): + _storage["by_endpoint"] = { + ep: list(ids) + for ep, ids in by_endpoint.items() + if isinstance(ep, str) and isinstance(ids, list) + } + else: + for eid, entry in _storage["entries"].items(): + ep = entry.get("endpoint") + if isinstance(ep, str): + _storage["by_endpoint"].setdefault(ep, []).append(eid) + logger.info( + "test_log hydrated from %s (%d entries)", + _path, + len(_storage["entries"]), + ) + + +def _persist() -> None: + path = _path + if path is None: + return + try: + payload = json.dumps(_storage, ensure_ascii=False, default=str) + path.parent.mkdir(parents=True, exist_ok=True) + with ( + _lock, + tempfile.NamedTemporaryFile( + mode="w", + encoding="utf-8", + dir=str(path.parent), + prefix=f".{path.name}.", + suffix=".tmp", + delete=False, + ) as tmp, + ): + tmp.write(payload) + tmp_path = Path(tmp.name) + tmp_path.replace(path) + except Exception: + logger.exception("test_log persist to %s failed", path) + + +def _agent_meta_from(ctx: RunContextWrapper) -> tuple[str, str]: + inner = ctx.context if isinstance(ctx.context, dict) else {} + agent_id = str(inner.get("agent_id") or "unknown") + agent_name = str(inner.get("agent_name") or agent_id) + return agent_id, agent_name + + +def _normalize_endpoint(endpoint: str) -> str: + return endpoint.strip() + + +def _record_test_impl( + endpoint: str, + vuln_class: str, + outcome: str, + payloads: list[str] | None, + notes: str, + tags: list[str] | None, + agent_id: str, + agent_name: str, +) -> dict[str, Any]: + with _lock: + if not endpoint or not endpoint.strip(): + return {"success": False, "error": "endpoint cannot be empty"} + vc = vuln_class.strip().lower() + if vc not in _VALID_VULN_CLASSES: + return { + "success": False, + "error": ( + f"unknown vuln_class '{vuln_class}'. valid: " + f"{sorted(_VALID_VULN_CLASSES)}" + ), + } + oc = outcome.strip().lower() + if oc not in _VALID_OUTCOMES: + return { + "success": False, + "error": ( + f"unknown outcome '{outcome}'. valid: {sorted(_VALID_OUTCOMES)}" + ), + } + + ep = _normalize_endpoint(endpoint) + timestamp = datetime.now(UTC).isoformat() + + existing_id: str | None = None + for eid in _storage["by_endpoint"].get(ep, []): + entry = _storage["entries"].get(eid) + if entry and entry.get("vuln_class") == vc: + existing_id = eid + break + + if existing_id is not None: + entry = _storage["entries"][existing_id] + if payloads: + merged = list(entry.get("payloads") or []) + for p in payloads: + if p and p not in merged: + merged.append(p) + entry["payloads"] = merged + if notes: + prior = entry.get("notes") or "" + if notes not in prior: + entry["notes"] = (prior + "\n" + notes).strip() if prior else notes + if tags: + merged_tags = list(entry.get("tags") or []) + for t in tags: + if t and t not in merged_tags: + merged_tags.append(t) + entry["tags"] = merged_tags + entry["outcome"] = oc + entry["updated_at"] = timestamp + entry.setdefault("agent_history", []).append( + {"agent_id": agent_id, "agent_name": agent_name, "at": timestamp} + ) + _persist() + return { + "success": True, + "entry_id": existing_id, + "merged": True, + "endpoint": ep, + "vuln_class": vc, + "outcome": oc, + "total_entries": len(_storage["entries"]), + } + + entry_id = uuid.uuid4().hex[:8] + entry = { + "endpoint": ep, + "vuln_class": vc, + "payloads": list(payloads or []), + "outcome": oc, + "notes": notes or "", + "tags": list(tags or []), + "agent_id": agent_id, + "agent_name": agent_name, + "agent_history": [ + {"agent_id": agent_id, "agent_name": agent_name, "at": timestamp} + ], + "created_at": timestamp, + "updated_at": timestamp, + } + _storage["entries"][entry_id] = entry + _storage["by_endpoint"].setdefault(ep, []).append(entry_id) + _persist() + return { + "success": True, + "entry_id": entry_id, + "merged": False, + "endpoint": ep, + "vuln_class": vc, + "outcome": oc, + "total_entries": len(_storage["entries"]), + } + + +def _query_tests_impl( + endpoint: str | None, + vuln_class: str | None, + outcome: str | None, + agent_id: str | None, + tag: str | None, + limit: int, +) -> dict[str, Any]: + with _lock: + results: list[dict[str, Any]] = [] + ep_norm = _normalize_endpoint(endpoint) if endpoint else None + vc_norm = vuln_class.strip().lower() if vuln_class else None + oc_norm = outcome.strip().lower() if outcome else None + + for eid, entry in _storage["entries"].items(): + if ep_norm and ep_norm not in entry.get("endpoint", ""): + continue + if vc_norm and entry.get("vuln_class") != vc_norm: + continue + if oc_norm and entry.get("outcome") != oc_norm: + continue + if agent_id and entry.get("agent_id") != agent_id: + continue + if tag and tag not in (entry.get("tags") or []): + continue + results.append({"entry_id": eid, **entry}) + + results.sort(key=lambda r: r.get("updated_at", ""), reverse=True) + truncated = len(results) > limit + return { + "success": True, + "entries": results[:limit], + "returned": min(len(results), limit), + "matched": len(results), + "total_in_log": len(_storage["entries"]), + "truncated": truncated, + } + + +def _summary_impl() -> dict[str, Any]: + with _lock: + by_class: dict[str, int] = {} + by_outcome: dict[str, int] = {} + for entry in _storage["entries"].values(): + by_class[entry.get("vuln_class", "?")] = ( + by_class.get(entry.get("vuln_class", "?"), 0) + 1 + ) + by_outcome[entry.get("outcome", "?")] = ( + by_outcome.get(entry.get("outcome", "?"), 0) + 1 + ) + return { + "success": True, + "total_entries": len(_storage["entries"]), + "unique_endpoints": len(_storage["by_endpoint"]), + "by_vuln_class": by_class, + "by_outcome": by_outcome, + } + + +@function_tool(timeout=30) +async def record_test( + ctx: RunContextWrapper, + endpoint: str, + vuln_class: str, + outcome: str, + payloads: list[str] | None = None, + notes: str = "", + tags: list[str] | None = None, +) -> str: + """Record that you tested an endpoint for a vulnerability class. + + Call this **after every meaningful test attempt** so other agents + (and this same agent after a ``--resume``) can see what's already + been covered and skip duplicate work. + + If an entry for the same ``endpoint`` + ``vuln_class`` already + exists, this **merges** into it: payloads and tags are + de-duplicated and appended, notes are concatenated, and ``outcome`` + is overwritten with the new value (so re-testing can promote + ``needs_more_testing`` to ``vulnerable``). + + Args: + endpoint: Canonical request identifier. Prefer + ``"METHOD /path"`` (e.g. ``"GET /api/v2/orders/{id}"``). + Keep path-param shapes (``{id}``) instead of concrete + values so the entry de-duplicates across IDs. + vuln_class: One of the supported classes (``idor``, ``ssrf``, + ``xss``, ``sqli``, ``rce``, ``lfi``, ``rfi``, ``xxe``, + ``ssti``, ``auth_bypass``, ``broken_access_control``, + ``csrf``, ``open_redirect``, ``file_upload``, + ``race_condition``, ``business_logic``, + ``mass_assignment``, ``nosql_injection``, + ``header_injection``, ``http_smuggling``, + ``cache_poison``, ``subdomain_takeover``, + ``info_disclosure``, ``secrets_exposure``, ``graphql``, + ``oauth``, ``saml``, ``mfa_bypass``, + ``prototype_pollution``, ``other``). + outcome: One of ``vulnerable``, ``not_vulnerable``, + ``needs_more_testing``, ``inconclusive``, ``blocked``. + payloads: Short list of payload summaries you actually sent, + each ideally one line including the response signal + (``"id=2 -> 200 victim PII returned"``). Skip raw + multi-line HTTP — keep it scannable. + notes: Free-form notes that the next agent would benefit from + (auth required, WAF behavior, rate limit observed, etc.). + tags: Free-form tags for later filtering + (e.g. ``["auth", "production", "v2-api"]``). + """ + agent_id, agent_name = _agent_meta_from(ctx) + return json.dumps( + await asyncio.to_thread( + _record_test_impl, + endpoint, + vuln_class, + outcome, + payloads, + notes, + tags, + agent_id, + agent_name, + ), + ensure_ascii=False, + default=str, + ) + + +@function_tool(timeout=30) +async def query_tests( + ctx: RunContextWrapper, + endpoint: str | None = None, + vuln_class: str | None = None, + outcome: str | None = None, + agent_id: str | None = None, + tag: str | None = None, + limit: int = 50, +) -> str: + """Look up prior test attempts before spending time re-testing. + + Call this **before** any vulnerability testing on an endpoint — + another agent (or yourself in a prior run) may have already covered + the same surface. Filters compose with AND semantics. + + Args: + endpoint: Substring match against recorded endpoint strings + (e.g. ``"/api/v2/orders"`` matches + ``"GET /api/v2/orders/{id}"``). + vuln_class: Exact match against vuln class. + outcome: Exact match against outcome. + agent_id: Filter to a specific recording agent. + tag: Match entries that include this tag. + limit: Cap on returned entries (default 50). + """ + return json.dumps( + await asyncio.to_thread( + _query_tests_impl, + endpoint, + vuln_class, + outcome, + agent_id, + tag, + max(1, min(limit, 500)), + ), + ensure_ascii=False, + default=str, + ) + + +@function_tool(timeout=30) +async def test_log_summary(ctx: RunContextWrapper) -> str: + """Coverage snapshot: how many endpoints have been tested, broken + down by vuln class and outcome. Useful for the root agent to decide + where to spawn the next specialist.""" + return json.dumps( + await asyncio.to_thread(_summary_impl), + ensure_ascii=False, + default=str, + )