LLM/pentestagent
1
0
Fork 0
mirror of https://github.com/GH05TCREW/pentestagent.git synced 2026-03-07 14:23:20 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore(mcp): remove docs refs to vendored adapters; add example adapter scaffold and tests

Browse Source
This commit is contained in:
giveen
2026-01-21 11:31:03 -07:00
parent db06dcfa63
commit a417522f1a
6 changed files with 127 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
.env.example
View File

@@ -17,21 +17,15 @@ PENTESTAGENT_DEBUG=true
# vendored MCP servers and helper daemons. Set to `true` to enable auto-start.
# - Defaults are `false` to avoid automatically running networked services.
# Vendored HexStrike MCP adapter (legacy name support: LAUNCH_HEXSTRIKE)
LAUNCH_HEXTRIKE=false
#LAUNCH_HEXSTRIKE=false # alternate spelling (kept for compatibility)
# Metasploit MCP (MetasploitMCP)
# When `LAUNCH_METASPLOIT_MCP=true` the setup script may attempt to start
# `msfrpcd` (Metasploit RPC daemon) and then start the vendored MetasploitMCP
# HTTP/SSE server. Provide `MSF_PASSWORD` if you want the setup script to
# auto-launch `msfrpcd` (it will never invoke sudo).
LAUNCH_METASPLOIT_MCP=false
# When set to `true`, the subtree helper scripts (e.g. scripts/add_metasploit_subtree.sh)
# will force a pull/update of vendored subtrees. Useful when you want to refresh
# the third_party trees during setup. Set to `true` to enable.
FORCE_SUBTREE_PULL=true
# MCP adapters and vendored integrations
# The project no longer vendors external MCP adapters such as HexStrike
# or MetasploitMCP. Operators who need external adapters should install
# and run them manually (for example under `third_party/`) and then
# configure `mcp_servers.json` to reference the adapter.
#
# A minimal example adapter scaffold is provided at
# `pentestagent/mcp/example_adapter.py` to help implement adapters that
# match the expected adapter interface.
# Metasploit RPC (msfrpcd) connection settings
# - `MSF_USER`/`MSF_PASSWORD`: msfrpcd credentials (keep password secret)

9
pentestagent/interface/tui.py
View File

@@ -1314,11 +1314,10 @@ class PentestAgentTUI(App):
self.rag_engine = None
# MCP: automatic install/start has been removed. Operators should
# run the scripts in `third_party/` manually to install and start
# any vendored MCP servers (e.g., HexStrike, MetasploitMCP) and
# then configure `mcp_servers.json` accordingly. If a config is
# present, `MCPManager` will auto-connect and register tools so
# the TUI can display MCP-provided tools.
# install and run any external MCP adapters themselves (for
# example under `third_party/`) and then configure
# `mcp_servers.json` accordingly. A minimal example adapter is
# available at `pentestagent/mcp/example_adapter.py`.
try:
from ..mcp import MCPManager

83
pentestagent/mcp/example_adapter.py Normal file
View File

@@ -0,0 +1,83 @@
"""Minimal MCP adapter scaffold for PentestAgent.
This module provides a small example adapter and a base interface that
adapter implementers can follow. Adapters are expected to provide a
lightweight set of methods so the `MCPManager` or external tools can
manage adapter lifecycle and issue tool calls. This scaffold intentionally
does not auto-start external processes; it's a development aid only.
Implemented surface (example):
- `BaseAdapter` (abstract interface)
- `ExampleAdapter` (in-process mock adapter for testing)
Usage:
- Use `ExampleAdapter` as a working reference when implementing real
adapters under `third_party/` or when wiring an adapter into
`mcp_servers.json`.
"""
from __future__ import annotations
from typing import Any, Dict, List, Optional
class BaseAdapter:
"""Minimal adapter interface.
Implementers should provide at least these methods. Real adapters may
expose additional methods such as `stop_sync` or an underlying
`_process` attribute that the manager may inspect when cleaning up.
"""
name: str = "base"
async def start(self) -> None: # pragma: no cover - interface
raise NotImplementedError()
async def stop(self) -> None: # pragma: no cover - interface
raise NotImplementedError()
def stop_sync(self) -> None: # pragma: no cover - optional
raise NotImplementedError()
async def list_tools(self) -> List[Dict[str, Any]]: # pragma: no cover - interface
raise NotImplementedError()
async def call_tool(self, name: str, arguments: Dict[str, Any]) -> Any: # pragma: no cover - interface
raise NotImplementedError()
class ExampleAdapter(BaseAdapter):
"""A trivial in-process adapter useful for tests and development.
- `list_tools()` returns a single example tool definition.
- `call_tool()` returns a simple echo response.
"""
name = "example"
def __init__(self):
self._running = False
async def start(self) -> None:
self._running = True
async def stop(self) -> None:
self._running = False
def stop_sync(self) -> None:
# Synchronous stop helper for manager cleanup code paths
self._running = False
async def list_tools(self) -> List[Dict[str, Any]]:
return [
{
"name": "ping",
"description": "Return a ping response",
"inputSchema": {"type": "object", "properties": {}},
}
]
async def call_tool(self, name: str, arguments: Dict[str, Any]) -> Any:
if name == "ping":
return [{"type": "text", "text": "pong"}]
raise ValueError(f"Unknown tool: {name}")

