77 lines
2.9 KiB
Python
77 lines
2.9 KiB
Python
"""采集指定 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())
|