"""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 通道同时在线。""" class IntentLaunchUnconfirmedError(DeviceConnectionError): """`am start -W` 没有给出可确认的启动成功结果。""" @dataclass(frozen=True) class CommandResult: """可注入命令执行器的最小、可离线构造结果。""" stdout: str stderr: str = "" returncode: int = 0 @dataclass(frozen=True) class IntentLaunchSummary: """不含 Activity、页面内容或 ADB 输出的受限启动摘要。""" status: str returncode: int 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 start_pdd_view_intent(self, serial: str, goods_id: str) -> IntentLaunchSummary: """以参数数组启动唯一允许的拼多多 ACTION_VIEW Intent。 这里刻意不提供任意 shell 或任意 package 的执行接口。调用方必须先完成 ``inspect`` 和应用版本核验；本方法在本层从纯数字 ``goods_id`` 重建 URL，调用方不能 把另一个 URL 直接交给 ADB。本方法既不点击控件，也不解析 Activity 或页面文本。 """ selected_serial = _require_serial(serial) if ( not isinstance(goods_id, str) or not goods_id or any(character < "0" or character > "9" for character in goods_id) ): raise ValueError("goods_id 必须是纯数字") canonical_url = f"https://mobile.yangkeduo.com/goods.html?goods_id={goods_id}" result = self._run_checked( ( "-s", selected_serial, "shell", "am", "start", "-W", "-a", "android.intent.action.VIEW", "-d", canonical_url, "-p", "com.xunmeng.pinduoduo", ) ) if not any(line.strip() == "Status: ok" for line in result.stdout.splitlines()): raise IntentLaunchUnconfirmedError("商品链接启动结果无法确认，已停止后续取证。") return IntentLaunchSummary(status="ok", returncode=result.returncode) 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("指定设备不处于可用状态。")