查询 37 次,耗时 0.2255 秒

从零实现一个安全可控的 MCP 工具服务:协议流程、Python Server/Client 与权限边界

speedhacker

New member

speedhacker Rep
0
0
0
Rep
0
speedhacker Vouches
Vouches
0
4 MONTHS
4 4 MONTHS OF SERVICE
speedhacker 获得
0
0
0
获得
0
speedhacker 凭证
0
0
0
凭证
0
主题
1
点赞
0
HakB
4 月
在社区成长 4 月
LEVEL 100 XP
 
一、背景:MCP 解决的到底是什么问题

MCP(Model Context Protocol)用于在模型应用与外部能力之间建立标准化连接。这里的外部能力可以是文件、数据库、内部 API、搜索服务,也可以是经过封装的业务操作。与为每个模型客户端单独编写插件相比,MCP 将双方的职责拆分得更清楚:

1. Host 是用户直接使用的模型应用,负责会话、授权提示和整体安全策略。
2. Client 运行在 Host 一侧,与某个 MCP Server 建立会话。
3. Server 暴露工具、资源或提示模板,但不应假设自己天然拥有无限权限。

一个常见误区是把 MCP 理解为“让模型自由调用本机命令”。实际上,MCP 只是通信协议,真正的安全性取决于 Server 暴露了什么能力、以什么系统身份运行,以及 Client 是否在调用前执行授权和参数检查。

本文实现一个本地笔记查询 Server。它只允许读取指定目录中的 Markdown 文件,不提供任意路径读取和命令执行能力,并使用独立 Client 验证初始化、工具发现和调用流程。所有测试均在本地目录中完成。

二、协议原理与一次完整调用的生命周期

MCP 消息基于 JSON-RPC 2.0。一次正常连接通常包含以下阶段:

1. Client 发送 initialize,声明其支持的协议版本、客户端信息和能力。
2. Server 返回协商后的协议版本、服务端信息及能力。
3. Client 发送 initialized 通知,表示初始化完成。
4. Client 使用 tools/list、resources/list 或 prompts/list 发现服务能力。
5. Client 使用 tools/call 等方法执行具体操作。
6. 会话结束后关闭底层传输及相关资源。

工具定义至少包括名称、描述和输入结构。描述不是装饰文字:模型通常会依据描述判断何时调用工具,因此应写清用途、限制和副作用。例如,“删除文件”不能只写成“管理文件”,而应明确说明会产生不可逆修改,并由 Host 在调用前要求用户确认。

本地集成最常见的是 stdio 传输:Client 启动 Server 子进程,并通过标准输入、标准输出交换协议消息。远程部署通常使用 Streamable HTTP。较旧实现可能仍采用历史上的 HTTP+SSE 方式,接入时应先核对双方 SDK 和协议版本,不要在业务代码中硬编码某个版本字符串。使用官方或兼容 SDK,可以避免手工处理消息关联、初始化顺序和传输细节。

三、前置条件

本文示例需要:

- Python 3.10 或更高版本;
- 一个独立虚拟环境;
- 本地可读的测试笔记目录;
- Node.js 可选,仅在使用 MCP Inspector 调试时需要。

创建实验环境:

mkdir mcp-safe-notes
cd mcp-safe-notes
python -m venv .venv

Linux 或 macOS 激活环境:

source .venv/bin/activate

Windows PowerShell 激活环境:

.venv\Scripts\Activate.ps1

安装 Python MCP SDK:

python -m pip install --upgrade pip
python -m pip install "mcp[cli]"

创建测试数据:

mkdir notes
printf "# MCP\nMCP Server 可以暴露 tools、resources 和 prompts。\n" > notes/mcp.md
printf "# 安全\n服务端必须自行实施目录白名单和参数校验。\n" > notes/security.md

Windows 用户也可以直接用编辑器创建上述两个 UTF-8 文件。

四、实现带目录边界的 MCP Server

新建 server.py:

import logging
import os
from pathlib import Path

from mcp.server.fastmcp import FastMCP

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("safe-notes")

mcp = FastMCP("safe-notes")

raw_base = os.environ.get("MCP_NOTES_DIR", "./notes")
BASE_DIR = Path(raw_base).expanduser().resolve(strict=True)
MAX_FILE_BYTES = 1024 * 1024

def confined_file(relative_path: str) -> Path:
if not relative_path or "\x00" in relative_path:
raise ValueError("文件路径为空或包含非法字符")

candidate = (BASE_DIR / relative_path).resolve(strict=True)
try:
candidate.relative_to(BASE_DIR)
except ValueError as exc:
raise ValueError("路径超出允许的笔记目录") from exc

if not candidate.is_file():
raise ValueError("目标不是普通文件")
if candidate.suffix.lower() != ".md":
raise ValueError("只允许读取 Markdown 文件")
if candidate.stat().st_size > MAX_FILE_BYTES:
raise ValueError("文件超过 1 MiB 限制")
return candidate

@mcp.tool()
def list_notes(limit: int = 50) -> dict:
"""列出授权目录中的 Markdown 笔记,不读取文件内容。limit 范围为 1 到 100。"""
limit = max(1, min(limit, 100))
items = []
for item in sorted(BASE_DIR.rglob("*.md")):
try:
resolved = item.resolve(strict=True)
resolved.relative_to(BASE_DIR)
except (OSError, ValueError):
continue
if resolved.is_file():
items.append(str(resolved.relative_to(BASE_DIR)))
if len(items) >= limit:
break
return {"base": BASE_DIR.name, "notes": items, "count": len(items)}

@mcp.tool()
def read_note(path: str) -> dict:
"""读取授权目录中的一个 UTF-8 Markdown 文件。path 必须来自 list_notes 的结果。"""
target = confined_file(path)
text = target.read_text(encoding="utf-8")
return {
"path": str(target.relative_to(BASE_DIR)),
"content": text,
"characters": len(text),
}

@mcp.tool()
def search_notes(query: str, limit: int = 20) -> dict:
"""在授权目录的 Markdown 文件中进行不区分大小写的纯文本搜索。"""
query = query.strip()
if not query:
raise ValueError("query 不能为空")
if len(query) > 200:
raise ValueError("query 不能超过 200 个字符")

limit = max(1, min(limit, 50))
needle = query.casefold()
matches = []

for item in sorted(BASE_DIR.rglob("*.md")):
try:
resolved = item.resolve(strict=True)
resolved.relative_to(BASE_DIR)
if not resolved.is_file() or resolved.stat().st_size > MAX_FILE_BYTES:
continue
text = resolved.read_text(encoding="utf-8")
except (OSError, UnicodeError, ValueError):
continue

for line_number, line in enumerate(text.splitlines(), start=1):
if needle in line.casefold():
matches.append({
"path": str(resolved.relative_to(BASE_DIR)),
"line": line_number,
"text": line[:500],
})
if len(matches) >= limit:
return {"query": query, "matches": matches, "truncated": True}

return {"query": query, "matches": matches, "truncated": False}

if __name__ == "__main__":
mcp.run(transport="stdio")

这段实现有几个关键点。

第一,目录边界必须由 Server 自己执行。即使 Host 向 Server 提供了 roots 等范围信息,也不能将其当作操作系统级沙箱。Server 仍需对最终解析后的真实路径做白名单判断。

第二,仅检查字符串是否以某个目录开头是不安全的。例如 /srv/note 与 /srv/notes-archive 可能产生错误匹配,../ 和符号链接也会改变真实目标。示例先调用 resolve,再使用 relative_to 判断候选路径是否确实位于 BASE_DIR 内。

