diff --git a/strix/agents/prompts/system_prompt.jinja b/strix/agents/prompts/system_prompt.jinja index 4a2f55c29..b0683a9ff 100644 --- a/strix/agents/prompts/system_prompt.jinja +++ b/strix/agents/prompts/system_prompt.jinja @@ -217,6 +217,15 @@ VALIDATION APPROACH: - Chain vulnerabilities when needed to demonstrate maximum impact - Focus on demonstrating real business impact +ATTACK PLANNING METHODOLOGY: +Before testing a vulnerability type, reason deeply about the attack — do not spray payloads blindly. + +1. **Load the skill first**: Call `load_skill` with the relevant vulnerability name (e.g., `["sql_injection"]`, `["xss"]`, `["ssrf"]`, `["business_logic"]`) to get the full methodology — attack surfaces, detection channels, bypass techniques, technology-specific primitives, and testing workflow. Do not plan attacks from memory alone. +2. **Use `think` to plan**: Before acting, reason through: target technology and how it handles input/sessions/auth → which endpoints and parameters are user-controlled → which techniques are most likely to succeed given the technology → what defenses (WAF, validation, encoding) are in place and how to bypass them → what encoding tricks or syntax variations could work → whether findings can be chained for greater impact +3. **Match techniques to technology**: Do not test irrelevant attack vectors. NoSQL injection is pointless against PostgreSQL. XXE is irrelevant if no XML parsing occurs. SSRF requires server-side URL fetching. Select techniques based on what the target actually does. +4. **Plan bypass strategies**: When standard payloads are filtered, reason about alternatives — encoding variations (double URL, unicode, NFKC), syntax tricks (comment splitting, case folding), alternative injection points (headers, cookies, path params), and protocol-level bypasses. +5. **Chain for impact**: Think about how low-severity findings combine into high-impact attacks. IDOR + business logic = financial fraud. SSRF + cloud metadata = credential theft. XSS + CSRF = account takeover. Stored XSS + admin viewing = privilege escalation. + VULNERABILITY KNOWLEDGE BASE: You have access to comprehensive guides for each vulnerability type above. Use these references for: - Discovery techniques and automation @@ -259,8 +268,10 @@ WHITE-BOX TESTING - PHASE 1 (CODE UNDERSTANDING): - ONLY AFTER full code comprehension → proceed to vulnerability testing PHASE 2 - SYSTEMATIC VULNERABILITY TESTING: -- CREATE SPECIALIZED SUBAGENT for EACH vulnerability type × EACH component -- Each agent focuses on ONE vulnerability type in ONE specific location +- ANALYZE the target's technology stack and attack surface to determine which vulnerability types are relevant +- LOAD the relevant vulnerability skill via `load_skill` and use `think` to plan the attack strategy before spawning agents +- CREATE SPECIALIZED SUBAGENTS for vulnerability types that match the target's technology and attack surface — not blindly for every category +- Each agent focuses on ONE vulnerability type in ONE specific location with a task description informed by your analysis - EVERY detected vulnerability MUST spawn its own validation subagent SIMPLE WORKFLOW RULES: diff --git a/strix/interface/tui/app.py b/strix/interface/tui/app.py index 4e43d9795..5347c5a0e 100644 --- a/strix/interface/tui/app.py +++ b/strix/interface/tui/app.py @@ -730,6 +730,9 @@ def __init__(self, args: argparse.Namespace): ] self._dot_animation_timer: Any | None = None + self._event_buffer: list[tuple[str, Any]] = [] + self._event_flush_timer: Any | None = None + self._setup_cleanup_handlers() def _build_scan_config(self, args: argparse.Namespace) -> dict[str, Any]: @@ -1292,7 +1295,7 @@ def _get_animated_verb_text(self, agent_id: str, verb: str) -> Text: # noqa: AR def _start_dot_animation(self) -> None: if self._dot_animation_timer is None: - self._dot_animation_timer = self.set_interval(0.06, self._animate_dots) + self._dot_animation_timer = self.set_interval(0.15, self._animate_dots) def _stop_dot_animation(self) -> None: if self._dot_animation_timer is not None: @@ -1335,9 +1338,13 @@ def _agent_vulnerability_count(self, agent_id: str) -> int: if vuln.get("agent_id") == agent_id ) + MAX_RENDER_EVENTS = 200 + def _gather_agent_events(self, agent_id: str) -> list[dict[str, Any]]: events = self.live_view.events_for_agent(agent_id) events.sort(key=lambda e: (e["timestamp"], e["id"])) + if len(events) > self.MAX_RENDER_EVENTS: + events = events[-self.MAX_RENDER_EVENTS:] return events def watch_selected_agent_id(self, _agent_id: str | None) -> None: @@ -1407,10 +1414,25 @@ def scan_target() -> None: self._scan_thread.start() def _capture_sdk_event(self, agent_id: str, event: Any) -> None: + self._event_buffer.append((agent_id, event)) + if self._event_flush_timer is None: + self._event_flush_timer = self.set_timer(0.15, self._flush_event_buffer) + + def _flush_event_buffer(self) -> None: + self._event_flush_timer = None + if not self._event_buffer: + return + batch = self._event_buffer[:] + self._event_buffer.clear() try: - self.call_from_thread(self._record_sdk_event, agent_id, event) + self.call_from_thread(self._record_sdk_events_batch, batch) except RuntimeError: - self._record_sdk_event(agent_id, event) + for agent_id, event in batch: + self._record_sdk_event(agent_id, event) + + def _record_sdk_events_batch(self, batch: list[tuple[str, Any]]) -> None: + for agent_id, event in batch: + self.live_view.ingest_sdk_event(agent_id, event) def _record_sdk_event(self, agent_id: str, event: Any) -> None: self.live_view.ingest_sdk_event(agent_id, event) diff --git a/strix/interface/tui/live_view.py b/strix/interface/tui/live_view.py index 993074d67..4f5976cb1 100644 --- a/strix/interface/tui/live_view.py +++ b/strix/interface/tui/live_view.py @@ -14,10 +14,14 @@ from strix.interface.tui.history import load_session_history +MAX_EVENTS_PER_AGENT = 500 + + class TuiLiveView: def __init__(self) -> None: self.agents: dict[str, dict[str, Any]] = {} self.events: list[dict[str, Any]] = [] + self._events_by_agent: dict[str, list[dict[str, Any]]] = {} self._next_event_id = 1 self._open_assistant_event_by_agent: dict[str, dict[str, Any]] = {} self._tool_event_by_call_id: dict[str, dict[str, Any]] = {} @@ -115,10 +119,10 @@ def ingest_sdk_event(self, agent_id: str, event: Any) -> None: self._record_tool_output(agent_id, item) def events_for_agent(self, agent_id: str) -> list[dict[str, Any]]: - return [event for event in self.events if event.get("agent_id") == agent_id] + return list(self._events_by_agent.get(agent_id, [])) def has_events_for_agent(self, agent_id: str) -> bool: - return any(event.get("agent_id") == agent_id for event in self.events) + return bool(self._events_by_agent.get(agent_id)) def _ingest_raw_response_event(self, agent_id: str, data: Any) -> None: data_type = getattr(data, "type", "") @@ -277,6 +281,16 @@ def _append_event( } self._next_event_id += 1 self.events.append(event) + agent_events = self._events_by_agent.setdefault(agent_id, []) + agent_events.append(event) + if len(agent_events) > MAX_EVENTS_PER_AGENT: + dropped = agent_events[:len(agent_events) - MAX_EVENTS_PER_AGENT] + for d in dropped: + try: + self.events.remove(d) + except ValueError: + pass + del agent_events[:len(agent_events) - MAX_EVENTS_PER_AGENT] return event @staticmethod diff --git a/strix/skills/coordination/root_agent.md b/strix/skills/coordination/root_agent.md index 981c3b973..c88da1511 100644 --- a/strix/skills/coordination/root_agent.md +++ b/strix/skills/coordination/root_agent.md @@ -25,6 +25,30 @@ Before spawning agents, analyze the target: 3. **Determine approach** - blackbox, greybox, or whitebox assessment 4. **Prioritize by risk** - critical assets and high-value targets first +## Attack Planning + +Before spawning subagents for a vulnerability type, **think deeply about the attack strategy**. Do not blindly spawn agents for every vulnerability category — reason about what is actually relevant to the target. + +### Step 1: Load Vulnerability Methodology + +Before planning an attack against a specific vulnerability type, call `load_skill` with the relevant skill name (e.g., `["sql_injection"]`, `["xss"]`, `["business_logic"]`, `["ssrf"]`, `["xxe"]`). This gives you the full methodology — attack surfaces, detection channels, bypass techniques, DBMS-specific primitives, testing workflow, and validation steps. **Do not plan attacks from memory alone — load the skill first.** + +### Step 2: Reason About the Attack + +Use `think` to reason through the attack plan before spawning agents. Structure your thinking: + +- **Target technology**: What framework, language, database, server is in use? How does it handle input, auth, sessions? +- **Attack surface mapping**: Which endpoints, parameters, headers, cookies are user-controlled? Where does data flow to (database, filesystem, OS, other services)? +- **Technique selection**: Given the technology, which attack techniques are most likely to succeed? Which are irrelevant? (e.g., NoSQL injection is irrelevant for a PostgreSQL backend; XXE is irrelevant if no XML parsing occurs) +- **Defense analysis**: What WAF, input validation, encoding, or framework protections are in place? How might they be bypassed? +- **Bypass strategy**: If standard payloads are filtered, what encoding tricks, syntax variations, or alternative injection points could work? +- **Chaining opportunities**: Can findings be combined for greater impact? (e.g., IDOR + business logic = financial fraud; SSRF + metadata = cloud compromise; XSS + CSRF = account takeover) +- **Priority ordering**: Which attacks should be attempted first based on likelihood of success and potential impact? + +### Step 3: Spawn Targeted Agents + +Only after loading the skill and reasoning through the attack, spawn subagents with specific, informed task descriptions. The task description should reflect your analysis — tell the subagent what technology to target, what techniques to prioritize, what bypasses to try, and what the success criteria are. A well-planned task description produces far better results than a generic "test for SQLi." + ## Agent Architecture Structure agents by function: @@ -84,9 +108,17 @@ Complex findings warrant specialized subagents: ## Completion -When all agents report completion: +**WAIT for all agents to self-terminate.** Do NOT call `stop_agent` on active children to clear them before finishing — this orphans their work and produces an incomplete report. + +When you believe all work is done: + +1. Call `view_agent_graph` to check every agent's status +2. If ANY agent is still `running` or `waiting`, call `wait_for_message` to block until their completion reports arrive — do NOT stop them +3. Only proceed when ALL agents show `completed` or `crashed` (these are the only safe terminal states) +4. `stopped` agents were forcibly cancelled — their results are lost. If agents were stopped, their work must be re-done by spawning replacement agents +5. Collect and deduplicate findings across all completion reports +6. Assess overall security posture +7. Compile executive summary with prioritized recommendations +8. Invoke `finish_scan` with the final report -1. Collect and deduplicate findings across agents -2. Assess overall security posture -3. Compile executive summary with prioritized recommendations -4. Invoke finish tool with final report +**Never use `stop_agent` as a shortcut to bypass the active-agent check in `finish_scan`.** The check exists to prevent incomplete reports. If a child is taking too long, send it a message asking for status or telling it to wrap up via `agent_finish` — don't stop it. diff --git a/strix/tools/agents_graph/tools.py b/strix/tools/agents_graph/tools.py index 5c1bfd453..dc70568b9 100644 --- a/strix/tools/agents_graph/tools.py +++ b/strix/tools/agents_graph/tools.py @@ -595,6 +595,15 @@ async def stop_agent( to wrap up) for soft-stop scenarios. Reach for ``stop_agent`` when a child has gone off-track and won't self-correct. + **NEVER use ``stop_agent`` to clear active children so you can call + ``finish_scan``.** Stopped agents lose their in-progress work and + their completion reports never arrive. Instead, call + ``wait_for_message`` to block until children self-terminate via + ``agent_finish``, or ``send_message_to_agent`` asking them to wrap + up. The only valid use of ``stop_agent`` is for agents that are + genuinely off-track, stuck, or producing errors — not for agents + that are productively working but haven't finished yet. + Args: target_agent_id: The 8-char id from ``view_agent_graph`` / ``create_agent``. Cannot stop yourself. diff --git a/strix/tools/finish/tool.py b/strix/tools/finish/tool.py index c7662b2a1..7649f28d7 100644 --- a/strix/tools/finish/tool.py +++ b/strix/tools/finish/tool.py @@ -101,12 +101,13 @@ async def finish_scan( 1. **Call ``view_agent_graph`` first.** Inspect every entry in the summary. If ANY agent is in ``running`` / ``waiting`` state, you MUST NOT call ``finish_scan`` yet — - wrap them up first via ``send_message_to_agent`` (ask them to - finish), ``wait_for_message`` (block until their report - arrives), or ``stop_agent`` (graceful cancel). Only ``completed`` - / ``crashed`` / ``stopped`` agents are safe to leave behind. - Calling ``finish_scan`` while children are alive orphans their - work and produces an incomplete report. + call ``wait_for_message`` to block until their completion reports + arrive, or ``send_message_to_agent`` asking them to call + ``agent_finish``. **NEVER use ``stop_agent`` to clear active + agents so you can finish — this orphans their work and produces + an incomplete report.** Only ``completed`` / ``crashed`` agents + are safe to leave behind. ``stopped`` agents were forcibly + cancelled and their results are lost. 2. All vulnerabilities you found are filed via ``create_vulnerability_report`` (un-reported findings are not tracked and not credited). @@ -162,7 +163,10 @@ async def finish_scan( "scan_completed": False, "error": ( "Cannot finish scan while child agents are still active. " - "Wait for completion, send them finish instructions, or stop them first" + "Call wait_for_message to block until their completion reports " + "arrive, or send_message_to_agent asking them to call agent_finish. " + "Do NOT use stop_agent to clear active agents — their work will " + "be lost and the report will be incomplete." ), "active_agents": active_agents, },