Exploitation capability improvements (12 phases) + CVE/quality follow-ups

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

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

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
 

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(

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."""
         ...

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,