Decouple and separate MCP server into standalone cape-mcp package

cape_mcp/README.md

@@ -0,0 +1,42 @@
+# CAPE Sandbox MCP Server
+
+Standalone Model Context Protocol (MCP) server for CAPE Sandbox.
+
+Connect your favorite LLM client (like Claude Desktop) to a CAPE Sandbox instance seamlessly.
+
+## Quick Start (Standalone / No Cloning Required)
+
+If you have `uv` installed, you can run the server directly without cloning the repository:
+
+```bash
+uvx --from cape-mcp cape-mcp --url "https://cape.yourcompany.com/apiv2" --token "YOUR_API_TOKEN"
+```
+
+## Configuration for Claude Desktop
+
+Add this to your Claude Desktop configuration file (e.g., `~/Library/Application Support/Claude/claude_desktop_config.json` on macOS or `%APPDATA%\Claude\claude_desktop_config.json` on Windows):
+
+```json
+{
+  "mcpServers": {
+    "cape-sandbox": {
+      "command": "uvx",
+      "args": [
+        "--from", "cape-mcp",
+        "cape-mcp",
+        "--url", "https://cape.yourcompany.com/apiv2",
+        "--token", "YOUR_API_TOKEN"
+      ]
+    }
+  }
+}
+```
+
+## Environment Variables
+
+The server can also be configured using environment variables:
+
+- `CAPE_API_URL`: URL to your CAPE instance APIdoc (e.g., `http://localhost:8000/apiv2`).
+- `CAPE_API_TOKEN`: CAPE API Authorization Token.
+- `CAPE_ENABLED_MCP_TOOLS`: A comma-separated list of enabled tool names (e.g. `filecreate,tasklist`), or `*` to enable all (default in standalone mode).
+- `CAPE_AUTH_REQUIRED`: Set to `true` to require token authentication for all requests.

cape_mcp/pyproject.toml

@@ -0,0 +1,18 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "cape-mcp"
+version = "0.1.0"
+description = "Model Context Protocol (MCP) server for CAPE Sandbox"
+readme = "README.md"
+requires-python = ">=3.10"
+license = { text = "MIT" }
+dependencies = [
+    "fastmcp>=0.4.1",
+    "httpx>=0.24.0",
+]
+
+[project.scripts]
+cape-mcp = "cape_mcp.server:main"

cape_mcp/schemas.py

@@ -0,0 +1,130 @@
+# Static definitions of search schemas decoupled from lib.cuckoo core
+# This allows running the MCP server standalone without importing the CAPE backend codebase.
+
+search_term_map = {
+    "id": "info.id",
+    "ids": "info.id",
+    "tags_tasks": "info.id",
+    "package": "info.package",
+    "ttp": "ttps.ttp",
+    "malscore": "malscore",
+    "name": "target.file.name",
+    "type": "target.file.type",
+    "file": "behavior.summary.files",
+    "command": "behavior.summary.executed_commands",
+    "configs": "CAPE.configs",
+    "resolvedapi": "behavior.summary.resolved_apis",
+    "key": "behavior.summary.keys",
+    "mutex": "behavior.summary.mutexes",
+    "domain": "network.domains.domain",
+    "ip": "network.hosts.ip",
+    "asn": "network.hosts.asn",
+    "asn_name": "network.hosts.asn_name",
+    "signature": "signatures.description",
+    "signame": "signatures.name",
+    "detections": "detections.family",
+    "url": "target.url",
+    "iconhash": "static.pe.icon_hash",
+    "iconfuzzy": "static.pe.icon_fuzzy",
+    "surihttp": "suricata.http",
+    "suritls": "suricata.tls",
+    "surisid": "suricata.alerts.sid",
+    "surialert": "suricata.alerts.signature",
+    "surimsg": "suricata.alerts.signature",
+    "suriurl": "suricata.http.uri",
+    "suriua": "suricata.http.ua",
+    "surireferrer": "suricata.http.referrer",
+    "surihost": "suricata.http.hostname",
+    "suritlssubject": "suricata.tls.subject",
+    "suritlsissuerdn": "suricata.tls.issuer",
+    "suritlsfingerprint": "suricata.tls.fingerprint",
+    "procmemyara": ("procmemory.yara.name", "procmemory.cape_yara.name"),
+    "procdumpyara": ("procdump.yara.name", "procdump.cape_yara.name"),
+    "virustotal": "virustotal.results.sig",
+    "machinename": "info.machine.name",
+    "machinelabel": "info.machine.label",
+    "comment": "info.comments.Data",
+    "custom": "info.custom",
+    "target_sha256": "target.file.sha256",
+    "tlp": "info.tlp",
+    "ja3_hash": "suricata.tls.ja3.hash",
+    "ja3_string": "suricata.tls.ja3.string",
+    "dhash": "static.pe.icon_dhash",
+    "dport": ("network.tcp.dport", "network.udp.dport", "network.smtp_ex.dport"),
+    "sport": ("network.tcp.dport", "network.udp.dport", "network.smtp_ex.dport"),
+    "port": (
+        "network.tcp.dport",
+        "network.udp.dport",
+        "network.smtp_ex.dport",
+        "network.tcp.dport",
+        "network.udp.dport",
+        "network.smtp_ex.dport",
+    ),
+    "extracted_tool": (
+        "info.parent_sample.selfextract",
+        "target.file.selfextract",
+        "dropped.selfextract",
+        "procdump.selfextract",
+        "CAPE.payloads.selfextract",
+    ),
+}
+
+# Add standard repetitive blocks (like those created in loops dynamically in web_utils.py)
+search_term_map_base_naming = (
+    "target.file",
+    "dropped",
+    "procdump",
+    "CAPE.payloads",
+)
+search_term_map_repetetive_blocks = {
+    "md5": "md5",
+    "sha1": "sha1",
+    "sha256": "sha256",
+    "sha512": "sha512",
+    "ssdeep": "ssdeep",
+    "crc32": "crc32",
+    "yara": "yara.name",
+    "capeyara": "cape_yara.name",
+}
+
+for key, value in search_term_map_repetetive_blocks.items():
+    search_term_map.update({key: [f"{path}.{value}" for path in search_term_map_base_naming]})
+
+perform_search_filters = {
+    "info": 1,
+    "virustotal_summary": 1,
+    "detections.family": 1,
+    "malfamily_tag": 1,
+    "malscore": 1,
+    "network.pcap_sha256": 1,
+    "mlist_cnt": 1,
+    "f_mlist_cnt": 1,
+    "target.file.clamav": 1,
+    "target.file.sha256": 1,
+}
+
+hash_searches = {
+    "ssdeep": "ssdeep",
+    "crc32": "crc32",
+    "md5": "md5",
+    "sha1": "sha1",
+    "sha3": "sha3_384",
+    "sha256": "_id",
+    "sha512": "sha512",
+}
+
+normalized_lower_terms = (
+    "target_sha256",
+    "md5",
+    "sha1",
+    "sha3",
+    "sha256",
+    "sha512",
+    "ip",
+    "domain",
+    "ja3_hash",
+    "dhash",
+    "iconhash",
+    "imphash",
+    "package",
+)

