Security: close second-order Telegram parsing boundaries

README.md

@@ -295,14 +295,15 @@ is validated separately.
 - **入站只有一条窄协议** `ask:<nonce>:<index>`（tg_ask_choice 的按钮回调），
   单独校验：nonce 精确 fullmatch、index 必须落在选项区间内、私聊只认聊天对面
   那个人。Telegram 的消息内容**从不**被映射成 shell / 文件系统 / 任意工具调用。
-- **出站三道有界闸**（见下表）：blocks 结构有界、媒体目录可限、目标 chat 可限。
+- **四道边界闸**（见下表）：JSON-RPC 请求先限字节，blocks 再限结构，媒体目录与目标 chat 均可收紧。
   目标是 **bounding，不是复刻 Telegram 的 schema**——未知 block type 照样透传，
   我们不维护一份会跟 Telegram 漂移的白名单。
 
 | 闸 | env | 默认 | 作用 |
 |---|---|---|---|
-| blocks 有界结构闸 | 常开；`TG_RICH_BLOCKS_MAX_NODES` / `TG_RICH_BLOCKS_MAX_CHARS` 可调 | 深度 16 / 节点 2000 / 单数组 4096 / 单串 10 万 / 总字符 100 万 | 迭代遍历框住 blocks 的深度/节点/数组/字符串/总字符；dict 键必须是 str、只放行 JSON 兼容类型；map 经纬度·缩放、attach 索引、`file://` 本地 scheme 就地判死。拒绝一律在任何网络请求之前，报错不整段回显 payload。 |
-| 媒体目录白名单 | `TG_RICH_MEDIA_ROOTS`（多目录按 `os.pathsep` 分隔：**Unix `:` / Windows `;`**） | **未配＝不限目录** | 只允许发这些目录（及子目录）里的文件；按 `resolve()` 后的真实路径做父子判定（不是字符串前缀），symlink 借链也逃不出去。凭证文件名 guard 仍作第二层。 |
+| JSON-RPC 请求字节闸 | `TG_RICH_MAX_REQUEST_BYTES` | 4 MiB | 在 `json.loads` **之前**按字节有界读取；超长行只保留 `limit+1`，其余用固定窗口 drain 到换行后继续服务，避免巨型请求先把 parser/内存吃满。 |
+| blocks 有界结构闸 | 常开；`TG_RICH_BLOCKS_MAX_NODES` / `TG_RICH_BLOCKS_MAX_CHARS` 可调 | 深度 16 / 节点 2000 / 单数组 4096 / 单串 10 万 / 总字符 100 万 | 迭代遍历框住 blocks 的深度/节点/数组/字符串/总字符；dict 键必须是 str、只放行 JSON 兼容类型。`attach://` / `file:` 只在真正的 `media` / `url` 字段解释，普通 `text` / `pre` 里的字面量不误杀；map 经纬度·缩放仍就地判死。 |
+| 媒体目录白名单 | `TG_RICH_MEDIA_ROOTS`（多目录按 `os.pathsep` 分隔：**Unix `:` / Windows `;`**） | **未配＝不限目录** | 稳定路径下按 `resolve()` 后真实目标做父子判定，普通 symlink 越界会被拦；凭证文件名 guard 仍作第二层。**这不是 race-hard filesystem sandbox**：不承诺抵抗另一个本地进程在检查与 `open()` 之间并发替换路径组件（TOCTOU）。 |
 | chat 白名单 | `TG_RICH_ALLOWED_CHATS`（逗号分隔） | **未配＝不限** | 配了之后 send / edit / draft / sticker / ask 的目标 chat（**含默认 chat**）都必须在名单内，否则拒发；报错不泄露名单内容。 |
 
 > ⚠️ 两个「未配＝不限」是**刻意的默认兼容取舍**：老配置不动，行为与今日一字不差。

test_security_boundaries_r2.py

@@ -0,0 +1,133 @@
+import io
+import json
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+import tg_rich_mcp as mcp
+
+
+class SemanticStringScope(unittest.TestCase):
+    def test_plain_text_literals_are_not_interpreted(self):
+        mcp.guard_blocks([
+            {"type": "pre", "text": "file:///etc/passwd"},
+            {"type": "paragraph", "text": "example attach://f999999 only"},
+        ])
+
+    def test_media_attach_still_checked(self):
+        with self.assertRaisesRegex(ValueError, "attach"):
+            mcp.guard_blocks(
+                [{"type": "photo", "photo": {"media": "attach://f3"}}], media_count=1
+            )
+
+    def test_media_and_url_file_scheme_still_rejected(self):
+        cases = (
+            [{"type": "photo", "photo": {"media": "file:///etc/passwd"}}],
+            [{"type": "paragraph", "text": [
+                {"type": "url", "text": "x", "url": "file:///etc/passwd"}
+            ]}],
+        )
+        for blocks in cases:
+            with self.subTest(blocks=blocks):
+                with self.assertRaisesRegex(ValueError, "本地 scheme"):
+                    mcp.guard_blocks(blocks)
+
+    def test_valid_media_attach_requires_explicit_capability(self):
+        blocks = [{"type": "photo", "photo": {"media": "attach://f0"}}]
+        mcp.build_rich({"blocks": blocks}, media_count=1)
+        with self.assertRaisesRegex(ValueError, "attach"):
+            mcp.build_rich({"blocks": blocks})
+
+
+class MediaCapabilityBoundary(unittest.TestCase):
+    def test_edit_cannot_fake_media_paths(self):
+        args = {
+            "chat_id": "10001", "message_id": 1,
+            "blocks": [{"type": "photo", "photo": {"media": "attach://f0"}}],
+            "media_paths": ["/tmp/not-really-uploaded.jpg"],
+        }
+        with mock.patch.object(mcp, "call_api") as api:
+            with self.assertRaisesRegex(ValueError, "只支持 tg_rich_send"):
+                mcp.tool_edit(args)
+            api.assert_not_called()
+
+    def test_draft_cannot_fake_media_paths(self):
+        args = {
+            "chat_id": "10001", "draft_id": 1,
+            "blocks": [{"type": "photo", "photo": {"media": "attach://f0"}}],
+            "media_paths": ["/tmp/not-really-uploaded.jpg"],
+        }
+        with mock.patch.object(mcp, "call_api") as api:
+            with self.assertRaisesRegex(ValueError, "只支持 tg_rich_send"):
+                mcp.tool_draft(args)
+            api.assert_not_called()
+
+    def test_send_real_media_path_authorizes_attach(self):
+        # Windows 的 NamedTemporaryFile 默认持有独占句柄；load_media 再 open 会被拒。
+        # 用目录 + 已关闭普通文件，测试的是产品代码的跨平台读路径。
+        with tempfile.TemporaryDirectory() as td:
+            media_path = Path(td) / "pic.jpg"
+            media_path.write_bytes(b"jpg")
+            args = {
+                "chat_id": "10001",
+                "blocks": [{"type": "photo", "photo": {"media": "attach://f0"}}],
+                "media_paths": [str(media_path)],
+            }
+            with mock.patch.object(
+                mcp, "call_api", return_value={"result": {"message_id": 7}}
+            ) as api:
+                mcp.tool_send(args)
+            self.assertEqual(api.call_args.args[0], "sendRichMessage")
+            self.assertIn("f0", api.call_args.kwargs["files"])
+
+
+class TransportByteBound(unittest.TestCase):
+    def test_oversized_line_is_drained_and_next_request_survives(self):
+        stream = io.BytesIO(b"x" * 200_000 + b"\n" + b'{"jsonrpc":"2.0"}\n')
+        line, oversized = mcp._read_request_line(stream, 64)
+        self.assertTrue(oversized)
+        self.assertEqual(line, b"")
+        line, oversized = mcp._read_request_line(stream, 64)
+        self.assertFalse(oversized)
+        self.assertEqual(line, b'{"jsonrpc":"2.0"}\n')
+
+    def test_limit_counts_utf8_bytes(self):
+        stream = io.BytesIO("猫猫\n".encode("utf-8"))
+        _line, oversized = mcp._read_request_line(stream, 6)
+        self.assertTrue(oversized)
+
+    def test_main_returns_error_then_continues(self):
+        oversized = b"x" * 100 + b"\n"
+        ping = b'{"jsonrpc":"2.0","id":9,"method":"ping"}\n'
+
+        class FakeStdin:
+            def __init__(self, data: bytes):
+                self.buffer = io.BytesIO(data)
+
+        out = io.StringIO()
+        with mock.patch.dict(os.environ, {"TG_RICH_MAX_REQUEST_BYTES": "64"}), \
+             mock.patch.object(mcp.sys, "stdin", FakeStdin(oversized + ping)), \
+             mock.patch.object(mcp.sys, "stdout", out):
+            mcp.main()
+        responses = [json.loads(line) for line in out.getvalue().splitlines()]
+        self.assertEqual(responses[0]["error"]["code"], -32600)
+        self.assertEqual(responses[1]["id"], 9)
+        self.assertEqual(responses[1]["result"], {})
+
+    def test_blocks_string_is_bounded_before_second_json_parse(self):
+        raw = '[{"type":"paragraph","text":"' + ("x" * 1000) + '"}]'
+        with mock.patch.dict(os.environ, {"TG_RICH_MAX_REQUEST_BYTES": "128"}), \
+             mock.patch.object(mcp.json, "loads", wraps=mcp.json.loads) as loads:
+            with self.assertRaisesRegex(ValueError, "解析前字节上限"):
+                mcp.build_rich({"blocks": raw})
+            loads.assert_not_called()
+
+    def test_bad_env_falls_back(self):
+        with mock.patch.dict(os.environ, {"TG_RICH_MAX_REQUEST_BYTES": "nope"}):
+            self.assertEqual(mcp._request_max_bytes(), mcp.REQUEST_MAX_BYTES_DEFAULT)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)