From efa63b200020882471916d273e57e82e727a01a6 Mon Sep 17 00:00:00 2001 From: root Date: Sun, 5 Apr 2026 15:34:20 +0200 Subject: [PATCH] Add Deep Phases system + coverage heartbeat + stagnation detection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Solves premature scan termination without any manual prompting after start. ## Deep Phases Gate (finish_actions.py) - finish_scan is intercepted on phases 0–2 and returns a "phase_gate" response with escalating deep-dive objectives instead of completing. - Phase 1: endpoint exhaustion, parameter fuzzing, second-order attacks, UI expansion, auth re-test. - Phase 2: attack chaining, business logic, advanced injection, HTTP-level, JWT, SSRF escalation. - Phase 3: final validation, false-positive audit, attack chain maximization, coverage confirmation. - Only phase 3 (the 4th call to finish_scan) completes the scan for real. - Each gate resets max_iterations_warning_sent so the agent doesn't panic-finish in the next phase. ## Scan Mode Budgets (strix_agent.py) - quick: 300 iterations, 2 phases - standard: 800 iterations, 3 phases - deep: 1500 iterations, 4 phases - scan_mode wired through cli.py and tui.py (hardcoded 300 removed). ## Coverage Heartbeat (base_agent.py) - Every 30 iterations, root agent receives a status pulse: phase, vulns found, tool exec count, and 4 yes/no coverage questions to self-check. ## Stagnation Detection (base_agent.py) - If 80%+ of last 15 tool calls are the same tool, agent receives a redirect prompt forcing it to switch to a completely different attack surface. ## Approaching-max Warning (state.py) - Threshold moved from 85% to 97% so the panic-finish warning fires much later. - Warning reframed: "you have time, keep testing" instead of "finish ASAP". Co-Authored-By: Claude Sonnet 4.6 --- strix/agents/StrixAgent/strix_agent.py | 22 ++- strix/agents/base_agent.py | 125 ++++++++++++++-- strix/agents/state.py | 8 +- strix/interface/cli.py | 4 +- strix/interface/tui.py | 3 +- strix/tools/finish/finish_actions.py | 188 +++++++++++++++++++++++++ 6 files changed, 335 insertions(+), 15 deletions(-) diff --git a/strix/agents/StrixAgent/strix_agent.py b/strix/agents/StrixAgent/strix_agent.py index 1519ed51..c3abce97 100644 --- a/strix/agents/StrixAgent/strix_agent.py +++ b/strix/agents/StrixAgent/strix_agent.py @@ -5,7 +5,16 @@ from strix.llm.config import LLMConfig class StrixAgent(BaseAgent): - max_iterations = 300 + # Default iterations per scan mode. Deep mode gets a large budget so the + # phase gate system can run 4 full phases without hitting the iteration cap. + max_iterations = 1500 + + # Map scan-mode names to iteration budgets and phase counts. + _SCAN_MODE_CONFIGS: dict[str, dict] = { + "quick": {"max_iterations": 300, "max_phases": 2}, + "standard": {"max_iterations": 800, "max_phases": 3}, + "deep": {"max_iterations": 1500, "max_phases": 4}, + } def __init__(self, config: dict[str, Any]): default_skills = [] @@ -16,8 +25,19 @@ class StrixAgent(BaseAgent): self.default_llm_config = LLMConfig(skills=default_skills) + # Apply scan-mode budget before super().__init__ reads self.max_iterations + scan_mode = config.get("scan_mode", "deep") + mode_cfg = self._SCAN_MODE_CONFIGS.get(scan_mode, self._SCAN_MODE_CONFIGS["deep"]) + if "max_iterations" not in config: + self.max_iterations = mode_cfg["max_iterations"] + super().__init__(config) + # Configure phase count on the state after BaseAgent sets it up. + # Only root agents use phases (sub-agents complete on first finish). + if self.state.parent_id is None: + self.state.max_phases = mode_cfg["max_phases"] + async def execute_scan(self, scan_config: dict[str, Any]) -> dict[str, Any]: # noqa: PLR0912 user_instructions = scan_config.get("user_instructions", "") targets = scan_config.get("targets", []) diff --git a/strix/agents/base_agent.py b/strix/agents/base_agent.py index b1909e9f..e475ad0f 100644 --- a/strix/agents/base_agent.py +++ b/strix/agents/base_agent.py @@ -190,30 +190,56 @@ class BaseAgent(metaclass=AgentMeta): self.state.increment_iteration() + # ------------------------------------------------------------------ + # Coverage heartbeat — every 30 iterations inject a status pulse + # showing the agent how far it is and encouraging continued testing. + # Only injected for root agents (parent_id is None). + # ------------------------------------------------------------------ + if ( + self.state.parent_id is None + and self.state.iteration > 0 + and self.state.iteration % 30 == 0 + ): + self._inject_coverage_heartbeat(tracer) + + # ------------------------------------------------------------------ + # Stagnation detection — if the last 15 tool calls are all the same + # tool, the agent is spinning. Kick it in a new direction. + # Only for root agents; only if we have enough history. + # ------------------------------------------------------------------ + if ( + self.state.parent_id is None + and len(self.state.actions_taken) >= 15 + ): + self._check_and_break_stagnation() + + # ------------------------------------------------------------------ + # Approaching-max warning — pushed to 97% so it fires very late. + # Framed as "you still have time" rather than "finish now". + # ------------------------------------------------------------------ if ( self.state.is_approaching_max_iterations() and not self.state.max_iterations_warning_sent ): self.state.max_iterations_warning_sent = True remaining = self.state.max_iterations - self.state.iteration + current_phase = getattr(self.state, "current_phase", 0) + max_phases = getattr(self.state, "max_phases", 4) warning_msg = ( - f"URGENT: You are approaching the maximum iteration limit. " - f"Current: {self.state.iteration}/{self.state.max_iterations} " + f"NOTICE: You are at iteration {self.state.iteration}/{self.state.max_iterations} " f"({remaining} iterations remaining). " - f"Please prioritize completing your required task(s) and calling " - f"the appropriate finish tool (finish_scan for root agent, " - f"agent_finish for sub-agents) as soon as possible." + f"Current phase: {current_phase + 1}/{max_phases}. " + f"Use remaining iterations to complete all untested endpoints and UI sections. " + f"Only call finish_scan when you have completed Phase {max_phases}/{max_phases} " + f"and have tested everything. Do NOT rush to finish — exhaustive coverage matters." ) self.state.add_message("user", warning_msg) if self.state.iteration == self.state.max_iterations - 3: final_warning_msg = ( - "CRITICAL: You have only 3 iterations left! " - "Your next message MUST be the tool call to the appropriate " - "finish tool: finish_scan if you are the root agent, or " - "agent_finish if you are a sub-agent. " - "No other actions should be taken except finishing your work " - "immediately." + "CRITICAL: Only 3 iterations left. " + "Call finish_scan NOW with your complete findings report. " + "Include all vulnerabilities discovered across all phases." ) self.state.add_message("user", final_warning_msg) @@ -638,6 +664,83 @@ class BaseAgent(metaclass=AgentMeta): tracer.update_agent_status(self.state.agent_id, "error") return True + def _inject_coverage_heartbeat(self, tracer: Optional["Tracer"]) -> None: + """Inject a periodic status pulse so the agent knows its progress. + + Pulls live data from the tracer (vulnerability count, tool execution + count) and from phase state so the agent has concrete numbers to act on. + """ + try: + vuln_count = 0 + tool_exec_count = 0 + if tracer: + vuln_count = len(getattr(tracer, "vulnerability_reports", [])) + tool_exec_count = len(getattr(tracer, "tool_executions", {})) + + current_phase = getattr(self.state, "current_phase", 0) + max_phases = getattr(self.state, "max_phases", 4) + phase_iter_start = getattr(self.state, "phase_iteration_start", 0) + phase_iters_used = self.state.iteration - phase_iter_start + remaining = self.state.max_iterations - self.state.iteration + + heartbeat = ( + f"[SCAN HEARTBEAT — iteration {self.state.iteration}]\n" + f"Phase: {current_phase + 1}/{max_phases}\n" + f"Iterations in this phase: {phase_iters_used}\n" + f"Iterations remaining: {remaining}\n" + f"Vulnerabilities reported so far: {vuln_count}\n" + f"Total tool executions: {tool_exec_count}\n\n" + f"STATUS CHECK — before continuing, answer these mentally:\n" + f"• Have you tested EVERY discovered endpoint for auth bypass?\n" + f"• Have you tested EVERY form input for injection?\n" + f"• Have you opened and tested EVERY UI section/page/modal?\n" + f"• Have you tested privilege escalation across all user roles?\n" + f"If the answer to ANY of the above is 'no', keep testing. " + f"Do NOT call finish_scan until this phase's objectives are complete." + ) + self.state.add_message("user", heartbeat) + except Exception: # noqa: BLE001 + pass # heartbeat is non-fatal + + def _check_and_break_stagnation(self) -> None: + """Detect if the agent is repeating the same tool and kick it out. + + If the last 15 tool calls are all the same tool type, the agent is + spinning. Inject a redirect prompt to force a change of approach. + """ + try: + recent = self.state.actions_taken[-15:] + tool_names = [] + for entry in recent: + action = entry.get("action", {}) + # Tool invocations can be dicts with a 'name' or 'tool' key + name = action.get("name") or action.get("tool") or action.get("function", {}).get("name", "") + if name: + tool_names.append(name) + + if len(tool_names) < 10: + return + + # If 80%+ of recent tools are the same, we're stagnating + if tool_names: + most_common = max(set(tool_names), key=tool_names.count) + ratio = tool_names.count(most_common) / len(tool_names) + if ratio >= 0.8: + redirect = ( + f"[STAGNATION DETECTED] You have called '{most_common}' " + f"{tool_names.count(most_common)} times in the last {len(tool_names)} " + f"actions. You are stuck in a loop.\n\n" + f"STOP what you are doing and switch to a completely different attack surface:\n" + f"• If you were fuzzing parameters → switch to UI navigation and click new pages\n" + f"• If you were browsing the UI → switch to API endpoint testing\n" + f"• If you were testing one endpoint → move to a different endpoint\n" + f"• If you were running automated tools → try manual testing instead\n\n" + f"Pick a new area you have NOT yet tested and start there immediately." + ) + self.state.add_message("user", redirect) + except Exception: # noqa: BLE001 + pass # stagnation check is non-fatal + def cancel_current_execution(self) -> None: self._force_stop = True if self._current_task and not self._current_task.done(): diff --git a/strix/agents/state.py b/strix/agents/state.py index da04ee7f..361d3168 100644 --- a/strix/agents/state.py +++ b/strix/agents/state.py @@ -29,6 +29,12 @@ class AgentState(BaseModel): final_result: dict[str, Any] | None = None max_iterations_warning_sent: bool = False + # Deep Phases system — finish_scan is intercepted per-phase and only + # completes on the final phase. 0-indexed: phases 0..max_phases-1. + current_phase: int = 0 + max_phases: int = 4 + phase_iteration_start: int = 0 + messages: list[dict[str, Any]] = Field(default_factory=list) context: dict[str, Any] = Field(default_factory=dict) @@ -113,7 +119,7 @@ class AgentState(BaseModel): def has_reached_max_iterations(self) -> bool: return self.iteration >= self.max_iterations - def is_approaching_max_iterations(self, threshold: float = 0.85) -> bool: + def is_approaching_max_iterations(self, threshold: float = 0.97) -> bool: return self.iteration >= int(self.max_iterations * threshold) def has_waiting_timeout(self) -> bool: diff --git a/strix/interface/cli.py b/strix/interface/cli.py index 7bf88f91..3db97eba 100644 --- a/strix/interface/cli.py +++ b/strix/interface/cli.py @@ -195,7 +195,9 @@ async def run_cli(args: Any) -> None: # noqa: PLR0915 llm_config = LLMConfig(scan_mode=scan_mode) agent_config: dict[str, Any] = { "llm_config": llm_config, - "max_iterations": 300, + "scan_mode": scan_mode, + # max_iterations is intentionally NOT set here — StrixAgent picks + # the right budget based on scan_mode (300/800/1500 for quick/standard/deep). } if getattr(args, "local_sources", None): diff --git a/strix/interface/tui.py b/strix/interface/tui.py index c30cca7f..aa822039 100644 --- a/strix/interface/tui.py +++ b/strix/interface/tui.py @@ -766,7 +766,8 @@ class StrixTUIApp(App): # type: ignore[misc] config: dict[str, Any] = { "llm_config": llm_config, - "max_iterations": 300, + "scan_mode": scan_mode, + # max_iterations intentionally omitted — StrixAgent sets it from scan_mode. } if getattr(args, "local_sources", None): diff --git a/strix/tools/finish/finish_actions.py b/strix/tools/finish/finish_actions.py index 79f48e79..40053f46 100644 --- a/strix/tools/finish/finish_actions.py +++ b/strix/tools/finish/finish_actions.py @@ -3,6 +3,160 @@ from typing import Any from strix.tools.registry import register_tool +# --------------------------------------------------------------------------- +# Deep-dive phase prompts injected when finish_scan is called too early. +# Phase 0 completes normally (no gate). Phases 1-3 are progressively harder. +# --------------------------------------------------------------------------- + +_PHASE_PROMPTS = { + 1: """ +╔══════════════════════════════════════════════════════════════════════╗ +║ PHASE 1 COMPLETE — ADVANCING TO PHASE 2 (DEEP DIVE) ║ +╚══════════════════════════════════════════════════════════════════════╝ + +You attempted to finish, but the scan is NOT complete. Phase 1 covered broad +discovery. Phase 2 requires deep exploitation of everything you found. + +MANDATORY PHASE 2 OBJECTIVES — execute ALL of these before calling finish_scan again: + +1. ENDPOINT EXHAUSTION + - List every URL/endpoint you discovered in Phase 1. + - Test each one for IDOR, BAC, injection, and auth bypass independently. + - Do not skip any endpoint because it "seems safe" — test it anyway. + +2. PARAMETER FUZZING + - For every POST/PUT endpoint: fuzz all parameters individually with boundary + values, operator objects, SQL metacharacters, and null/empty values. + - For every GET endpoint: fuzz all query parameters with the same payloads. + +3. SECOND-ORDER ATTACKS + - If Phase 1 found stored data anywhere: access that data from a DIFFERENT + user/role to check if stored XSS or IDOR applies. + - Test every "view your own profile/order/message" endpoint with another + user's IDs substituted. + +4. UI SECTION EXPANSION + - Open the app in the browser. Click EVERY button, link, dropdown, and tab + you have not yet clicked. + - Fill out and submit every form you have not yet submitted. + - Navigate to every page section: dashboard, profile, settings, admin, + billing, notifications, API keys, integrations. + +5. AUTHENTICATION RE-TEST + - Try accessing every authenticated endpoint WITHOUT a session token. + - Try accessing every admin endpoint with a standard user token. + - Test password reset flow end-to-end if you haven't already. + +DO NOT call finish_scan until all 5 objectives above are fully completed. +""", + + 2: """ +╔══════════════════════════════════════════════════════════════════════╗ +║ PHASE 2 COMPLETE — ADVANCING TO PHASE 3 (EXPERT MODE) ║ +╚══════════════════════════════════════════════════════════════════════╝ + +Phase 2 is complete. Phase 3 requires expert-level techniques that go beyond +standard testing. You must now assume the application has hidden vulnerabilities +that basic testing cannot reveal. + +MANDATORY PHASE 3 OBJECTIVES — ALL must be executed: + +1. CHAINED ATTACK PATHS + - Combine every finding from Phases 1 and 2. Example: Use a low-severity + info-disclosure to get a user ID, then use that ID in an IDOR attack. + - Identify every privilege escalation path: low-priv → admin, anonymous → user. + - Map the kill chain: what is the maximum damage achievable by chaining + all discovered vulnerabilities? + +2. BUSINESS LOGIC ATTACKS + - Test every numeric field for negative values, overflow, and race conditions. + - Test every workflow for step-skipping: can you reach step 5 without step 3? + - Test every price/quantity field: can you purchase at $0? Can you submit + negative quantities to get credits? + - Test coupon codes, referral codes, and discount logic for reuse/bypass. + +3. ADVANCED INJECTION + - For every JSON body endpoint: send MongoDB operator objects + {"$ne": null}, {"$gt": ""}, {"$regex": ".*"} — check for boolean differential. + - For every search field: test SSTI with {{7*7}}, ${7*7}, #{7*7}. + - For every file upload: test path traversal in filename, polyglot files, + SVG with XSS, and content-type bypass. + +4. HTTP-LEVEL ATTACKS + - Test Host header injection on all endpoints. + - Test X-Forwarded-For, X-Real-IP, X-Original-URL header injection. + - Test HTTP method override: X-HTTP-Method-Override: DELETE on POST endpoints. + - Try HTTP/1.1 request smuggling (CL.TE or TE.CL) on any reverse-proxied endpoint. + +5. JWT & SESSION ATTACKS (if auth uses tokens) + - Decode every JWT and check algorithm. Try alg:none bypass. + - Test algorithm confusion: RS256 → HS256 with public key as secret. + - Test token reuse after logout. + - Test predictable session IDs by collecting 5+ tokens and checking entropy. + +6. SSRF ESCALATION + - If Phase 1/2 found any SSRF: escalate it. Try accessing internal metadata + endpoints (169.254.169.254, fd00:ec2::254), internal services on 127.0.0.1:*, + and cloud IMDS endpoints. + - Try SSRF via file://, gopher://, dict:// protocols. + +DO NOT call finish_scan until all 6 objectives above are fully documented with evidence. +""", + + 3: """ +╔══════════════════════════════════════════════════════════════════════╗ +║ PHASE 3 COMPLETE — ADVANCING TO PHASE 4 (FINAL VALIDATION) ║ +╚══════════════════════════════════════════════════════════════════════╝ + +Phases 1-3 are complete. Phase 4 is the FINAL VALIDATION sweep. Your job now +is to prove everything, consolidate everything, and maximize impact. + +MANDATORY PHASE 4 OBJECTIVES — this is your last pass before reporting: + +1. VALIDATE EVERY FINDING + - Reproduce EVERY vulnerability found in Phases 1-3. Confirm it still works. + - For each finding: capture the EXACT request/response pair as evidence. + - Assign correct severity: does it require auth? Does it require user interaction? + Is impact limited or full application compromise? + - PURGE any finding where you cannot reproduce it or where the response + is indistinguishable from normal application behavior. + +2. FALSE POSITIVE AUDIT — DISCARD findings that match ANY of these: + - XSS reported only because payload appeared in a JSON API response + (not rendered in HTML — JSON never executes JavaScript). + - NoSQL injection reported only because HTTP 500 occurred on operator object + (type mismatch, not injection — must prove boolean differential or auth bypass). + - SSRF reported only because DNS resolution occurred with no internal access. + - CORS reported on a public endpoint that intentionally serves all origins. + - Missing security headers reported as High/Critical severity. + - Rate limiting absence on non-sensitive endpoints. + +3. ATTACK CHAIN MAXIMIZATION + - Build the highest-impact attack chain from all findings combined. + - Describe the exact sequence: step 1 (exploit A) → step 2 (use result to + exploit B) → step 3 (achieve full account takeover / data exfiltration / RCE). + - This chain MUST appear in your final report as a "Critical Attack Path." + +4. COVERAGE CONFIRMATION + - List every endpoint discovered during the scan. + - Confirm each one was tested. + - List every UI section/page visited. + - List every form submitted. + - If any endpoint was NOT tested, test it now before finishing. + +5. FINAL REPORT REQUIREMENTS + Your finish_scan call MUST include: + - executive_summary: overall risk rating, top 3 findings, business impact + - methodology: all 4 phases, tools used, coverage percentage, what was tested + - technical_analysis: every finding with title, severity, proof-of-concept + request/response, CWE ID, and reproduction steps + - recommendations: specific remediation for each finding, ordered by priority + +NOW call finish_scan with a complete, validated, evidence-backed report. +""", +} + + def _validate_root_agent(agent_state: Any) -> dict[str, Any] | None: if agent_state and hasattr(agent_state, "parent_id") and agent_state.parent_id is not None: return { @@ -113,6 +267,40 @@ def finish_scan( if validation_errors: return {"success": False, "message": "Validation failed", "errors": validation_errors} + # ------------------------------------------------------------------ + # DEEP PHASES GATE — intercept finish_scan until the final phase. + # Each interception resets the max-iterations warning so the agent + # doesn't get panicked into finishing early in the next phase. + # ------------------------------------------------------------------ + if agent_state and hasattr(agent_state, "current_phase"): + current_phase = agent_state.current_phase + max_phases = getattr(agent_state, "max_phases", 4) + + if current_phase < max_phases - 1: + next_phase = current_phase + 1 + agent_state.current_phase = next_phase + agent_state.phase_iteration_start = agent_state.iteration + # Reset warning flag so the agent doesn't panic-finish in Phase N+1 + agent_state.max_iterations_warning_sent = False + + phase_prompt = _PHASE_PROMPTS.get(next_phase, _PHASE_PROMPTS[1]) + return { + "success": False, + "phase_gate": True, + "current_phase": current_phase, + "next_phase": next_phase, + "total_phases": max_phases, + "message": ( + f"Phase {current_phase + 1}/{max_phases} summary accepted. " + f"Advancing to Phase {next_phase + 1}/{max_phases}. " + f"You must complete the objectives below before calling finish_scan again." + ), + "phase_objectives": phase_prompt.strip(), + } + + # ------------------------------------------------------------------ + # Final phase (or phases disabled) — complete the scan for real. + # ------------------------------------------------------------------ try: from strix.telemetry.tracer import get_global_tracer