第三,工具能力采用白名单设计。示例没有提供 shell(command)、read_any_file(path) 之类的通用高权限工具。需要新增写入功能时,建议单独定义 create_note 或 update_note,并限制文件名、扩展名、大小和覆盖策略。

五、运行与调试 Server

直接运行:

MCP_NOTES_DIR="$PWD/notes" python server.py

stdio 模式下,进程看起来可能没有输出,这是正常现象。协议数据占用标准输入和标准输出,因此不要使用 print 向 stdout 输出调试信息,否则可能破坏消息流。日志应写入 stderr,Python logging 默认即可满足这一点。

使用 MCP Inspector 检查工具列表和参数结构:

npx -y @modelcontextprotocol/inspector python server.py

如果 Server 依赖 MCP_NOTES_DIR,应在启动 Inspector 前设置环境变量,或者把示例的默认目录改成绝对路径。Inspector 仅用于本地或授权环境调试,不应将带有真实凭据的服务暴露到公共网络。

如果要接入支持 MCP 的桌面 Host,配置通常包含 command、args 和 env 三部分。概念示例如下:

{
"command": "/absolute/path/.venv/bin/python",
"args": ["/absolute/path/server.py"],
"env": {
"MCP_NOTES_DIR": "/absolute/path/notes"
}
}

实际配置文件位置和字段外层结构由具体 Host 决定。排障时优先使用绝对路径,因为桌面程序启动子进程时的工作目录和终端往往不同。

六、编写独立 MCP Client 验证调用

新建 client.py:

import asyncio
import sys

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

async def main():
params = StdioServerParameters(
command=sys.executable,
args=["server.py"],
env={"MCP_NOTES_DIR": "./notes"},
)

async with stdio_client(params) as (read_stream, write_stream):
async with ClientSession(read_stream, write_stream) as session:
init_result = await session.initialize()
print("Server:", init_result.serverInfo)

tools = await session.list_tools()
print("Tools:", [tool.name for tool in tools.tools])

result = await session.call_tool(
"search_notes",
arguments={"query": "MCP", "limit": 10},
)
print("Result:", result)

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

执行:

python client.py

这个 Client 明确执行 initialize 后再列举和调用工具。不要假定连接建立就等于协议初始化完成,也不要跳过能力协商后直接发送任意方法。面向生产的 Client 还应增加调用超时、取消处理、Server 退出检测,以及对返回内容大小的限制。

七、权限与安全设计清单

1. 最小权限运行

Server 进程应使用专用低权限账户,只挂载必要目录。即使路径校验存在缺陷,操作系统权限也应构成第二道边界。容器不是自动安全边界;仍需使用只读挂载、非 root 用户、能力裁剪和网络出口限制。

2. 将工具返回视为不可信输入

文件、网页和第三方 API 的内容可能包含提示注入文本,例如要求模型忽略用户指令或调用其他高风险工具。Host 不应因为这些内容出现在工具结果中,就自动扩大权限或执行其中的操作。数据内容与控制指令必须分离。

3. 高风险调用需要确认

读取公开知识库通常可以低摩擦处理,但发送邮件、转账、删除数据、部署代码等有外部副作用的操作,应向用户展示目标、关键参数和预计影响,并在确认后执行。确认应绑定到具体参数,不能用一次模糊授权覆盖后续所有调用。

4. 凭据不能交给模型转述

远程 Server 的访问令牌应由 Host 或安全的凭据代理注入传输层,不应放进提示词、工具描述或普通工具参数。日志中也要过滤 Authorization、Cookie、API Key 和个人数据。

5. 远程传输必须逐请求鉴权

采用 Streamable HTTP 时,应使用 TLS、验证授权信息、限制请求体大小、设置超时和并发上限。不能仅依赖一个可猜测的会话标识。服务还应按规范要求处理 Origin 校验,避免本机服务受到 DNS rebinding 等攻击;仅供本机使用的服务应绑定回环地址,而不是 0.0.0.0。

6. 输出与审计

