Merge pull request #6 from barlanyado/main

mcp security scanner

Author: oroxenberg

README.md

@@ -1,15 +1,31 @@
-# MCP Gateway
 
-![Hugging Face Token Masking Example](docs/MCP_Flow.png)
+<div align="center">
+  <a href="https://pypi.org/project/mcp-gateway/">
+    <img src="https://img.shields.io/pypi/v/mcp-gateway.svg?color=blue" alt="PyPI version">
+  </a>
+  <a href="https://pypi.org/project/mcp-gateway/">
+    <img src="https://img.shields.io/pypi/pyversions/mcp-gateway.svg" alt="Python Versions">
+  </a>
+  <a href="./LICENSE">
+    <img src="https://img.shields.io/github/license/lasso-security/mcp-gateway" alt="License">
+  </a>
+  
+  # MCP Gateway
+
+</div>
+
+# Overview
+![](docs/MCP_Flow.png)
 
 MCP Gateway is an advanced intermediary solution for Model Context Protocol (MCP) servers that centralizes and enhances your AI infrastructure.
 
 MCP Gateway acts as an intermediary between LLMs and other MCP servers. It:
 
-1. Reads server configurations from a `mcp.json` file located in your root directory.
-2. Manages the lifecycle of configured MCP servers.
-3. Intercepts requests and responses to sanitize sensitive information.
-4. Provides a unified interface for discovering and interacting with all proxied MCPs.
+1. 📄 Reads server configurations from a `mcp.json` file located in your root directory.
+2. ⚙️ Manages the lifecycle of configured MCP servers.
+3. 🛡️ Intercepts requests and responses to sanitize sensitive information.
+4. 🔗 Provides a unified interface for discovering and interacting with all proxied MCPs.
+5. 🔒 **Security Scanner** - Analyzes server reputation and security risks before loading MCP servers.
 
 ## Installation
 
