Tufin MCP Server - Open Source Community Project

Tufin MCP 服务 - 开源社区项目

简介

欢迎使用 Tufin MCP（Multi-Controller Protocol，多控制器协议）服务项目！这项开源计划旨在为 Tufin 用户提供强大的工具，弥合 Tufin 的强大 API（SecureTrack 和 SecureChange）与现代 AI 开发工作流之间的差距。

该服务通过具有基于角色的访问控制（RBAC）的安全、标准化 REST API 暴露 Tufin 功能，让你能够轻松地将 Tufin 数据和操作集成到自定义脚本、应用程序和 AI Agent 中（例如运行在 Cursor、ChatGPT Actions、Ollama 等平台上的 Agent）。

这是一个由社区驱动的项目。 我们欢迎大家贡献！无论是添加新的端点覆盖范围、改进安全性、增强客户端库，还是修复错误，你的贡献都很有价值。详情请查看贡献部分。

在 GitHub 上查看项目仓库：https://github.com/stonecircle82/tufin-mcp

为什么使用它？

• 简化 AI 集成： 在 Tufin 现有 API 之上提供现代 REST API 层，让它更容易与 AI 工具、聊天机器人和自动化脚本集成。
• 集中控制： 在一个位置管理 Tufin 访问权限和身份验证。
• 增强安全性： 在 Tufin 身份验证之上增加 API Key 和基于角色的访问控制。
• 可扩展性： 旨在扩展以覆盖更多 Tufin API，并可能支持其他安全平台。
• 未来潜力： 致力于支持更新的 Tufin API（如 GraphQL）和批量操作等功能（例如从文件中添加设备），这些功能通过直接 API 调用实现可能较为复杂。

概述

此服务充当 Tufin API（目标版本为 v25.1）的安全代理和抽象层。

主要功能：

• 标准 REST/JSON API 接口（通过 OpenAPI 编写文档）。
• 集中管理 Tufin 身份验证（使用通过环境变量配置的 Basic Auth）。
• 为 MCP 客户端提供 API Key 身份验证（密钥使用 bcrypt 哈希）。
• 可配置的基于角色的访问控制（管理员、工单管理者、用户）。
• 提供 SecureChange 和 SecureTrack 关键操作的端点（工单、设备、拓扑）。
• 结构化 JSON 日志，包含请求 ID。
• 基于 IP 的速率限制。
• 基础 Python 客户端库。
• 支持 Docker。

开始使用

先决条件

• Python 3.8+
• 可访问正在运行的 Tufin SecureTrack 和 SecureChange 实例（兼容 v25.1）。
• Docker（如果通过容器运行）。
• Git。

安装

1. 克隆仓库：

   git clone https://github.com/<your-username>/tufin-mcp.git # ** Replace URL **
   cd tufin-mcp
   

2. 创建并激活虚拟环境：

   python -m venv venv
   source venv/bin/activate # On Windows use `venv\Scripts\activate`
   

3. 安装依赖：

   pip install -r requirements.txt
   

配置

通过环境变量或项目根目录中的 .env 文件管理配置。

1. 创建 .env 文件： 如果提供了 .env.example，请复制该文件；也可以手动创建 .env。

