diff --git a/CLAUDE.md b/CLAUDE.md
index 2c2bde2..c1a215f 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -107,8 +107,9 @@ live in `pentestgpt/experimental/agents/components/` after the 2026-05 finishing
to `StrategyAgent.validate_plan` and `MITREGuidedPlanner.suggest_techniques_via_llm`.
`pentestgpt/agents/parsers/` provides a registry-backed parser system (`base.py` + `registry.py`) with concrete parsers
-for nmap, masscan, nessus, nikto, nuclei, sqlmap, hydra, and gobuster. The registry is the canonical way to look up a
-parser by tool name.
+for nmap, masscan, nessus, nikto, nuclei, sqlmap, hydra, and gobuster, plus nine added in the exploitation-capability
+work: secretsdump, getuserspns, certipy, netexec, responder, wpscan, ffuf, bloodhound, and enum4linux_ng. The registry
+is the canonical way to look up a parser by tool name.
MITRE ATT&CK integration spans four files — `pentestgpt/utils/mitre_planner.py` (task mapping),
`pentestgpt/utils/mitre_attack.py` (full taxonomy), `pentestgpt/utils/mitre_kb.py` (knowledge base), and
diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index 7bec89e..9e79b9e 100644
--- a/docs/CHANGELOG.md
+++ b/docs/CHANGELOG.md
@@ -7,6 +7,32 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
## [Unreleased]
+### Exploitation Capability Improvements — Phases 1–12
+
+A 12-phase effort (branch `feature/exploitation-capability-improvements`) extending the orchestration loop with
+finding-typed chaining, failure-driven replanning, additional tool parsers, target-aware exploitation, and new attack
+knowledge domains.
+
+- **Finding chaining & replanning** (`pentestgpt/orchestrator.py`) — per-session findings accumulator, finding-typed
+ chaining between iterations, dead-end replanning, and exploitation/technique-selector wiring so structured findings
+ drive subsequent task planning.
+- **Failure feedback** — failure-classification fix with failure-reason rerouting, a within-session failure gate, and
+ configurable retirement thresholds (`pentestgpt/utils/learning_context_manager.py`,
+ `pentestgpt/agents/strategy_agent.py`, `pentestgpt/config/knowledge_config.py`,
+ `pentestgpt/knowledge/learning_engine.py`).
+- **Nine new tool-output parsers** (`pentestgpt/agents/parsers/`, all registered in `registry.py`): `secretsdump`
+ (Impacket → credentials), `getuserspns` (Impacket → kerberoastable SPNs), `netexec`/CrackMapExec (credentials/valid
+ logins), `certipy` (vulnerable cert templates), `responder` (NTLMv2 credentials), `bloodhound` (AD attack paths), and
+ tier-2 `ffuf`, `wpscan`, `enum4linux_ng` web/SMB parsers. Registry now covers 17 tools.
+- **Exploitation wiring** — target-aware exploit generation and real-time defense adaptation
+ (`agents/exploit_developer.py`, `agents/advanced_exploitation_agent.py`, `knowledge/adaptive_technique_selector.py`);
+ tool-allowlist additions and a nuclei stdout-output fix (`agents/tool_executor.py`).
+- **Four new knowledge domains** (`pentestgpt/knowledge/domains/`, exported in `__init__.py`, registered by
+ `OrchestratorFactory._register_domain_modules`): `ApiGraphqlModule`, `DeserializationModule`, `CicdModule`,
+ `IdentityProviderModule` — bringing the registered domain total to 12.
+- **RAG corpus loader** (`pentestgpt/utils/rag.py`) — now ingests Markdown, plain-text, and JSON corpus files alongside
+ PDFs.
+
### Added
- **AdvancedAgentManager container** — `pentestgpt/agents/advanced_agent_manager.py` rewritten from a 168-line stub to a
diff --git a/docs/development/CODE_QUALITY.md b/docs/development/CODE_QUALITY.md
index b0a409d..d8d882d 100644
--- a/docs/development/CODE_QUALITY.md
+++ b/docs/development/CODE_QUALITY.md
@@ -2,7 +2,8 @@
## 🎯 Overview
-This guide establishes comprehensive code quality gates for the PentestGPT project to ensure consistent standards, maintain high quality, and improve security and reliability.
+This guide establishes comprehensive code quality gates for the PentestGPT project to ensure consistent standards,
+maintain high quality, and improve security and reliability.
## 📋 Quality Gate Requirements
@@ -20,7 +21,8 @@ This guide establishes comprehensive code quality gates for the PentestGPT proje
- **Purpose**: Sorts Python imports automatically
- **Configuration**: Profile = black, Line length = 120
- **Usage**: `isort pentestgpt`
-- **Note**: Configured in `pyproject.toml` but not in pre-commit hooks (Ruff handles import sorting via its `I` rule set)
+- **Note**: Configured in `pyproject.toml` but not in pre-commit hooks (Ruff handles import sorting via its `I` rule
+ set)
#### **Ruff Linter**
@@ -109,7 +111,8 @@ pydocstyle pentestgpt --convention=google
#### **Dependency Security**
- **Tools**: Safety for dependency vulnerability scanning
-- **Usage**: `safety check --full-report`
+- **Usage**: `safety scan --policy-file .safety-policy.yml` (the legacy `safety check` command is deprecated and cannot
+ parse the v3 policy file)
## 🔧 Configuration Files
diff --git a/pentestgpt/agents/advanced_exploitation_agent.py b/pentestgpt/agents/advanced_exploitation_agent.py
index e4dd3ab..8db1cf1 100644
--- a/pentestgpt/agents/advanced_exploitation_agent.py
+++ b/pentestgpt/agents/advanced_exploitation_agent.py
@@ -77,15 +77,18 @@ def __init__(
self,
model_router: ModelRouter,
agent_id: str = "advanced_exploitation_agent",
+ tool_executor: Any | None = None,
):
"""Initialize Advanced Exploitation Agent.
Args:
model_router: ModelRouter for LLM access
agent_id: Unique agent identifier
+ tool_executor: Optional ToolExecutor for sandboxed exploit testing
"""
self.agent_id = agent_id
self.router = model_router
+ self.tool_executor = tool_executor
# Initialize components
self.knowledge_manager = KnowledgeManager()
@@ -95,6 +98,8 @@ def __init__(
self.exploit_developer = ExploitDeveloper(
self.knowledge_manager.get_exploit_knowledge_base(),
self.knowledge_manager.get_payload_library(),
+ model_router=self.router,
+ tool_executor=tool_executor,
)
self.post_exploitation_handler = PostExploitationHandler()
self.session_manager = SessionManager(
diff --git a/pentestgpt/agents/agent_abstract.py b/pentestgpt/agents/agent_abstract.py
index 2c25d9c..d252343 100644
--- a/pentestgpt/agents/agent_abstract.py
+++ b/pentestgpt/agents/agent_abstract.py
@@ -165,6 +165,8 @@ def suggest_alternative_technique(
self,
failed_technique_id: str,
target_info: dict[str, Any],
+ failure_reason: str | None = None,
+ failure_type: str | None = None,
) -> str | None:
"""Suggest alternative technique when one fails."""
...
diff --git a/pentestgpt/agents/command_generator.py b/pentestgpt/agents/command_generator.py
index d0a61dd..f90e758 100644
--- a/pentestgpt/agents/command_generator.py
+++ b/pentestgpt/agents/command_generator.py
@@ -16,6 +16,8 @@
from pentestgpt.llms.router import ModelRouter, TaskType
from pentestgpt.mitre import MITRETechnique
from pentestgpt.prompts.mitre_prompts import MITREPrompts
+from pentestgpt.prompts.prompt_class_v3 import PentestGPTPromptV3
+from pentestgpt.utils.learning_context_manager import redact_mapping
from .agent_abstract import AgentMetadata, BaseAgent, CommandGeneratorInterface
@@ -120,7 +122,8 @@ def generate_commands(
# Get RAG context if not provided
if not rag_context and self.rag and technique:
- findings: list[Any] = target_info.get("findings", [])
+ session_findings = target_info.get("session_findings", {})
+ findings: list[Any] = session_findings.get("findings", [])
rag_context = self._get_rag_context(technique, findings=findings)
# --- Phase 4: History-guided skip check ---
@@ -129,8 +132,13 @@ def generate_commands(
effectiveness = self.learning_engine.get_technique_effectiveness(
technique.id, target_os=target_info.get("os", "unknown")
)
- if effectiveness and effectiveness.get("sample_count", 0) >= 5:
- if effectiveness.get("success_rate", 1.0) < 0.2:
+ cfg_learning = getattr(getattr(self.learning_engine, "config", None), "learning", None)
+ raw_min = getattr(cfg_learning, "skip_min_samples", 5)
+ raw_max = getattr(cfg_learning, "skip_max_success_rate", 0.2)
+ skip_min = raw_min if isinstance(raw_min, int) else 5
+ skip_max = raw_max if isinstance(raw_max, (int, float)) else 0.2
+ if effectiveness and effectiveness.get("sample_count", 0) >= skip_min:
+ if effectiveness.get("success_rate", 1.0) < skip_max:
logger.info(
f"Skipping {technique.id} — low historical success rate "
f"({effectiveness['success_rate']:.1%})"
@@ -144,6 +152,16 @@ def generate_commands(
if self.knowledge_manager and technique:
knowledge_context = self._get_knowledge_context(technique)
+ # A chained finding (e.g. credential -> lateral movement) overrides RAG context with a targeted plan
+ findings_obj = getattr(task, "findings", None)
+ chain_context = findings_obj.get("chain_context") if isinstance(findings_obj, dict) else None
+ if isinstance(chain_context, dict) and chain_context.get("prompt_type") == "lateral_movement":
+ rag_context = PentestGPTPromptV3.get_lateral_movement_prompt(
+ current_access=chain_context.get("current_access", {}),
+ discovered_hosts=chain_context.get("discovered_hosts", []),
+ credentials=redact_mapping(chain_context.get("credentials", [])),
+ )
+
# Generate prompt
if technique:
prompt = MITREPrompts.generate_command_prompt(
@@ -335,17 +353,33 @@ def _get_knowledge_context(self, technique: MITRETechnique) -> str | None:
try:
loop = asyncio.get_running_loop()
- # If we're in a loop, we can't block on it if it's in the same thread.
- # Assuming this is called from another thread if loop is present, or we risk deadlock.
- # However, fixing the TypeError: run_coroutine_threadsafe expects a coroutine.
- coro = self.knowledge_manager.search_knowledge(
- query=query,
- knowledge_type="technique_implementation",
- max_results=5,
- )
- results = asyncio.run_coroutine_threadsafe(coro, loop).result()
except RuntimeError:
- # No running loop, safe to use asyncio.run()
+ loop = None
+
+ if loop is not None and loop.is_running():
+ # We're on the event loop thread — cannot block. Run in a new thread.
+ import concurrent.futures
+
+ pool = concurrent.futures.ThreadPoolExecutor(max_workers=1)
+ future = pool.submit(
+ asyncio.run,
+ self.knowledge_manager.search_knowledge(
+ query=query,
+ knowledge_type="technique_implementation",
+ max_results=5,
+ ),
+ )
+ timed_out = False
+ try:
+ results = future.result(timeout=10)
+ except concurrent.futures.TimeoutError:
+ timed_out = True
+ future.cancel()
+ logger.warning("Knowledge context lookup timed out")
+ return None
+ finally:
+ pool.shutdown(wait=not timed_out, cancel_futures=True)
+ else:
results = asyncio.run(
self.knowledge_manager.search_knowledge(
query=query,
diff --git a/pentestgpt/agents/exploit_developer.py b/pentestgpt/agents/exploit_developer.py
index 02082ee..519e271 100644
--- a/pentestgpt/agents/exploit_developer.py
+++ b/pentestgpt/agents/exploit_developer.py
@@ -3,8 +3,9 @@
Handles exploit generation, testing, and payload creation.
"""
-import secrets
+import copy
import time
+import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from enum import Enum
@@ -85,15 +86,21 @@ def __init__(
self,
exploit_knowledge_base: dict[str, Any],
payload_library: dict[str, Any],
+ model_router: Any | None = None,
+ tool_executor: Any | None = None,
):
"""Initialize Exploit Developer.
Args:
exploit_knowledge_base: Reference to exploit knowledge base
payload_library: Reference to payload library
+ model_router: Optional ModelRouter for LLM-driven exploit synthesis
+ tool_executor: Optional ToolExecutor for sandboxed exploit testing
"""
self.exploit_knowledge_base = exploit_knowledge_base
self.payload_library = payload_library
+ self.model_router = model_router
+ self.tool_executor = tool_executor
self.exploit_candidates: dict[str, ExploitCandidate] = {}
self.exploitation_results: list[ExploitationResult] = []
@@ -142,20 +149,37 @@ def develop_and_test_exploits(
},
)
- # Test exploit
- test_result = self._test_exploit(
- exploit_candidate,
- target,
- vulnerability,
- )
+ # Bounded mutate-retry: re-test up to 3 attempts, adapting the
+ # vulnerability (and regenerating the candidate) from the failure
+ # reason each time. Stop at the first success.
+ current_candidate = exploit_candidate
+ current_vuln = vulnerability
+ test_result: dict[str, Any] = {}
+ for _attempt in range(3):
+ test_result = self._test_exploit(
+ current_candidate,
+ target,
+ current_vuln,
+ )
+ if test_result.get("success", False):
+ break
+ reason = test_result.get("reason", "unknown")
+ current_vuln = self._mutate_vulnerability(current_vuln, reason)
+ regenerated = self._generate_exploit_candidate(
+ current_vuln,
+ target,
+ )
+ if regenerated is None:
+ break
+ current_candidate = regenerated
if test_result.get("success", False):
successful_exploits.append(
{
- "exploit_id": exploit_candidate.exploit_id,
+ "exploit_id": current_candidate.exploit_id,
"vulnerability_id": vulnerability["vulnerability_id"],
"exploitation_result": test_result,
- "payload_delivered": exploit_candidate.payload,
+ "payload_delivered": current_candidate.payload,
"execution_details": test_result.get("details", {}),
},
)
@@ -164,7 +188,7 @@ def develop_and_test_exploits(
exploitation_result = ExploitationResult(
result_id=f"result_{int(time.time())}_{len(self.exploitation_results)}",
session_id=session_id,
- exploit_id=exploit_candidate.exploit_id,
+ exploit_id=current_candidate.exploit_id,
target=target,
success=True,
execution_details=test_result.get("details", {}),
@@ -203,7 +227,7 @@ def _generate_exploit_candidate(
ExploitCandidate or None
"""
try:
- exploit_id = f"exploit_{vulnerability['vulnerability_id']}_{int(time.time())}"
+ exploit_id = f"exploit_{vulnerability['vulnerability_id']}_{uuid.uuid4().hex}"
# Determine exploit type based on vulnerability
vuln_type = vulnerability.get("type", "unknown")
@@ -271,7 +295,11 @@ def _generate_exploit_code(
vulnerability: dict[str, Any],
target: str,
) -> str:
- """Generate exploit code for specific type.
+ """Generate exploit code for a specific type.
+
+ Uses the wired model router to synthesize a target-aware exploit. When no
+ router is available (or generation fails), degrades gracefully to a static
+ literal so the pipeline keeps working.
Args:
exploit_type: Type of exploit
@@ -282,24 +310,129 @@ def _generate_exploit_code(
Exploit code string
"""
try:
- if exploit_type == ExploitType.SQL_INJECTION:
- return "' UNION SELECT 1,2,3,4,5,6,7,8,9,10--"
- if exploit_type == ExploitType.XSS:
- return ""
- if exploit_type == ExploitType.COMMAND_INJECTION:
- return f"'; {self._get_reverse_shell_payload('127.0.0.1', 4444)}"
- if exploit_type == ExploitType.REMOTE_CODE_EXECUTION:
- return f"{self._get_reverse_shell_payload('127.0.0.1', 4444)}"
- if exploit_type == ExploitType.LOCAL_PRIVILEGE_ESCALATION:
- return "sudo -u#-1 /bin/bash"
- return "# Generic exploit placeholder"
-
+ if self.model_router is None:
+ return self._static_fallback(exploit_type)
+ from pentestgpt.llms.router import TaskType
+
+ prompt = self._build_exploit_prompt(exploit_type, vulnerability, target)
+ code, _provider, _cost = self.model_router.route_task(
+ TaskType.COMMAND_GENERATION,
+ prompt,
+ )
+ return code or self._static_fallback(exploit_type)
except Exception as e:
logger.error(f"Error generating exploit code: {e}")
- return "# Error generating exploit code"
+ return self._static_fallback(exploit_type)
+
+ def _static_fallback(self, exploit_type: ExploitType) -> str:
+ """Static literal fallback used when no model router is wired."""
+ if exploit_type == ExploitType.SQL_INJECTION:
+ return "' UNION SELECT 1,2,3,4,5,6,7,8,9,10--"
+ if exploit_type == ExploitType.XSS:
+ return ""
+ if exploit_type == ExploitType.COMMAND_INJECTION:
+ return f"'; {self._get_reverse_shell_payload('127.0.0.1', 4444)}"
+ if exploit_type == ExploitType.REMOTE_CODE_EXECUTION:
+ return self._get_reverse_shell_payload("127.0.0.1", 4444)
+ if exploit_type == ExploitType.LOCAL_PRIVILEGE_ESCALATION:
+ return "sudo -u#-1 /bin/bash"
+ return "# Generic exploit placeholder"
+
+ def _build_exploit_prompt(
+ self,
+ exploit_type: ExploitType,
+ vulnerability: dict[str, Any],
+ target: str,
+ ) -> str:
+ """Build an LLM instruction prompt for target-aware exploit synthesis.
+
+ Inlines the concrete vulnerability fields so target specifics reach the
+ model, and encodes expert adaptation rules as plain-text guidance keyed on
+ the exploit type.
+
+ Args:
+ exploit_type: Type of exploit to generate
+ vulnerability: Vulnerability data (provides dialect, columns, libs, etc.)
+ target: Target system (host/IP)
+
+ Returns:
+ A single instruction prompt string.
+ """
+ lhost = str(vulnerability.get("lhost") or self.payload_library.get("lhost") or "127.0.0.1")
+ lport = str(vulnerability.get("lport") or self.payload_library.get("lport") or 4444)
+
+ lines: list[str] = [
+ "You are an expert exploit developer working an authorized penetration test.",
+ f"Target host: {target}",
+ f"Exploit type: {exploit_type.value}",
+ f"Vulnerability name: {vulnerability.get('name', 'unknown')}",
+ f"Vulnerability id: {vulnerability.get('vulnerability_id', 'unknown')}",
+ "Vulnerability details (use these specifics verbatim where relevant):",
+ f" dbms={vulnerability.get('dbms')} columns={vulnerability.get('columns')} "
+ f"waf={vulnerability.get('waf')} filtered={vulnerability.get('filtered')} "
+ f"classpath_libs={vulnerability.get('classpath_libs')} "
+ f"framework={vulnerability.get('framework')} os={vulnerability.get('os')} "
+ f"arch={vulnerability.get('arch')} gadget_index={vulnerability.get('gadget_index')}",
+ "Output ONLY the raw exploit payload/command, no commentary or fences.",
+ "Adaptation rules:",
+ ]
+
+ if exploit_type == ExploitType.SQL_INJECTION:
+ lines.append(
+ f"- Build a dialect-aware UNION-based injection for dbms='{vulnerability.get('dbms')}' "
+ f"with a column count of {vulnerability.get('columns')}.",
+ )
+ if vulnerability.get("waf"):
+ lines.append(
+ "- A WAF is present: drive sqlmap with a tamper chain "
+ "'space2comment,randomcase,charencode,equaltolike,between'; for MySQL add "
+ "'modsecurityversioned'; for MSSQL add 'space2mssqlblank'; and escalate "
+ "--level/--risk.",
+ )
+ elif exploit_type == ExploitType.DESERIALIZATION:
+ lines.append(
+ f"- Select a ysoserial gadget chain from classpath_libs={vulnerability.get('classpath_libs')} "
+ "(CommonsCollections1-7 by CC version and JDK; Spring1/2; Groovy1; Jdk7u21 for "
+ "pure-JDK). For .NET use ysoserial.net TypeConfuseDelegate / ObjectStateFormatter for "
+ "ViewState. For PHP use phpggc keyed on the framework "
+ f"({vulnerability.get('framework')}). gadget_index={vulnerability.get('gadget_index')}.",
+ )
+ elif exploit_type in (
+ ExploitType.REMOTE_CODE_EXECUTION,
+ ExploitType.COMMAND_INJECTION,
+ ):
+ lines.append(
+ f"- Generate a reverse shell with msfvenom selected from os={vulnerability.get('os')} "
+ f"arch={vulnerability.get('arch')} (-p //..., -f exe|elf|raw, and "
+ f"-e x86/shikata_ga_nai -i N to evade AV). Use real LHOST={lhost} LPORT={lport} "
+ "(never hardcode 127.0.0.1:4444).",
+ )
+ if exploit_type == ExploitType.COMMAND_INJECTION and vulnerability.get("filtered"):
+ lines.append(
+ "- Input is filtered: apply encoding variants (HTML-entity, \\uXXXX, base64 "
+ "eval(atob()),