@@ -293,6 +309,7 @@ The Lasso guardrail checks content through Lasso's API for security violations b
 
 Read more on our website 👉 [Lasso Security](https://www.lasso.security/).
 
+
 ## Tracing
 
 ### Xetrack
@@ -398,6 +415,82 @@ D SELECT server_name,capability_name,path,content_text FROM db.events LIMIT 1;
 
 Of course you can use another MCP server to query the sqlite database 😊
 
+# Scanner
+
+The Security Scanner analyzes MCP servers for potential security risks before loading, providing an additional layer of protection through reputation analysis and tool description scanning.
+
+```bash
+mcp-gateway --scan -p basic
+```
+
+**Features:**
+- 🔍 **Reputation Analysis** - Evaluates server reputation using marketplace (Smithery, NPM) and GitHub data
+- 🛡️ **Tool Description Scanning** - Detects hidden instructions, sensitive file patterns, and malicious actions
+- ⚡ **Automatic Blocking** - Blocks risky MCPs based on reputation scores (threshold: 30) and security analysis
+- 📝 **Configuration Updates** - Automatically updates your MCP configuration file with scan results
+
+## Quickstart
+Initial configuration:
+```json
+{
+    "mcpServers": {
+        "mcp-gateway": {
+            "command": "mcp-gateway",
+            "args": [
+                "--mcp-json-path",
+                "~/.cursor/mcp.json",
+                "--scan"
+            ],
+            "servers": {
+                "filesystem": {
+                    "command": "npx",
+                    "args": [
+                        "-y",
+                        "@modelcontextprotocol/server-filesystem",
+                        "."
+                    ]
+                }
+            }
+        }
+    }
+}
+```
+After the first run, the scanner will analyze all configured MCP servers and add a `blocked` status to your configuration:
+```json
+{
+    "mcpServers": {
+        "mcp-gateway": {
+            "command": "mcp-gateway",
+            "args": [
+                "--mcp-json-path",
+                "~/.cursor/mcp.json",
+                "--scan"
+            ],
+            "servers": {
+                "filesystem": {
+                    "command": "npx",
+                    "args": [
+                        "-y",
+                        "@modelcontextprotocol/server-filesystem",
+                        "."
+                    ],
+                    "blocked": "passed"
+                }
+            }
+        }
+    }
+}
+```
+**Status Values:**
+- `"passed"` - Server passed all security checks and is safe to use
+- `"blocked"` - Server failed security checks and will be blocked from loading
+- `"skipped"` - Server scanning was skipped (manual override)
+- `null` - Server not yet scanned or previously blocked server now considered safe
+
+> **Note:** You can manually change a blocked server to `"skipped"` if you're confident it's safe.
+
+
+
 ## How It Works
 Your agent interacts directly with our MCP Gateway, which functions as a central router and management system. Each underlying MCP is individually wrapped and managed.
 
@@ -413,6 +506,9 @@ Key Features
 * Includes intelligent risk assessment with MCP risk scoring.
 * Delivers real-time status monitoring and performance metrics.
 
+**Security Scanner**
+* Analyzes MCP server reputation and tool descriptions for security risks before loading.
+
 **Advanced Tracking**
 * Maintains detailed logs of all requests and responses for each guardrail.
 * Offers cost evaluation tools for MCPs requiring paid tokens.

mcp_gateway/__init__.py

@@ -4,4 +4,4 @@
 This package provides functionality to manage and route requests to multiple MCP servers.
 """
 
-__version__ = "0.1.0"
+__version__ = "1.1.0"

mcp_gateway/config.py

@@ -2,7 +2,8 @@
 import logging
 import os
 from pathlib import Path
-from typing import Dict, Any
+from typing import Dict, Any, List, Tuple
+from mcp import types
 
 CONFIG_FILE_NAME = "mcp.json"
 
@@ -11,6 +12,10 @@
 # logging.basicConfig(level=logging.INFO) # Removed basicConfig here
 
 
+class Constants:
+    SERVERS = "servers"
+    
+    
 def find_config_file(mcp_json_path: str) -> Path | None:
     """Uses the provided mcp_json_path to locate the configuration file.
 
@@ -90,21 +95,21 @@ def load_servers_config_from_path(config_path: Path) -> Dict[str, Any]:
             return {}
 
         # 3. Extract the nested 'servers' key from the gateway's config
-        nested_servers_config = gateway_config.get("servers", {})
+        nested_servers_config = gateway_config.get(Constants.SERVERS, {})
         if not isinstance(nested_servers_config, dict):
             logger.warning(
-                f"Nested 'servers' key within '{gateway_config_key}' config in {config_path} is not a dictionary. Treating as empty."
+                f"Nested '{Constants.SERVERS}' key within '{gateway_config_key}' config in {config_path} is not a dictionary. Treating as empty."
             )
             return {}  # Return empty dict if nested 'servers' isn't a dict
 
         if not nested_servers_config:
             logger.warning(
-                f"Nested 'servers' key within '{gateway_config_key}' config in {config_path} is missing or empty. No proxied servers will be configured."
+                f"Nested '{Constants.SERVERS}' key within '{gateway_config_key}' config in {config_path} is missing or empty. No proxied servers will be configured."
             )
             # Fall through to return the (potentially empty) nested_servers_config
 
         logger.info(
-            f"Successfully extracted nested 'servers' config for proxied servers from {config_path}."
+            f"Successfully extracted nested '{Constants.SERVERS}' config for proxied servers from {config_path}."
         )
         # Return the extracted dict directly
         return nested_servers_config
@@ -161,3 +166,31 @@ def load_config(mcp_json_path: str) -> Dict[str, Any]:
         )
         logger.warning("Using empty configuration for proxied servers.")
         return {}  # Return empty dict
+
+
+def get_tool_params_description(tool: types.Tool) -> List[Tuple[str, Any, str]]:
+    param_signatures = []
+
+    # Tool has inputSchema (JSON Schema) instead of arguments
+    if hasattr(tool, "inputSchema") and tool.inputSchema:
+        # Try to extract properties from JSON Schema
+        properties = tool.inputSchema.get("properties", {})
+        for param_name, param_schema in properties.items():
+            param_type = Any  # Default type
+            param_description = param_schema.get("description", "")
+
+            # Map JSON Schema types to Python types
+            json_type = param_schema.get("type")
+            if json_type:
+                type_mapping = {
+                    "string": str,
+                    "integer": int,
+                    "boolean": bool,
+                    "number": float,
+                    "object": Dict[str, Any],
+                    "array": List[Any],
+                }
+                param_type = type_mapping.get(json_type, Any)
+
+            param_signatures.append((param_name, param_type, param_description))
+    return param_signatures