2. 设置必需的变量：

   # Server Settings
   MCP_PORT=8000
   LOG_LEVEL="INFO" # DEBUG, INFO, WARNING, ERROR
   
   # Tufin Connection (REPLACE with your details)
   TUFIN_SECURETRACK_URL="https://your-securetrack-host"
   TUFIN_SECURECHANGE_URL="https://your-securechange-host"
   TUFIN_USERNAME="your_tufin_username"
   TUFIN_PASSWORD="your_tufin_password"
   TUFIN_SSL_VERIFY="True" # Set to False only if absolutely necessary (insecure!)
   TUFIN_API_TIMEOUT="30.0" # Timeout for Tufin API calls in seconds
   
   # --- Development API Keys (Insecure - For Dev/Test Only!) ---
   # For the default In-Memory secure store, provide initial RAW keys and roles via this JSON string.
   # The server will HASH the keys on startup and store them in memory.
   # DO NOT USE PRODUCTION KEYS HERE.
   DEV_API_KEYS='[{"key":"admin_key_abc", "role":"admin"}, {"key":"manager_key_xyz", "role":"ticket_manager"}, {"key":"user_key_123", "role":"user"}]'
   

   SSL 验证说明（TUFIN_SSL_VERIFY）： 默认情况下，服务会尝试验证 Tufin 实例的 SSL 证书（TUFIN_SSL_VERIFY="True"）。将其设置为 "False" 会禁用验证，这对于使用自签名证书的环境可能是必要的，但在生产环境中极不安全，因为连接会暴露于中间人攻击之下。如果使用内部 CA，可考虑配置底层系统或提供自定义 CA 证书包路径（需要修改代码）。

   生产环境安全说明： API 密钥使用 bcrypt 进行哈希处理。默认的 InMemorySecureStore 仅用于开发环境，会从 DEV_API_KEYS 加载原始密钥。在生产环境中，必须：

   1. 将 src/app/core/secure_store.py 中的 InMemorySecureStore 替换为使用安全数据库或密钥管理器（例如 HashiCorp Vault）的实现。
   2. 实现安全的 API 密钥管理流程或端点（生成、存储哈希值及角色、撤销）。
   3. 不要在生产环境中使用 DEV_API_KEYS。

运行服务

本地开发（Uvicorn）

# From the project root directory
uvicorn src.app.main:app --reload --port ${MCP_PORT:-8000}

• --reload 标志可在代码发生更改时自动重新加载。
• 使用 .env 中定义的端口；如果未定义，则默认为 8000。

Docker 容器

1. 构建镜像：

   # From the project root directory
   docker build -t tufin-mcp-server:latest .
   

2. 运行容器：

   docker run --rm -d \
     --name tufin-mcp \
     --env-file .env \
     -p ${MCP_PORT:-8000}:${MCP_PORT:-8000} \
     tufin-mcp-server:latest
   

   • 从本地 .env 文件加载环境变量。
   • 将主机端口映射到容器端口（由 MCP_PORT 定义；默认值为 8000）。

3. 访问： http://localhost:${MCP_PORT:-8000}

4. 日志： docker logs tufin-mcp（使用 -f 持续查看）

5. 停止： docker stop tufin-mcp

测试

本项目使用 pytest 进行测试。

1. 安装测试依赖：

   pip install -r requirements-dev.txt
   

2. 运行测试：

   # From the project root directory
   pytest
   

   • 可以运行指定文件：pytest tests/api/v1/test_endpoints.py
   • 使用 -v 输出详细信息：pytest -v
   • 使用 -k 按名称筛选测试：pytest -k list_devices

3. 测试覆盖率（可选）： 安装 pytest-cov（pip install pytest-cov）并运行：

   pytest --cov=src/app --cov-report=term-missing
   

API 使用

身份验证

除 /health 外，所有请求都应在 HTTP 标头 X-API-Key 中传入生成的 API 密钥。

授权（RBAC）

访问权限由创建 API 密钥时分配的角色控制（通过安全的生产流程，或在开发环境中使用 DEV_API_KEYS）。每个端点允许的角色在 src/app/core/config.py 的 ENDPOINT_PERMISSIONS 中配置。

• 角色： admin、ticket_manager、user。
• 尝试执行未经授权的操作会返回 403 Forbidden。

配置端点权限：

src/app/core/config.py 中的 ENDPOINT_PERMISSIONS 字典定义了哪些角色可以访问哪些逻辑操作（权限 ID）。示例：

# src/app/core/config.py
ENDPOINT_PERMISSIONS: Dict[str, List[UserRole]] = {
    "list_devices": [UserRole.ADMIN, UserRole.TICKET_MANAGER, UserRole.USER],
    "get_device": [UserRole.ADMIN, UserRole.TICKET_MANAGER, UserRole.USER],
    "create_ticket": [UserRole.ADMIN, UserRole.TICKET_MANAGER],
    # ... other permissions ...
}

• 为什么采用这种方式？ 这种方式将访问控制策略集中在一处，便于审核和修改权限，而无需更改端点代码本身。
• 工作原理： 每个 API 端点路由都使用 require_permission("permission_id") 依赖项。该依赖项会检查与用户已验证的 API 密钥关联的角色，是否包含在 ENDPOINT_PERMISSIONS 字典中为相应 permission_id 定义的列表内。
• 自定义： 你可以直接在 config.py 中修改任意权限 ID 对应的 UserRole 枚举列表，轻松自定义访问权限。添加新端点时，也可添加新的权限 ID。

速率限制

适用基于 IP 的速率限制（默认值：每分钟 60 次）。超出限制会返回 429 Too Many Requests。

端点概览

API 结构定义在 openapi.yaml 中。你可以直接查看此文件，也可以使用 Swagger UI 等工具（如果应用提供了该规范；这需要在 main.py 中添加静态文件服务）。

当前端点摘要：

• GET /health：健康检查。
• POST /api/v1/tickets：创建 SecureChange 工单。
• GET /api/v1/tickets：列出 SecureChange 工单（支持按 status 筛选）。
• GET /api/v1/tickets/{ticket_id}：获取指定的 SecureChange 工单。
• PUT /api/v1/tickets/{ticket_id}：更新 SecureChange 工单。
• GET /api/v1/devices：列出 SecureTrack 设备（支持按 status、name、vendor 筛选）。
• GET /api/v1/devices/{device_id}：获取 SecureTrack 设备详情。
• POST /api/v1/devices/bulk：添加一个或多个设备（请求正文中需要包含特定于供应商的 device_data）。
• POST /api/v1/devices/bulk/import：将受管设备（DG、ADOM、上下文等）导入现有管理设备。
• GET /api/v1/topology/map：获取 SecureTrack 拓扑图。
• POST /api/v1/topology/query：运行 SecureTrack 拓扑查询。
• GET /api/v1/topology/path：运行 SecureTrack 拓扑路径查询。返回摘要结果，其中包含 traffic_allowed、is_fully_routed 和 path_device_names（仅在流量允许或路径已完全路由时返回）。
• GET /api/v1/topology/path/image：以图像形式（例如 PNG）获取拓扑路径。
• POST /api/v1/graphql/rules：使用 GraphQL 和 TQL 筛选器查询 SecureTrack 规则。

注意： 实现时需要根据 Tufin 25.1 REST API 文档进行验证。筛选器的实现需要依据 Tufin 特定 API 语法进行检查。

基本示例（curl）

# Set environment variables for convenience
export MCP_URL="http://localhost:8000"
export MCP_API_KEY="your_api_key_here"

# Health Check
curl "$MCP_URL/health"

# List Devices
curl -H "X-API-Key: $MCP_API_KEY" "$MCP_URL/api/v1/devices?limit=5"

# Create Ticket (Requires appropriate role for the key)
curl -X POST -H "X-API-Key: $MCP_API_KEY" -H "Content-Type: application/json" \
     -d '{"subject": "API: Allow Port 443", "description": "...details..."}' \
     "$MCP_URL/api/v1/tickets"

# Example Topology Path Query
curl -G -H "X-API-Key: $MCP_API_KEY" "$MCP_URL/api/v1/topology/path" \
     --data-urlencode "src=1.1.1.1" \
     --data-urlencode "dst=8.8.8.8" \
     --data-urlencode "service=tcp:443"
# Example Topology Path Image Request (save to file)
curl -H "X-API-Key: $MCP_API_KEY" "$MCP_URL/api/v1/topology/path/image?src=1.1.1.1&dst=8.8.8.8&service=tcp:443" -o topology_path.png

