Tool Family: agent-modules

Agent-planning MCP adapters split out of the attack-modules god file into tools/mcp_tools/modules/. Each file defines one register_*_tools(mcp, *, ctx) registrar; tools/mcp_tools/registry.py auto-discovers every register_*_tools in the package including subpackages (modules/, terminal/) via _discover_tool_registrars, and collect_tools() AST-validates that every @mcp.tool also carries @audit_tool or @require_allowlist (CI fails otherwise). No edit to mcp_exploit_server.py is needed for a new module file.

• Registration sources: register_planning_tools (modules/planning.py:18), register_campaign_tools (modules/campaign.py:32), register_adaptive_tools (modules/adaptive.py:17), register_synthesis_tools (modules/synthesis.py:16), register_hash_tools (modules/hash.py:128), register_web_tools (modules/web.py:170).
• Context: ToolContext (tools/mcp_tools/registry.py:45-61) — workspace, config, search (ExploitSearch), nvd (NVDClient), researcher (WebResearcher), audit_tool, require_allowlist, plus sandbox / sandbox_notice for the execution plane.
• Gate pattern: target-touching tools use @require_allowlist(); advisory/local-only tools use @audit_tool. run_campaign_step is @audit_tool + a manual check_targets_allowlist on the target read back from state.json (LLM-writable path).
• Shared helper: _identify_hash_modes(h) (modules/hash.py:11) — single source for the hash-type → hashcat-mode mapping, re-exported via tools/mcp_tools/attack_modules.py for cracking.py and modules/synthesis.py.

Architecture

mcp_exploit_server.create_mcp_server
  └─ collect_tools() ─► _discover_tool_registrars()  (package walk incl. modules/)
       ├─ register_planning_tools  → create_attack_plan / get_current_plan / replan
       ├─ register_campaign_tools  → start_autonomous_campaign / get_campaign_status /
       │                              run_campaign_step / stop_campaign
       ├─ register_adaptive_tools  → craft_exploit / mutate_exploit
       ├─ register_synthesis_tools → cve_to_exploit_synth (+ 18 _render_* templates)
       ├─ register_hash_tools      → hash_crack_identify (+ _identify_hash_modes)
       └─ register_web_tools       → jwt_tamper / ssti_probe / graphql_introspect /
                                      race_request / timing_oracle /
                                      request_smuggling_probe / password_spray

Backing engines live outside mcp_tools: tools/attack_planner.py (plan model + prompts), tools/autonomous_orchestrator.py (campaign runner), tools/exploit_mutator.py + tools/payload_crafter.py (adaptive generation), tools/attack_modules/ (pre-packaged module registry), tools/cve_lookup.py + ExploitSearch.cve_to_poc (synth evidence), tools/recon_pipeline.py (campaign step recon).

Planning (modules/planning.py)

def register_planning_tools(mcp: Any, *, ctx: ToolContext) -> None
def create_attack_plan(target_ip: str, target_os: str = "", known_cves: str = "") -> str
def get_current_plan(target_ip: str) -> str
def replan(target_ip: str, failure_reason: str) -> str

Lifecycle: create_attack_plan → workspace/plans/<ip_underscored>_plan.json → get_current_plan reads → replan mutates + re-saves. Kill chain phases follow RECON → ENUMERATE → EXPLOIT → ESCALATE → LOOT → PIVOT → DONE (per the tool docstring; phase enum lives in tools/attack_planner.py).

Campaign (modules/campaign.py)

_running_campaign_tasks: set = set()
_campaign_orchestrators: dict[str, Any] = {}
def register_campaign_tools(mcp: Any, *, ctx: ToolContext) -> None
async def start_autonomous_campaign(target_ip: str, goal: str = "initial_access", aggression_level: str = "normal") -> str
def get_campaign_status(campaign_id: str) -> str
async def run_campaign_step(campaign_id: str) -> str
def stop_campaign(campaign_id: str) -> str

Adaptive (modules/adaptive.py)

def register_adaptive_tools(mcp: Any, *, ctx: ToolContext) -> None
def craft_exploit(target_ip: str, service_name: str, version: str = "", os_hint: str = "", module_name: str = "") -> str
def mutate_exploit(script_id: str, failure_output: str) -> str