记录工具名称、调用时间、授权主体、结果状态和耗时即可。除非有明确合规需求,不要完整记录文件正文和敏感参数。审计日志应防篡改,并配置保留期限。

八、常见问题与排查

问题一:Host 显示 Server 启动后立即退出。

先在同一虚拟环境中手动执行 python server.py,检查依赖和路径。桌面 Host 常使用不同的 PATH,因此 command 应填写虚拟环境 Python 的绝对路径。同时确认 server.py 的绝对路径正确。

问题二:出现 JSON 解析错误或连接无响应。

检查是否有 print、第三方库横幅或异常栈写入 stdout。stdio Server 的 stdout 必须留给协议消息,普通日志一律写 stderr。还应确认 Client 与 Server 使用兼容的 SDK和传输方式。

问题三:工具能列出,但调用参数总是失败。

查看 tools/list 返回的输入结构,核对参数名称和类型。不要把 JSON 字符串当作 arguments 对象传入。例如应传 {"limit": 10},而不是 "{\"limit\":10}"。服务端仍要独立校验范围,不能只依赖 JSON Schema。

问题四:配置了目录范围,为什么仍能读到符号链接之外的文件?

只检查输入路径前缀无法防御符号链接逃逸。应对候选路径执行真实路径解析,再确认解析结果位于授权目录中。对于安全要求更高的写入场景,还要考虑检查与使用之间的竞态条件,必要时使用操作系统提供的目录文件描述符和禁止跟随符号链接选项。

问题五:是否应该直接提供通用 shell 工具?

通常不应该。通用 shell 很难建立稳定的参数和权限边界。更合理的做法是把确实需要的操作封装为窄接口,例如 get_service_status、restart_test_service,并固定目标集合、超时时间和可执行程序。涉及重启或修改时再增加用户确认和审计。

问题六:升级 SDK 后 Client 无法连接旧 Server。

先核对传输类型与协议版本协商结果。不要在应用中绕过 initialize 或强制覆盖服务端协商值。对远程服务升级时,应先在测试环境验证 tools/list、结构化输出、错误返回和断线重连,并为旧客户端保留明确的迁移窗口。

九、总结

一个可靠的 MCP 接入不只是“让工具能被模型调用”。完整实现至少包括正确的初始化和能力发现、清晰且窄化的工具定义、服务端强制权限边界、客户端授权策略、传输层鉴权,以及可审计但不过度收集的日志。

本文的安全笔记 Server 适合作为最小基线:它使用 stdio 完成本地通信,通过真实路径解析限制文件范围,并对文件类型、大小、查询长度和结果数量设置上限。继续扩展时,建议保持“一项业务动作对应一个明确工具”的原则,不要为了省代码退化成任意文件、任意 URL 或任意命令接口。MCP 提供了统一协议,但最小权限和用户知情确认仍必须由 Server、Client 与 Host 共同落实。
 

1371340

New member

1371340 Rep
0
0
0
Rep
0
1371340 Vouches
Vouches
0
1 MONTH
1 1 MONTH OF SERVICE
1371340 获得
0
0
0
获得
0
1371340 凭证
0
0
0
凭证
0
主题
2
点赞
0
HakB
1 月
在社区成长 1 月
LEVEL 1 1 XP
 
建议补充两个边界测试。其一是符号链接竞态:当前先 resolve 校验、再 read_text,校验后文件可能被替换为指向目录外的链接。高风险场景可使用基于目录文件描述符的 openat,并启用 O_NOFOLLOW,打开后再用 fstat 校验普通文件及大小。其二是 stdio 传输下严禁向 stdout 打日志,否则会污染 JSON-RPC 帧;应统一写 stderr。测试用例至少覆盖 ../ 越界、绝对路径、目录内外符号链接、超大文件、非法 UTF-8,以及调用超时和 Server 异常退出时 Client 能否正确回收子进程。
 
顶部