# Example Add Device (Cisco ASA - requires ADMIN role & correct device_data)
curl -X POST -H "X-API-Key: $MCP_API_KEY" -H "Content-Type: application/json" \
     -d '{
           "devices": [
             {
               "display_name": "MCP-ASA-Test",
               "ip_address": "10.1.2.3",
               "vendor": "Cisco",
               "model": "ASA",
               "securetrack_domain": "Default",
               "enable_topology": true,
               "device_data": {
                 "user_name": "tufin-api-user", 
                 "password": "tufin-api-password",
                 "enable_password": "tufin-enable-password"
                 # Add other ASA specific fields from Tufin docs here
               }
             }
           ]
         }' \
     "$MCP_URL/api/v1/devices/bulk"

# Example Import Managed Devices (Panorama DG)
curl -X POST -H "X-API-Key: $MCP_API_KEY" -H "Content-Type: application/json" \
     -d '{
           "devices": [
             {
               "device_id": "1", 
               "device_data": {
                 "import_all": false,
                 "import_devices": [
                   {"name": "DG1", "import_all": false, "managed_devices": ["fw1"]}
                 ]
               }
             }
           ]
         }' \
     "$MCP_URL/api/v1/devices/bulk/import"

# Example GraphQL Rule Query (using curl and jq for readability)
# Find firewall rules allowing any source to 8.8.8.8
export TQL_FILTER="destination.ip 8.8.8.8"
curl -X POST -H "X-API-Key: $MCP_API_KEY" -H "Content-Type: application/json" \
     -d "{\"tql_filter\": \"$TQL_FILTER\"}" \
     "$MCP_URL/api/v1/graphql/rules" | jq

OpenAPI 规范

根目录中包含 OpenAPI 3.0 规范文件（openapi.yaml）。该文件正式描述了 API，包括端点、参数、请求和响应架构以及安全要求。它可与各种工具配合使用：

• 代码生成： 生成不同语言的客户端库。
• API 文档： 使用 Swagger UI 或 Redoc 等工具生成交互式文档。
• AI 集成： 导入 ChatGPT Actions 等平台以启用交互。
• 测试工具： 与 Postman 或 Insomnia 等工具配合使用。

（可选增强）： 可考虑配置 FastAPI，使其自动在 /openapi.yaml 提供此规范，并提供交互式文档（Swagger UI/Redoc）。

Python 客户端库

client_libs/python/ 中提供了基本的 Python 客户端库结构。

安装（从本地源码安装）

# From the project root directory
cd client_libs/python
pip install .

使用示例

from tufin_mcp_client import TufinMCPClient, TufinMCPClientError

SERVER_URL = "http://localhost:8000" 
API_KEY = "your_api_key_here"

# Use as a context manager (recommended)
with TufinMCPClient(base_url=SERVER_URL, api_key=API_KEY) as client:
    try:
        health = client.get_health()
        print(f"Health: {health}")
        
        devices = client.list_devices()
        print(f"Devices Found: {devices.total}")
        
        # Example: Get first device if list is not empty
        if devices.devices:
            first_device_id = devices.devices[0].id
            device_details = client.get_device(first_device_id)
            print(f"Device {first_device_id}: {device_details.name} ({device_details.vendor})")
        
        # Example: Create Ticket 
        ticket_data = {
            "workflow_name": "Example Firewall Workflow", # Check configured workflows
            "subject": "Client Lib Test", 
            "details": { # Workflow specific fields go here
                "description": "Testing ticket creation via client",
                "priority": "Medium"
                 # Add other fields required by the specific workflow
            }
        }
        created_ticket = client.create_ticket(ticket_data)
        print(f"Created Ticket ID: {created_ticket.id}, Status: {created_ticket.status}")
        ticket_id = created_ticket.id
        
        if ticket_id:
            retrieved_ticket = client.get_ticket(ticket_id)
            print(f"Retrieved Ticket {ticket_id}: Subject: {retrieved_ticket.subject}")
            
            updated_data = {"status": "In Progress"} # Check actual updatable fields
            updated_ticket = client.update_ticket(ticket_id, updated_data)
            print(f"Updated Ticket {ticket_id}: Status: {updated_ticket.status}")
            
            # Example Add Device
            asa_device = {
              "display_name": "MCP-ASA-Test-Client",
              "ip_address": "10.1.2.4",
              "vendor": "Cisco",
              "model": "ASA",
              "securetrack_domain": "Default",
              "enable_topology": True,
              "device_data": {
                 "user_name": "tufin-api-user", 
                 "password": "tufin-api-password",
                 "enable_password": "tufin-enable-password"
              }
            }
            client.add_devices([asa_device]) # Pass as a list
            print("Device add request accepted.")
            
        # Example Import Managed Device
        import_details = {
            "devices": [
                {
                    "device_id": "1", # Panorama ID
                    "device_data": {
                        "import_all": False,
                        "import_devices": [
                            {"name": "DG_CLIENT", "import_all": True}
                        ]
                    }
                }
            ]
        }
        client.import_managed_devices(import_details)
        print("Managed device import request accepted.")

        # Example GraphQL Rule Query
        rule_filter = "action accept and source.ip 192.168.1.0/24"
        rules_response = client.query_rules_graphql(tql_filter=rule_filter)
        print(f"\nFound {rules_response.rules.count} rules matching filter:")
        for rule in rules_response.rules.values:
            print(f"  - ID: {rule.id}, Name: {rule.name}, Action: {rule.action}")
            
    except TufinMCPClientError as e:
        print(f"MCP Client Error: {e}")
        if e.status_code:
            print(f"  Status Code: {e.status_code}")
        if e.response_text:
            print(f"  Response: {e.response_text}")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")

**注意：**客户端库提供基本功能和错误处理（TufinMCPClientError），并以 Pydantic 模型的形式返回响应。后续计划进一步完善（请参阅路线图）。

JavaScript / TypeScript 客户端库

client_libs/javascript/ 中提供了一个基本的 TypeScript 客户端库结构，用于与 MCP 服务 API 交互。

安装

# From your JS/TS project directory
npm install <path_to_client_libs/javascript> 
# or link for local development
# cd client_libs/javascript && npm link && cd ../../ && npm link tufin-mcp-client-js

# Or if published to npm:
# npm install tufin-mcp-client-js 

如果从源码安装，请确保先构建客户端：cd client_libs/javascript && npm run build && cd ../..

使用示例（TypeScript）

import { TufinMCPClient, TufinMCPClientError } from 'tufin-mcp-client-js'; // Adjust import path

const SERVER_URL = 'http://localhost:8000'; // Your MCP Server URL
const API_KEY = 'your_api_key_here';

const client = new TufinMCPClient(SERVER_URL, API_KEY);

async function runClient() {
    try {
        const health = await client.getHealth();
        console.log('Health:', health);

        const devices = await client.listDevices({ limit: 5 }); // Example param
        console.log(`Devices Found: ${devices.total}`);
        console.log('First device:', devices.devices[0]);
        
        // Example: Create Ticket
        const ticketData = {
            workflow_name: 'Example Firewall Workflow', // Check configured workflows
            subject: 'TS Client Test',
            details: {
                description: 'Testing ticket from TS client',
                priority: 'Low'
                // Add other workflow-specific fields
            }
        };
        const createdTicket = await client.createTicket(ticketData);
        console.log(`Created Ticket ID: ${createdTicket.id}, Status: ${createdTicket.status}`);

        // Example Add Device
        const asaDevice = {
          display_name: "MCP-ASA-Test-JS",
          ip_address: "10.1.2.5",
          vendor: "Cisco",
          model: "ASA",
          securetrack_domain: "Default",
          enable_topology: true,
          device_data: {
             user_name: "tufin-api-user", 
             password: "tufin-api-password",
             enable_password: "tufin-enable-password"
          }
        };
        await client.addDevices([asaDevice]); // Pass as an array
        console.log("Device add request accepted.");

        // Example Import Managed Device
        const importDetails = {
          devices: [
            {
              device_id: "1", // Panorama ID
              device_data: {
                import_all: false,
                import_devices: [
                  { name: "DG_JS_CLIENT", import_all: true }
                ]
              }
            }
          ]
        };
        await client.importManagedDevices(importDetails);
        console.log("Managed device import request accepted.");

        // Example GraphQL Rule Query
        const ruleFilter = "disabled true";
        const rulesResponse = await client.queryRulesGraphQL({tql_filter: ruleFilter});
        console.log(`\nFound ${rulesResponse.rules.count} disabled rules:`);
        rulesResponse.rules.values.forEach(rule => {
            console.log(`  - ID: ${rule.id}, Name: ${rule.name}`);
        });

    } catch (error) {
        if (error instanceof TufinMCPClientError) {
            console.error('MCP Client Error:', error.message);
            if (error.status) {
                console.error('  Status Code:', error.status);
            }
            if (error.data) {
                console.error('  Response Data:', error.data);
            }
        } else {
            console.error('An unexpected error occurred:', error);
        }
    }
}

runClient();

**注意：**此客户端功能基础。要实现完善的错误处理、潜在的模型解析（如果不依赖 any）和更可靠的类型安全，还需要进一步开发。

直接 Tufin API 客户端（替代方案）

如果需要绕过 MCP 服务，使用基本身份验证直接连接 Tufin API，可使用 client_libs/javascript_direct/ 中另一个更精简的 JS/TS 客户端库。详情请参阅该目录中的 README。

与 AI 工具集成

• **核心要求：**MCP 服务 URL、API 密钥。
• **方法：**Python 客户端库或直接调用 API。
• **工具指南：**Cursor、ChatGPT Actions（使用 openapi.yaml）、Copilot（通过代码上下文）、Ollama/OpenRouter（通过中间脚本）。
• **安全注意事项：**密钥管理、HTTPS、最小权限、监控。

参与贡献

欢迎贡献！有关如何参与、报告错误、提出功能建议和提交拉取请求的详情，请参阅贡献指南。

我们也遵守行为准则。

路线图 / 未来改进

此列表反映了已知的待办事项和潜在的未来改进：

• 验证并完善 Tufin REST API 逻辑：
  • 根据 Tufin 25.1 官方 REST 文档，仔细核对 src/app/clients/tufin.py 中使用的所有 Tufin REST API 端点路径、方法、请求参数/正文（尤其是筛选语法/功能）和响应结构。
  • 确保 src/app/models/ 中的 Pydantic 模型能够准确解析已验证的 Tufin REST API 响应。
• GraphQL API 集成：
  • 探索并实现利用 Tufin SecureTrack GraphQL API 的端点（文档），以便检索可能更丰富的数据（例如设备、规则、USP）。（已添加规则查询）
  • 支持通过 MCP API 动态选择 GraphQL 查询中的字段。
• 批量操作：
  • 实现批量操作端点，例如从文件（如 CSV/Excel）添加/更新多个设备，以及导入受管设备。
• 生产环境安全：
  • 将 InMemorySecureStore 替换为可用于生产环境的解决方案（数据库、Vault），并实现安全的 API 密钥管理（生成、撤销）。
  • 在 src/app/core/logging_config.py 中实现可靠的敏感数据屏蔽。
  • 确保为生产部署正确配置 SSL 证书验证（TUFIN_SSL_VERIFY）。
• 测试：
  • 添加全面的单元测试和集成测试，覆盖客户端逻辑、API 端点、安全性和错误处理。
• 客户端库：
  • 改进 Python 客户端库（错误处理、响应模型/解析）。
  • 改进 JS/TS 客户端库（错误处理、构建流程）。
• Tufin 客户端改进：
  • 为 src/app/clients/tufin.py 中的 httpx 客户端添加可配置的超时和重试逻辑。
• 部署与开发者体验：
  • 提供更多部署指南（例如 Kubernetes 清单、生产环境 Gunicorn 设置）。
  • 配置 FastAPI 以提供 openapi.yaml 和交互式文档（Swagger UI/Redoc）。

许可证

本项目根据 MIT 许可证授权；详情请参阅 LICENSE 文件。