Add Deep Phases system + coverage heartbeat + stagnation detection

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 <noreply@anthropic.com>
This commit is contained in:
root 2026-04-05 15:34:20 +02:00
parent b962a8d6b0
commit efa63b2000
6 changed files with 335 additions and 15 deletions

View file

@ -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", [])

View file

@ -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():

View file

@ -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:

View file

@ -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):

View file

@ -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):

View file

@ -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