54
pentestagent/mcp/manager.py
View File

@@ -142,19 +142,11 @@ class MCPManager:
start_on_launch=config.get("start_on_launch", False),
description=config.get("description", ""),
)
# Allow override via environment variables for vendored MCP servers.
# Per-adapter overrides supported:
# - Hexstrike: LAUNCH_HEXTRIKE or LAUNCH_HEXSTRIKE
# - Metasploit: LAUNCH_METASPLOIT_MCP
# If set to a truthy value (1,true,y), force-enable auto-start for matching vendored server.
# If set to a falsy value (0,false,n), force-disable auto-start for matching vendored server.
# NOTE: Removed automatic LAUNCH_* overrides. MCP vendored adapters and
# auto-start behavior is intentionally disabled. MCP servers should be
# installed and configured manually via scripts in `third_party/` and
# `mcp_servers.json` should be prepared by the operator. Automatic
# start-on-launch logic previously controlled by environment variables
# is deprecated to simplify setup and avoid unexpected background
# processes.
# Environment-based auto-start overrides (previously supported via
# LAUNCH_* variables) have been removed. MCP adapters are no
# longer auto-started by the manager; operators should install and
# run adapters manually and add them to `mcp_servers.json` if they
# want `MCPManager` to connect to them.
return servers
except json.JSONDecodeError as e:
@@ -330,39 +322,9 @@ class MCPManager:
if name not in servers_config:
return None
config = servers_config[name]
# If this appears to be a vendored Metasploit MCP entry, attempt to auto-start
# the vendored adapter so `pentestagent mcp test metasploit-local` works
try:
args_joined = " ".join(config.args or [])
cmd_str = config.command or ""
is_msf = (
(name and "metasploit" in name.lower())
or ("third_party/MetasploitMCP" in cmd_str)
or ("third_party/MetasploitMCP" in args_joined)
)
if is_msf:
launch_msf_env = os.environ.get("LAUNCH_METASPLOIT_MCP")
launch_disabled = False
if launch_msf_env is not None:
v = str(launch_msf_env).strip().lower()
if v in ("0", "false", "no", "n"):
launch_disabled = True
if not launch_disabled:
try:
from .metasploit_adapter import MetasploitAdapter
adapter = MetasploitAdapter()
started = await adapter.start()
if started:
try:
self._started_adapters[name] = adapter
except Exception:
pass
print(f"[MCP] Auto-started vendored server for {name}")
except Exception:
pass
except Exception:
pass
# Auto-start of vendored adapters (HexStrike/MetasploitMCP) has been
# removed. Operators should install and run any third-party MCP adapters
# manually under `third_party/` and configure `mcp_servers.json`.
server = await self._connect_server(config)
if server:

2
pentestagent/workspaces/utils.py
View File

@@ -36,7 +36,7 @@ def get_loot_base(root: Optional[Path] = None) -> Path:
def get_loot_file(relpath: str, root: Optional[Path] = None) -> Path:
"""Return a Path for a file under the loot base, creating parent dirs.
Example: get_loot_file('artifacts/hexstrike.log')
Example: get_loot_file('artifacts/example.log')
"""
base = get_loot_base(root=root)
p = base / relpath

22
tests/test_mcp_scaffold.py Normal file
View File

@@ -0,0 +1,22 @@
import asyncio
from pentestagent.mcp.example_adapter import ExampleAdapter
def test_example_adapter_list_and_call():
adapter = ExampleAdapter()
async def run():
await adapter.start()
tools = await adapter.list_tools()
assert isinstance(tools, list)
assert any(t.get("name") == "ping" for t in tools)
result = await adapter.call_tool("ping", {})
assert isinstance(result, list)
assert result[0].get("text") == "pong"
await adapter.stop()
asyncio.run(run())