Files
pentestagent/ghostcrew/interface/main.py

287 lines
9.0 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() -> argparse.Namespace:
"""Parse command line arguments."""
parser = argparse.ArgumentParser(
description="GhostCrew - AI Penetration Testing",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
ghostcrew Launch TUI
ghostcrew -t 192.168.1.1 Launch TUI with target
ghostcrew -n -t example.com Non-interactive run
ghostcrew tools list List available tools
ghostcrew mcp list List MCP servers
""",
)
# Subcommands
subparsers = parser.add_subparsers(dest="command", help="Commands")
# 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", "-d", 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")
# Target option
parser.add_argument("--target", "-t", help="Target (IP, hostname, or URL)")
# Non-interactive mode
parser.add_argument(
"-n",
"--headless",
action="store_true",
help="Run without TUI (requires --target)",
)
# Task for non-interactive mode
parser.add_argument("--task", help="Task to run in non-interactive mode")
# Report output (saves to loot/reports/ by default)
parser.add_argument(
"--report",
"-r",
nargs="?",
const="auto",
help="Generate report (default: loot/reports/<target>_<timestamp>.md)",
)
# Max tool calls limit
parser.add_argument(
"--max", type=int, default=50, help="Max calls before stopping (default: 50)"
)
# Model options
parser.add_argument(
"--model",
"-m",
default=DEFAULT_MODEL,
help="LLM model (set GHOSTCREW_MODEL in .env)",
)
# Docker mode
parser.add_argument(
"--docker",
"-d",
action="store_true",
help="Run tools inside Docker container (requires Docker)",
)
# Version
parser.add_argument("--version", action="version", version="GhostCrew 0.2.0")
return 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."""
args = parse_arguments()
# Handle subcommands
if args.command == "tools":
handle_tools_command(args)
return
if args.command == "mcp":
handle_mcp_command(args)
return
# 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
# Determine interface mode
if args.headless:
if not args.target:
print("Error: --target is required for headless mode")
return
asyncio.run(
run_cli(
target=args.target,
model=args.model,
task=args.task,
report=args.report,
max_tools=args.max,
use_docker=args.docker,
)
)
else:
# TUI doesn't need asyncio.run - it runs its own event loop
run_tui(target=args.target, model=args.model, use_docker=args.docker)
if __name__ == "__main__":
main()