diff --git a/client/requirements.txt b/client/requirements.txt index b38251b..ccca7c5 100644 --- a/client/requirements.txt +++ b/client/requirements.txt @@ -1,7 +1,9 @@ # -*- coding: utf-8 -*- # 桌面界面（Qt 官方 Python 绑定）。 PySide6 -# 后续真机取证会使用；本阶段不导入或连接设备。 +# T-101 基线取证使用；只连接显式 serial，不打开或操作拼多多页面。 uiautomator2 +# T-101 直接使用当前 ADB server 的已列出设备对象交给 uiautomator2，禁止 WiFi 自动重连。 +adbutils>=2.11,<3 # 后续截图完整性检查会使用。 Pillow diff --git a/client/scripts/capture_device_baseline.py b/client/scripts/capture_device_baseline.py new file mode 100644 index 0000000..4c72149 --- /dev/null +++ b/client/scripts/capture_device_baseline.py @@ -0,0 +1,76 @@ +"""采集指定 Android 设备的本地基线证据；不打开或操作拼多多页面。""" + +from __future__ import annotations + +import argparse +from pathlib import Path +import sys + + +CLIENT_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(CLIENT_ROOT / "src")) + +from cmbuyer_client.device.adb import AdbClient, DeviceConnectionError, SubprocessAdbRunner +from cmbuyer_client.device.baseline import ( + BaselineCaptureError, + DeviceBaselineCapturer, + NoReconnectUiautomatorConnector, +) + + +def parse_arguments(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description="采集显式指定 Android 设备的本地基线证据。") + parser.add_argument("--serial", required=True, help="ADB device serial；禁止自动选择。") + parser.add_argument("--output-dir", required=True, type=Path, help="新建的本地证据目录；不得覆盖已有目录。") + parser.add_argument("--timeout", type=float, default=10.0, help="ADB、uiautomator2 RPC 与 ADB socket 超时（秒）。") + parser.add_argument("--adb", default="adb", help="adb 可执行文件路径。") + return parser.parse_args(argv) + + +def validate_arguments(arguments: argparse.Namespace) -> None: + """在导入设备库前拒绝危险或无效输入，便于离线测试。""" + + if not arguments.serial.strip(): + raise ValueError("必须显式提供非空 --serial。") + if arguments.timeout <= 0: + raise ValueError("--timeout 必须大于 0。") + + +def main(argv: list[str] | None = None) -> int: + arguments = parse_arguments(argv) + try: + validate_arguments(arguments) + except ValueError as error: + print(f"失败：{error}", file=sys.stderr) + return 2 + + try: + import adbutils + import uiautomator2 as u2 + except ImportError: + print("失败：缺少 uiautomator2；请在采购工具虚拟环境中运行。", file=sys.stderr) + return 2 + + client = AdbClient(SubprocessAdbRunner(arguments.adb), timeout_seconds=arguments.timeout) + connector = NoReconnectUiautomatorConnector( + adbutils.AdbClient(socket_timeout=arguments.timeout).device_list, + u2.connect, + ) + capturer = DeviceBaselineCapturer(client, connector, timeout_seconds=arguments.timeout) + try: + result = capturer.capture(arguments.serial, arguments.output_dir) + except (DeviceConnectionError, BaselineCaptureError) as error: + # 错误类型只表达状态，不打印 ADB 输出、serial、XML 或页面正文。 + print(f"基线取证失败：{error}", file=sys.stderr) + return 1 + except OSError: + print("基线取证失败：无法创建或发布本地证据目录。", file=sys.stderr) + return 1 + + print(f"基线取证完成：{result.output_directory}") + print(f"manifest：{result.manifest_path}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/client/src/cmbuyer_client/device/__init__.py b/client/src/cmbuyer_client/device/__init__.py new file mode 100644 index 0000000..22d39aa --- /dev/null +++ b/client/src/cmbuyer_client/device/__init__.py @@ -0,0 +1,15 @@ +"""设备连接与基线取证边界。 + +本包只提供显式设备选择、非敏感身份核验和本地基线采集；不包含任何采购页面或订单操作。 +""" + +from .adb import AdbClient, AdbDevice, CommandResult +from .baseline import BaselineCaptureResult, DeviceBaselineCapturer + +__all__ = [ + "AdbClient", + "AdbDevice", + "BaselineCaptureResult", + "CommandResult", + "DeviceBaselineCapturer", +] diff --git a/client/src/cmbuyer_client/device/adb.py b/client/src/cmbuyer_client/device/adb.py new file mode 100644 index 0000000..a4d4f04 --- /dev/null +++ b/client/src/cmbuyer_client/device/adb.py @@ -0,0 +1,234 @@ +"""ADB 设备清单与物理设备冲突的 fail-closed 边界。""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +import subprocess +from typing import Protocol, Sequence + + +class DeviceConnectionError(RuntimeError): + """显式设备连接边界的基础错误，不携带命令输出或设备敏感内容。""" + + +class SerialRequiredError(DeviceConnectionError): + """调用方没有明确指定设备 serial。""" + + +class DeviceNotFoundError(DeviceConnectionError): + """指定 serial 不在 ADB 当前清单中。""" + + +class DeviceOfflineError(DeviceConnectionError): + """指定设备处于 offline 状态。""" + + +class DeviceUnauthorizedError(DeviceConnectionError): + """指定设备尚未授权此电脑。""" + + +class DeviceStateError(DeviceConnectionError): + """指定设备处于其他不可用状态。""" + + +class DeviceCommandTimeoutError(DeviceConnectionError): + """ADB 命令超过调用方指定的超时。""" + + +class DeviceCommandError(DeviceConnectionError): + """ADB 命令失败；错误文本刻意不回显设备输出。""" + + +class DeviceIdentityUnconfirmedError(DeviceConnectionError): + """多条在线通道无法完成同机身份判断，必须由人处理。""" + + +class DuplicatePhysicalDeviceError(DeviceConnectionError): + """同一物理手机通过多个 ADB 通道同时在线。""" + + +@dataclass(frozen=True) +class CommandResult: + """可注入命令执行器的最小、可离线构造结果。""" + + stdout: str + stderr: str = "" + returncode: int = 0 + + +class CommandRunner(Protocol): + """运行 ADB 子命令的可替换边界。""" + + def run(self, arguments: Sequence[str], timeout_seconds: float) -> CommandResult: + """运行参数，不得通过 shell 拼接。""" + + +class SubprocessAdbRunner: + """使用 subprocess 的生产执行器，所有调用必须带超时。""" + + def __init__(self, executable: str | Path = "adb") -> None: + self._executable = str(executable) + + def run(self, arguments: Sequence[str], timeout_seconds: float) -> CommandResult: + try: + completed = subprocess.run( + [self._executable, *arguments], + check=False, + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + timeout=timeout_seconds, + ) + except subprocess.TimeoutExpired as error: + raise DeviceCommandTimeoutError("ADB 命令超时，请检查设备连接后由人工重试。") from error + except OSError as error: + raise DeviceCommandError("无法启动 ADB，请检查 adb 路径与本机工具链。") from error + + return CommandResult( + stdout=completed.stdout, + stderr=completed.stderr, + returncode=completed.returncode, + ) + + +@dataclass(frozen=True) +class AdbDevice: + """`adb devices -l` 的单行非敏感传输元数据。""" + + serial: str + state: str + product: str | None = None + model: str | None = None + device: str | None = None + + +@dataclass(frozen=True) +class DeviceInspection: + """选定通道的只读身份结果，原始硬件标识只在内存中参与比较。""" + + device: AdbDevice + model: str + android_version: str + + +def parse_adb_devices(output: str) -> list[AdbDevice]: + """解析 `adb devices -l`，忽略标题、空行和 adb 附加提示。""" + + devices: list[AdbDevice] = [] + for raw_line in output.splitlines(): + line = raw_line.strip() + if not line or line.startswith("List of devices attached") or line.startswith("*"): + continue + fields = line.split() + if len(fields) < 2: + continue + details = { + key: value + for field in fields[2:] + if ":" in field + for key, value in [field.split(":", 1)] + } + devices.append( + AdbDevice( + serial=fields[0], + state=fields[1], + product=details.get("product"), + model=details.get("model"), + device=details.get("device"), + ) + ) + return devices + + +class AdbClient: + """显式 serial 的 ADB 只读查询。 + + 多个在线通道必须完成硬件身份比对。比对失败时不能用相同 model/product 猜测同一台手机， + 因为那会把不确定性隐藏成错误的安全结论。 + """ + + def __init__(self, runner: CommandRunner, timeout_seconds: float = 10.0) -> None: + if timeout_seconds <= 0: + raise ValueError("timeout_seconds 必须大于 0") + self._runner = runner + self._timeout_seconds = timeout_seconds + + def inspect(self, serial: str) -> DeviceInspection: + """确认指定通道在线且不与另一在线通道指向同一物理设备。""" + + selected_serial = _require_serial(serial) + devices = self.devices() + selected = next((device for device in devices if device.serial == selected_serial), None) + if selected is None: + raise DeviceNotFoundError("指定设备不在 ADB 清单中，请显式检查 serial。") + _raise_for_state(selected.state) + + online_devices = [device for device in devices if device.state == "device"] + if len(online_devices) > 1: + identities: dict[str, frozenset[str]] = {} + for candidate in online_devices: + try: + identities[candidate.serial] = self._physical_identity(candidate) + except DeviceConnectionError as error: + raise DeviceIdentityUnconfirmedError( + "存在多个在线 ADB 通道且身份无法确认，已拒绝选择设备。" + ) from error + + selected_identity = identities[selected.serial] + if any( + candidate_serial != selected.serial and selected_identity.intersection(candidate_identity) + for candidate_serial, candidate_identity in identities.items() + ): + raise DuplicatePhysicalDeviceError( + "同一物理手机的多个 ADB 通道同时在线，已拒绝继续；请仅保留一个通道。" + ) + + model = self._getprop(selected.serial, "ro.product.model") or selected.model or "unknown" + android_version = self._getprop(selected.serial, "ro.build.version.release") or "unknown" + return DeviceInspection(device=selected, model=model, android_version=android_version) + + def devices(self) -> list[AdbDevice]: + """读取并解析 ADB 设备清单。""" + + result = self._run_checked(("devices", "-l")) + return parse_adb_devices(result.stdout) + + def _physical_identity(self, device: AdbDevice) -> frozenset[str]: + serialno = self._getprop(device.serial, "ro.serialno") + boot_serialno = self._getprop(device.serial, "ro.boot.serialno") + identifiers = frozenset(value for value in (serialno, boot_serialno) if value) + if identifiers: + return identifiers + # model/product/device 只能作为展示元数据，不能证明两台同型号设备是同一物理机。 + raise DeviceIdentityUnconfirmedError("无法读取设备硬件身份摘要。") + + def _getprop(self, serial: str, property_name: str) -> str: + result = self._run_checked(("-s", serial, "shell", "getprop", property_name)) + return result.stdout.strip() + + def _run_checked(self, arguments: Sequence[str]) -> CommandResult: + try: + result = self._runner.run(arguments, self._timeout_seconds) + except subprocess.TimeoutExpired as error: + raise DeviceCommandTimeoutError("ADB 命令超时，请检查设备连接后由人工重试。") from error + if result.returncode != 0: + raise DeviceCommandError("ADB 命令失败，请检查设备连接或授权状态。") + return result + + +def _require_serial(serial: str) -> str: + if not isinstance(serial, str) or not serial.strip(): + raise SerialRequiredError("必须显式提供设备 serial，禁止自动选择设备。") + return serial.strip() + + +def _raise_for_state(state: str) -> None: + if state == "device": + return + if state == "offline": + raise DeviceOfflineError("指定设备处于 offline 状态。") + if state == "unauthorized": + raise DeviceUnauthorizedError("指定设备尚未授权此电脑。") + raise DeviceStateError("指定设备不处于可用状态。") diff --git a/client/src/cmbuyer_client/device/baseline.py b/client/src/cmbuyer_client/device/baseline.py new file mode 100644 index 0000000..b6e5529 --- /dev/null +++ b/client/src/cmbuyer_client/device/baseline.py @@ -0,0 +1,233 @@ +"""只读设备基线取证，严格限制在元数据、截图和完整节点树。""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +from datetime import UTC, datetime +from hashlib import sha256 +import base64 +import binascii +from io import BytesIO +import json +import os +from pathlib import Path +import shutil +from typing import Any, Protocol +from uuid import uuid4 +from xml.etree import ElementTree + +from adbutils.errors import AdbTimeout +from PIL import Image, UnidentifiedImageError +from uiautomator2.exceptions import HTTPTimeoutError + +from .adb import AdbClient, DeviceInspection + + +PDD_PACKAGE = "com.xunmeng.pinduoduo" +SCREENSHOT_PARAMS = [1, 80] +HIERARCHY_PARAMS = [False, 50] + + +class BaselineCaptureError(RuntimeError): + """基线取证无法完整落盘时的失败，不会伪造成功产物。""" + + +class BaselineCaptureTimeoutError(BaselineCaptureError): + """设备基线取证超时；底层异常文本不向 CLI 或日志泄露。""" + + +class UiAutomatorDevice(Protocol): + """本任务唯一需要的 uiautomator2 只读能力。""" + + def app_info(self, package_name: str) -> dict[str, Any]: + """读取已安装应用元数据。""" + + def jsonrpc_call(self, method: str, params: Any = None, timeout: float = 10) -> Any: + """调用公开 uiautomator2 JSON-RPC 接口。""" + + +class NoReconnectUiautomatorConnector: + """只把当前 ADB server 已列出的设备对象交给 uiautomator2。 + + uiautomator2 直接接收 IP serial 时会在内部尝试 adb disconnect/connect。这里先从已列出设备中 + 取对象再调用 ``u2.connect(device_object)``，避免连接阶段隐式重连已经掉线的 WiFi 通道。 + """ + + def __init__(self, list_devices: Callable[[], list[Any]], connect: Callable[[Any], UiAutomatorDevice]) -> None: + self._list_devices = list_devices + self._connect = connect + + def __call__(self, serial: str) -> UiAutomatorDevice: + device = next((item for item in self._list_devices() if item.serial == serial), None) + if device is None: + raise BaselineCaptureError("设备在连接前已从 ADB 清单消失，已拒绝自动重连。") + return self._connect(device) + + +@dataclass(frozen=True) +class BaselineCaptureResult: + """已原子发布的基线取证摘要，不包含页面正文或原始 serial。""" + + output_directory: Path + manifest_path: Path + screenshot_path: Path + hierarchy_path: Path + + +class DeviceBaselineCapturer: + """以先校验通道、后连接、最后原子发布的顺序采集基线。 + + 截图和 XML 可能包含页面敏感内容，因此仅落在调用方明确指定的本地目录；manifest 只写 + 哈希、设备非敏感元数据和脱敏后的 serial 摘要，绝不嵌入 XML 或页面文本。 + """ + + def __init__( + self, + adb_client: AdbClient, + connector: Callable[[str], UiAutomatorDevice], + timeout_seconds: float, + ) -> None: + if timeout_seconds <= 0: + raise ValueError("timeout_seconds 必须大于 0") + self._adb_client = adb_client + self._connector = connector + self._timeout_seconds = timeout_seconds + + def capture(self, serial: str, output_directory: Path) -> BaselineCaptureResult: + """采集完整基线，任何一步失败均不发布 output_directory。""" + + inspection = self._adb_client.inspect(serial) + target = Path(output_directory) + if target.exists(): + raise BaselineCaptureError("输出目录已存在；为防止混入旧证据，拒绝覆盖。") + if not target.name: + raise BaselineCaptureError("输出目录必须是明确的新目录。") + + target.parent.mkdir(parents=True, exist_ok=True) + staging = target.parent / f".{target.name}.staging-{uuid4().hex}" + staging.mkdir() + try: + device = self._connector(serial) + app_info = device.app_info(PDD_PACKAGE) + version = _extract_version(app_info) + + screenshot_path = staging / "screenshot.png" + screenshot_base64 = device.jsonrpc_call( + "takeScreenshot", + SCREENSHOT_PARAMS, + timeout=self._timeout_seconds, + ) + _save_base64_screenshot(screenshot_base64, screenshot_path) + + hierarchy = device.jsonrpc_call( + "dumpWindowHierarchy", + HIERARCHY_PARAMS, + timeout=self._timeout_seconds, + ) + _validate_hierarchy(hierarchy) + hierarchy_path = staging / "hierarchy.xml" + hierarchy_path.write_text(hierarchy, encoding="utf-8") + + manifest_path = staging / "manifest.json" + manifest_path.write_text( + json.dumps( + _manifest(inspection, serial, version, screenshot_path, hierarchy_path), + ensure_ascii=False, + indent=2, + sort_keys=True, + ) + + "\n", + encoding="utf-8", + ) + os.replace(staging, target) + except BaselineCaptureError: + # 仅删除本次创建、名称带随机标识的暂存目录，绝不触碰调用方原有输出目录。 + if staging.exists(): + shutil.rmtree(staging) + raise + except (AdbTimeout, HTTPTimeoutError, TimeoutError) as error: + if staging.exists(): + shutil.rmtree(staging) + raise BaselineCaptureTimeoutError("设备基线取证超时，未发布任何证据产物。") from error + except Exception as error: + if staging.exists(): + shutil.rmtree(staging) + # uiautomator2/adbutils 可能把 serial、路径或远端响应放入异常文本，不能直接传播到 CLI。 + raise BaselineCaptureError("设备基线取证未完成，未发布任何证据产物。") from error + + return BaselineCaptureResult( + output_directory=target, + manifest_path=target / "manifest.json", + screenshot_path=target / "screenshot.png", + hierarchy_path=target / "hierarchy.xml", + ) + + +def _extract_version(app_info: dict[str, Any]) -> str: + version = app_info.get("versionName") or app_info.get("version_name") + if not isinstance(version, str) or not version.strip(): + raise BaselineCaptureError("无法读取拼多多版本，拒绝发布不完整取证。") + return version.strip() + + +def _save_base64_screenshot(value: Any, target: Path) -> None: + """严格解码 RPC 截图，并用 Pillow 验证后保存 PNG；没有 adb fallback。""" + + if not isinstance(value, str) or not value: + raise BaselineCaptureError("截图 RPC 未返回 base64 数据，拒绝发布不完整取证。") + try: + raw_image = base64.b64decode(value.encode("ascii"), validate=True) + 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 + + +def _validate_hierarchy(value: Any) -> None: + """确认 RPC 返回的是完整节点树，不把原始 XML 放进错误或日志。""" + + if not isinstance(value, str) or not value: + raise BaselineCaptureError("节点树导出为空，拒绝发布不完整取证。") + try: + root = ElementTree.fromstring(value) + except ElementTree.ParseError as error: + raise BaselineCaptureError("节点树不是有效 XML，拒绝发布不完整取证。") from error + if root.tag != "hierarchy": + raise BaselineCaptureError("节点树根节点无效，拒绝发布不完整取证。") + + +def _manifest( + inspection: DeviceInspection, + serial: str, + pdd_version: str, + screenshot_path: Path, + hierarchy_path: Path, +) -> dict[str, Any]: + """只序列化审计摘要；页面内容留在 XML 文件，不进入日志或 manifest。""" + + return { + "schema_version": 1, + "captured_at": datetime.now(UTC).isoformat(), + "channel": "wifi" if ":" in serial else "usb", + "serial_sha256": sha256(serial.encode("utf-8")).hexdigest(), + "device": { + "model": inspection.model, + "android_version": inspection.android_version, + "pdd_package": PDD_PACKAGE, + "pdd_version": pdd_version, + }, + "artifacts": [ + {"path": screenshot_path.name, "sha256": _sha256_file(screenshot_path)}, + {"path": hierarchy_path.name, "sha256": _sha256_file(hierarchy_path)}, + ], + } + + +def _sha256_file(path: Path) -> str: + digest = sha256() + with path.open("rb") as source: + for chunk in iter(lambda: source.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() diff --git a/client/tests/device/__init__.py b/client/tests/device/__init__.py new file mode 100644 index 0000000..b8cb2b6 --- /dev/null +++ b/client/tests/device/__init__.py @@ -0,0 +1 @@ +"""设备连接与基线取证的离线测试。""" diff --git a/client/tests/device/test_adb.py b/client/tests/device/test_adb.py new file mode 100644 index 0000000..814b784 --- /dev/null +++ b/client/tests/device/test_adb.py @@ -0,0 +1,182 @@ +"""ADB 设备边界测试：所有命令执行器均为 mock，不连接真机。""" + +from __future__ import annotations + +from collections.abc import Sequence +from pathlib import Path +import subprocess +import sys +import unittest + + +CLIENT_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(CLIENT_ROOT / "src")) + +from cmbuyer_client.device.adb import ( + AdbClient, + CommandResult, + DeviceIdentityUnconfirmedError, + DeviceNotFoundError, + DeviceOfflineError, + DeviceStateError, + DeviceUnauthorizedError, + DuplicatePhysicalDeviceError, + SerialRequiredError, +) + + +USB_SERIAL = "3B65BD02H7F00000" +WIFI_SERIAL = "192.168.0.173:5555" + + +class FakeRunner: + def __init__(self, devices_output: str, properties: dict[tuple[str, str], CommandResult | str]) -> None: + self.devices_output = devices_output + self.properties = properties + self.calls: list[tuple[str, ...]] = [] + + def run(self, arguments: Sequence[str], timeout_seconds: float) -> CommandResult: + self.calls.append(tuple(arguments)) + if tuple(arguments) == ("devices", "-l"): + return CommandResult(stdout=self.devices_output) + key = (arguments[1], arguments[-1]) + value = self.properties.get(key, "") + return value if isinstance(value, CommandResult) else CommandResult(stdout=value) + + +def _properties(serials: tuple[str, ...]) -> dict[tuple[str, str], str]: + values: dict[tuple[str, str], str] = {} + for serial in serials: + values[(serial, "ro.serialno")] = "physical-phone-1" + values[(serial, "ro.boot.serialno")] = "physical-phone-1" + values[(serial, "ro.product.model")] = "PKG110" + values[(serial, "ro.product.name")] = "PKG110" + values[(serial, "ro.product.device")] = "OP5D2BL1" + values[(serial, "ro.build.version.release")] = "16" + return values + + +class AdbClientTests(unittest.TestCase): + def test_requires_explicit_serial(self) -> None: + runner = FakeRunner("List of devices attached\n", {}) + + with self.assertRaises(SerialRequiredError): + AdbClient(runner).inspect(" ") + + self.assertEqual(runner.calls, []) + + def test_missing_offline_and_unauthorized_are_distinct(self) -> None: + missing = AdbClient(FakeRunner("List of devices attached\n", {})) + with self.assertRaises(DeviceNotFoundError): + missing.inspect(USB_SERIAL) + + offline = AdbClient(FakeRunner(f"List of devices attached\n{USB_SERIAL}\toffline\n", {})) + with self.assertRaises(DeviceOfflineError): + offline.inspect(USB_SERIAL) + + unauthorized = AdbClient(FakeRunner(f"List of devices attached\n{USB_SERIAL}\tunauthorized\n", {})) + with self.assertRaises(DeviceUnauthorizedError): + unauthorized.inspect(USB_SERIAL) + + def test_two_channels_with_same_physical_identity_fail_closed(self) -> None: + output = ( + "List of devices attached\n" + f"{WIFI_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n" + f"{USB_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n" + ) + runner = FakeRunner(output, _properties((WIFI_SERIAL, USB_SERIAL))) + + with self.assertRaises(DuplicatePhysicalDeviceError): + AdbClient(runner).inspect(USB_SERIAL) + + self.assertIn(("-s", WIFI_SERIAL, "shell", "getprop", "ro.serialno"), runner.calls) + self.assertIn(("-s", USB_SERIAL, "shell", "getprop", "ro.serialno"), runner.calls) + + def test_multiple_online_devices_with_failed_identity_fail_closed(self) -> None: + output = ( + "List of devices attached\n" + f"{WIFI_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n" + f"{USB_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n" + ) + properties = _properties((WIFI_SERIAL, USB_SERIAL)) + properties[(WIFI_SERIAL, "ro.serialno")] = CommandResult(stdout="", returncode=1) + runner = FakeRunner(output, properties) + + with self.assertRaises(DeviceIdentityUnconfirmedError): + AdbClient(runner).inspect(USB_SERIAL) + + def test_online_explicit_serial_reads_non_sensitive_metadata(self) -> None: + output = f"List of devices attached\n{USB_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n" + inspection = AdbClient(FakeRunner(output, _properties((USB_SERIAL,)))).inspect(USB_SERIAL) + + self.assertEqual(inspection.device.serial, USB_SERIAL) + self.assertEqual(inspection.model, "PKG110") + self.assertEqual(inspection.android_version, "16") + + def test_single_online_device_does_not_require_hardware_identity(self) -> None: + output = f"List of devices attached\n{USB_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n" + properties = _properties((USB_SERIAL,)) + properties[(USB_SERIAL, "ro.serialno")] = "" + properties[(USB_SERIAL, "ro.boot.serialno")] = "" + + inspection = AdbClient(FakeRunner(output, properties)).inspect(USB_SERIAL) + + self.assertEqual(inspection.model, "PKG110") + + def test_multiple_online_devices_without_hardware_identity_are_unconfirmed(self) -> None: + output = ( + "List of devices attached\n" + f"{WIFI_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n" + f"{USB_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n" + ) + properties = _properties((WIFI_SERIAL, USB_SERIAL)) + properties[(WIFI_SERIAL, "ro.serialno")] = "" + properties[(WIFI_SERIAL, "ro.boot.serialno")] = "" + + with self.assertRaises(DeviceIdentityUnconfirmedError): + AdbClient(FakeRunner(output, properties)).inspect(USB_SERIAL) + + def test_multiple_online_devices_with_different_identity_keep_explicit_selection(self) -> None: + output = ( + "List of devices attached\n" + f"{WIFI_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n" + f"{USB_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n" + ) + properties = _properties((WIFI_SERIAL, USB_SERIAL)) + properties[(WIFI_SERIAL, "ro.serialno")] = "physical-phone-2" + properties[(WIFI_SERIAL, "ro.boot.serialno")] = "physical-phone-2" + + inspection = AdbClient(FakeRunner(output, properties)).inspect(USB_SERIAL) + + self.assertEqual(inspection.device.serial, USB_SERIAL) + + def test_shared_boot_serial_is_duplicate_even_when_ro_serial_differs(self) -> None: + output = ( + "List of devices attached\n" + f"{WIFI_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n" + f"{USB_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n" + ) + properties = _properties((WIFI_SERIAL, USB_SERIAL)) + properties[(WIFI_SERIAL, "ro.serialno")] = "wifi-transport-serial" + properties[(USB_SERIAL, "ro.serialno")] = "usb-transport-serial" + properties[(WIFI_SERIAL, "ro.boot.serialno")] = "shared-hardware-serial" + properties[(USB_SERIAL, "ro.boot.serialno")] = "shared-hardware-serial" + + with self.assertRaises(DuplicatePhysicalDeviceError): + AdbClient(FakeRunner(output, properties)).inspect(USB_SERIAL) + + def test_unknown_adb_state_is_rejected(self) -> None: + client = AdbClient(FakeRunner(f"List of devices attached\n{USB_SERIAL}\trecovery\n", {})) + + with self.assertRaises(DeviceStateError): + client.inspect(USB_SERIAL) + + def test_runner_timeout_is_a_distinct_connection_error(self) -> None: + class TimeoutRunner: + def run(self, arguments: Sequence[str], timeout_seconds: float) -> CommandResult: + raise subprocess.TimeoutExpired(arguments, timeout_seconds) + + from cmbuyer_client.device.adb import DeviceCommandTimeoutError + + with self.assertRaises(DeviceCommandTimeoutError): + AdbClient(TimeoutRunner()).inspect(USB_SERIAL) diff --git a/client/tests/device/test_baseline.py b/client/tests/device/test_baseline.py new file mode 100644 index 0000000..9505581 --- /dev/null +++ b/client/tests/device/test_baseline.py @@ -0,0 +1,213 @@ +"""基线取证测试：mock ADB/uiautomator2，不连接手机。""" + +from __future__ import annotations + +from pathlib import Path +import base64 +from io import BytesIO +import sys +import tempfile +import unittest + +from PIL import Image +from uiautomator2.exceptions import HTTPTimeoutError + + +CLIENT_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(CLIENT_ROOT / "src")) +sys.path.insert(0, str(CLIENT_ROOT / "scripts")) + +from cmbuyer_client.device.adb import AdbDevice, DeviceInspection +from cmbuyer_client.device.baseline import ( + BaselineCaptureError, + BaselineCaptureTimeoutError, + DeviceBaselineCapturer, + NoReconnectUiautomatorConnector, + PDD_PACKAGE, +) +from capture_device_baseline import parse_arguments, validate_arguments + + +SERIAL = "USB-serial-for-test" + + +class StaticAdbClient: + def __init__(self) -> None: + self.serials: list[str] = [] + + def inspect(self, serial: str) -> DeviceInspection: + self.serials.append(serial) + return DeviceInspection( + device=AdbDevice(serial=serial, state="device", model="Test Model"), + model="Test Model", + android_version="16", + ) + + +class FakeUiDevice: + def __init__(self, fail_dump: bool = False) -> None: + self.fail_dump = fail_dump + self.rpc_calls: list[tuple[str, object, float]] = [] + self.app_info_calls: list[str] = [] + + def app_info(self, package_name: str) -> dict[str, str]: + self.app_info_calls.append(package_name) + return {"versionName": "8.17.0"} + + def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str: + self.rpc_calls.append((method, params, timeout)) + if method == "takeScreenshot": + image_data = BytesIO() + Image.new("RGB", (1, 1), color="white").save(image_data, format="PNG") + return base64.b64encode(image_data.getvalue()).decode("ascii") + if method != "dumpWindowHierarchy": + raise AssertionError(f"unexpected method: {method}") + if self.fail_dump: + raise RuntimeError("mock dump failed") + return "" + + +class BaselineCaptureTests(unittest.TestCase): + def test_capture_writes_hashes_without_xml_or_raw_serial_in_manifest(self) -> None: + adb = StaticAdbClient() + device = FakeUiDevice() + capturer = DeviceBaselineCapturer(adb, lambda serial: device, timeout_seconds=7.5) + + with tempfile.TemporaryDirectory() as directory: + output = Path(directory) / "baseline" + result = capturer.capture(SERIAL, output) + manifest = result.manifest_path.read_text(encoding="utf-8") + + self.assertEqual(adb.serials, [SERIAL]) + self.assertEqual(device.app_info_calls, [PDD_PACKAGE]) + self.assertEqual( + device.rpc_calls, + [ + ("takeScreenshot", [1, 80], 7.5), + ("dumpWindowHierarchy", [False, 50], 7.5), + ], + ) + self.assertTrue(result.screenshot_path.is_file()) + self.assertTrue(result.hierarchy_path.is_file()) + self.assertIn('"sha256"', manifest) + self.assertNotIn("page body must stay out of manifest", manifest) + self.assertNotIn(SERIAL, manifest) + self.assertIn('"channel": "usb"', manifest) + + def test_capture_failure_cleans_staging_and_does_not_publish_partial_output(self) -> None: + device = FakeUiDevice(fail_dump=True) + capturer = DeviceBaselineCapturer(StaticAdbClient(), lambda serial: device, timeout_seconds=5) + + with tempfile.TemporaryDirectory() as directory: + parent = Path(directory) + output = parent / "baseline" + with self.assertRaises(BaselineCaptureError) as raised: + capturer.capture(SERIAL, output) + + self.assertFalse(output.exists()) + self.assertEqual(list(parent.iterdir()), []) + self.assertNotIn("mock dump failed", str(raised.exception)) + + def test_existing_output_is_never_overwritten(self) -> None: + with tempfile.TemporaryDirectory() as directory: + output = Path(directory) / "baseline" + output.mkdir() + sentinel = output / "keep.txt" + sentinel.write_text("preserve", encoding="utf-8") + capturer = DeviceBaselineCapturer(StaticAdbClient(), lambda serial: FakeUiDevice(), timeout_seconds=5) + + with self.assertRaises(BaselineCaptureError): + capturer.capture(SERIAL, output) + self.assertEqual(sentinel.read_text(encoding="utf-8"), "preserve") + + def test_no_reconnect_connector_passes_only_current_adb_device_object(self) -> None: + class ListedDevice: + serial = SERIAL + + listed = ListedDevice() + connected: list[object] = [] + + connector = NoReconnectUiautomatorConnector(lambda: [listed], lambda device: connected.append(device) or FakeUiDevice()) + connector(SERIAL) + + self.assertEqual(connected, [listed]) + + def test_no_reconnect_connector_refuses_disappeared_serial(self) -> None: + connector = NoReconnectUiautomatorConnector(lambda: [], lambda device: FakeUiDevice()) + + with self.assertRaises(BaselineCaptureError) as raised: + connector(SERIAL) + self.assertIn("拒绝自动重连", str(raised.exception)) + + def test_connector_exception_is_redacted_and_publishes_no_partial_output(self) -> None: + def failing_connector(serial: str) -> FakeUiDevice: + raise RuntimeError(f"third party leaked {serial}") + + capturer = DeviceBaselineCapturer(StaticAdbClient(), failing_connector, timeout_seconds=5) + with tempfile.TemporaryDirectory() as directory: + output = Path(directory) / "baseline" + with self.assertRaises(BaselineCaptureError) as raised: + capturer.capture(SERIAL, output) + + self.assertNotIn(SERIAL, str(raised.exception)) + self.assertFalse(output.exists()) + self.assertEqual(list(Path(directory).iterdir()), []) + + def test_invalid_screenshot_base64_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" + 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): + capturer.capture(SERIAL, output) + + self.assertFalse(output.exists()) + self.assertEqual(list(Path(directory).iterdir()), []) + + 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: + if method == "dumpWindowHierarchy": + return "" + return super().jsonrpc_call(method, params, timeout) + + capturer = DeviceBaselineCapturer(StaticAdbClient(), lambda serial: InvalidHierarchyDevice(), timeout_seconds=5) + with tempfile.TemporaryDirectory() as directory: + output = Path(directory) / "baseline" + with self.assertRaises(BaselineCaptureError) as raised: + capturer.capture(SERIAL, output) + + self.assertNotIn("not-hierarchy", str(raised.exception)) + self.assertFalse(output.exists()) + self.assertEqual(list(Path(directory).iterdir()), []) + + def test_rpc_timeout_is_distinct_redacted_and_does_not_publish_partial_output(self) -> None: + class TimeoutRpcDevice(FakeUiDevice): + def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str: + raise HTTPTimeoutError(f"raw serial={SERIAL} xml=") + + capturer = DeviceBaselineCapturer(StaticAdbClient(), lambda serial: TimeoutRpcDevice(), timeout_seconds=5) + with tempfile.TemporaryDirectory() as directory: + output = Path(directory) / "baseline" + with self.assertRaises(BaselineCaptureTimeoutError) as raised: + capturer.capture(SERIAL, output) + + self.assertIn("超时", str(raised.exception)) + self.assertNotIn(SERIAL, str(raised.exception)) + self.assertNotIn("hierarchy", str(raised.exception)) + self.assertFalse(output.exists()) + self.assertEqual(list(Path(directory).iterdir()), []) + + def test_cli_validation_rejects_empty_serial_and_nonpositive_timeout(self) -> None: + empty_serial = parse_arguments(["--serial", "", "--output-dir", "baseline"]) + with self.assertRaisesRegex(ValueError, "非空 --serial"): + validate_arguments(empty_serial) + + nonpositive_timeout = parse_arguments(["--serial", SERIAL, "--output-dir", "baseline", "--timeout", "0"]) + with self.assertRaisesRegex(ValueError, "必须大于 0"): + validate_arguments(nonpositive_timeout) diff --git a/docs/03-tech-stack.md b/docs/03-tech-stack.md index de01962..3976761 100644 --- a/docs/03-tech-stack.md +++ b/docs/03-tech-stack.md @@ -72,6 +72,36 @@ | 测试 | `go test ./...` | `.\.venv\Scripts\python.exe -m unittest discover -s tests -t .` | | 静态检查 | `go vet ./...` | `.\.venv\Scripts\python.exe -m compileall -q src tests scripts` | +### T-101 设备基线取证（代码已完成，等待人工真机验收） + +`client/scripts/capture_device_baseline.py` 只允许对**手工明确填写**的 ADB serial 做连接前核验、 +设备型号 / Android / 拼多多版本读取、截图和 `dump_hierarchy(compressed=False)`。它不打开商品、 +不读取页面判据，也不执行采购、下单或付款动作。多个在线通道必须完成 `getprop` 物理身份比对： +同一手机 USB + WiFi 同时在线，或任一在线通道的身份读取失败，都会 fail closed，不能随机继续。 + +人工验收前，先由人把手机切换到不含收货地址、手机号、支付信息或其他无关隐私的安全页面，再在 +`adb devices -l` 中**手工复制**一个在线 serial；USB 和 WiFi 分别验收，且每次只保留一个通道在线。 +WiFi 通道必须由人先行建立；脚本禁止 `adb connect`、`adb disconnect` 或自动重连。以下命令中的尖括号 +必须替换为该次人工确认的实际 serial，不能省略或改成自动选择： + +```powershell +# 仓库根目录；先手工确认设备状态，命令本身只读 ADB 清单 +D:\Portable\adb\adb.exe devices -l + +# USB：粘贴该次 devices -l 显示的 USB serial +.\client\.venv\Scripts\python.exe client\scripts\capture_device_baseline.py --serial --output-dir "$env:LOCALAPPDATA\cmbuyer\artifacts\T-101\usb-baseline" --timeout 10 --adb D:\Portable\adb\adb.exe + +# WiFi：由人先建立 WiFi ADB 通道、断开 USB 后，粘贴该次 devices -l 显示的 WiFi serial +.\client\.venv\Scripts\python.exe client\scripts\capture_device_baseline.py --serial --output-dir "$env:LOCALAPPDATA\cmbuyer\artifacts\T-101\wifi-baseline" --timeout 10 --adb D:\Portable\adb\adb.exe +``` + +成功时输出目录仅包含截图、完整 XML 与不含页面正文的 `manifest.json`（设备元数据、通道、时间、 +文件 SHA-256 和 serial 哈希）。`--timeout` 约束 ADB 命令、ADB socket 及 `takeScreenshot` / +`dumpWindowHierarchy(compressed=False, max_depth=50)` 的公开 JSON-RPC 调用；uiautomator2 初始化仍有 +上游固定启动上限。XML 仅留在本机明确指定的证据目录；人工必须先在本地检查截图/XML，再只记录路径 +和哈希，不得把原始证据提交 Git；之后把设备型号、Android、拼多多版本及 USB/WiFi 结论记录到 T-101， +才能完成该任务。 + Windows 的标准入口是仓库根 `./init.ps1`。它要求 Go、两端目录及其哨兵文件存在；已有合规 `client/.venv` 时，所有采购工具检查与 validator 都使用该解释器。只有 venv 不存在时，才从 `py -0p` 枚举的版本中确定性选择最高的 Python 3.11+ 创建它；没有合规版本时明确失败，绝不回退默认 `python`。 diff --git a/docs/04-architecture.md b/docs/04-architecture.md index 3bb7794..02775da 100644 --- a/docs/04-architecture.md +++ b/docs/04-architecture.md @@ -386,7 +386,7 @@ PENDING_RETRIAL ─────────────┴─claim→ CLAIMED | 同一商品两趟结果不一致 | 第二趟价格变了、规格选项变了或商品下架 | 闸门二拦截；一律转人工，不自动放弃也不自动继续 | | 图搜结果含跨类目商品（V2） | 搜服装出现纸巾 | B 路径只产 goods_id 且限 5 个；后续用 VLM 看截图筛同款 | | WiFi ADB 稳定性 | 息屏、换网、DHCP 续租会断连 | 超时可配置；断连视为技术失败并保留现场，不重试点击 | -| 同一手机 USB + WiFi 同时在线 | `adb devices` 列出两条，自动选设备会失败 | 设备档案必须显式指定 serial，不允许留空自动选 | +| 同一手机 USB + WiFi 同时在线 | `adb devices` 列出两条，自动选设备会失败 | serial 必填；多在线通道必须读到 `ro.serialno` 或 `ro.boot.serialno` 才能比对。身份一致或任一身份读取失败时都 fail closed，不能以相同 model/product 猜测后继续 | | 不可逆动作的重试 | 点击「现在买」后超时，无法判断订单是否已创建 | 一律转人工并预留金额额度，**禁止自动重试点击** | | 双端契约漂移 | 两端独立演进会静默不兼容 | 契约改动必跑完整门禁；[api.md](api.md) 是唯一权威 | diff --git a/docs/current-state.md b/docs/current-state.md index f1edd6e..d543215 100644 --- a/docs/current-state.md +++ b/docs/current-state.md @@ -18,17 +18,18 @@ 使用 Python 3.11+ / uiautomator2 / PySide6。 详见 [`03-tech-stack.md`](03-tech-stack.md) - 生产代码：`admin/` 已有最小 Go 服务、健康检查、核心领域模型、SQLite 迁移与任务状态机； - `client/` 已有 Python 包、PySide6 最小入口、运行目录与日志脱敏策略；尚无真机采购流程 -- 测试：采购服务已覆盖健康检查、核心模型、迁移与状态机等离线包级测试；采购工具 6 项离线单元测试 - （不连接真机） + `client/` 已有 Python 包、PySide6 最小入口、运行目录与日志脱敏策略，以及显式 serial 的 ADB + 连接边界与本地基线取证 CLI；尚无真机采购流程 +- 测试：采购服务已覆盖健康检查、核心模型、迁移与状态机等离线包级测试；采购工具 27 项离线单元测试 + （全部 mock，不连接真机） - 数据：SQLite 核心表与迁移已落成；无业务实例数据 - 标准启动路径：Windows PowerShell 运行 `./init.ps1`，Unix shell 运行 `./init.sh`。Windows 入口 优先使用合规的既有 venv；仅在其缺失时才从 Python Launcher 已安装版本中选择最高的 Python 3.11+， 并且不覆盖低版本环境；成功后打印真实启动命令。 - 标准验证路径：`./init.ps1` 已实际跑通 admin 的 mod download / test / vet / build、client 的 editable install / 包导入 / unittest / compileall，以及仓库上下文校验。可单独运行两端命令诊断。 -- 当前 blocker：无外部 blocker。T-002、T-003、T-004 已完成；T-101 真机环境盘点已就绪， - 但尚未真机验收。桌面 GUI 与真机流程尚未验收。 +- 当前 blocker：T-101 等待人工分别完成 USB 与 WiFi 真机验收；这是 Phase 1 后续任务的门禁。 + T-002、T-003、T-004 已完成，桌面 GUI 与真机流程尚未验收。 ## 当前目录要点 @@ -39,7 +40,7 @@ | `docs/design/` | 已有（6 个原型） | web 登录 / 建单 / 工作台 / 详情，desk 采购执行 / 配置；均已人工确认 | | `scripts/` | 已有 | 上下文门禁、Vikunja 单向导出与 MCP 启动包装 | | `admin/` | 已初始化 | Go 1.23+ / gin / SQLite，含核心模型、迁移与状态机；无真机采购执行 | -| `client/` | 已初始化 | Python 3.11+ 包、依赖源、PySide6 最小入口、离线测试与 wheel 元数据检查；无真机或采购流程 | +| `client/` | 已初始化 | Python 3.11+ 包、依赖源、PySide6 最小入口、显式 serial 的设备基线取证、离线测试与 wheel 元数据检查；无采购流程 | | `init.ps1` / `init.sh` | 已完成 | 统一安装与离线验证入口；PowerShell 优先复用合规 venv，缺失时自动选择最高的 Python 3.11+，Unix 缺工具链明确失败 | ## 任务状态 @@ -50,8 +51,8 @@ 源码目录契约）、T-008（Vikunja 任务权威与单向导出）、T-009（MVP 关键路径与并行波次）， 以及 T-001（采购服务 Go 骨架）。 - 已完成：T-002（采购工具 Python 骨架）、T-003（双端统一初始化与验证入口）、 - T-004（核心数据模型）。T-101（真机环境盘点）已建档并就绪，尚未真机验收，接续推进 - T-101 → T-102 → T-103。 + T-004（核心数据模型）。T-101（真机环境盘点）已完成离线实现和 mock 测试，仍为 `DOING`， + 必须等待人工 USB、WiFi 两通道真机验收后才能完成；接续推进 T-101 → T-102 → T-103。 - T-103 是当前最高优先级和 MVP 生死线。通过前不开发依赖真机可读字段的 Phase 2 生产页面。 - 已确认原型继续只作信息架构依据；原型假数据不调用真实接口、不驱动真机。真机结论改变 可读字段时必须先回修原型与交互清单。 @@ -93,10 +94,28 @@ cd client ``` `client/requirements.txt` 是唯一依赖来源，`client/pyproject.toml` 动态读取它生成 wheel 的 -`Requires-Dist`。Python 3.12 已验证 6 项离线测试、编译与 wheel 元数据；完整运行时依赖安装 +`Requires-Dist`。Python 3.12 已验证 27 项离线测试、编译与 wheel 元数据；完整运行时依赖安装 （`pip install -e .`）已通过。`init.ps1` 优先使用合规既有 venv，缺失时自动选择最高的 Python 3.11+； 本机现有 venv 实际验证为 Python 3.12。桌面 GUI 与真机流程未作为 T-003 验收执行。 +T-101 的人工真机验收命令（先把手机切到不含收货地址、手机号、支付信息或其他无关隐私的安全页面； +必须从 `adb devices -l` 手工复制在线 serial，不能留空或自动选择）： + +```powershell +# 仓库根目录；USB 和 WiFi 分开执行，每次只保留一个通道在线 +D:\Portable\adb\adb.exe devices -l +.\client\.venv\Scripts\python.exe client\scripts\capture_device_baseline.py --serial --output-dir "$env:LOCALAPPDATA\cmbuyer\artifacts\T-101\usb-baseline" --timeout 10 --adb D:\Portable\adb\adb.exe +# WiFi 必须由人先建立通道、断开 USB 后再手工粘贴在线 WiFi serial；脚本不自动 connect/reconnect +.\client\.venv\Scripts\python.exe client\scripts\capture_device_baseline.py --serial --output-dir "$env:LOCALAPPDATA\cmbuyer\artifacts\T-101\wifi-baseline" --timeout 10 --adb D:\Portable\adb\adb.exe +``` + +脚本只读取非敏感设备元数据、拼多多版本、截图和完整 XML；不会打开商品或写页面判据。它对同机双 +通道、身份读取失败、offline、unauthorized、超时或 serial 不存在均 fail closed。成功目录的 +`manifest.json` 只记录元数据、产物路径和 SHA-256，不记录 XML 页面正文或原始 serial。人工记录 +两次验收结果前，T-101 必须保持 `DOING`。`--timeout` 约束 ADB 命令、ADB socket 以及截图/节点树的 +公开 JSON-RPC 调用；uiautomator2 初始化仍有上游固定启动上限。截图/XML 只在本地人工检查，执行记录 +只写路径和 SHA-256，原始证据不得提交 Git。 + ## 关键背景 本项目是 `cmroubao`（Go 后端 + Android AccessibilityService）与 `cmpdd` diff --git a/docs/tasks/T-101.md b/docs/tasks/T-101.md index 96f2a79..623f3ba 100644 --- a/docs/tasks/T-101.md +++ b/docs/tasks/T-101.md @@ -23,7 +23,7 @@ write_paths: - docs/current-state.md --- - + ## 问题 / 背景 T-002 已建立采购工具骨架，但本项目还没有对实际 Android 手机、ADB 通道或 uiautomator2 做过取证。后续所有拼多多页面判据都依赖稳定、显式且可审计的设备连接；若复用前序项目结论或自动猜设备，会把错误设备和旧页面事实带入生产流程。 @@ -55,6 +55,10 @@ T-002 已建立采购工具骨架，但本项目还没有对实际 Android 手 ### 2026-08-03T10:56:19Z · ila T-101 已领取，Git 状态将提交为 DOING。只读环境盘点：adb 位于 D:\Portable\adb\adb.exe；当前同时列出 WiFi serial 192.168.0.173:5555 与 USB serial 3B65BD02H7F00000，二者 product/model/device 均为 PKG110/PKG110/OP5D2BL1。该现场必须由实现识别为同机双通道并 fail closed。尚未截图、dump 或操作手机，本记录不是人工真机验收；needs_device 规则继续生效。 + +### 2026-08-03T11:27:08Z · ila + +2026-08-03 离线实现与主审完成：新增显式 serial 的 ADB 边界、USB/WiFi 同机多通道 fail-closed、禁止 uiautomator2 隐式重连的连接器，以及只读基线 CLI（型号、Android、拼多多版本、截图、compressed=false XML、manifest SHA-256）。主 agent 三轮审查后补齐：截图/XML JSON-RPC 可配置超时、adbutils/uiautomator2 类型化超时、完整硬件身份集合比对、非法截图/XML fail-closed、异常脱敏、暂存清理，以及人工安全页面/原始证据不入 Git 的操作要求。独立验证：init.ps1 通过；admin go test/vet/build 通过；client 27 项 mock 单测、compileall、wheel METADATA、validator、bash -n 与 diff check 通过。未连接或操作真机，T-101 继续保持 DOING；等待人分别完成 USB 与 WiFi 取证并记录 manifest 中的设备型号、Android、拼多多版本、路径与 SHA-256。已知边界：ADB/ADB socket/截图与节点树 RPC 超时可配置，uiautomator2 初始化仍受上游固定启动上限约束。 ## 边界