feat: v0.2.0 文件传输(MCP 工具 + HTTP 直传)与安全/正确性修复

文件功能: - 新增 file_read / file_write / file_stat 工具(文本/base64/图片,分块读取,原子替换写入,SHA-256) - 新增 file_link 工具与 GET/PUT /files 端点:HMAC 签名限时链接或 Bearer 认证, 支持 Range、HEAD、chunked、Expect: 100-continue,大文件不经过模型上下文 - 新增 --file-root / --max-upload;配置白名单但未限定目录时文件功能自动禁用安全: - 白名单改为程序名精确匹配,shell 模式禁止元字符,shell_write 输入逐行受检;黑名单检查每一段 - HTTP 签发 Mcp-Session-Id,按会话隔离 shell 会话与取消,支持 DELETE 终止会话 - HTTP 客户端不能再用 notifications/exit 关停服务;token 常量时间比较正确性: - 后台进程占用输出管道时 run_command 不再挂起(detached=true) - shell_read 改用绝对偏移,修复缓冲裁剪后错位;被信号杀死的会话能报告退出;阻塞读可取消 - 输出与分块读取不再切断 UTF-8 多字节字符 - shell_* 结果改为标准 content + structuredContent - POST + SSE 的 tools/call 在同一响应流中推送进度与结果 - /mcp 支持 Expect: 100-continue(curl 发送大请求时不再白等 1 秒) - CLI 参数优先于环境变量;SIGTERM 在 stdio 模式下也能清理退出;Windows 支持 Ctrl+C 清理与优雅退出资源:并发工具调用 / HTTP 连接 / shell 会话数量上限,空闲会话自动回收 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-18 10:23:15 +08:00
parent 63d48ec951
commit 84653bec61
9 changed files with 3083 additions and 752 deletions
@@ -1,9 +1,13 @@
 #!/usr/bin/env python3
 """bolt HTTP 模式端到端测试"""
-import http.client, json, subprocess, threading, time
+import hashlib, http.client, json, os, shutil, subprocess, sys, tempfile, threading, time
+from urllib.parse import urlsplit

-BOLT = "/root/bolt/target/release/bolt"
+# 二进制路径:环境变量 BOLT_BIN 优先,默认为仓库内的 release 产物
+BOLT = os.environ.get("BOLT_BIN") or os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "target", "release", "bolt")
 PORT = 18080
+PORT2 = 18081  # 带 token + --file-root 的第二个实例
+TOKEN = "e2e-secret"
 passed = failed = 0
 def check(name, cond, detail=""):
    global passed, failed
@@ -14,26 +18,50 @@ def check(name, cond, detail=""):
        failed += 1
        print(f"  FAIL {name} {detail}")

+tmpdir = tempfile.mkdtemp(prefix="bolt-http-e2e-")
 proc = subprocess.Popen([BOLT, "--http", f"127.0.0.1:{PORT}"], stderr=subprocess.PIPE)
+proc2 = subprocess.Popen([BOLT, "--http", f"127.0.0.1:{PORT2}", "--token", TOKEN, "--file-root", tmpdir], stderr=subprocess.PIPE)
 time.sleep(0.5)

-def post(obj, auth=None, origin=None, accept="application/json", path="/mcp"):
-    c = http.client.HTTPConnection("127.0.0.1", PORT, timeout=30)
+def request(method, path, body=None, headers=None, port=PORT, timeout=30):
+    c = http.client.HTTPConnection("127.0.0.1", port, timeout=timeout)
+    c.request(method, path, body=body, headers=headers or {})
+    r = c.getresponse()
+    data = r.read()
+    hdrs = {k.lower(): v for k, v in r.getheaders()}
+    c.close()
+    return r.status, hdrs, data
+
+def post(obj, auth=None, origin=None, accept="application/json", path="/mcp", sid=None, port=PORT):
    headers = {"Content-Type": "application/json", "Accept": accept}
    if auth: headers["Authorization"] = auth
    if origin: headers["Origin"] = origin
-    body = json.dumps(obj)
-    c.request("POST", path, body=body, headers=headers)
-    r = c.getresponse()
-    data = r.read()
-    ct = r.getheader("Content-Type", "")
-    c.close()
-    return r.status, ct, data
+    if sid: headers["Mcp-Session-Id"] = sid
+    st, hdrs, data = request("POST", path, json.dumps(obj), headers, port=port)
+    return st, hdrs.get("content-type", ""), data

-# 1. initialize(JSON 响应)
-st, ct, data = post({"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "t", "version": "1"}}})
+def post_h(obj, sid=None, accept="application/json", port=PORT, auth=None):
+    """返回 (status, headers, data)"""
+    headers = {"Content-Type": "application/json", "Accept": accept}
+    if sid: headers["Mcp-Session-Id"] = sid
+    if auth: headers["Authorization"] = auth
+    return request("POST", "/mcp", json.dumps(obj), headers, port=port)
+
+def call(name, args, rid=100, sid=None, port=PORT, auth=None):
+    st, _, data = post_h({"jsonrpc": "2.0", "id": rid, "method": "tools/call", "params": {"name": name, "arguments": args}}, sid=sid, port=port, auth=auth)
+    return json.loads(data)
+
+def sse_events(data):
+    return [json.loads(line[len(b"data: "):]) for line in data.split(b"\r\n") if line.startswith(b"data: ")]
+
+INIT = {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "t", "version": "1"}}}
+
+# 1. initialize(JSON 响应 + 签发 Mcp-Session-Id)
+st, hdrs, data = post_h(INIT)
 m = json.loads(data)
-check("init json", st == 200 and "application/json" in ct and m["result"]["serverInfo"]["name"] == "bolt", f"{st} {ct} {data[:100]}")
+sid_a = hdrs.get("mcp-session-id")
+check("init json", st == 200 and "application/json" in hdrs.get("content-type", "") and m["result"]["serverInfo"]["name"] == "bolt", f"{st} {data[:100]}")
+check("init issues session id", bool(sid_a) and len(sid_a) == 32, str(hdrs))

 # 2. SSE 响应(accept 含 text/event-stream)
 st, ct, data = post({"jsonrpc": "2.0", "id": 2, "method": "ping"}, accept="application/json, text/event-stream")
@@ -51,8 +79,7 @@ m = json.loads(data)
 sc = m["result"]["structuredContent"]
 check("run over http", sc["exit_code"] == 5 and sc["stdout"].strip() == "HTTP_OK", str(sc))

-# 5. GET SSE 流:开流收心跳,并发命令进度通知广播
-got_event = threading.Event()
+# 5. GET SSE 流:开流收心跳,同作用域命令的进度通知推送到流上
 got_heartbeat = threading.Event()
 msgs = []
 def sse_get():
@@ -71,6 +98,7 @@ def sse_get():
            buf = b""
            if msgs and (": ping" in msgs[-1]):
                got_heartbeat.set()
+                break
    c.close()
 t = threading.Thread(target=sse_get)
 t.start()