mcp/server.py → cape_mcp/server.py

@@ -5,82 +5,75 @@
 import re
 from typing import Any, Dict
 
-# Ensure CAPE root is in path for lib imports
-CAPE_ROOT = os.path.join(os.path.abspath(os.path.dirname(__file__)), "..")
-sys.path.append(CAPE_ROOT)
-
+# Try to import FastMCP and httpx
 try:
     import httpx
     from fastmcp import FastMCP
 except ImportError:
-    sys.exit("poetry run pip install .[mcp]")
+    sys.exit("pip install fastmcp httpx")
 
+# Try to import schemas from the local package/directory first (for standalone use)
 try:
-    from lib.cuckoo.common.config import Config
-    from lib.cuckoo.common.web_utils import (
+    from cape_mcp.schemas import (
         search_term_map,
         perform_search_filters,
         hash_searches,
         normalized_lower_terms,
     )
 except ImportError:
-    sys.exit("Could not import lib.cuckoo.common.config. Ensure you are running from CAPE root.")
-
-# Initialize CAPE Config
-api_config = Config("api")
-
-# Configuration from Environment or Config File
-# Run with: CAPE_API_URL=http://127.0.0.1:8000/apiv2 CAPE_API_TOKEN=your_token poetry run python mcp/server.py
-API_URL = os.environ.get("CAPE_API_URL")
-if not API_URL:
-    # Try to get from api.conf [api] url
     try:
-        base_url = api_config.api.url.rstrip("/")
-        API_URL = f"{base_url}/apiv2"
-    except AttributeError:
-        API_URL = "http://127.0.0.1:8000/apiv2"
-
+        from schemas import (
+            search_term_map,
+            perform_search_filters,
+            hash_searches,
+            normalized_lower_terms,
+        )
+    except ImportError:
+        # Fallback to local import from lib.cuckoo.common.web_utils if running from CAPE root
+        try:
+            from lib.cuckoo.common.web_utils import (
+                search_term_map,
+                perform_search_filters,
+                hash_searches,
+                normalized_lower_terms,
+            )
+        except ImportError:
+            sys.exit("Could not import search schemas.")
+
+# Configuration from Environment or command line
+API_URL = os.environ.get("CAPE_API_URL", "http://127.0.0.1:8000/apiv2")
 API_TOKEN = os.environ.get("CAPE_API_TOKEN", "")
 
-# Proactively map enabled MCP tools. Default is NO.
+# Determine which tools are enabled (default: enable ALL in standalone mode)
 ENABLED_MCP_TOOLS = set()
-for section_name in api_config.get_config():
-    if section_name == "api":
-        continue
-    try:
-        section = api_config.get(section_name)
-        if getattr(section, "mcp", False):
-            ENABLED_MCP_TOOLS.add(section_name)
-    except Exception:
-        continue
+env_enabled_tools = os.environ.get("CAPE_ENABLED_MCP_TOOLS")
+
+if env_enabled_tools:
+    if env_enabled_tools == "*":
+        ENABLED_MCP_TOOLS = None  # None indicates "All are enabled"
+    else:
+        ENABLED_MCP_TOOLS = set(name.strip() for name in env_enabled_tools.split(",") if name.strip())
+else:
+    # Default to enabling all tools
+    ENABLED_MCP_TOOLS = None
 
 def check_mcp_enabled(section: str) -> bool:
     """Check if a specific section is enabled for MCP."""
+    if ENABLED_MCP_TOOLS is None:
+        return True
     return section in ENABLED_MCP_TOOLS
 
 def mcp_tool(section: str):
     """
     Conditional decorator that only registers the tool with FastMCP
-    if the corresponding section is enabled in api.conf.
+    if the corresponding section is enabled.
     """
     def decorator(func):
         if check_mcp_enabled(section):
             return mcp.tool()(func)
         return func
     return decorator
 
-def is_auth_required() -> bool:
-    """Check if token authorization is enabled globally."""
-    try:
-        return api_config.api.token_auth_enabled
-    except AttributeError:
-        return False
-
-# Startup Check: Warn if Auth is enabled but no default token is provided
-if is_auth_required() and not API_TOKEN:
-    print("WARNING: Token authentication is enabled in api.conf, but CAPE_API_TOKEN is not set.", file=sys.stderr)
-    print("         All MCP tool calls must include a valid 'token' argument.", file=sys.stderr)
-
 # Initialize FastMCP
 mcp = FastMCP("cape-sandbox")
 
@@ -97,12 +90,6 @@ def get_headers(token: str = "") -> Dict[str, str]:
     return headers
 
 async def _request(method: str, endpoint: str, token: str = "", **kwargs) -> Any:
-    # Auth Check
-    if is_auth_required():
-        auth_token = token if token else API_TOKEN
-        if not auth_token:
-             return {"error": True, "message": "Authentication required but no token provided."}
-
     url = f"{API_URL.rstrip('/')}/{endpoint.lstrip('/')}"
     async with httpx.AsyncClient() as client:
         try:
@@ -125,12 +112,6 @@ async def _request(method: str, endpoint: str, token: str = "", **kwargs) -> Any
 
 async def _download_file(endpoint: str, destination: str, default_filename: str = "downloaded_file.bin", token: str = "") -> str:
     """Helper to download a file from an API endpoint."""
-    # Auth Check
-    if is_auth_required():
-        auth_token = token if token else API_TOKEN
-        if not auth_token:
-             return json.dumps({"error": True, "message": "Authentication required but no token provided."}, indent=2)
-
     if not os.path.isdir(destination):
          return json.dumps({"error": True, "message": "Destination directory does not exist"})
 
@@ -198,12 +179,6 @@ async def submit_file(
     """
     Submit a local file for analysis.
     """
-    # Auth Check (Manual check needed here because we stream file)
-    if is_auth_required():
-        auth_token = token if token else API_TOKEN
-        if not auth_token:
-             return json.dumps({"error": True, "message": "Authentication required but no token provided."})
-
     if not os.path.exists(file_path):
         return json.dumps({"error": True, "message": "File not found"})
 
@@ -300,12 +275,6 @@ async def submit_static(
     token: str = ""
 ) -> str:
     """Submit a file for static extraction only."""
-    # Auth Check (Manual check needed here because we stream file)
-    if is_auth_required():
-        auth_token = token if token else API_TOKEN
-        if not auth_token:
-             return json.dumps({"error": True, "message": "Authentication required but no token provided."})
-
     if not os.path.exists(file_path):
         return json.dumps({"error": True, "message": "File not found"})
 
@@ -626,7 +595,7 @@ async def verify_auth(token: str = "") -> str:
 
     return json.dumps({"authenticated": True, "message": "Token is valid.", "user": "Authenticated User"}, indent=2)
 
-if __name__ == "__main__":
+def main():
     import argparse
 
     parser = argparse.ArgumentParser(description="CAPE MCP Server")
@@ -640,3 +609,6 @@ async def verify_auth(token: str = "") -> str:
         mcp.run(transport=args.transport, host=args.host, port=args.port)
     else:
         mcp.run(transport="stdio")
+
+if __name__ == "__main__":
+    main()