CalvinBackup/fuzzforge_ai
1
0
Fork 0
mirror of https://github.com/FuzzingLabs/fuzzforge_ai.git synced 2026-03-18 11:03:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #48 from FuzzingLabs/dev

Browse Source
This commit is contained in:
AFredefon
2026-03-17 08:15:42 +01:00
committed by GitHub
parent 07c32de294 2e96517d11
commit e7022c2c82
9 changed files with 326 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@@ -10,3 +10,6 @@ __pycache__
# Podman/Docker container storage artifacts
~/.fuzzforge/
# User-specific hub config (generated at runtime)
hub-config.json

242
README.md
View File

@@ -17,9 +17,9 @@
<sub>
<a href="#-overview"><b>Overview</b></a> •
<a href="#-features"><b>Features</b></a> •
<a href="#-mcp-security-hub"><b>Security Hub</b></a> •
<a href="#-installation"><b>Installation</b></a> •
<a href="USAGE.md"><b>Usage Guide</b></a> •
<a href="#-modules"><b>Modules</b></a> •
<a href="#-contributing"><b>Contributing</b></a>
</sub>
</p>
@@ -32,39 +32,44 @@
## 🚀 Overview
**FuzzForge AI** is an open-source runtime that enables AI agents (GitHub Copilot, Claude, etc.) to orchestrate security research workflows through the **Model Context Protocol (MCP)**.
**FuzzForge AI** is an open-source MCP server that enables AI agents (GitHub Copilot, Claude, etc.) to orchestrate security research workflows through the **Model Context Protocol (MCP)**.
### The Core: Modules
FuzzForge connects your AI assistant to **MCP tool hubs** — collections of containerized security tools that the agent can discover, chain, and execute autonomously. Instead of manually running security tools, describe what you want and let your AI assistant handle it.
At the heart of FuzzForge are **modules** - containerized security tools that AI agents can discover, configure, and orchestrate. Each module encapsulates a specific security capability (static analysis, fuzzing, crash analysis, etc.) and runs in an isolated container.
### The Core: Hub Architecture
- **🔌 Plug & Play**: Modules are self-contained - just pull and run
- **🤖 AI-Native**: Designed for AI agent orchestration via MCP
- **🔗 Composable**: Chain modules together into automated workflows
- **📦 Extensible**: Build custom modules with the Python SDK
FuzzForge acts as a **meta-MCP server** — a single MCP endpoint that gives your AI agent access to tools from multiple MCP hub servers. Each hub server is a containerized security tool (Binwalk, YARA, Radare2, Nmap, etc.) that the agent can discover at runtime.
FuzzForge AI handles module discovery, execution, and result collection. Security modules (developed separately) provide the actual security tooling - from static analyzers to fuzzers to crash triagers.
- **🔍 Discovery**: The agent lists available hub servers and discovers their tools
- **🤖 AI-Native**: Hub tools provide agent context — usage tips, workflow guidance, and domain knowledge
- **🔗 Composable**: Chain tools from different hubs into automated pipelines
- **📦 Extensible**: Add your own MCP servers to the hub registry
Instead of manually running security tools, describe what you want and let your AI assistant handle it.
### 🎬 Use Case: Firmware Vulnerability Research
> **Scenario**: Analyze a firmware image to find security vulnerabilities — fully automated by an AI agent.
```
User: "Search for vulnerabilities in firmware.bin"
Agent → Binwalk: Extract filesystem from firmware image
Agent → YARA: Scan extracted files for vulnerability patterns
Agent → Radare2: Trace dangerous function calls in prioritized binaries
Agent → Report: 8 vulnerabilities found (2 critical, 4 high, 2 medium)
```
### 🎬 Use Case: Rust Fuzzing Pipeline
> **Scenario**: Fuzz a Rust crate to discover vulnerabilities using AI-assisted harness generation and parallel fuzzing.
<table align="center">
<tr>
<th>1️⃣ Analyze, Generate & Validate Harnesses</th>
<th>2️⃣ Run Parallel Continuous Fuzzing</th>
</tr>
<tr>
<td><img src="assets/demopart2.gif" alt="FuzzForge Demo - Analysis Pipeline" width="100%"></td>
<td><img src="assets/demopart1.gif" alt="FuzzForge Demo - Parallel Fuzzing" width="100%"></td>
</tr>
<tr>
<td align="center"><sub>AI agent analyzes code, generates harnesses, and validates they compile</sub></td>
<td align="center"><sub>Multiple fuzzing sessions run in parallel with live metrics</sub></td>
</tr>
</table>
```
User: "Fuzz the blurhash crate for vulnerabilities"
Agent → Rust Analyzer: Identify fuzzable functions and attack surface
Agent → Harness Gen: Generate and validate fuzzing harnesses
Agent → Cargo Fuzzer: Run parallel coverage-guided fuzzing sessions
Agent → Crash Analysis: Deduplicate and triage discovered crashes
```
---
@@ -82,13 +87,13 @@ If you find FuzzForge useful, please **star the repo** to support development!
| Feature | Description |
|---------|-------------|
| 🤖 **AI-Native** | Built for MCP - works with GitHub Copilot, Claude, and any MCP-compatible agent |
| 📦 **Containerized** | Each module runs in isolation via Docker or Podman |
| 🔄 **Continuous Mode** | Long-running tasks (fuzzing) with real-time metrics streaming |
| 🔗 **Workflows** | Chain multiple modules together in automated pipelines |
| 🛠️ **Extensible** | Create custom modules with the Python SDK |
| 🏠 **Local First** | All execution happens on your machine - no cloud required |
| 🔒 **Secure** | Sandboxed containers with no network access by default |
| 🤖 **AI-Native** | Built for MCP — works with GitHub Copilot, Claude, and any MCP-compatible agent |
| 🔌 **Hub System** | Connect to MCP tool hubs — each hub brings dozens of containerized security tools |
| 🔍 **Tool Discovery** | Agents discover available tools at runtime with built-in usage guidance |
| 🔗 **Pipelines** | Chain tools from different hubs into automated multi-step workflows |
| 🔄 **Persistent Sessions** | Long-running tools (Radare2, fuzzers) with stateful container sessions |
| 🏠 **Local First** | All execution happens on your machine — no cloud required |
| 🔒 **Sandboxed** | Every tool runs in an isolated container via Docker or Podman |
---
@@ -102,27 +107,57 @@ If you find FuzzForge useful, please **star the repo** to support development!
▼
┌─────────────────────────────────────────────────────────────────┐
│ FuzzForge MCP Server │
│ ┌─────────────┐ ┌──────────────┐ ┌────────────────────────┐ │
│ │list_modules │ │execute_module│ │start_continuous_module │ │
│ └─────────────┘ └──────────────┘ └────────────────────────┘ │
│ │
│ Projects Hub Discovery Hub Execution │
│ ┌──────────────┐ ┌──────────────────┐ ┌───────────────────┐ │
│ │init_project │ │list_hub_servers │ │execute_hub_tool │ │
│ │set_assets │ │discover_hub_tools│ │start_hub_server │ │
│ │list_results │ │get_tool_schema │ │stop_hub_server │ │
│ └──────────────┘ └──────────────────┘ └───────────────────┘ │
└───────────────────────────┬─────────────────────────────────────┘
│
│ Docker/Podman
▼
┌─────────────────────────────────────────────────────────────────┐
│ FuzzForge Runner │
│ Container Engine (Docker/Podman) │
└───────────────────────────┬─────────────────────────────────────┘
│
┌───────────────────┼───────────────────┐
▼ ▼ ▼
┌───────────────┐ ┌───────────────┐ ┌───────────────┐
│ Module A │ │ Module B │ │ Module C │
│ (Container) │ │ (Container) │ │ (Container) │
└───────────────┘ └───────────────┘ └───────────────┘
│ MCP Hub Servers │
│ │
│ ┌───────────┐ ┌───────────┐ ┌───────────┐ ┌───────────┐ │
│ │ Binwalk │ │ YARA │ │ Radare2 │ │ Nmap │ │
│ │ 6 tools │ │ 5 tools │ │ 32 tools │ │ 8 tools │ │
│ └───────────┘ └───────────┘ └───────────┘ └───────────┘ │
│ ┌───────────┐ ┌───────────┐ ┌───────────┐ ┌───────────┐ │
│ │ Nuclei │ │ SQLMap │ │ Trivy │ │ ... │ │
│ │ 7 tools │ │ 8 tools │ │ 7 tools │ │ 36 hubs │ │
│ └───────────┘ └───────────┘ └───────────┘ └───────────┘ │
└─────────────────────────────────────────────────────────────────┘
```
---
## 🔧 MCP Security Hub
FuzzForge ships with built-in support for the **[MCP Security Hub](https://github.com/FuzzingLabs/mcp-security-hub)** — a collection of 36 production-ready, Dockerized MCP servers covering offensive security:
| Category | Servers | Examples |
|----------|---------|----------|
| 🔍 **Reconnaissance** | 8 | Nmap, Masscan, Shodan, WhatWeb |
| 🌐 **Web Security** | 6 | Nuclei, SQLMap, ffuf, Nikto |
| 🔬 **Binary Analysis** | 6 | Radare2, Binwalk, YARA, Capa, Ghidra |
| ⛓️ **Blockchain** | 3 | Medusa, Solazy, DAML Viewer |
| ☁️ **Cloud Security** | 3 | Trivy, Prowler, RoadRecon |
| 💻 **Code Security** | 1 | Semgrep |
| 🔑 **Secrets Detection** | 1 | Gitleaks |
| 💥 **Exploitation** | 1 | SearchSploit |
| 🎯 **Fuzzing** | 2 | Boofuzz, Dharma |
| 🕵️ **OSINT** | 2 | Maigret, DNSTwist |
| 🛡️ **Threat Intel** | 2 | VirusTotal, AlienVault OTX |
| 🏰 **Active Directory** | 1 | BloodHound |
> 185+ individual tools accessible through a single MCP connection.
The hub is open source and can be extended with your own MCP servers. See the [mcp-security-hub repository](https://github.com/FuzzingLabs/mcp-security-hub) for details.
---
## 📦 Installation
### Prerequisites
@@ -140,11 +175,20 @@ cd fuzzforge_ai
# Install dependencies
uv sync
# Build module images
make build-modules
```
### Link the Security Hub
```bash
# Clone the MCP Security Hub
git clone https://github.com/FuzzingLabs/mcp-security-hub.git ~/.fuzzforge/hubs/mcp-security-hub
# Build the Docker images for the hub tools
./scripts/build-hub-images.sh
```
Or use the terminal UI (`uv run fuzzforge ui`) to link hubs interactively.
### Configure MCP for Your AI Agent
```bash
@@ -165,81 +209,20 @@ uv run fuzzforge mcp status
---
## 📦 Modules
## 🧑‍💻 Usage
FuzzForge modules are containerized security tools that AI agents can orchestrate. The module ecosystem is designed around a simple principle: **the OSS runtime orchestrates, enterprise modules execute**.
Once installed, just talk to your AI agent:
### Module Ecosystem
| | FuzzForge AI | FuzzForge Enterprise Modules |
|---|---|---|
| **What** | Runtime & MCP server | Security research modules |
| **License** | Apache 2.0 | BSL 1.1 (Business Source License) |
| **Compatibility** | ✅ Runs any compatible module | ✅ Works with FuzzForge AI |
**Enterprise modules** are developed separately and provide production-ready security tooling:
| Category | Modules | Description |
|----------|---------|-------------|
| 🔍 **Static Analysis** | Rust Analyzer, Solidity Analyzer, Cairo Analyzer | Code analysis and fuzzable function detection |
| 🎯 **Fuzzing** | Cargo Fuzzer, Honggfuzz, AFL++ | Coverage-guided fuzz testing |
| 💥 **Crash Analysis** | Crash Triager, Root Cause Analyzer | Automated crash deduplication and analysis |
| 🔐 **Vulnerability Detection** | Pattern Matcher, Taint Analyzer | Security vulnerability scanning |
| 📝 **Reporting** | Report Generator, SARIF Exporter | Automated security report generation |
> 💡 **Build your own modules!** The FuzzForge SDK allows you to create custom modules that integrate seamlessly with FuzzForge AI. See [Creating Custom Modules](#-creating-custom-modules).
### Execution Modes
Modules run in two execution modes:
#### One-shot Execution
Run a module once and get results:
```python
result = execute_module("my-analyzer", assets_path="/path/to/project")
```
"What security tools are available?"
"Scan this firmware image for vulnerabilities"
"Analyze this binary with radare2"
"Run nuclei against https://example.com"
```
#### Continuous Execution
The agent will use FuzzForge to discover the right hub tools, chain them into a pipeline, and return results — all without you touching a terminal.
For long-running tasks like fuzzing, with real-time metrics:
```python
# Start continuous execution
session = start_continuous_module("my-fuzzer",
assets_path="/path/to/project",
configuration={"target": "my_target"})
# Check status with live metrics
status = get_continuous_status(session["session_id"])
# Stop and collect results
stop_continuous_module(session["session_id"])
```
---
## 🛠️ Creating Custom Modules
Build your own security modules with the FuzzForge SDK:
```python
from fuzzforge_modules_sdk import FuzzForgeModule, FuzzForgeModuleResults
class MySecurityModule(FuzzForgeModule):
def _run(self, resources):
self.emit_event("started", target=resources[0].path)
# Your analysis logic here
results = self.analyze(resources)
self.emit_progress(100, status="completed",
message=f"Analysis complete")
return FuzzForgeModuleResults.SUCCESS
```
📖 See the [Module SDK Guide](fuzzforge-modules/fuzzforge-modules-sdk/README.md) for details.
See the [Usage Guide](USAGE.md) for detailed setup and advanced workflows.
---
@@ -247,26 +230,17 @@ class MySecurityModule(FuzzForgeModule):
```
fuzzforge_ai/
├── fuzzforge-cli/ # Command-line interface
├── fuzzforge-mcp/ # MCP server — the core of FuzzForge
├── fuzzforge-cli/ # Command-line interface & terminal UI
├── fuzzforge-common/ # Shared abstractions (containers, storage)
├── fuzzforge-mcp/ # MCP server for AI agents
├── fuzzforge-modules/ # Security modules
│ └── fuzzforge-modules-sdk/ # Module development SDK
├── fuzzforge-runner/ # Local execution engine
├── fuzzforge-types/ # Type definitions & schemas
└── demo/ # Demo projects for testing
├── fuzzforge-runner/ # Container execution engine (Docker/Podman)
├── fuzzforge-tests/ # Integration tests
├── mcp-security-hub/ # Default hub: 36 offensive security MCP servers
└── scripts/ # Hub image build scripts
```
---
## 🗺️ What's Next
**[MCP Security Hub](https://github.com/FuzzingLabs/mcp-security-hub) integration** — Bridge 175+ offensive security tools (Nmap, Nuclei, Ghidra, and more) into FuzzForge workflows, all orchestrated by AI agents.
See [ROADMAP.md](ROADMAP.md) for the full roadmap.
---
## 🤝 Contributing
We welcome contributions from the community!
@@ -274,7 +248,7 @@ We welcome contributions from the community!
- 🐛 Report bugs via [GitHub Issues](../../issues)
- 💡 Suggest features or improvements
- 🔧 Submit pull requests
- 📦 Share your custom modules
- 🔌 Add new MCP servers to the [Security Hub](https://github.com/FuzzingLabs/mcp-security-hub)
See [CONTRIBUTING.md](CONTRIBUTING.md) for guidelines.

BIN
assets/demopart1.gif
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 360 KiB

BIN
assets/demopart2.gif
View File

Binary file not shown.
Side by Side

Before

Width:  |  Height:  |  Size: 2.1 MiB

41
fuzzforge-mcp/src/fuzzforge_mcp/application.py
View File

@@ -47,15 +47,46 @@ FuzzForge is a security research orchestration platform. Use these tools to:
Typical workflow:
1. Initialize a project with `init_project`
2. Set project assets with `set_project_assets` (optional, only needed once for the source directory)
2. Set project assets with `set_project_assets` — path to the directory containing
target files (firmware images, binaries, source code, etc.)
3. List available hub servers with `list_hub_servers`
4. Discover tools from servers with `discover_hub_tools`
5. Execute hub tools with `execute_hub_tool`
Hub workflow:
1. List available hub servers with `list_hub_servers`
2. Discover tools from servers with `discover_hub_tools`
3. Execute hub tools with `execute_hub_tool`
Agent context convention:
When you call `discover_hub_tools`, some servers return an `agent_context` field
with usage tips, known issues, rule templates, and workflow guidance. Always read
this context before using the server's tools.
File access in containers:
- Assets set via `set_project_assets` are mounted read-only at `/app/uploads/` and `/app/samples/`
- A writable output directory is mounted at `/app/output/` — use it for extraction results, reports, etc.
- Always use container paths (e.g. `/app/uploads/file`) when passing file arguments to hub tools
Stateful tools:
- Some tools (e.g. radare2-mcp) require multi-step sessions. Use `start_hub_server` to launch
a persistent container, then `execute_hub_tool` calls reuse that container. Stop with `stop_hub_server`.
Firmware analysis pipeline (when analyzing firmware images):
1. **binwalk-mcp** (`binwalk_scan` + `binwalk_extract`) — identify and extract filesystem from firmware
2. **yara-mcp** (`yara_scan_with_rules`) — scan extracted files with vulnerability rules to prioritize targets
3. **radare2-mcp** (persistent session) — confirm dangerous code paths
4. **searchsploit-mcp** (`search_exploitdb`) — query version strings from radare2 against ExploitDB
Run steps 3 and 4 outputs feed into a final triage summary.
radare2-mcp agent context (upstream tool — no embedded context):
- Start a persistent session with `start_hub_server("radare2-mcp")` before any calls.
- IMPORTANT: the `open_file` tool requires the parameter name `file_path` (with underscore),
not `filepath`. Example: `execute_hub_tool("hub:radare2-mcp:open_file", {"file_path": "/app/output/..."})`
- Workflow: `open_file` → `analyze` → `list_imports` → `xrefs_to` → `run_command` with `pdf @ <addr>`.
- Static binary fallback: firmware binaries are often statically linked. When `list_imports`
returns an empty result, fall back to `list_symbols` and search for dangerous function names
(system, strcpy, gets, popen, sprintf) in the output. Then use `xrefs_to` on their addresses.
- For string extraction, use `run_command` with `iz` (data section strings).
The `list_all_strings` tool may return garbled output for large binaries.
- For decompilation, use `run_command` with `pdc @ <addr>` (pseudo-C) or `pdf @ <addr>`
(annotated disassembly). The `decompile` tool may fail with "not available in current mode".
- Stop the session with `stop_hub_server("radare2-mcp")` when done.
""",
lifespan=lifespan,
)

6
fuzzforge-mcp/src/fuzzforge_mcp/resources/executions.py
View File

@@ -30,10 +30,10 @@ async def list_executions() -> list[dict[str, Any]]:
return [
{
"execution_id": exec_id,
"has_results": storage.get_execution_results(project_path, exec_id) is not None,
"execution_id": entry["execution_id"],
"has_results": storage.get_execution_results(project_path, entry["execution_id"]) is not None,
}
for exec_id in execution_ids
for entry in execution_ids
]
except Exception as exception:

80
fuzzforge-mcp/src/fuzzforge_mcp/storage.py
View File

@@ -13,9 +13,11 @@ from __future__ import annotations
import json
import logging
from datetime import UTC, datetime
from pathlib import Path
from tarfile import open as Archive # noqa: N812
from typing import Any
from uuid import uuid4
logger = logging.getLogger("fuzzforge-mcp")
@@ -79,6 +81,7 @@ class LocalStorage:
storage_path = self._get_project_path(project_path)
storage_path.mkdir(parents=True, exist_ok=True)
(storage_path / "runs").mkdir(parents=True, exist_ok=True)
(storage_path / "output").mkdir(parents=True, exist_ok=True)
# Create .gitignore to avoid committing large files
gitignore_path = storage_path / ".gitignore"
@@ -86,6 +89,7 @@ class LocalStorage:
gitignore_path.write_text(
"# FuzzForge storage - ignore large/temporary files\n"
"runs/\n"
"output/\n"
"!config.json\n"
)
@@ -141,17 +145,85 @@ class LocalStorage:
logger.info("Set project assets: %s -> %s", project_path.name, assets_path)
return assets_path
def list_executions(self, project_path: Path) -> list[str]:
"""List all execution IDs for a project.
def get_project_output_path(self, project_path: Path) -> Path | None:
"""Get the output directory path for a project.
Returns the path to the writable output directory that is mounted
into hub tool containers at /app/output.
:param project_path: Path to the project directory.
:returns: List of execution IDs.
:returns: Path to output directory, or None if project not initialized.
"""
output_path = self._get_project_path(project_path) / "output"
if output_path.exists():
return output_path
return None
def record_execution(
self,
project_path: Path,
server_name: str,
tool_name: str,
arguments: dict[str, Any],
result: dict[str, Any],
) -> str:
"""Record an execution result to the project's runs directory.
:param project_path: Path to the project directory.
:param server_name: Hub server name.
:param tool_name: Tool name that was executed.
:param arguments: Arguments passed to the tool.
:param result: Execution result dictionary.
:returns: Execution ID.
"""
execution_id = f"{datetime.now(tz=UTC).strftime('%Y%m%dT%H%M%SZ')}_{uuid4().hex[:8]}"
run_dir = self._get_project_path(project_path) / "runs" / execution_id
run_dir.mkdir(parents=True, exist_ok=True)
metadata = {
"execution_id": execution_id,
"timestamp": datetime.now(tz=UTC).isoformat(),
"server": server_name,
"tool": tool_name,
"arguments": arguments,
"success": result.get("success", False),
"result": result,
}
(run_dir / "metadata.json").write_text(json.dumps(metadata, indent=2, default=str))
logger.info("Recorded execution %s: %s:%s", execution_id, server_name, tool_name)
return execution_id
def list_executions(self, project_path: Path) -> list[dict[str, Any]]:
"""List all executions for a project with summary metadata.
:param project_path: Path to the project directory.
:returns: List of execution summaries (id, timestamp, server, tool, success).
"""
runs_dir = self._get_project_path(project_path) / "runs"
if not runs_dir.exists():
return []
return [d.name for d in runs_dir.iterdir() if d.is_dir()]
executions: list[dict[str, Any]] = []
for run_dir in sorted(runs_dir.iterdir(), reverse=True):
if not run_dir.is_dir():
continue
meta_path = run_dir / "metadata.json"
if meta_path.exists():
meta = json.loads(meta_path.read_text())
executions.append({
"execution_id": meta.get("execution_id", run_dir.name),
"timestamp": meta.get("timestamp"),
"server": meta.get("server"),
"tool": meta.get("tool"),
"success": meta.get("success"),
})
else:
executions.append({"execution_id": run_dir.name})
return executions
def get_execution_results(
self,

92
fuzzforge-mcp/src/fuzzforge_mcp/tools/hub.py
View File

@@ -20,10 +20,41 @@ from fuzzforge_mcp.dependencies import get_project_path, get_settings, get_stora
mcp: FastMCP = FastMCP()
# Name of the convention tool that hub servers can implement to provide
# rich usage context for AI agents (known issues, workflow tips, rules, etc.).
_AGENT_CONTEXT_TOOL = "get_agent_context"
# Global hub executor instance (lazy initialization)
_hub_executor: HubExecutor | None = None
async def _fetch_agent_context(
executor: HubExecutor,
server_name: str,
tools: list[Any],
) -> str | None:
"""Call get_agent_context if the server provides it.
Returns the context string, or None if the server doesn't implement
the convention or the call fails.
"""
if not any(t.name == _AGENT_CONTEXT_TOOL for t in tools):
return None
try:
result = await executor.execute_tool(
identifier=f"hub:{server_name}:{_AGENT_CONTEXT_TOOL}",
arguments={},
)
if result.success and result.result:
content = result.result.get("content", [])
if content and isinstance(content, list):
text: str = content[0].get("text", "")
return text
except Exception: # noqa: BLE001, S110 - best-effort context fetch
pass
return None
def _get_hub_executor() -> HubExecutor:
"""Get or create the hub executor instance.
@@ -50,12 +81,15 @@ def _get_hub_executor() -> HubExecutor:
@mcp.tool
async def list_hub_servers() -> dict[str, Any]:
async def list_hub_servers(category: str | None = None) -> dict[str, Any]:
"""List all registered MCP hub servers.
Returns information about configured hub servers, including
their connection type, status, and discovered tool count.
:param category: Optional category to filter by (e.g. "binary-analysis",
"web-security", "reconnaissance"). Only servers in this category
are returned.
:return: Dictionary with list of hub servers.
"""
@@ -63,6 +97,9 @@ async def list_hub_servers() -> dict[str, Any]:
executor = _get_hub_executor()
servers = executor.list_servers()
if category:
servers = [s for s in servers if s.get("category") == category]
return {
"servers": servers,
"count": len(servers),
@@ -93,7 +130,14 @@ async def discover_hub_tools(server_name: str | None = None) -> dict[str, Any]:
if server_name:
tools = await executor.discover_server_tools(server_name)
return {
# Convention: auto-fetch agent context if server provides it.
agent_context = await _fetch_agent_context(executor, server_name, tools)
# Hide the convention tool from the agent's tool list.
visible_tools = [t for t in tools if t.name != "get_agent_context"]
result: dict[str, Any] = {
"server": server_name,
"tools": [
{
@@ -102,15 +146,24 @@ async def discover_hub_tools(server_name: str | None = None) -> dict[str, Any]:
"description": t.description,
"parameters": [p.model_dump() for p in t.parameters],
}
for t in tools
for t in visible_tools
],
"count": len(tools),
"count": len(visible_tools),
}
if agent_context:
result["agent_context"] = agent_context
return result
else:
results = await executor.discover_all_tools()
all_tools = []
contexts: dict[str, str] = {}
for server, tools in results.items():
ctx = await _fetch_agent_context(executor, server, tools)
if ctx:
contexts[server] = ctx
for tool in tools:
if tool.name == "get_agent_context":
continue
all_tools.append({
"identifier": tool.identifier,
"name": tool.name,
@@ -119,11 +172,14 @@ async def discover_hub_tools(server_name: str | None = None) -> dict[str, Any]:
"parameters": [p.model_dump() for p in tool.parameters],
})
return {
result = {
"servers_discovered": len(results),
"tools": all_tools,
"count": len(all_tools),
}
if contexts:
result["agent_contexts"] = contexts
return result
except Exception as e:
if isinstance(e, ToolError):
@@ -183,6 +239,11 @@ async def execute_hub_tool(
Always use /app/uploads/<filename> or /app/samples/<filename> when
passing file paths to hub tools — do NOT use the host path.
Tool outputs are persisted to a writable shared volume:
- /app/output/ (writable — extraction results, reports, etc.)
Files written here survive container destruction and are available
to subsequent tool calls. The host path is .fuzzforge/output/.
"""
try:
executor = _get_hub_executor()
@@ -191,6 +252,7 @@ async def execute_hub_tool(
# Mounts the assets directory at the standard paths used by hub tools:
# /app/uploads — binwalk, and other tools that use UPLOAD_DIR
# /app/samples — yara, capa, and other tools that use SAMPLES_DIR
# /app/output — writable volume for tool outputs (persists across calls)
extra_volumes: list[str] = []
try:
storage = get_storage()
@@ -202,6 +264,9 @@ async def execute_hub_tool(
f"{assets_str}:/app/uploads:ro",
f"{assets_str}:/app/samples:ro",
]
output_path = storage.get_project_output_path(project_path)
if output_path:
extra_volumes.append(f"{output_path!s}:/app/output:rw")
except Exception: # noqa: BLE001 - never block tool execution due to asset injection failure
extra_volumes = []
@@ -212,6 +277,20 @@ async def execute_hub_tool(
extra_volumes=extra_volumes or None,
)
# Record execution history for list_executions / get_execution_results.
try:
storage = get_storage()
project_path = get_project_path()
storage.record_execution(
project_path=project_path,
server_name=result.server_name,
tool_name=result.tool_name,
arguments=arguments or {},
result=result.to_dict(),
)
except Exception: # noqa: BLE001, S110 - never fail the tool call due to recording issues
pass
return result.to_dict()
except Exception as e:
@@ -372,6 +451,9 @@ async def start_hub_server(server_name: str) -> dict[str, Any]:
f"{assets_str}:/app/uploads:ro",
f"{assets_str}:/app/samples:ro",
]
output_path = storage.get_project_output_path(project_path)
if output_path:
extra_volumes.append(f"{output_path!s}:/app/output:rw")
except Exception: # noqa: BLE001 - never block server start due to asset injection failure
extra_volumes = []

26
fuzzforge-mcp/src/fuzzforge_mcp/tools/projects.py
View File

@@ -15,15 +15,14 @@ mcp: FastMCP = FastMCP()
@mcp.tool
async def init_project(project_path: str | None = None) -> dict[str, Any]:
"""Initialize a new FuzzForge project.
"""Initialize a new FuzzForge project workspace.
Creates a `.fuzzforge/` directory inside the project for storing:
- config.json: Project configuration
- runs/: Execution results
Creates a `.fuzzforge/` directory for storing configuration and execution results.
Call this once before using hub tools. The project path is a working directory
for FuzzForge state — it does not need to contain the files you want to analyze.
Use `set_project_assets` separately to specify the target files.
This should be called before executing hub tools.
:param project_path: Path to the project directory. If not provided, uses current directory.
:param project_path: Working directory for FuzzForge state. Defaults to current directory.
:return: Project initialization result.
"""
@@ -51,12 +50,13 @@ async def init_project(project_path: str | None = None) -> dict[str, Any]:
@mcp.tool
async def set_project_assets(assets_path: str) -> dict[str, Any]:
"""Set the initial assets (source code) for a project.
"""Set the directory containing target files to analyze.
This sets the DEFAULT source directory that will be mounted into
hub tool containers via volume mounts.
Points FuzzForge to the directory with your analysis targets
(firmware images, binaries, source code, etc.). This directory
is mounted read-only into hub tool containers.
:param assets_path: Path to the project source directory.
:param assets_path: Path to the directory containing files to analyze.
:return: Result including stored assets path.
"""
@@ -85,9 +85,9 @@ async def set_project_assets(assets_path: str) -> dict[str, Any]:
async def list_executions() -> dict[str, Any]:
"""List all executions for the current project.
Returns a list of execution IDs that can be used to retrieve results.
Returns execution summaries including server, tool, timestamp, and success status.
:return: List of execution IDs.
:return: List of execution summaries.
"""
storage = get_storage()