@@ -80,20 +108,29 @@ st, ct, data = post({"jsonrpc": "2.0", "id": 4, "method": "tools/call", "params"
 m = json.loads(data)
 check("long cmd done", m["result"]["structuredContent"]["stdout"].count("h-") == 4, str(m)[:100])
 saw_progress = any("notifications/progress" in x and "h1" in x for x in msgs)
-check("sse got progress broadcast", saw_progress, str(msgs)[:200])
+check("sse got progress on GET stream", saw_progress, str(msgs)[:200])
 t.join(timeout=25)
-saw_hb = got_heartbeat.is_set()
-check("sse heartbeat", saw_hb, str(msgs)[:200])
+check("sse heartbeat", got_heartbeat.is_set(), str(msgs)[:200])

-# 6. 取消:连接 A 长命令,连接 B 发 cancelled
-st, ct, data = post({"jsonrpc": "2.0", "id": 5, "method": "tools/call", "params": {"name": "run_command", "arguments": {"command": "echo nothing"}}})  # 预热(连接池无关,每次新连接)
+# 5b. POST + SSE:进度通知与结果在同一个响应流里
+st, hdrs, data = post_h({"jsonrpc": "2.0", "id": 41, "method": "tools/call", "params": {"name": "run_command", "arguments": {"command": "for x in 1 2 3; do echo p-$x; sleep 0.3; done"}, "_meta": {"progressToken": "p1"}}}, accept="application/json, text/event-stream", sid=sid_a)
+evs = sse_events(data)
+prog = [e for e in evs if e.get("method") == "notifications/progress" and e["params"]["progressToken"] == "p1"]
+final = [e for e in evs if e.get("id") == 41]
+check("post sse progress + result", st == 200 and prog and final and final[0]["result"]["structuredContent"]["stdout"].count("p-") == 3, str(evs)[:300])
+check("post sse result is last", evs and evs[-1].get("id") == 41, str(evs)[-200:])

+# 6. 取消:连接 A 长命令,连接 B 发 cancelled(同一作用域)
 def long_cmd():
    return post({"jsonrpc": "2.0", "id": 6, "method": "tools/call", "params": {"name": "run_command", "arguments": {"command": "sleep 30"}}})
 res = {}
 t2 = threading.Thread(target=lambda: res.update(r=long_cmd()))
 t2.start()
 time.sleep(0.5)
+# 其他会话发同 id 的取消不能误杀
+st, _, _ = post({"jsonrpc": "2.0", "method": "notifications/cancelled", "params": {"requestId": 6}}, sid=sid_a)
+time.sleep(0.3)
+check("cancel from other scope ignored", t2.is_alive())
 st, ct, data = post({"jsonrpc": "2.0", "method": "notifications/cancelled", "params": {"requestId": 6}})
 check("cancel notif 202", st == 202)
 t2.join(timeout=10)
@@ -112,24 +149,16 @@ check("http concurrent", json.loads(results[10][2])["result"]["structuredContent
 # 8. 错误路径
 st, _, _ = post({"jsonrpc": "2.0", "id": 99, "method": "nope"})
 check("unknown method", st == 200)
-c = http.client.HTTPConnection("127.0.0.1", PORT, timeout=5)
-c.request("POST", "/mcp", body=b"this is {not json", headers={"Content-Type": "application/json", "Accept": "application/json"})
-r = c.getresponse(); bd = r.read(); c.close()
-check("bad json -> 400", r.status == 400, str(r.status))
+st, _, _ = request("POST", "/mcp", b"this is {not json", {"Content-Type": "application/json", "Accept": "application/json"})
+check("bad json -> 400", st == 400, str(st))
 st, _, data = post({"jsonrpc": "2.0", "id": 99, "method": "tools/call", "params": {"name": "nope", "arguments": {}}})
 check("unknown tool -32602", json.loads(data)["error"]["code"] == -32602)
-c = http.client.HTTPConnection("127.0.0.1", PORT, timeout=5)
-c.request("GET", "/other", headers={"Accept": "text/event-stream"})
-r = c.getresponse(); r.read(); c.close()
-check("404 other path", r.status == 404)
-c = http.client.HTTPConnection("127.0.0.1", PORT, timeout=5)
-c.request("PUT", "/mcp")
-r = c.getresponse(); r.read(); c.close()
-check("405 method", r.status == 405)
-c = http.client.HTTPConnection("127.0.0.1", PORT, timeout=5)
-c.request("OPTIONS", "/mcp")
-r = c.getresponse(); r.read(); c.close()
-check("options cors", r.status == 204 and r.getheader("Access-Control-Allow-Origin") == "*")
+st, _, _ = request("GET", "/other", headers={"Accept": "text/event-stream"})
+check("404 other path", st == 404)
+st, _, _ = request("PUT", "/mcp")
+check("405 method", st == 405)
+st, hdrs, _ = request("OPTIONS", "/mcp")
+check("options cors", st == 204 and hdrs.get("access-control-allow-origin") == "*" and "Mcp-Session-Id" in hdrs.get("access-control-allow-headers", ""))

 # 9. origin 校验
 st, ct, data = post({"jsonrpc": "2.0", "id": 50, "method": "ping"}, origin="https://evil.example.com")
@@ -144,10 +173,81 @@ r = c.getresponse()
 c.close()
 check("get sse default", r.status == 200 and "text/event-stream" in r.getheader("Content-Type"))

+# 11. MCP 会话:隔离、未知 ID → 404、DELETE 终止
+st, hdrs, _ = post_h(INIT)
+sid_b = hdrs.get("mcp-session-id")
+r = call("shell_open", {}, sid=sid_a)
+shell_id = r["result"]["structuredContent"]["session_id"]
+r = call("shell_read", {"session_id": shell_id, "timeout_ms": 10}, sid=sid_b)
+check("shell session isolated", r["result"].get("isError") is True, str(r))
+r = call("shell_read", {"session_id": shell_id, "timeout_ms": 10}, sid=sid_a)
+check("shell session owner ok", r["result"].get("isError") is None, str(r))
+st, _, _ = post_h({"jsonrpc": "2.0", "id": 60, "method": "ping"}, sid="0" * 32)
+check("unknown session 404", st == 404, str(st))
+st, _, _ = request("DELETE", "/mcp", headers={"Mcp-Session-Id": sid_a})
+check("delete session", st == 200, str(st))
+st, _, _ = post_h({"jsonrpc": "2.0", "id": 61, "method": "ping"}, sid=sid_a)
+check("deleted session 404", st == 404, str(st))
+r = call("shell_read", {"session_id": shell_id, "timeout_ms": 10}, sid=sid_b)
+check("deleted session shells closed", r["result"].get("isError") is True)
+
+# 12. HTTP 客户端不能用 notifications/exit 关停服务
+post({"jsonrpc": "2.0", "method": "notifications/exit"})
+time.sleep(0.4)
+st, _, _ = post({"jsonrpc": "2.0", "id": 70, "method": "ping"})
+check("exit notif ignored over http", st == 200 and proc.poll() is None)
+
+# 13. tools/list 含 file_link
+st, _, data = post({"jsonrpc": "2.0", "id": 71, "method": "tools/list"})
+names = [t["name"] for t in json.loads(data)["result"]["tools"]]
+check("tools/list http", names[-4:] == ["file_read", "file_write", "file_stat", "file_link"], str(names))
+
+# 14. /files:Bearer 认证、签名链接、Range、chunked 上传、目录限制
+AUTH = {"Authorization": f"Bearer {TOKEN}"}
+fpath = os.path.join(tmpdir, "data.bin")
+payload = os.urandom(300_000)
+# 未认证请求在读取 body 前即返回 401 并关闭连接,这里用小 body 避免客户端写入时 broken pipe
+st, _, _ = request("PUT", f"/files?path={fpath}", b"x", port=PORT2)
+check("files put needs token", st == 401 and not os.path.exists(fpath), str(st))
+st, _, data = request("PUT", f"/files?path={fpath}", payload, AUTH, port=PORT2)
+info = json.loads(data)
+check("files put", st == 201 and info["sha256"] == hashlib.sha256(payload).hexdigest(), f"{st} {data[:200]}")
+st, hdrs, data = request("GET", f"/files?path={fpath}", headers=AUTH, port=PORT2)
+check("files get", st == 200 and data == payload and hdrs.get("accept-ranges") == "bytes")
+st, hdrs, data = request("GET", f"/files?path={fpath}", headers={**AUTH, "Range": "bytes=100-199"}, port=PORT2)
+check("files range", st == 206 and data == payload[100:200] and hdrs.get("content-range") == f"bytes 100-199/{len(payload)}", f"{st} {hdrs}")
+st, _, _ = request("GET", "/files?path=/etc/hosts", headers=AUTH, port=PORT2)
+check("files outside root 403", st == 403, str(st))
+# chunked 上传(http.client 对迭代器 body 使用 chunked 编码)
+cpath = os.path.join(tmpdir, "chunked.txt")
+c = http.client.HTTPConnection("127.0.0.1", PORT2, timeout=10)
+c.request("PUT", f"/files?path={cpath}&mkdirs=1", body=iter([b"hello ", b"chunked ", b"world"]), headers=AUTH, encode_chunked=True)
+r = c.getresponse(); r.read(); c.close()
+check("files chunked put", r.status == 201 and open(cpath, "rb").read() == b"hello chunked world", str(r.status))
+# 签名链接:不带 token 也能访问,且仅限签名的路径/方法
+st, _, data = post_h(INIT, port=PORT2, auth=AUTH["Authorization"])
+r = call("file_link", {"path": fpath}, port=PORT2, auth=AUTH["Authorization"])
+url = r["result"]["structuredContent"]["url"]
+u = urlsplit(url)
+check("file_link url", u.netloc == f"127.0.0.1:{PORT2}" and u.path == "/files" and "curl" in r["result"]["structuredContent"]["curl"], url)
+st, _, data = request("GET", f"{u.path}?{u.query}", port=PORT2)
+check("signed get", st == 200 and data == payload, str(st))
+st, _, _ = request("PUT", f"{u.path}?{u.query}", b"x", port=PORT2)
+check("signed get cannot put", st == 403, str(st))
+r = call("file_link", {"path": os.path.join(tmpdir, "up", "new.txt"), "method": "PUT", "mkdirs": True}, port=PORT2, auth=AUTH["Authorization"])
+u = urlsplit(r["result"]["structuredContent"]["url"])
+st, _, data = request("PUT", f"{u.path}?{u.query}", b"uploaded via link", {"Expect": "100-continue"}, port=PORT2)
+check("signed put", st == 201 and open(os.path.join(tmpdir, "up", "new.txt"), "rb").read() == b"uploaded via link", f"{st} {data[:200]}")
+st, _, _ = request("GET", f"{u.path}?{u.query.replace('sig=', 'sig=0')}", port=PORT2)
+check("bad signature 403", st == 403, str(st))
+r = call("file_link", {"path": "/etc/hosts"}, port=PORT2, auth=AUTH["Authorization"])
+check("file_link outside root", r["result"].get("isError") is True)
+
 proc.kill()
+proc2.kill()
 err = proc.stderr.read().decode()
+shutil.rmtree(tmpdir, ignore_errors=True)
 print(f"\n=== HTTP: {passed} passed, {failed} failed ===")
-if err.strip():
+if failed and err.strip():
    print("bolt stderr:", err[:400])
-sys_exit = 1 if failed else 0
-import sys; sys.exit(sys_exit)
+sys.exit(1 if failed else 0)