mirror of
https://github.com/GH05TCREW/pentestagent.git
synced 2026-03-08 06:44:11 +00:00
308 lines
9.9 KiB
Python
308 lines
9.9 KiB
Python
"""Main entry point for GhostCrew."""
|
|
|
|
import argparse
|
|
import asyncio
|
|
|
|
from ..config.constants import DEFAULT_MODEL
|
|
from .cli import run_cli
|
|
from .tui import run_tui
|
|
|
|
|
|
def parse_arguments():
|
|
"""Parse command line arguments."""
|
|
parser = argparse.ArgumentParser(
|
|
description="GhostCrew - AI Penetration Testing",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="""
|
|
Examples:
|
|
ghostcrew tui Launch TUI
|
|
ghostcrew tui -t 192.168.1.1 Launch TUI with target
|
|
ghostcrew run -t localhost --task "scan" Headless run
|
|
ghostcrew tools list List available tools
|
|
ghostcrew mcp list List MCP servers
|
|
""",
|
|
)
|
|
|
|
parser.add_argument("--version", action="version", version="GhostCrew 0.2.0")
|
|
|
|
# Subcommands
|
|
subparsers = parser.add_subparsers(dest="command", help="Commands")
|
|
|
|
# Common arguments for runtime modes
|
|
runtime_parent = argparse.ArgumentParser(add_help=False)
|
|
runtime_parent.add_argument("--target", "-t", help="Target (IP, hostname, or URL)")
|
|
runtime_parent.add_argument(
|
|
"--model",
|
|
"-m",
|
|
default=DEFAULT_MODEL,
|
|
help="LLM model (set GHOSTCREW_MODEL in .env)",
|
|
)
|
|
runtime_parent.add_argument(
|
|
"--docker",
|
|
"-d",
|
|
action="store_true",
|
|
help="Run tools inside Docker container (requires Docker)",
|
|
)
|
|
|
|
# TUI subcommand
|
|
subparsers.add_parser(
|
|
"tui", parents=[runtime_parent], help="Launch TUI (Interactive Mode)"
|
|
)
|
|
|
|
# Run subcommand (Headless)
|
|
run_parser = subparsers.add_parser(
|
|
"run", parents=[runtime_parent], help="Run in headless mode"
|
|
)
|
|
run_parser.add_argument("task", nargs="+", help="Task to run")
|
|
run_parser.add_argument(
|
|
"--report",
|
|
"-r",
|
|
nargs="?",
|
|
const="auto",
|
|
help=(
|
|
"Generate report. "
|
|
"If used without value, auto-generates path under loot/reports/. "
|
|
"If omitted, no report is generated."
|
|
),
|
|
)
|
|
run_parser.add_argument(
|
|
"--max-loops",
|
|
type=int,
|
|
default=50,
|
|
help="Max agent loops before stopping (default: 50)",
|
|
)
|
|
|
|
# Tools subcommand
|
|
tools_parser = subparsers.add_parser("tools", help="Manage tools")
|
|
tools_subparsers = tools_parser.add_subparsers(
|
|
dest="tools_command", help="Tool commands"
|
|
)
|
|
|
|
# tools list
|
|
tools_subparsers.add_parser("list", help="List all available tools")
|
|
|
|
# tools info
|
|
tools_info = tools_subparsers.add_parser("info", help="Show tool details")
|
|
tools_info.add_argument("name", help="Tool name")
|
|
|
|
# MCP subcommand
|
|
mcp_parser = subparsers.add_parser("mcp", help="Manage MCP servers")
|
|
mcp_subparsers = mcp_parser.add_subparsers(dest="mcp_command", help="MCP commands")
|
|
|
|
# mcp list
|
|
mcp_subparsers.add_parser("list", help="List configured MCP servers")
|
|
|
|
# mcp add
|
|
mcp_add = mcp_subparsers.add_parser("add", help="Add an MCP server")
|
|
mcp_add.add_argument("name", help="Server name")
|
|
mcp_add.add_argument("command", help="Command to run (e.g., npx)")
|
|
mcp_add.add_argument("args", nargs="*", help="Command arguments")
|
|
mcp_add.add_argument("--description", default="", help="Server description")
|
|
|
|
# mcp remove
|
|
mcp_remove = mcp_subparsers.add_parser("remove", help="Remove an MCP server")
|
|
mcp_remove.add_argument("name", help="Server name to remove")
|
|
|
|
# mcp test
|
|
mcp_test = mcp_subparsers.add_parser("test", help="Test MCP server connection")
|
|
mcp_test.add_argument("name", help="Server name to test")
|
|
|
|
return parser, parser.parse_args()
|
|
|
|
|
|
def handle_tools_command(args: argparse.Namespace):
|
|
"""Handle tools subcommand."""
|
|
from rich.console import Console
|
|
from rich.table import Table
|
|
|
|
from ..tools import get_all_tools, get_tool
|
|
|
|
console = Console()
|
|
|
|
if args.tools_command == "list":
|
|
tools = get_all_tools()
|
|
|
|
if not tools:
|
|
console.print("[yellow]No tools found[/]")
|
|
return
|
|
|
|
table = Table(title="Available Tools")
|
|
table.add_column("Name", style="cyan")
|
|
table.add_column("Category", style="green")
|
|
table.add_column("Description")
|
|
|
|
for tool in sorted(tools, key=lambda t: t.name):
|
|
desc = (
|
|
tool.description[:50] + "..."
|
|
if len(tool.description) > 50
|
|
else tool.description
|
|
)
|
|
table.add_row(tool.name, tool.category, desc)
|
|
|
|
console.print(table)
|
|
console.print(f"\nTotal: {len(tools)} tools")
|
|
|
|
elif args.tools_command == "info":
|
|
tool = get_tool(args.name)
|
|
if not tool:
|
|
console.print(f"[red]Tool not found: {args.name}[/]")
|
|
return
|
|
|
|
console.print(f"\n[bold cyan]{tool.name}[/]")
|
|
console.print(f"[dim]Category:[/] {tool.category}")
|
|
console.print(f"\n{tool.description}")
|
|
|
|
if tool.schema.properties:
|
|
console.print("\n[bold]Parameters:[/]")
|
|
for name, props in tool.schema.properties.items():
|
|
required = (
|
|
"required" if name in (tool.schema.required or []) else "optional"
|
|
)
|
|
ptype = props.get("type", "any")
|
|
desc = props.get("description", "")
|
|
console.print(f" [cyan]{name}[/] ({ptype}, {required}): {desc}")
|
|
|
|
else:
|
|
console.print("[yellow]Use 'ghostcrew tools --help' for commands[/]")
|
|
|
|
|
|
def handle_mcp_command(args: argparse.Namespace):
|
|
"""Handle MCP subcommand."""
|
|
from rich.console import Console
|
|
from rich.table import Table
|
|
|
|
from ..mcp.manager import MCPManager
|
|
|
|
console = Console()
|
|
manager = MCPManager()
|
|
|
|
if args.mcp_command == "list":
|
|
servers = manager.list_configured_servers()
|
|
|
|
if not servers:
|
|
console.print("[yellow]No MCP servers configured[/]")
|
|
console.print(
|
|
"\nAdd a server with: ghostcrew mcp add <name> <command> <args...>"
|
|
)
|
|
return
|
|
|
|
table = Table(title="Configured MCP Servers")
|
|
table.add_column("Name", style="cyan")
|
|
table.add_column("Command", style="green")
|
|
table.add_column("Args")
|
|
table.add_column("Connected", style="yellow")
|
|
|
|
for server in servers:
|
|
args_str = " ".join(server["args"][:3])
|
|
if len(server["args"]) > 3:
|
|
args_str += "..."
|
|
connected = "+" if server.get("connected") else "-"
|
|
table.add_row(server["name"], server["command"], args_str, connected)
|
|
|
|
console.print(table)
|
|
console.print(f"\nConfig file: {manager.config_path}")
|
|
|
|
elif args.mcp_command == "add":
|
|
manager.add_server(
|
|
name=args.name,
|
|
command=args.command,
|
|
args=args.args or [],
|
|
description=args.description,
|
|
)
|
|
console.print(f"[green]Added MCP server: {args.name}[/]")
|
|
console.print(f" Command: {args.command} {' '.join(args.args or [])}")
|
|
|
|
elif args.mcp_command == "remove":
|
|
if manager.remove_server(args.name):
|
|
console.print(f"[yellow]Removed MCP server: {args.name}[/]")
|
|
else:
|
|
console.print(f"[red]Server not found: {args.name}[/]")
|
|
|
|
elif args.mcp_command == "test":
|
|
console.print(f"[bold]Testing MCP server: {args.name}[/]\n")
|
|
|
|
async def test_server():
|
|
server = await manager.connect_server(args.name)
|
|
if server and server.connected:
|
|
console.print("[green]+ Connected successfully![/]")
|
|
console.print(f"\n[bold]Available tools ({len(server.tools)}):[/]")
|
|
for tool in server.tools:
|
|
desc = tool.get("description", "No description")[:60]
|
|
console.print(f" [cyan]{tool['name']}[/]: {desc}")
|
|
await manager.disconnect_all()
|
|
else:
|
|
console.print("[red]x Failed to connect[/]")
|
|
|
|
asyncio.run(test_server())
|
|
|
|
else:
|
|
console.print("[yellow]Use 'ghostcrew mcp --help' for available commands[/]")
|
|
|
|
|
|
def main():
|
|
"""Main entry point."""
|
|
parser, args = parse_arguments()
|
|
|
|
# Handle subcommands
|
|
if args.command == "tools":
|
|
handle_tools_command(args)
|
|
return
|
|
|
|
if args.command == "mcp":
|
|
handle_mcp_command(args)
|
|
return
|
|
|
|
if args.command == "run":
|
|
# Check model configuration
|
|
if not args.model:
|
|
print("Error: No model configured.")
|
|
print("Set GHOSTCREW_MODEL in .env file or use --model flag.")
|
|
print(
|
|
"Example: GHOSTCREW_MODEL=gpt-5 or GHOSTCREW_MODEL=claude-sonnet-4-20250514"
|
|
)
|
|
return
|
|
|
|
if not args.target:
|
|
print("Error: --target is required for run mode")
|
|
return
|
|
|
|
# Join task arguments
|
|
task_description = " ".join(args.task)
|
|
|
|
try:
|
|
asyncio.run(
|
|
run_cli(
|
|
target=args.target,
|
|
model=args.model,
|
|
task=task_description,
|
|
report=args.report,
|
|
max_loops=args.max_loops,
|
|
use_docker=args.docker,
|
|
)
|
|
)
|
|
except KeyboardInterrupt:
|
|
print("\n[!] Interrupted by user.")
|
|
return
|
|
|
|
if args.command == "tui":
|
|
# Check model configuration
|
|
if not args.model:
|
|
print("Error: No model configured.")
|
|
print("Set GHOSTCREW_MODEL in .env file or use --model flag.")
|
|
print(
|
|
"Example: GHOSTCREW_MODEL=gpt-5 or GHOSTCREW_MODEL=claude-sonnet-4-20250514"
|
|
)
|
|
return
|
|
|
|
run_tui(target=args.target, model=args.model, use_docker=args.docker)
|
|
return
|
|
|
|
# If no command provided, default to TUI
|
|
if args.command is None:
|
|
run_tui(target=None, model=DEFAULT_MODEL, use_docker=False)
|
|
return
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|