Lifecycle: craft_exploit (generation 1, parent_id=None) → run → mutate_exploit with failure output → child generation with parent_id linkage, up to adaptive_exploits.max_mutations.

Synthesis (modules/synthesis.py)

def register_synthesis_tools(mcp: Any, *, ctx: ToolContext) -> None
def cve_to_exploit_synth(target_ip: str, cve_id: str, service_name: str = "", version: str = "") -> str
def _render_log4j_exploit(target_ip: str, port: int, svc: str, ver: str, cve: str) -> str
# + 17 more _render_*_template with the same signature (see table)
def _render_generic_probe(target_ip: str, port: int, svc: str, ver: str, cve: str) -> str

Dispatch table (18 branches; TEMPLATE_DISPATCHED: <name> is echoed so the agent can see which branch fired):

cve_to_exploit_synth pipeline: validate target (validate_target_or_ip) + CVE format (CVE-\d{4}-\d{4,}) + reject newline/quote/backslash in service_name/version (they are interpolated into generated-source comments) → nvd.search_sync(cve) + format_cve_results (capped at 2000 chars) → verified PoC lookup via search.cve_to_poc(cve, nvd_refs) with search_web_exploit fallback (capped at 1500 chars) → port hint from service_name (http 80, https 443, smb 445, rdp 3389, ssh 22, activemq 61616, confluence 8090, jenkins 8080, …) → renderer → optional inline syntax gate (poc_verification.enabled → syntax_check(exploit_body) → SYNTAX_OK: true/false + fix-and-verify_poc instructions) → INSTRUCTIONS: Use write_python_file ... then run_python_file.

All templates are detection-only probes (banner/version checks, safe payloads, non-routable default callbacks) meant for the LLM to refine — not weaponized exploits.

Implementation note: CVEToExploit (tools.attack_modules.modules.synthesis) and _identify_hash_modes are imported at the top of synthesis.py as the single-source reuse seam, but the tool body builds its output from the local _render_* templates plus nvd/search evidence rather than calling the module class directly.

Hash (modules/hash.py)

def _identify_hash_modes(h: str) -> list[tuple[str, str, str]]
def register_hash_tools(mcp: Any, *, ctx: ToolContext) -> None
def hash_crack_identify(hash_value: str) -> str

_identify_hash_modes returns [(name, hashcat_mode, sample_cmd), ...]. Order matters: a 32-hex string reports both NTLM (mode 1000) and MD5 (mode 0) since they are format-indistinguishable (context decides — NTLM from SMB, MD5 from a web app). LM (16 hex, mode 3000) only fires when nothing else matched, so fragments of longer hashes do not false-positive.

hash_crack_identify (@audit_tool, local-only) prints each (Type, mode, Command) plus a rule-based-attack hint (-r best64.rule / OneRuleToRuleThemAll); unknown formats suggest hashid/hash-identifier with an 80-char preview.

Web probes (modules/web.py)

_MIN_PORT = 1; _MAX_PORT = 65535
_MIN_CONCURRENT = 2; _MAX_CONCURRENT = 100
_MIN_TOOL_TIMEOUT = 5; _MAX_TOOL_TIMEOUT = 300; _DEFAULT_TOOL_TIMEOUT = 90
_MAX_OUTPUT_CHARS = 20000; _MAX_ENDPOINT_CHARS = 2048
def register_web_tools(mcp: Any, *, ctx: ToolContext) -> None
def jwt_tamper(target_ip: str, jwt_token: str = "", timeout: int = 90) -> str
def ssti_probe(target_ip: str, port: int = 80, timeout: int = 90) -> str
def graphql_introspect(target_ip: str, port: int = 80, timeout: int = 90) -> str
def race_request(target_ip: str, port: int = 80, endpoint: str = "/api/redeem", concurrent: int = 20, timeout: int = 90) -> str
def timing_oracle(target_ip: str, port: int = 80, timeout: int = 90) -> str
def request_smuggling_probe(target_ip: str, port: int = 80, timeout: int = 90) -> str
def password_spray(target_ip: str, port: int = 80, password: str = "Password1", timeout: int = 90) -> str

