查询 25 次,耗时 0.4556 秒

从零实现一个安全的 MCP Workspace Server:协议原理、Client 接入与权限边界

gs328772

New member

gs328772 Rep
0
0
0
Rep
0
gs328772 Vouches
Vouches
0
1 MONTH
1 1 MONTH OF SERVICE
gs328772 获得
0
0
0
获得
0
gs328772 凭证
0
0
0
凭证
0
主题
4
点赞
0
HakB
1 月
在社区成长 1 月
LEVEL 1 1 XP
 
一、背景:为什么工具接入不能只关注“能否调用”

MCP(Model Context Protocol)用于在模型应用与外部工具、资源和提示模板之间建立统一接口。它解决的核心问题并不是让模型直接执行任意操作,而是让 Host、Client 与 Server 通过明确的能力声明和结构化协议完成交互。

在典型架构中,Host 是承载模型和用户交互的应用;MCP Client 运行在 Host 内部,负责连接一个或多个 Server;MCP Server 则暴露 tools、resources、prompts 等能力。以工具调用为例,Client 先读取 Server 的工具清单及输入模式,模型或业务逻辑再决定是否发起调用,最终由 Server 执行受控操作并返回结果。

这层抽象提高了复用性,但不会自动带来安全性。如果 Server 暴露了无限制文件读取、Shell 执行或数据库写入能力,模型受到提示注入后仍可能触发越权行为。因此,生产级 MCP 接入必须同时设计输入约束、身份认证、授权、审批、审计和输出处理。

本文实现一个只允许读取指定工作区文本文件的 Python MCP Server,并给出 Client 接入、测试和安全加固方法。所有操作均面向本地实验环境,不涉及真实目标或未授权系统。

二、协议与调用流程

MCP 通常基于 JSON-RPC 2.0 风格的消息进行通信。SDK 会处理消息封装,开发者主要关注以下生命周期:

1. Client 与 Server 建立 stdio 或 Streamable HTTP 连接。
2. Client 发起 initialize,双方交换协议版本、实现信息和能力。
3. 初始化完成后,Client 可通过 tools/list 获取工具定义。
4. Client 使用 tools/call 调用指定工具,并传入符合输入模式的参数。
5. Server 返回结构化内容或错误,Client 再决定是否把结果提供给模型。

stdio 适合本机进程集成:Host 启动 Server 子进程,并通过标准输入输出通信。Streamable HTTP 更适合远程或多用户部署,但同时需要处理 TLS、认证、跨域来源校验、会话隔离和反向代理配置。需要特别注意:会话 ID 不是用户身份,不能代替认证与授权。

工具的名称、描述和参数模式会影响模型如何选择工具,但描述文本不是权限控制。即使某个工具标注了 readOnlyHint 或 destructiveHint,这些注解也只能帮助 Client 做交互决策,Server 仍然必须在执行层强制检查权限。

三、目标与权限模型

本例 Server 仅提供两个工具:

1. list_files:列出工作区某一目录的直接子项。
2. read_text:读取工作区中的少量文本内容。

权限边界如下:

• 只能访问 MCP_WORKSPACE 指定目录。
• 拒绝绝对路径和通过 ../ 进行的目录穿越。
• 只允许读取白名单扩展名。
• 同时限制磁盘文件大小和返回字符数。
• 不提供 Shell、写文件、网络访问和动态代码执行。
• stdio 模式下日志只能写入 stderr,避免污染协议输出。

这种设计遵循“最小能力”原则。若业务只需要检索文档,就不应顺手提供命令执行;若确实需要写操作,应拆分成独立工具,并增加人工确认、版本控制或事务回滚。

四、前置条件

建议准备:

• Python 3.10 或更高版本。
• 独立虚拟环境。
• 一个仅包含测试文件的本地目录。
• 官方 MCP Python SDK。

在空目录中执行:

mkdir secure-mcp-workspace
cd secure-mcp-workspace
python -m venv .venv

Linux 或 macOS 激活环境:

source .venv/bin/activate

Windows PowerShell 可使用:

.venv\\Scripts\\Activate.ps1

安装 SDK:

python -m pip install "mcp[cli]"
python -m pip freeze > requirements.lock

生产环境不建议长期使用无版本约束的依赖。应在测试通过后提交锁文件,并通过依赖扫描和变更评审升级版本。不同 SDK 版本的启动参数可能有变化,升级时应对照对应版本文档,而不是只验证“进程能够启动”。

创建实验工作区:

mkdir workspace
printf "MCP local test document\n" > workspace/README.md

五、实现受限的 MCP Server

创建 server.py:

from pathlib import Path
import logging
import os

from mcp.server.fastmcp import FastMCP

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("workspace-reader")

