Spaces:
Running
Running
| """ | |
| AI HACKING TEAM — FastAPI + WebSocket Backend | |
| Real C2 server: routes commands to real AI agents with real tools | |
| """ | |
| import os | |
| import json | |
| import asyncio | |
| import traceback | |
| from datetime import datetime | |
| from typing import Optional | |
| from pathlib import Path | |
| from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException | |
| from fastapi.responses import FileResponse | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from dotenv import load_dotenv | |
| # Load environment | |
| load_dotenv() | |
| # Import agents (lazy to avoid startup errors if key missing) | |
| team = None | |
| def get_team(): | |
| global team | |
| if team is None: | |
| from agents.agents import AgentTeam | |
| team = AgentTeam() | |
| return team | |
| app = FastAPI(title="AI Hacking Team — Hermes C2", version="4.2.1") | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=["*"], | |
| allow_methods=["*"], | |
| allow_headers=["*"], | |
| ) | |
| # Base directory (all static assets live here) | |
| BASE_DIR = Path(__file__).parent | |
| # ───────────────────────────────────────────────── | |
| # MISSION STATE (in-memory, real session state) | |
| # ───────────────────────────────────────────────── | |
| missions = [ | |
| {"id": "OP-2024-0047", "name": "OPERATION GHOSTWIRE", "priority": "critical", "progress": 82, "agents": ["ARES","ARTEMIS","HERACLES"], "status": "ACTIVE"}, | |
| {"id": "OP-2024-0051", "name": "OPERATION MIRAGE", "priority": "high", "progress": 47, "agents": ["APOLLO","ATHENA"], "status": "ACTIVE"}, | |
| {"id": "OP-2024-0039", "name": "OPERATION SILENT HERALD", "priority": "critical", "progress": 91, "agents": ["HEPHAESTUS","HERACLES"], "status": "ACTIVE"}, | |
| {"id": "OP-2024-0058", "name": "OPERATION PHANTOM CIRCUIT", "priority": "medium", "progress": 12, "agents": ["ARTEMIS","ARES"], "status": "QUEUED"}, | |
| {"id": "OP-2024-0055", "name": "OPERATION DARK ORACLE", "priority": "high", "progress": 63, "agents": ["ATHENA","ARES","HEPHAESTUS"], "status": "ACTIVE"}, | |
| {"id": "OP-2024-0060", "name": "OPERATION WHISPER NET", "priority": "low", "progress": 28, "agents": ["ATHENA","ARTEMIS"], "status": "ACTIVE"}, | |
| ] | |
| agent_status = { | |
| "hermes": {"status": "ACTIVE", "tasks_completed": 247, "current_task": "Orchestrating team"}, | |
| "artemis": {"status": "ACTIVE", "tasks_completed": 89, "current_task": "Network recon"}, | |
| "athena": {"status": "ACTIVE", "tasks_completed": 134, "current_task": "OSINT gathering"}, | |
| "ares": {"status": "STANDBY", "tasks_completed": 67, "current_task": "Awaiting recon data"}, | |
| "hephaestus": {"status": "IDLE", "tasks_completed": 45, "current_task": "Payload queue empty"}, | |
| "heracles": {"status": "STANDBY", "tasks_completed": 56, "current_task": "C2 beacon monitoring"}, | |
| "apollo": {"status": "IDLE", "tasks_completed": 38, "current_task": "No active campaigns"}, | |
| } | |
| # ───────────────────────────────────────────────── | |
| # REST ENDPOINTS | |
| # ───────────────────────────────────────────────── | |
| async def serve_frontend(): | |
| return FileResponse(str(BASE_DIR / "index.html"), media_type="text/html") | |
| # ── Explicit static file routes (must be before the catch-all mount) ────────── | |
| async def serve_css(): | |
| return FileResponse(str(BASE_DIR / "style.css"), media_type="text/css") | |
| async def serve_js(): | |
| return FileResponse(str(BASE_DIR / "app.js"), media_type="application/javascript") | |
| async def serve_hermes_js(): | |
| return FileResponse(str(BASE_DIR / "hermes.js"), media_type="application/javascript") | |
| async def serve_hermes_img(): | |
| return FileResponse(str(BASE_DIR / "hermes_agent.png"), media_type="image/png") | |
| async def serve_bg_img(): | |
| return FileResponse(str(BASE_DIR / "ops_center_bg.png"), media_type="image/png") | |
| async def serve_team_img(): | |
| return FileResponse(str(BASE_DIR / "team_agents.png"), media_type="image/png") | |
| async def get_status(): | |
| return { | |
| "system": "OPERATIONAL", | |
| "version": "4.2.1", | |
| "timestamp": datetime.utcnow().isoformat(), | |
| "api_key_configured": bool(os.getenv("LLM_API_KEY")), | |
| "agents": agent_status, | |
| "active_missions": len([m for m in missions if m["status"] == "ACTIVE"]), | |
| "total_missions": len(missions), | |
| } | |
| async def get_missions(): | |
| return {"missions": missions} | |
| async def update_mission(mission_id: str, progress: int): | |
| for m in missions: | |
| if m["id"] == mission_id: | |
| m["progress"] = min(max(progress, 0), 100) | |
| if m["progress"] >= 100: | |
| m["status"] = "COMPLETE" | |
| return {"success": True, "mission": m} | |
| raise HTTPException(status_code=404, detail="Mission not found") | |
| async def get_agents(): | |
| return {"agents": agent_status} | |
| # ───────────────────────────────────────────────── | |
| # WEBSOCKET — Real-time C2 Channel | |
| # ───────────────────────────────────────────────── | |
| async def terminal_ws(ws: WebSocket): | |
| await ws.accept() | |
| await ws.send_json({"type": "connected", "message": "HERMES C2 WebSocket established. All comms encrypted."}) | |
| try: | |
| while True: | |
| data = await ws.receive_json() | |
| cmd = data.get("command", "").strip() | |
| await process_terminal_command(ws, cmd) | |
| except WebSocketDisconnect: | |
| pass | |
| except Exception as e: | |
| try: | |
| await ws.send_json({"type": "error", "message": str(e)}) | |
| except Exception: | |
| pass | |
| async def send_lines(ws: WebSocket, lines: list, delay: float = 0.05): | |
| """Send multiple terminal lines with a small delay for realistic feel""" | |
| for line in lines: | |
| await ws.send_json({"type": "terminal_line", **line}) | |
| await asyncio.sleep(delay) | |
| async def process_terminal_command(ws: WebSocket, cmd: str): | |
| """Process real terminal commands — calls real agents/tools""" | |
| parts = cmd.lower().strip().split() | |
| if not parts: | |
| return | |
| base = parts[0] | |
| args = parts[1:] if len(parts) > 1 else [] | |
| ts = datetime.utcnow().strftime("%H:%M:%S") | |
| # ── HELP ────────────────────────────────────────────── | |
| if base in ["help", "?"]: | |
| await send_lines(ws, [ | |
| {"cls": "header", "text": "┌────────────────────────────────────────────────────────────┐"}, | |
| {"cls": "header", "text": "│ HERMES C2 — REAL COMMAND REFERENCE │"}, | |
| {"cls": "header", "text": "└────────────────────────────────────────────────────────────┘"}, | |
| {"cls": "info", "text": ""}, | |
| {"cls": "info", "text": " RECON COMMANDS (Real network tools):"}, | |
| {"cls": "output", "text": " recon <target> — Full recon: DNS, ports, HTTP headers, SSL"}, | |
| {"cls": "output", "text": " dns <target> — DNS resolution (A, AAAA, MX, NS, TXT)"}, | |
| {"cls": "output", "text": " ports <target> — Port scan (common ports via socket probe)"}, | |
| {"cls": "output", "text": " ssl <target> — SSL certificate inspection"}, | |
| {"cls": "output", "text": " ping <target> — ICMP ping check"}, | |
| {"cls": "output", "text": " headers <target> — HTTP/S header analysis"}, | |
| {"cls": "info", "text": ""}, | |
| {"cls": "info", "text": " OSINT COMMANDS (Real public APIs):"}, | |
| {"cls": "output", "text": " osint <target> — Full OSINT: subdomains, WHOIS, geolocation"}, | |
| {"cls": "output", "text": " subdomains <domain> — Certificate transparency subdomain enum"}, | |
| {"cls": "output", "text": " whois <domain> — WHOIS lookup"}, | |
| {"cls": "output", "text": " geoip <ip> — IP geolocation (country, ISP, org)"}, | |
| {"cls": "output", "text": " reverseip <ip> — Reverse IP: find hosted domains"}, | |
| {"cls": "info", "text": ""}, | |
| {"cls": "info", "text": " CVE / VULNERABILITY (Real NIST NVD API):"}, | |
| {"cls": "output", "text": " cve <keyword> — Search real CVEs by keyword"}, | |
| {"cls": "output", "text": " cve-id <CVE-XXXX-XXXX> — Get specific CVE details"}, | |
| {"cls": "output", "text": " vulns <target> <svc> — CVE search for specific software"}, | |
| {"cls": "info", "text": ""}, | |
| {"cls": "info", "text": " AI AGENT COMMANDS (Real LLM AI):"}, | |
| {"cls": "output", "text": " ask <agent> <question> — Ask any agent (hermes/artemis/ares/athena/hephaestus/heracles/apollo)"}, | |
| {"cls": "output", "text": " forge <task> — Hephaestus: generate a real script/tool"}, | |
| {"cls": "output", "text": " intel <target> — Athena: full AI OSINT analysis"}, | |
| {"cls": "info", "text": ""}, | |
| {"cls": "info", "text": " SYSTEM:"}, | |
| {"cls": "output", "text": " status — Team & system status"}, | |
| {"cls": "output", "text": " missions — List active missions"}, | |
| {"cls": "output", "text": " agents — List all agents and status"}, | |
| {"cls": "output", "text": " clear — Clear terminal"}, | |
| {"cls": "info", "text": ""}, | |
| ]) | |
| # ── STATUS ──────────────────────────────────────────── | |
| elif base == "status": | |
| api_ok = bool(os.getenv("LLM_API_KEY")) | |
| await send_lines(ws, [ | |
| {"cls": "info", "text": f"[{ts}] SYSTEM STATUS — HERMES C2 v4.2.1"}, | |
| {"cls": "success", "text": f" LLM API: {'✓ CONFIGURED' if api_ok else '✗ NOT SET (add LLM_API_KEY to .env)'}"}, | |
| {"cls": "success", "text": " Server: ✓ OPERATIONAL"}, | |
| {"cls": "success", "text": " WebSocket: ✓ CONNECTED"}, | |
| {"cls": "success", "text": " Tools: ✓ DNS, HTTP, Ports, CVE, OSINT"}, | |
| {"cls": "info", "text": ""}, | |
| {"cls": "info", "text": " AGENT ROSTER:"}, | |
| ]) | |
| for name, info in agent_status.items(): | |
| cls = "success" if info["status"] == "ACTIVE" else "warning" if info["status"] == "STANDBY" else "dim" | |
| await ws.send_json({"type": "terminal_line", "cls": cls, | |
| "text": f" {name.upper():12} [{info['status']:8}] Tasks: {info['tasks_completed']:3} | {info['current_task']}"}) | |
| # ── AGENTS ──────────────────────────────────────────── | |
| elif base == "agents": | |
| await send_lines(ws, [ | |
| {"cls": "info", "text": f"[{ts}] AGENT ROSTER — 7 ACTIVE AGENTS"}, | |
| ]) | |
| for name, info in agent_status.items(): | |
| cls = "success" if info["status"] == "ACTIVE" else "warning" if info["status"] == "STANDBY" else "dim" | |
| await ws.send_json({"type": "terminal_line", "cls": cls, | |
| "text": f" {name.upper():12} {info['status']:8} Tasks completed: {info['tasks_completed']}"}) | |
| # ── MISSIONS ────────────────────────────────────────── | |
| elif base == "missions": | |
| await send_lines(ws, [ | |
| {"cls": "info", "text": f"[{ts}] ACTIVE MISSIONS"}, | |
| ]) | |
| for m in missions: | |
| bar = "█" * (m["progress"] // 10) + "░" * (10 - m["progress"] // 10) | |
| cls = "error" if m["priority"] == "critical" else "warning" if m["priority"] == "high" else "output" | |
| await ws.send_json({"type": "terminal_line", "cls": cls, | |
| "text": f" {m['id']} [{bar}] {m['progress']:3}% {m['name']} ({m['status']})"}) | |
| # ── CLEAR ───────────────────────────────────────────── | |
| elif base == "clear": | |
| await ws.send_json({"type": "clear"}) | |
| # ── PING ───────────────────────────────────────────── | |
| elif base == "ping" and args: | |
| target = args[0] | |
| await ws.send_json({"type": "terminal_line", "cls": "info", "text": f"[{ts}] [ARTEMIS] Pinging {target}..."}) | |
| await asyncio.sleep(0.3) | |
| from tools.recon import ping_host | |
| result = await asyncio.get_event_loop().run_in_executor(None, ping_host, target) | |
| status = "success" if result.get("alive") else "error" | |
| alive_txt = "ALIVE" if result.get("alive") else "NO RESPONSE" | |
| await ws.send_json({"type": "terminal_line", "cls": status, | |
| "text": f" Host {target}: {alive_txt} (RTT: {result.get('rtt_ms', 'N/A')}ms)"}) | |
| if result.get("output"): | |
| for line in result["output"].splitlines()[:5]: | |
| if line.strip(): | |
| await ws.send_json({"type": "terminal_line", "cls": "output", "text": f" {line}"}) | |
| # ── DNS ─────────────────────────────────────────────── | |
| elif base == "dns" and args: | |
| target = args[0] | |
| await ws.send_json({"type": "terminal_line", "cls": "info", "text": f"[{ts}] [ARTEMIS] DNS resolution for {target}..."}) | |
| from tools.recon import resolve_dns | |
| result = await asyncio.get_event_loop().run_in_executor(None, resolve_dns, target) | |
| if result.get("error"): | |
| await ws.send_json({"type": "terminal_line", "cls": "error", "text": f" Error: {result['error']}"}) | |
| else: | |
| for rtype, records in result.get("records", {}).items(): | |
| for r in records: | |
| await ws.send_json({"type": "terminal_line", "cls": "success", | |
| "text": f" {target:35} {rtype:6} {r}"}) | |
| if not result.get("records"): | |
| await ws.send_json({"type": "terminal_line", "cls": "warning", "text": " No DNS records found"}) | |
| # ── PORTS ───────────────────────────────────────────── | |
| elif base == "ports" and args: | |
| target = args[0] | |
| await ws.send_json({"type": "terminal_line", "cls": "info", | |
| "text": f"[{ts}] [ARTEMIS] Port scanning {target} (common ports)..."}) | |
| await ws.send_json({"type": "terminal_line", "cls": "dim", | |
| "text": " Probing: 21,22,25,53,80,443,445,3306,3389,8080,8443 — Please wait..."}) | |
| from tools.recon import probe_ports | |
| result = await asyncio.get_event_loop().run_in_executor( | |
| None, probe_ports, target, [21,22,25,53,80,443,445,3306,3389,8080,8443] | |
| ) | |
| if result.get("ip"): | |
| await ws.send_json({"type": "terminal_line", "cls": "info", | |
| "text": f" Target IP: {result['ip']}"}) | |
| open_ports = result.get("open_ports", []) | |
| if open_ports: | |
| for p in open_ports: | |
| await ws.send_json({"type": "terminal_line", "cls": "success", | |
| "text": f" PORT {p['port']:5}/tcp OPEN {p['service']:12} {p.get('banner','')[:60]}"}) | |
| else: | |
| await ws.send_json({"type": "terminal_line", "cls": "warning", | |
| "text": " No open ports found (firewall or host down)"}) | |
| await ws.send_json({"type": "terminal_line", "cls": "dim", | |
| "text": f" Closed/filtered: {len(result.get('closed_ports', []))} ports"}) | |
| # ── SSL ─────────────────────────────────────────────── | |
| elif base == "ssl" and args: | |
| target = args[0] | |
| await ws.send_json({"type": "terminal_line", "cls": "info", | |
| "text": f"[{ts}] [ARTEMIS] SSL certificate inspection for {target}:443..."}) | |
| from tools.recon import get_ssl_cert_info | |
| result = await asyncio.get_event_loop().run_in_executor(None, get_ssl_cert_info, target, 443) | |
| if result.get("error"): | |
| await ws.send_json({"type": "terminal_line", "cls": "error", | |
| "text": f" Error: {result['error']}"}) | |
| else: | |
| cert = result.get("cert", {}) | |
| subj = cert.get("subject", {}) | |
| await ws.send_json({"type": "terminal_line", "cls": "success", | |
| "text": f" Subject CN: {subj.get('commonName', 'N/A')}"}) | |
| iss = cert.get("issuer", {}) | |
| await ws.send_json({"type": "terminal_line", "cls": "output", | |
| "text": f" Issuer: {iss.get('organizationName', 'N/A')}"}) | |
| await ws.send_json({"type": "terminal_line", "cls": "output", | |
| "text": f" Valid From: {cert.get('notBefore', 'N/A')}"}) | |
| await ws.send_json({"type": "terminal_line", "cls": "output", | |
| "text": f" Valid Until: {cert.get('notAfter', 'N/A')}"}) | |
| await ws.send_json({"type": "terminal_line", "cls": "info", | |
| "text": f" TLS: {result.get('tls_version', 'N/A')} | Cipher: {result.get('cipher', ['N/A'])[0]}"}) | |
| sans = cert.get("subjectAltNames", []) | |
| for san in sans[:8]: | |
| await ws.send_json({"type": "terminal_line", "cls": "dim", | |
| "text": f" SAN: {san[1] if isinstance(san, tuple) else san}"}) | |
| # ── HEADERS ─────────────────────────────────────────── | |
| elif base == "headers" and args: | |
| target = args[0] | |
| await ws.send_json({"type": "terminal_line", "cls": "info", | |
| "text": f"[{ts}] [ARTEMIS] Fetching HTTP headers from {target}..."}) | |
| from tools.recon import get_http_headers | |
| result = await asyncio.get_event_loop().run_in_executor(None, get_http_headers, target, 80) | |
| if result.get("error") and not result.get("status_code"): | |
| # Try HTTPS | |
| from tools.recon import get_http_headers as ghh | |
| result = await asyncio.get_event_loop().run_in_executor(None, ghh, target, 443, True) | |
| await ws.send_json({"type": "terminal_line", "cls": "success" if result.get("status_code") else "error", | |
| "text": f" HTTP Status: {result.get('status_code', 'N/A')}"}) | |
| await ws.send_json({"type": "terminal_line", "cls": "output", | |
| "text": f" Server: {result.get('server', 'N/A')}"}) | |
| for k, v in list(result.get("headers", {}).items())[:15]: | |
| await ws.send_json({"type": "terminal_line", "cls": "dim", | |
| "text": f" {k}: {v}"}) | |
| # ── GEOIP ───────────────────────────────────────────── | |
| elif base == "geoip" and args: | |
| target = args[0] | |
| await ws.send_json({"type": "terminal_line", "cls": "info", | |
| "text": f"[{ts}] [ATHENA] Geolocating {target}..."}) | |
| from tools.osint import geolocate_ip, resolve_to_ip | |
| ip = await asyncio.get_event_loop().run_in_executor(None, resolve_to_ip, target) | |
| result = await asyncio.get_event_loop().run_in_executor(None, geolocate_ip, ip) | |
| fields = ["query","country","regionName","city","zip","lat","lon","timezone","isp","org","as"] | |
| for f in fields: | |
| if f in result: | |
| await ws.send_json({"type": "terminal_line", "cls": "output", | |
| "text": f" {f:12}: {result[f]}"}) | |
| # ── REVERSEIP ───────────────────────────────────────── | |
| elif base == "reverseip" and args: | |
| target = args[0] | |
| await ws.send_json({"type": "terminal_line", "cls": "info", | |
| "text": f"[{ts}] [ATHENA] Reverse IP lookup for {target}..."}) | |
| from tools.osint import hackertarget_reverse_ip, resolve_to_ip | |
| ip = await asyncio.get_event_loop().run_in_executor(None, resolve_to_ip, target) | |
| result = await asyncio.get_event_loop().run_in_executor(None, hackertarget_reverse_ip, ip) | |
| domains = result.get("domains", []) | |
| await ws.send_json({"type": "terminal_line", "cls": "success" if domains else "warning", | |
| "text": f" Found {len(domains)} domains on {ip}:"}) | |
| for d in domains[:20]: | |
| await ws.send_json({"type": "terminal_line", "cls": "output", "text": f" ↳ {d}"}) | |
| # ── SUBDOMAINS ──────────────────────────────────────── | |
| elif base == "subdomains" and args: | |
| target = args[0] | |
| await ws.send_json({"type": "terminal_line", "cls": "info", | |
| "text": f"[{ts}] [ATHENA] Certificate transparency subdomain enum for {target}..."}) | |
| from tools.osint import crtsh_subdomain_enum | |
| result = await asyncio.get_event_loop().run_in_executor(None, crtsh_subdomain_enum, target) | |
| subs = result.get("subdomains", []) | |
| await ws.send_json({"type": "terminal_line", "cls": "success", | |
| "text": f" Found {len(subs)} subdomains via crt.sh:"}) | |
| for s in subs[:30]: | |
| await ws.send_json({"type": "terminal_line", "cls": "output", "text": f" ↳ {s}"}) | |
| if len(subs) > 30: | |
| await ws.send_json({"type": "terminal_line", "cls": "dim", | |
| "text": f" ... and {len(subs) - 30} more"}) | |
| # ── WHOIS ───────────────────────────────────────────── | |
| elif base == "whois" and args: | |
| target = args[0] | |
| await ws.send_json({"type": "terminal_line", "cls": "info", | |
| "text": f"[{ts}] [ATHENA] WHOIS lookup for {target}..."}) | |
| from tools.osint import hackertarget_whois | |
| result = await asyncio.get_event_loop().run_in_executor(None, hackertarget_whois, target) | |
| whois_text = result.get("whois", "No data") | |
| for line in whois_text.splitlines()[:40]: | |
| if line.strip(): | |
| await ws.send_json({"type": "terminal_line", "cls": "output", "text": f" {line}"}) | |
| # ── CVE SEARCH ──────────────────────────────────────── | |
| elif base == "cve" and args: | |
| keyword = " ".join(args) | |
| await ws.send_json({"type": "terminal_line", "cls": "info", | |
| "text": f"[{ts}] [ARES] Querying NIST NVD for CVEs: '{keyword}'..."}) | |
| from tools.cve import search_cves, format_cve_brief | |
| result = await asyncio.get_event_loop().run_in_executor(None, search_cves, keyword, None, None, 8) | |
| await ws.send_json({"type": "terminal_line", "cls": "success", | |
| "text": f" Total matching CVEs in NVD: {result.get('total_results', 0)}"}) | |
| for v in result.get("vulnerabilities", []): | |
| score = v.get('cvss_v3_score', 'N/A') | |
| sev = v.get('cvss_v3_severity', 'N/A') | |
| color = "error" if sev == "CRITICAL" else "warning" if sev == "HIGH" else "output" | |
| await ws.send_json({"type": "terminal_line", "cls": color, | |
| "text": f" [{v['id']}] CVSS:{score} ({sev}) — Published: {v['published']}"}) | |
| desc = v.get('description', '')[:120] | |
| await ws.send_json({"type": "terminal_line", "cls": "dim", "text": f" {desc}..."}) | |
| # ── CVE BY ID ───────────────────────────────────────── | |
| elif base == "cve-id" and args: | |
| cve_id = args[0].upper() | |
| await ws.send_json({"type": "terminal_line", "cls": "info", | |
| "text": f"[{ts}] [ARES] Fetching {cve_id} from NIST NVD..."}) | |
| from tools.cve import get_cve_detail | |
| result = await asyncio.get_event_loop().run_in_executor(None, get_cve_detail, cve_id) | |
| vulns = result.get("vulnerabilities", []) | |
| if not vulns: | |
| await ws.send_json({"type": "terminal_line", "cls": "error", "text": f" CVE not found: {cve_id}"}) | |
| else: | |
| v = vulns[0] | |
| await ws.send_json({"type": "terminal_line", "cls": "error" if v.get('cvss_v3_severity') == 'CRITICAL' else "warning", | |
| "text": f" {v['id']} CVSS v3: {v.get('cvss_v3_score','N/A')} ({v.get('cvss_v3_severity','N/A')})"}) | |
| await ws.send_json({"type": "terminal_line", "cls": "output", | |
| "text": f" Published: {v['published']} Modified: {v['modified']}"}) | |
| await ws.send_json({"type": "terminal_line", "cls": "output", | |
| "text": f" Vector: {v.get('cvss_v3_vector','N/A')}"}) | |
| await ws.send_json({"type": "terminal_line", "cls": "info", "text": " Description:"}) | |
| desc = v.get('description', 'N/A') | |
| for i in range(0, len(desc), 100): | |
| await ws.send_json({"type": "terminal_line", "cls": "dim", "text": f" {desc[i:i+100]}"}) | |
| if v.get('cpe'): | |
| await ws.send_json({"type": "terminal_line", "cls": "info", "text": " Affected (CPE):"}) | |
| for c in v['cpe'][:5]: | |
| await ws.send_json({"type": "terminal_line", "cls": "dim", "text": f" {c}"}) | |
| if v.get('references'): | |
| await ws.send_json({"type": "terminal_line", "cls": "info", "text": " References:"}) | |
| for r in v['references'][:3]: | |
| await ws.send_json({"type": "terminal_line", "cls": "dim", "text": f" {r}"}) | |
| # ── FULL RECON ──────────────────────────────────────── | |
| elif base == "recon" and args: | |
| target = args[0] | |
| await ws.send_json({"type": "terminal_line", "cls": "info", | |
| "text": f"[{ts}] [ARTEMIS] Starting full recon sweep on {target}..."}) | |
| await ws.send_json({"type": "terminal_line", "cls": "dim", | |
| "text": " Running: DNS → Ping → Ports → HTTP Headers → SSL → OSINT..."}) | |
| t = get_team() | |
| analysis, raw = await asyncio.get_event_loop().run_in_executor(None, t.artemis.run_recon, target) | |
| # Show raw data summary | |
| ports = raw.get("ports", {}).get("open_ports", []) | |
| ip = raw.get("ports", {}).get("ip", "N/A") | |
| await ws.send_json({"type": "terminal_line", "cls": "success", "text": f" IP: {ip}"}) | |
| for p in ports: | |
| await ws.send_json({"type": "terminal_line", "cls": "success", | |
| "text": f" PORT {p['port']}/tcp OPEN {p['service']}"}) | |
| await ws.send_json({"type": "terminal_line", "cls": "info", "text": ""}) | |
| await ws.send_json({"type": "terminal_line", "cls": "info", | |
| "text": "[ARTEMIS AI ANALYSIS]:"}) | |
| # Stream analysis line by line | |
| for line in analysis.splitlines(): | |
| await ws.send_json({"type": "terminal_line", "cls": "output", "text": f" {line}"}) | |
| await asyncio.sleep(0.02) | |
| # ── OSINT ───────────────────────────────────────────── | |
| elif base == "osint" and args: | |
| target = args[0] | |
| await ws.send_json({"type": "terminal_line", "cls": "info", | |
| "text": f"[{ts}] [ATHENA] Full OSINT sweep on {target}..."}) | |
| t = get_team() | |
| analysis, raw = await asyncio.get_event_loop().run_in_executor(None, t.athena.run_osint, target) | |
| subs = raw.get("subdomains", {}).get("subdomains", []) | |
| if subs: | |
| await ws.send_json({"type": "terminal_line", "cls": "success", | |
| "text": f" Subdomains found: {len(subs)}"}) | |
| for s in subs[:10]: | |
| await ws.send_json({"type": "terminal_line", "cls": "output", "text": f" ↳ {s}"}) | |
| await ws.send_json({"type": "terminal_line", "cls": "info", "text": "[ATHENA AI ANALYSIS]:"}) | |
| for line in analysis.splitlines(): | |
| await ws.send_json({"type": "terminal_line", "cls": "output", "text": f" {line}"}) | |
| await asyncio.sleep(0.02) | |
| # ── INTEL ───────────────────────────────────────────── | |
| elif base == "intel" and args: | |
| target = args[0] | |
| await ws.send_json({"type": "terminal_line", "cls": "info", | |
| "text": f"[{ts}] [ATHENA] Building intelligence profile for {target}..."}) | |
| t = get_team() | |
| response = await asyncio.get_event_loop().run_in_executor( | |
| None, t.athena.chat, f"Build a comprehensive intelligence profile for: {target}. Include attack surface analysis, organizational intel, and recommended social engineering vectors based on publicly available information.") | |
| for line in response.splitlines(): | |
| await ws.send_json({"type": "terminal_line", "cls": "output", "text": f" {line}"}) | |
| await asyncio.sleep(0.02) | |
| # ── ASK AGENT ───────────────────────────────────────── | |
| elif base == "ask" and len(args) >= 2: | |
| agent_name = args[0].lower() | |
| question = " ".join(args[1:]) | |
| t = get_team() | |
| agent = t.get(agent_name) | |
| if not agent: | |
| await ws.send_json({"type": "terminal_line", "cls": "error", | |
| "text": f" Unknown agent: {agent_name}. Available: {', '.join(t.all_names())}"}) | |
| return | |
| await ws.send_json({"type": "terminal_line", "cls": "info", | |
| "text": f"[{ts}] [{agent_name.upper()}] Processing: {question}"}) | |
| response = await asyncio.get_event_loop().run_in_executor(None, agent.chat, question) | |
| for line in response.splitlines(): | |
| await ws.send_json({"type": "terminal_line", "cls": "output", "text": f" {line}"}) | |
| await asyncio.sleep(0.02) | |
| # ── FORGE ───────────────────────────────────────────── | |
| elif base == "forge" and args: | |
| task = " ".join(args) | |
| await ws.send_json({"type": "terminal_line", "cls": "info", | |
| "text": f"[{ts}] [HEPHAESTUS] Forging script for: {task}..."}) | |
| t = get_team() | |
| script = await asyncio.get_event_loop().run_in_executor(None, t.hephaestus.forge_script, task) | |
| for line in script.splitlines(): | |
| await ws.send_json({"type": "terminal_line", "cls": "output", "text": f" {line}"}) | |
| await asyncio.sleep(0.01) | |
| # ── CVE + AGENT ANALYSIS ────────────────────────────── | |
| elif base == "vulns" and len(args) >= 2: | |
| keyword = " ".join(args) | |
| await ws.send_json({"type": "terminal_line", "cls": "info", | |
| "text": f"[{ts}] [ARES] CVE analysis for '{keyword}'..."}) | |
| t = get_team() | |
| analysis, raw = await asyncio.get_event_loop().run_in_executor(None, t.ares.analyze_cves, keyword) | |
| await ws.send_json({"type": "terminal_line", "cls": "success", | |
| "text": f" Found {raw.get('total_results',0)} CVEs in NVD. AI Analysis:"}) | |
| for line in analysis.splitlines(): | |
| await ws.send_json({"type": "terminal_line", "cls": "output", "text": f" {line}"}) | |
| await asyncio.sleep(0.02) | |
| # ── UNKNOWN ─────────────────────────────────────────── | |
| else: | |
| await ws.send_json({"type": "terminal_line", "cls": "error", | |
| "text": f" Unknown command: '{cmd}'. Type 'help' for available commands."}) | |
| # ───────────────────────────────────────────────── | |
| # WEBSOCKET — Hermes Chat Channel | |
| # ───────────────────────────────────────────────── | |
| async def hermes_ws(ws: WebSocket): | |
| await ws.accept() | |
| t = get_team() | |
| await ws.send_json({ | |
| "type": "message", | |
| "from": "hermes", | |
| "text": "HERMES ONLINE. Secure channel established. LLM AI active. All 6 field agents reporting nominal status. What do you need, Operator?", | |
| "timestamp": datetime.utcnow().isoformat() | |
| }) | |
| try: | |
| while True: | |
| data = await ws.receive_json() | |
| user_msg = data.get("message", "").strip() | |
| if not user_msg: | |
| continue | |
| await ws.send_json({"type": "typing", "agent": "hermes"}) | |
| # Determine if we need to gather real data first | |
| response_text = "" | |
| lower = user_msg.lower() | |
| # Route to specialized agents if needed | |
| if any(w in lower for w in ["recon", "scan", "port", "dns", "ssl", "ping"]) and any( | |
| c.isalpha() and "." in user_msg for c in user_msg): | |
| # Extract target-like words | |
| words = user_msg.split() | |
| target = next((w for w in words if "." in w and len(w) > 4), None) | |
| if target: | |
| raw = {} | |
| try: | |
| _, raw = await asyncio.get_event_loop().run_in_executor(None, t.artemis.run_recon, target) | |
| except Exception: | |
| pass | |
| context = f"Fresh recon data for {target}: {json.dumps(raw, default=str)[:3000]}" | |
| response_text = await asyncio.get_event_loop().run_in_executor( | |
| None, t.hermes.chat, user_msg, context) | |
| else: | |
| response_text = await asyncio.get_event_loop().run_in_executor( | |
| None, t.hermes.chat, user_msg) | |
| elif any(w in lower for w in ["cve", "exploit", "vulnerability", "vuln"]): | |
| # Get real CVE context | |
| words = user_msg.split() | |
| keyword = " ".join([w for w in words if w.lower() not in | |
| ["cve", "exploit", "vulnerability", "find", "show", "get", "about", "tell", "me", "the"]][:3]) | |
| if keyword: | |
| cve_data = {} | |
| try: | |
| cve_data = await asyncio.get_event_loop().run_in_executor( | |
| None, __import__('tools.cve', fromlist=['search_cves']).search_cves, | |
| keyword, None, None, 5) | |
| except Exception: | |
| pass | |
| context = f"Real CVE data from NIST NVD: {json.dumps(cve_data, default=str)[:3000]}" | |
| response_text = await asyncio.get_event_loop().run_in_executor( | |
| None, t.hermes.chat, user_msg, context) | |
| else: | |
| response_text = await asyncio.get_event_loop().run_in_executor( | |
| None, t.hermes.chat, user_msg) | |
| elif any(w in lower for w in ["osint", "subdomain", "whois", "intel", "information"]): | |
| words = user_msg.split() | |
| target = next((w for w in words if "." in w and len(w) > 4), None) | |
| if target: | |
| _, raw = await asyncio.get_event_loop().run_in_executor(None, t.athena.run_osint, target) | |
| context = f"Real OSINT data: {json.dumps(raw, default=str)[:3000]}" | |
| response_text = await asyncio.get_event_loop().run_in_executor( | |
| None, t.hermes.chat, user_msg, context) | |
| else: | |
| response_text = await asyncio.get_event_loop().run_in_executor( | |
| None, t.hermes.chat, user_msg) | |
| else: | |
| response_text = await asyncio.get_event_loop().run_in_executor( | |
| None, t.hermes.chat, user_msg) | |
| await ws.send_json({ | |
| "type": "message", | |
| "from": "hermes", | |
| "text": response_text, | |
| "timestamp": datetime.utcnow().isoformat() | |
| }) | |
| except WebSocketDisconnect: | |
| pass | |
| except Exception as e: | |
| try: | |
| await ws.send_json({ | |
| "type": "message", "from": "hermes", | |
| "text": f"[HERMES ERROR] {str(e)}. Check LLM_API_KEY in .env", | |
| "timestamp": datetime.utcnow().isoformat() | |
| }) | |
| except Exception: | |
| pass | |
| if __name__ == "__main__": | |
| import uvicorn | |
| import sys | |
| # Fix Windows console encoding | |
| if sys.platform == 'win32': | |
| import io | |
| sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace') | |
| print("\n" + "="*60) | |
| print(" AI HACKING TEAM -- HERMES C2 SERVER") | |
| print(" Real agents. Real tools. Real LLM AI.") | |
| print("="*60) | |
| print(f" LLM API: {'[OK] Configured' if os.getenv('LLM_API_KEY') else '[!!] Missing -- add to .env'}") | |
| port = int(os.getenv("PORT", 8000)) | |
| print(f" Dashboard: http://localhost:{port}") | |
| print(f" API Docs: http://localhost:{port}/docs") | |
| print("="*60 + "\n") | |
| uvicorn.run("server:app", host="0.0.0.0", port=port, reload=False, log_level="warning") | |