Shared pre-gates (web.py:43-167): _check_target (non-empty, no CR/LF, validate_target_or_ip), _check_port (1-65535, numeric strings accepted), _check_concurrent (2-100), _check_endpoint (absolute path, no CR/LF or whitespace, ≤2048 chars), _clamp_timeout (5-300 s, default 90), _http_host (brackets IPv6), _open_connection (socket.create_connection, always used as a context manager), _sock_budget (per-connection budget honoring the tool deadline), _finish (caps display at 20000 chars with a [truncated] marker).

Implementation note: web.py imports PasswordSpray, JWTTamper, GraphQLIntrospect, RaceRequest, RequestSmuggling, SSTIProbe, TimingOracle from tools.attack_modules.modules.* as the single-source seam, but the registered probe bodies are inline raw-socket implementations in web.py itself.

Config keys

Examples

# 1. Plan, inspect, adapt.
create_attack_plan("192.168.1.100", "linux", "CVE-2024-6387,CVE-2021-44228")
get_current_plan("192.168.1.100")
replan("192.168.1.100", "Log4j probe returned no response — service may be patched")

# 2. Autonomous campaign with monitoring.
start_autonomous_campaign("192.168.1.100", "full_compromise", "aggressive")
get_campaign_status("campaign-20260504_120000-abc12345")
run_campaign_step("campaign-20260504_120000-abc12345")  # single-step/debug pacing
stop_campaign("campaign-20260504_120000-abc12345")

# 3. Adaptive exploit generation loop.
craft_exploit("192.168.1.100", "ssh", "OpenSSH 8.9p1", "linux", "RegreSSHion")
mutate_exploit("gen-1712345678-abc12345", "ConnectionResetError: target closed connection")

# 4. CVE synthesis → save → run.
cve_to_exploit_synth("192.168.1.100", "CVE-2021-44228", "http", "")
# write_python_file the TEMPLATE_DISPATCHED script, then run_python_file.

# 5. Identify a hash, then probe the web surface.
hash_crack_identify("5d41402abc4b2a76b9719d911017c592")
jwt_tamper("192.168.1.100")
ssti_probe("192.168.1.100", 80)
graphql_introspect("192.168.1.100", 80)
race_request("192.168.1.100", 80, "/api/redeem", 20)
timing_oracle("192.168.1.100", 80)
request_smuggling_probe("192.168.1.100", 80)
password_spray("192.168.1.100", 80, "Password1")

Related documentation

• Attack-modules tool family — list_attack_modules / run_attack_module aggregate that campaigns step into
• Terminal tool family — target-IP lock primitives reused by campaign steps
• Sessions tool family — background execution plane campaigns run alongside
• Cracking tool family — run_hash_crack execution side of hash_crack_identify
• Research tool family — cve_to_poc verified-PoC resolver used by synthesis
• MCP security — allowlist + audit decorator contract
• MCP registration — collect_tools() auto-discovery

Source map

• tools/mcp_tools/modules/planning.py — register_planning_tools, create_attack_plan, get_current_plan, replan
• tools/mcp_tools/modules/campaign.py — register_campaign_tools, start_autonomous_campaign, get_campaign_status, run_campaign_step, stop_campaign, _running_campaign_tasks, _campaign_orchestrators
• tools/mcp_tools/modules/adaptive.py — register_adaptive_tools, craft_exploit, mutate_exploit
• tools/mcp_tools/modules/synthesis.py — register_synthesis_tools, cve_to_exploit_synth, 18 _render_* CVE-family templates
• tools/mcp_tools/modules/hash.py — register_hash_tools, hash_crack_identify, _identify_hash_modes
• tools/mcp_tools/modules/web.py — register_web_tools, jwt_tamper, ssti_probe, graphql_introspect, race_request, timing_oracle, request_smuggling_probe, password_spray, shared _check_*/_clamp_timeout/_finish helpers
• tools/mcp_tools/attack_modules.py — register_attack_module_tools aggregate, _identify_hash_modes re-export
• tools/mcp_tools/registry.py — ToolContext, _discover_tool_registrars, collect_tools, decorator validation
• tools/attack_planner.py — plan model, planning/replanning prompts and parsers
• tools/autonomous_orchestrator.py — campaign runner, AggressionLevel, TaskStatus
• tools/exploit_mutator.py + tools/payload_crafter.py — adaptive generation and mutation
• tools/attack_modules/ — pre-packaged module registry stepped by campaigns
• config.yaml — adaptive_exploits, swarm, autonomous, opsec, poc_verification, exploit