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

refactor: remove module system, migrate to MCP hub tools architecture

Browse Source
This commit is contained in:
AFredefon
2026-03-08 17:53:29 +01:00
parent 075b678e9d
commit 1d495cedce
141 changed files with 1182 additions and 8992 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
fuzzforge-cli/pyproject.toml
View File

@@ -6,7 +6,7 @@ authors = []
readme = "README.md"
requires-python = ">=3.14"
dependencies = [
"fuzzforge-runner==0.0.1",
"fuzzforge-mcp==0.0.1",
"rich>=14.0.0",
"typer==0.20.1",
]
@@ -25,4 +25,4 @@ tests = [
fuzzforge = "fuzzforge_cli.__main__:main"
[tool.uv.sources]
fuzzforge-runner = { workspace = true }
fuzzforge-mcp = { workspace = true }

52
fuzzforge-cli/src/fuzzforge_cli/application.py
View File

@@ -3,12 +3,12 @@
from pathlib import Path
from typing import Annotated
from fuzzforge_runner import Runner, Settings
from typer import Context as TyperContext
from typer import Option, Typer
from fuzzforge_cli.commands import mcp, modules, projects
from fuzzforge_cli.commands import mcp, projects
from fuzzforge_cli.context import Context
from fuzzforge_mcp.storage import LocalStorage
application: Typer = Typer(
name="fuzzforge",
@@ -27,15 +27,6 @@ def main(
help="Path to the FuzzForge project directory.",
),
] = Path.cwd(),
modules_path: Annotated[
Path,
Option(
"--modules",
"-m",
envvar="FUZZFORGE_MODULES_PATH",
help="Path to the modules directory.",
),
] = Path.home() / ".fuzzforge" / "modules",
storage_path: Annotated[
Path,
Option(
@@ -44,53 +35,20 @@ def main(
help="Path to the storage directory.",
),
] = Path.home() / ".fuzzforge" / "storage",
engine_type: Annotated[
str,
Option(
"--engine",
envvar="FUZZFORGE_ENGINE__TYPE",
help="Container engine type (docker or podman).",
),
] = "docker",
engine_socket: Annotated[
str,
Option(
"--socket",
envvar="FUZZFORGE_ENGINE__SOCKET",
help="Container engine socket path.",
),
] = "",
context: TyperContext = None, # type: ignore[assignment]
) -> None:
"""FuzzForge AI - Security research orchestration platform.
Execute security research modules in isolated containers.
Discover and execute MCP hub tools for security research.
"""
from fuzzforge_runner.settings import EngineSettings, ProjectSettings, StorageSettings
settings = Settings(
engine=EngineSettings(
type=engine_type, # type: ignore[arg-type]
socket=engine_socket,
),
storage=StorageSettings(
path=storage_path,
),
project=ProjectSettings(
default_path=project_path,
modules_path=modules_path,
),
)
runner = Runner(settings)
storage = LocalStorage(storage_path=storage_path)
context.obj = Context(
runner=runner,
storage=storage,
project_path=project_path,
)
application.add_typer(mcp.application)
application.add_typer(modules.application)
application.add_typer(projects.application)

32
fuzzforge-cli/src/fuzzforge_cli/commands/mcp.py
View File

@@ -137,7 +137,7 @@ def _find_fuzzforge_root() -> Path:
# Walk up to find fuzzforge-oss root
for parent in current.parents:
if (parent / "fuzzforge-mcp").is_dir() and (parent / "fuzzforge-runner").is_dir():
if (parent / "fuzzforge-mcp").is_dir():
return parent
# Fall back to cwd
@@ -146,14 +146,12 @@ def _find_fuzzforge_root() -> Path:
def _generate_mcp_config(
fuzzforge_root: Path,
modules_path: Path,
engine_type: str,
engine_socket: str,
) -> dict:
"""Generate MCP server configuration.
:param fuzzforge_root: Path to fuzzforge-oss installation.
:param modules_path: Path to the modules directory.
:param engine_type: Container engine type (podman or docker).
:param engine_socket: Container engine socket path.
:returns: MCP configuration dictionary.
@@ -181,7 +179,6 @@ def _generate_mcp_config(
"args": args,
"cwd": str(fuzzforge_root),
"env": {
"FUZZFORGE_MODULES_PATH": str(modules_path),
"FUZZFORGE_ENGINE__TYPE": engine_type,
"FUZZFORGE_ENGINE__GRAPHROOT": str(graphroot),
"FUZZFORGE_ENGINE__RUNROOT": str(runroot),
@@ -266,14 +263,6 @@ def generate(
help="AI agent to generate config for (copilot, claude-desktop, or claude-code).",
),
],
modules_path: Annotated[
Path | None,
Option(
"--modules",
"-m",
help="Path to the modules directory.",
),
] = None,
engine: Annotated[
str,
Option(
@@ -287,16 +276,12 @@ def generate(
:param context: Typer context.
:param agent: Target AI agent.
:param modules_path: Override modules path.
:param engine: Container engine type.
"""
console = Console()
fuzzforge_root = _find_fuzzforge_root()
# Use defaults if not specified
resolved_modules = modules_path or (fuzzforge_root / "fuzzforge-modules")
# Detect socket
if engine == "podman":
socket = _detect_podman_socket()
@@ -306,7 +291,6 @@ def generate(
# Generate config
server_config = _generate_mcp_config(
fuzzforge_root=fuzzforge_root,
modules_path=resolved_modules,
engine_type=engine,
engine_socket=socket,
)
@@ -350,14 +334,6 @@ def install(
help="AI agent to install config for (copilot, claude-desktop, or claude-code).",
),
],
modules_path: Annotated[
Path | None,
Option(
"--modules",
"-m",
help="Path to the modules directory.",
),
] = None,
engine: Annotated[
str,
Option(
@@ -382,7 +358,6 @@ def install(
:param context: Typer context.
:param agent: Target AI agent.
:param modules_path: Override modules path.
:param engine: Container engine type.
:param force: Overwrite existing configuration.
@@ -401,9 +376,6 @@ def install(
config_path = _get_claude_desktop_mcp_path()
servers_key = "mcpServers"
# Use defaults if not specified
resolved_modules = modules_path or (fuzzforge_root / "fuzzforge-modules")
# Detect socket
if engine == "podman":
socket = _detect_podman_socket()
@@ -413,7 +385,6 @@ def install(
# Generate server config
server_config = _generate_mcp_config(
fuzzforge_root=fuzzforge_root,
modules_path=resolved_modules,
engine_type=engine,
engine_socket=socket,
)
@@ -453,7 +424,6 @@ def install(
console.print(f"[bold]Configuration file:[/bold] {config_path}")
console.print()
console.print("[bold]Settings:[/bold]")
console.print(f" Modules Path: {resolved_modules}")
console.print(f" Engine: {engine}")
console.print(f" Socket: {socket}")
console.print(f" Hub Config: {fuzzforge_root / 'hub-config.json'}")

166
fuzzforge-cli/src/fuzzforge_cli/commands/modules.py
View File

@@ -1,166 +0,0 @@
"""Module management commands for FuzzForge CLI."""
import asyncio
from pathlib import Path
from typing import Annotated, Any
from rich.console import Console
from rich.table import Table
from typer import Argument, Context, Option, Typer
from fuzzforge_cli.context import get_project_path, get_runner
application: Typer = Typer(
name="modules",
help="Module management commands.",
)
@application.command(
help="List available modules.",
name="list",
)
def list_modules(
context: Context,
) -> None:
"""List all available modules.
:param context: Typer context.
"""
runner = get_runner(context)
modules = runner.list_modules()
console = Console()
if not modules:
console.print("[yellow]No modules found.[/yellow]")
console.print(f" Modules directory: {runner.settings.modules_path}")
return
table = Table(title="Available Modules")
table.add_column("Identifier", style="cyan")
table.add_column("Available")
table.add_column("Description")
for module in modules:
table.add_row(
module.identifier,
"✓" if module.available else "✗",
module.description or "-",
)
console.print(table)
@application.command(
help="Execute a module.",
name="run",
)
def run_module(
context: Context,
module_identifier: Annotated[
str,
Argument(
help="Identifier of the module to execute.",
),
],
assets_path: Annotated[
Path | None,
Option(
"--assets",
"-a",
help="Path to input assets.",
),
] = None,
config: Annotated[
str | None,
Option(
"--config",
"-c",
help="Module configuration as JSON string.",
),
] = None,
) -> None:
"""Execute a module.
:param context: Typer context.
:param module_identifier: Module to execute.
:param assets_path: Optional path to input assets.
:param config: Optional JSON configuration.
"""
import json
runner = get_runner(context)
project_path = get_project_path(context)
configuration: dict[str, Any] | None = None
if config:
try:
configuration = json.loads(config)
except json.JSONDecodeError as e:
console = Console()
console.print(f"[red]✗[/red] Invalid JSON configuration: {e}")
return
console = Console()
console.print(f"[blue]→[/blue] Executing module: {module_identifier}")
async def execute() -> None:
result = await runner.execute_module(
module_identifier=module_identifier,
project_path=project_path,
configuration=configuration,
assets_path=assets_path,
)
if result.success:
console.print(f"[green]✓[/green] Module execution completed")
console.print(f" Execution ID: {result.execution_id}")
console.print(f" Results: {result.results_path}")
else:
console.print(f"[red]✗[/red] Module execution failed")
console.print(f" Error: {result.error}")
asyncio.run(execute())
@application.command(
help="Show module information.",
name="info",
)
def module_info(
context: Context,
module_identifier: Annotated[
str,
Argument(
help="Identifier of the module.",
),
],
) -> None:
"""Show information about a specific module.
:param context: Typer context.
:param module_identifier: Module to get info for.
"""
runner = get_runner(context)
module = runner.get_module_info(module_identifier)
console = Console()
if module is None:
console.print(f"[red]✗[/red] Module not found: {module_identifier}")
return
table = Table(title=f"Module: {module.identifier}")
table.add_column("Property", style="cyan")
table.add_column("Value")
table.add_row("Identifier", module.identifier)
table.add_row("Available", "Yes" if module.available else "No")
table.add_row("Description", module.description or "-")
table.add_row("Version", module.version or "-")
console.print(table)

28
fuzzforge-cli/src/fuzzforge_cli/commands/projects.py
View File

@@ -7,7 +7,7 @@ from rich.console import Console
from rich.table import Table
from typer import Argument, Context, Option, Typer
from fuzzforge_cli.context import get_project_path, get_runner
from fuzzforge_cli.context import get_project_path, get_storage
application: Typer = Typer(
name="project",
@@ -36,10 +36,10 @@ def init_project(
:param path: Path to initialize (defaults to current directory).
"""
runner = get_runner(context)
storage = get_storage(context)
project_path = path or get_project_path(context)
storage_path = runner.init_project(project_path)
storage_path = storage.init_project(project_path)
console = Console()
console.print(f"[green]✓[/green] Project initialized at {project_path}")
@@ -65,10 +65,10 @@ def set_assets(
:param assets_path: Path to assets.
"""
runner = get_runner(context)
storage = get_storage(context)
project_path = get_project_path(context)
stored_path = runner.set_project_assets(project_path, assets_path)
stored_path = storage.set_project_assets(project_path, assets_path)
console = Console()
console.print(f"[green]✓[/green] Assets stored from {assets_path}")
@@ -87,11 +87,11 @@ def show_info(
:param context: Typer context.
"""
runner = get_runner(context)
storage = get_storage(context)
project_path = get_project_path(context)
executions = runner.list_executions(project_path)
assets_path = runner.storage.get_project_assets_path(project_path)
executions = storage.list_executions(project_path)
assets_path = storage.get_project_assets_path(project_path)
console = Console()
table = Table(title=f"Project: {project_path.name}")
@@ -118,10 +118,10 @@ def list_executions(
:param context: Typer context.
"""
runner = get_runner(context)
storage = get_storage(context)
project_path = get_project_path(context)
executions = runner.list_executions(project_path)
executions = storage.list_executions(project_path)
console = Console()
@@ -134,7 +134,7 @@ def list_executions(
table.add_column("Has Results")
for exec_id in executions:
has_results = runner.get_execution_results(project_path, exec_id) is not None
has_results = storage.get_execution_results(project_path, exec_id) is not None
table.add_row(exec_id, "✓" if has_results else "-")
console.print(table)
@@ -168,10 +168,10 @@ def get_results(
:param extract_to: Optional directory to extract to.
"""
runner = get_runner(context)
storage = get_storage(context)
project_path = get_project_path(context)
results_path = runner.get_execution_results(project_path, execution_id)
results_path = storage.get_execution_results(project_path, execution_id)
console = Console()
@@ -182,5 +182,5 @@ def get_results(
console.print(f"[green]✓[/green] Results: {results_path}")
if extract_to:
extracted = runner.extract_results(results_path, extract_to)
extracted = storage.extract_results(results_path, extract_to)
console.print(f" Extracted to: {extracted}")

28
fuzzforge-cli/src/fuzzforge_cli/context.py
View File

@@ -5,35 +5,35 @@ from __future__ import annotations
from pathlib import Path
from typing import TYPE_CHECKING, cast
from fuzzforge_runner import Runner, Settings
from fuzzforge_mcp.storage import LocalStorage
if TYPE_CHECKING:
from typer import Context as TyperContext
class Context:
"""CLI context holding the runner instance and settings."""
"""CLI context holding the storage instance and settings."""
_runner: Runner
_storage: LocalStorage
_project_path: Path
def __init__(self, runner: Runner, project_path: Path) -> None:
def __init__(self, storage: LocalStorage, project_path: Path) -> None:
"""Initialize an instance of the class.
:param runner: FuzzForge runner instance.
:param storage: FuzzForge local storage instance.
:param project_path: Path to the current project.
"""
self._runner = runner
self._storage = storage
self._project_path = project_path
def get_runner(self) -> Runner:
"""Get the runner instance.
def get_storage(self) -> LocalStorage:
"""Get the storage instance.
:return: Runner instance.
:return: LocalStorage instance.
"""
return self._runner
return self._storage
def get_project_path(self) -> Path:
"""Get the current project path.
@@ -44,14 +44,14 @@ class Context:
return self._project_path
def get_runner(context: TyperContext) -> Runner:
"""Get runner from Typer context.
def get_storage(context: TyperContext) -> LocalStorage:
"""Get storage from Typer context.
:param context: Typer context.
:return: Runner instance.
:return: LocalStorage instance.
"""
return cast("Context", context.obj).get_runner()
return cast("Context", context.obj).get_storage()
def get_project_path(context: TyperContext) -> Path: