From bb3dbcb06ad24d272c3ecf4a3ab70277bc033846 Mon Sep 17 00:00:00 2001 From: chenwei Date: Thu, 30 Jul 2026 16:29:44 +0000 Subject: [PATCH] =?UTF-8?q?fix(mcp):=20decode=5Fimage=5Fsource=20=E9=9D=9E?= =?UTF-8?q?=E5=9B=BE=E7=89=87=E8=BE=93=E5=85=A5=20OSError=20=E5=BD=92?= =?UTF-8?q?=E4=B8=80=20ValueError=EF=BC=8C=E5=AE=88=20INVALID=5FPARAM=20?= =?UTF-8?q?=E5=A5=91=E7=BA=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 评审 fix round 1：Image.open 两处（路径/base64 分支）包 try/except OSError， 合法 base64 或存在文件但非可识别图片时不再让 PIL.UnidentifiedImageError 逃逸出 upload_image 炸 MCP 会话；补 decode 层与工具层覆盖测试各一。 Co-Authored-By: Claude --- src/ppclock/mcp_tools.py | 10 ++++++++-- tests/test_mcp_tools.py | 13 +++++++++++++ 2 files changed, 21 insertions(+), 2 deletions(-) diff --git a/src/ppclock/mcp_tools.py b/src/ppclock/mcp_tools.py index d057737..4a27490 100644 --- a/src/ppclock/mcp_tools.py +++ b/src/ppclock/mcp_tools.py @@ -23,7 +23,10 @@ _NOTFOUND_HINT = "未发现 NRF- 前缀设备；确认设备在位且未处于 def decode_image_source(source: str) -> Image.Image: """source = 服务器本地路径，或 base64（可带 data URI 前缀）。""" if os.path.exists(source): - return Image.open(source) + try: + return Image.open(source) + except OSError: # 文件存在但不是可识别图片（含截断） + raise ValueError("source 不是可识别的图片") from None data = source if data.startswith("data:"): if "," not in data: @@ -33,7 +36,10 @@ def decode_image_source(source: str) -> Image.Image: raw = base64.b64decode(data, validate=True) except (binascii.Error, ValueError): raise ValueError("source 既不是存在的文件路径，也不是合法 base64") from None - return Image.open(io.BytesIO(raw)) + try: + return Image.open(io.BytesIO(raw)) + except OSError: # 合法 base64 但不是可识别图片（PIL.UnidentifiedImageError 属 OSError） + raise ValueError("source 不是可识别的图片") from None def _err(tool: str, code: str, message: str, hint: str | None = None) -> dict: diff --git a/tests/test_mcp_tools.py b/tests/test_mcp_tools.py index d852e8e..110cf0d 100644 --- a/tests/test_mcp_tools.py +++ b/tests/test_mcp_tools.py @@ -101,6 +101,11 @@ class TestDecodeImageSource: with pytest.raises(ValueError): decode_image_source("!!!not-base64!!!") + def test_base64_not_image(self): + src = base64.b64encode(b"not an image").decode() + with pytest.raises(ValueError): + decode_image_source(src) + class TestTools: @pytest.mark.asyncio @@ -116,6 +121,14 @@ class TestTools: assert r["data"]["bytes_bw"] == 15000 assert r["data"]["slot"] == 0 + @pytest.mark.asyncio + async def test_upload_image_not_image(self): + src = base64.b64encode(b"not an image").decode() + r = await call(make_session_coro(FakeManager()), "upload_image", + {"source": src, "slot": 0}) + assert r["ok"] is False + assert r["error"]["code"] == "INVALID_PARAM" + @pytest.mark.asyncio async def test_set_mode_invalid_param(self): r = await call(make_session_coro(FakeManager()), "set_mode",