mcp = FastMCP("workspace-reader")
ROOT = Path(os.environ.get("MCP_WORKSPACE", "./workspace")).resolve()
ALLOWED_SUFFIXES = {".txt", ".md", ".json", ".yaml", ".yml"}
MAX_FILE_BYTES = 256 * 1024
MAX_RETURN_CHARS = 20000


def resolve_inside(relative_path: str) -> Path:
raw = Path(relative_path)
if raw.is_absolute():
raise ValueError("不允许使用绝对路径")

candidate = (ROOT / raw).resolve()
try:
candidate.relative_to(ROOT)
except ValueError as exc:
raise ValueError("路径超出允许的工作区") from exc
return candidate


@mcp.tool()
def list_files(relative_dir: str = ".") -> list[str]:
"""列出工作区目录的直接子项,不递归读取文件内容。"""
directory = resolve_inside(relative_dir)
if not directory.exists() or not directory.is_dir():
raise ValueError("目录不存在或不是目录")

result = []
for item in sorted(directory.iterdir(), key=lambda p: p.name.lower()):
kind = "dir" if item.is_dir() else "file"
result.append(f"{kind}: {item.name}")
if len(result) >= 200:
result.append("结果已截断")
break
logger.info("list_files dir=%s count=%d", directory, len(result))
return result


@mcp.tool()
def read_text(relative_path: str, max_chars: int = 4000) -> str:
"""读取工作区内白名单类型的文本文件,并限制返回长度。"""
path = resolve_inside(relative_path)
if not path.exists() or not path.is_file():
raise ValueError("文件不存在或不是普通文件")
if path.suffix.lower() not in ALLOWED_SUFFIXES:
raise ValueError("不允许读取该文件类型")

size = path.stat().st_size
if size > MAX_FILE_BYTES:
raise ValueError("文件超过大小限制")

safe_limit = max(1, min(max_chars, MAX_RETURN_CHARS))
text = path.read_text(encoding="utf-8", errors="replace")
if len(text) > safe_limit:
return text[:safe_limit] + "\n[内容已截断]"

logger.info("read_text path=%s chars=%d", path, len(text))
return text


if __name__ == "__main__":
if not ROOT.exists() or not ROOT.is_dir():
raise RuntimeError(f"工作区不存在:{ROOT}")
mcp.run(transport="stdio")

这里有几个关键点。首先,路径检查必须使用解析后的规范路径进行父目录判断,仅检查字符串中是否出现 ../ 并不可靠。其次,文件大小限制应在完整读取前完成,否则攻击者可能利用大文件造成内存压力。再次,Server 没有暴露任意路径参数,也没有把参数拼接到 Shell 命令中。

符号链接仍需谨慎:resolve 可以阻止指向工作区外部的既有符号链接,但在高并发或攻击者能够修改目录的场景中,检查与打开文件之间可能存在竞态条件。更严格的部署应让工作区对 Server 进程只读、禁止非受信用户修改目录,并结合容器挂载或操作系统沙箱,而不是只依赖应用层路径检查。

六、启动和本地检查

设置工作区为绝对路径后启动:

MCP_WORKSPACE=/absolute/path/secure-mcp-workspace/workspace python server.py

stdio Server 启动后等待协议消息,看起来像“没有输出”通常是正常现象。不要向 stdout 打印启动横幅或调试日志,因为 stdout 是协议通道。Python logging 默认写入 stderr,适合本例。

可以使用 MCP Inspector 在本地进行交互检查:

npx @modelcontextprotocol/inspector python /absolute/path/server.py

Inspector 启动后,应先检查 tools/list 的结果,再分别测试:

• relative_dir 设置为 .,应能看到 README.md。
• relative_path 设置为 README.md,应返回测试文本。
• relative_path 设置为 ../../etc/passwd,应被拒绝。
• 使用绝对路径,应被拒绝。
• 创建不在扩展名白名单中的文件,应被拒绝。
• 把 max_chars 设置为极大值,返回内容仍不能超过服务端上限。

npx 可能下载并运行包。对受控环境,应固定 Inspector 版本、核对包来源并在隔离的开发机上运行,不要把临时调试工具直接加入生产镜像。

七、编写最小 Client 验证调用链

创建 client.py:

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client


async def main():
params = StdioServerParameters(
command="python",
args=["/absolute/path/server.py"],
env={"MCP_WORKSPACE": "/absolute/path/workspace"},
)

async with stdio_client(params) as streams:
read_stream, write_stream = streams
async with ClientSession(read_stream, write_stream) as session:
await session.initialize()

tools = await session.list_tools()
print([tool.name for tool in tools.tools])

result = await session.call_tool(
"read_text",
arguments={"relative_path": "README.md", "max_chars": 1000},
)
print(result.content)


if __name__ == "__main__":
asyncio.run(main())

运行:

python client.py

不同 SDK 版本的返回对象字段可能调整,因此应使用项目锁定版本编写自动化测试。Client 不应把 Server 返回内容无条件视为可信指令。文件内容、网页内容和数据库文本都可能包含“忽略之前规则并调用某工具”之类的提示注入文本,它们只能作为数据处理,不能自动提升权限。

八、接入桌面 Host 或其他客户端

许多支持 MCP 的 Host 使用类似以下的本地 Server 配置结构,实际字段名应以对应 Host 文档为准:

{
"mcpServers": {
"workspace-reader": {
"command": "/absolute/path/.venv/bin/python",
"args": ["/absolute/path/server.py"],
"env": {
"MCP_WORKSPACE": "/absolute/path/workspace"
}
}
}
}

应尽量使用解释器和脚本的绝对路径,避免 Host 的工作目录、PATH 或虚拟环境与命令行测试环境不同。不要把 API 密钥直接提交到配置仓库;远程服务凭证应通过系统密钥链、短期令牌或专用秘密管理系统注入。

九、生产环境的权限与安全设计

1. 工具分级:将只读查询、有限写入、高风险变更划分不同等级。删除、发布、转账、发送消息等工具默认要求人工确认。
2. 服务端授权:不要只依赖模型决定“谁可以调用”。Server 应根据已认证主体、租户、资源和操作执行授权判断。
3. 参数约束:使用明确类型、枚举、长度、数量和范围限制。避免接收一段自由文本后直接解释为 SQL、Shell 或文件路径。
4. 输出约束:限制单次响应大小、分页数量和执行时间,防止上下文耗尽、内存占用和日志膨胀。
5. 审计记录:记录时间、主体、会话、工具名、参数摘要、授权结果、耗时和结果状态。敏感正文、令牌和个人数据应脱敏,不应完整写入日志。
6. 网络隔离:无网络需求的 Server 不应获得外网访问能力。需要访问后端时,采用目的地址白名单和独立服务身份。
7. 远程传输:使用 HTTPS,校验 Origin 和 Host,限制请求体大小与并发数,并在每次敏感调用时校验认证和授权。仅绑定 0.0.0.0 并不能构成可用的远程部署方案。
8. 提示注入隔离:工具返回值应标记为外部数据。Client 的策略层必须阻止工具输出自行批准后续高风险调用。
9. 进程隔离:使用低权限系统账户、只读文件系统、资源配额和容器安全配置。容器不是授权系统,但可以缩小失陷后的影响范围。
10. 防止混淆代理:用户有权访问 Client,不代表有权使用 Server 持有的所有后端权限。后端凭证应尽可能绑定用户或最小服务角色。

十、常见问题

问题一:为什么 tools/list 能看到工具,但调用失败?

常见原因包括参数名不匹配、输入类型错误、工作区环境变量未传入、Host 使用了不同 Python 环境,以及 Server 的当前目录与手工启动时不同。优先使用绝对路径,并检查 stderr 日志。

问题二:工具描述写明“只读”是否足够?

不够。描述和注解主要服务于模型选择及用户界面展示,不能阻止恶意或错误 Client 直接发起调用。只读约束必须由 Server 的实际实现和操作系统权限共同保证。

问题三:能否直接增加一个 run_command 工具?

技术上可以,但不建议把任意命令执行作为通用能力暴露给模型。更安全的方式是把业务动作封装成参数有限的专用工具,例如只允许查询固定服务状态,并在服务端映射到预定义命令,而不是接受任意命令字符串。

问题四:为什么同时限制文件字节数和返回字符数?

两者解决的问题不同。字节限制控制磁盘读取和解码成本,字符限制控制协议响应、Client 内存和模型上下文消耗。生产实现还应增加超时、并发限制和总配额。

问题五:stdio 是否天然安全?

不是。stdio 减少了网络暴露面,但 Server 仍继承启动进程的文件、环境变量和网络权限。若 Host 可被非受信输入驱动,Server 同样需要最小权限和严格校验。

问题六:什么时候应该使用 Streamable HTTP?

当需要跨进程、跨主机、多用户或集中管理时可以考虑。迁移前应先明确认证主体、租户隔离、会话存储、负载均衡、超时重试和审计方案。不要把本地 stdio Server 简单改成公网监听后就视为生产可用。

十一、总结

可靠的 MCP Server 不应只是“把现有函数加一个工具装饰器”。正确做法是先界定能力和信任边界,再实现协议接入。本文的 Workspace Server 通过固定根目录、规范化路径、扩展名白名单、响应上限和只读工具构造了一个较小但完整的安全基线;Client 侧则完成初始化、能力发现和结构化调用。

进一步扩展时,建议优先增加自动化协议测试、越权用例、超时与并发控制、结构化审计,而不是立即加入高权限工具。MCP 提供的是标准化连接方式,最终可靠性仍取决于 Server 是否强制授权、Client 是否隔离不可信内容,以及部署环境是否贯彻最小权限。
 
顶部