fix(client): accept whitespace in screenshot base64

client/src/cmbuyer_client/device/baseline.py

@@ -27,6 +27,7 @@ from .adb import AdbClient, DeviceInspection
 PDD_PACKAGE = "com.xunmeng.pinduoduo"
 SCREENSHOT_PARAMS = [1, 80]
 HIERARCHY_PARAMS = [False, 50]
+_BASE64_ASCII_WHITESPACE = " \t\r\n"
 
 
 class BaselineCaptureError(RuntimeError):
@@ -172,17 +173,21 @@ def _extract_version(app_info: dict[str, Any]) -> str:
 
 
 def _save_base64_screenshot(value: Any, target: Path) -> None:
-    """严格解码 RPC 截图，并用 Pillow 验证后保存 PNG；没有 adb fallback。"""
+    """规范化常见 ASCII Base64 空白后严格解码；没有 adb screenshot fallback。"""
 
     if not isinstance(value, str) or not value:
         raise BaselineCaptureError("截图 RPC 未返回 base64 数据，拒绝发布不完整取证。")
     try:
-        raw_image = base64.b64decode(value.encode("ascii"), validate=True)
+        normalized = value.translate({ord(character): None for character in _BASE64_ASCII_WHITESPACE})
+        raw_image = base64.b64decode(normalized.encode("ascii"), validate=True)
+    except (UnicodeEncodeError, ValueError, binascii.Error) as error:
+        raise BaselineCaptureError("截图 RPC Base64 语法无效，拒绝发布不完整取证。") from error
+    try:
         with Image.open(BytesIO(raw_image)) as image:
             image.load()
             image.save(target, format="PNG")
-    except (UnicodeEncodeError, ValueError, binascii.Error, UnidentifiedImageError, OSError) as error:
-        raise BaselineCaptureError("截图 RPC 返回的数据不是有效图像，拒绝发布不完整取证。") from error
+    except (UnidentifiedImageError, OSError) as error:
+        raise BaselineCaptureError("截图 RPC 图像数据无效，拒绝发布不完整取证。") from error
 
 
 def _validate_hierarchy(value: Any) -> None:

client/tests/device/test_baseline.py

@@ -153,22 +153,90 @@ class BaselineCaptureTests(unittest.TestCase):
             self.assertFalse(output.exists())
             self.assertEqual(list(Path(directory).iterdir()), [])
 
-    def test_invalid_screenshot_base64_fails_closed_without_partial_output(self) -> None:
+    def test_invalid_screenshot_base64_syntax_fails_closed_without_partial_output(self) -> None:
         class InvalidScreenshotDevice(FakeUiDevice):
             def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
                 if method == "takeScreenshot":
-                    return "not-valid-base64"
+                    valid = super().jsonrpc_call(method, params, timeout)
+                    return valid[:12] + "!" + valid[12:]
                 return super().jsonrpc_call(method, params, timeout)
 
         capturer = DeviceBaselineCapturer(StaticAdbClient(), lambda serial: InvalidScreenshotDevice(), timeout_seconds=5)
         with tempfile.TemporaryDirectory() as directory:
             output = Path(directory) / "baseline"
-            with self.assertRaises(BaselineCaptureError):
+            with self.assertRaises(BaselineCaptureError) as raised:
                 capturer.capture(SERIAL, output)
 
+            self.assertIn("Base64 语法无效", str(raised.exception))
+            self.assertNotIn("!", str(raised.exception))
             self.assertFalse(output.exists())
             self.assertEqual(list(Path(directory).iterdir()), [])
 
+    def test_invalid_padding_and_unapproved_ascii_whitespace_fail_closed(self) -> None:
+        invalid_insertions = {
+            "padding": lambda value: value[:-1],
+            "vertical-tab": lambda value: value[:12] + "\v" + value[12:],
+            "form-feed": lambda value: value[:12] + "\f" + value[12:],
+        }
+
+        for name, make_invalid in invalid_insertions.items():
+            with self.subTest(name=name), tempfile.TemporaryDirectory() as directory:
+                class InvalidScreenshotDevice(FakeUiDevice):
+                    def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                        value = super().jsonrpc_call(method, params, timeout)
+                        if method == "takeScreenshot":
+                            return make_invalid(value)
+                        return value
+
+                output = Path(directory) / "baseline"
+                capturer = DeviceBaselineCapturer(
+                    StaticAdbClient(),
+                    lambda serial: InvalidScreenshotDevice(),
+                    timeout_seconds=5,
+                )
+                with self.assertRaises(BaselineCaptureError) as raised:
+                    capturer.capture(SERIAL, output)
+
+                self.assertIn("Base64 语法无效", str(raised.exception))
+                self.assertNotIn(SERIAL, str(raised.exception))
+                self.assertFalse(output.exists())
+                self.assertEqual(list(Path(directory).iterdir()), [])
+
+    def test_base64_decoded_nonimage_fails_closed_without_partial_output(self) -> None:
+        class NonImageScreenshotDevice(FakeUiDevice):
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                if method == "takeScreenshot":
+                    return base64.b64encode(b"not an image").decode("ascii")
+                return super().jsonrpc_call(method, params, timeout)
+
+        capturer = DeviceBaselineCapturer(StaticAdbClient(), lambda serial: NonImageScreenshotDevice(), timeout_seconds=5)
+        with tempfile.TemporaryDirectory() as directory:
+            output = Path(directory) / "baseline"
+            with self.assertRaises(BaselineCaptureError) as raised:
+                capturer.capture(SERIAL, output)
+
+            self.assertIn("图像数据无效", str(raised.exception))
+            self.assertNotIn("not an image", str(raised.exception))
+            self.assertFalse(output.exists())
+            self.assertEqual(list(Path(directory).iterdir()), [])
+
+    def test_ascii_base64_whitespace_is_normalized_before_strict_decode(self) -> None:
+        class WhitespaceScreenshotDevice(FakeUiDevice):
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                value = super().jsonrpc_call(method, params, timeout)
+                if method == "takeScreenshot":
+                    return value[:10] + " \t\r\n" + value[10:30] + "\n" + value[30:]
+                return value
+
+        capturer = DeviceBaselineCapturer(StaticAdbClient(), lambda serial: WhitespaceScreenshotDevice(), timeout_seconds=5)
+        with tempfile.TemporaryDirectory() as directory:
+            output = Path(directory) / "baseline"
+            result = capturer.capture(SERIAL, output)
+
+            self.assertTrue(result.screenshot_path.is_file())
+            with Image.open(result.screenshot_path) as image:
+                self.assertEqual(image.size, (1, 1))
+
     def test_invalid_or_non_hierarchy_xml_fails_closed_without_partial_output(self) -> None:
         class InvalidHierarchyDevice(FakeUiDevice):
             def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str: