338 lines
13 KiB
Python
338 lines
13 KiB
Python
# -*- encoding=utf8 -*-
|
||
"""
|
||
PCAPDroid Manager - ADB 命令行控制版
|
||
通过 ADB 命令控制 PCAPDroid 的启停和状态监控，不依赖 Airtest/Poco。
|
||
"""
|
||
|
||
import time
|
||
import logging
|
||
import csv
|
||
import json
|
||
import base64
|
||
from collections import defaultdict
|
||
from pathlib import Path
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class AppBlocker:
|
||
"""应用阻塞器：根据包名或标签名导入域名屏蔽规则到 PCAPDroid blocklist
|
||
|
||
规则库由两个 CSV 文件组成：
|
||
- app_list: 包名 -> 应用名/标签映射
|
||
- url_lib: 应用名 -> 域名列表
|
||
"""
|
||
|
||
BLOCKLIST_ACTIVITY = (
|
||
"com.emanuelef.remote_capture/"
|
||
"com.emanuelef.remote_capture.activities.BlocklistCtrl"
|
||
)
|
||
|
||
def __init__(self, adb_helper, app_list_file=None, url_lib_file=None):
|
||
"""
|
||
Args:
|
||
adb_helper: ADBHelper 实例
|
||
app_list_file: 应用列表 CSV 路径（包含 package_name, app_name, final_tag_id 列）
|
||
url_lib_file: URL 库 CSV 路径（包含 app, url 列）
|
||
"""
|
||
self.adb_helper = adb_helper
|
||
self.app_list_file = app_list_file
|
||
self.url_lib_file = url_lib_file
|
||
self._apps = None
|
||
self._urls = None
|
||
|
||
# ------------------------------------------------------------------
|
||
# 规则加载
|
||
# ------------------------------------------------------------------
|
||
def _load_app_mapping(self):
|
||
if self._apps is not None:
|
||
return self._apps
|
||
self._apps = {}
|
||
if not self.app_list_file or not Path(self.app_list_file).exists():
|
||
logger.warning(f"[AppBlock] 应用列表文件不存在: {self.app_list_file}")
|
||
return self._apps
|
||
try:
|
||
with open(self.app_list_file, "r", encoding="utf-8") as f:
|
||
reader = csv.DictReader(f)
|
||
for row in reader:
|
||
if row.get("status", "Active") != "Active":
|
||
continue
|
||
pkg_names = row.get("package_name", "")
|
||
if pkg_names:
|
||
for pkg in pkg_names.split(","):
|
||
pkg = pkg.strip().strip('"')
|
||
if pkg:
|
||
self._apps[pkg] = {
|
||
"app_name": row.get("app_name", "Unknown"),
|
||
"category": row.get("final_tag_id", "Unknown"),
|
||
}
|
||
logger.info(f"[AppBlock] 已加载应用映射: {len(self._apps)} 条")
|
||
except Exception as e:
|
||
logger.error(f"[AppBlock] 加载应用列表失败: {e}")
|
||
return self._apps
|
||
|
||
def _load_url_lib(self):
|
||
if self._urls is not None:
|
||
return self._urls
|
||
self._urls = {}
|
||
if not self.url_lib_file or not Path(self.url_lib_file).exists():
|
||
logger.warning(f"[AppBlock] URL 库文件不存在: {self.url_lib_file}")
|
||
return self._urls
|
||
try:
|
||
with open(self.url_lib_file, "r", encoding="utf-8") as f:
|
||
reader = csv.DictReader(f)
|
||
for row in reader:
|
||
app_name = row.get("app", "")
|
||
url = row.get("url", "")
|
||
if app_name and url:
|
||
self._urls.setdefault(app_name, []).append(url)
|
||
logger.info(f"[AppBlock] 已加载 URL 库: {len(self._urls)} 个应用")
|
||
except Exception as e:
|
||
logger.error(f"[AppBlock] 加载 URL 库失败: {e}")
|
||
return self._urls
|
||
|
||
# ------------------------------------------------------------------
|
||
# 域名查找
|
||
# ------------------------------------------------------------------
|
||
def find_domains_by_package(self, package_name):
|
||
apps = self._load_app_mapping()
|
||
urls = self._load_url_lib()
|
||
if package_name not in apps:
|
||
return [], None
|
||
app_info = apps[package_name]
|
||
app_name = app_info["app_name"]
|
||
domains = urls.get(app_name, [])
|
||
return domains, app_info
|
||
|
||
def find_domains_by_tag(self, tag_name):
|
||
urls = self._load_url_lib()
|
||
if tag_name in urls:
|
||
return urls[tag_name], tag_name
|
||
for app_name in urls:
|
||
if app_name.lower() == tag_name.lower():
|
||
return urls[app_name], app_name
|
||
return [], None
|
||
|
||
# ------------------------------------------------------------------
|
||
# 域名简化
|
||
# ------------------------------------------------------------------
|
||
@staticmethod
|
||
def _extract_main_domain(domain):
|
||
parts = domain.lower().split(".")
|
||
if len(parts) >= 2:
|
||
return ".".join(parts[-2:])
|
||
return domain
|
||
|
||
@staticmethod
|
||
def simplify_domains(domains):
|
||
if not domains:
|
||
return []
|
||
groups = defaultdict(list)
|
||
for d in domains:
|
||
groups[AppBlocker._extract_main_domain(d)].append(d)
|
||
simplified = []
|
||
for main_domain, sub_domains in groups.items():
|
||
if len(sub_domains) >= 2:
|
||
simplified.append(main_domain)
|
||
else:
|
||
simplified.append(sub_domains[0])
|
||
return sorted(set(simplified))
|
||
|
||
# ------------------------------------------------------------------
|
||
# Blocklist 导入/清除
|
||
# ------------------------------------------------------------------
|
||
def import_blocklist(self, domains, mode="set"):
|
||
"""导入域名规则到 PCAPDroid blocklist
|
||
|
||
Args:
|
||
domains: 域名列表
|
||
mode: "set" (替换) 或 "append" (追加)
|
||
Returns:
|
||
bool: 是否成功
|
||
"""
|
||
if not domains:
|
||
logger.warning("[AppBlock] 无域名可导入")
|
||
return False
|
||
|
||
original_count = len(domains)
|
||
simplified = self.simplify_domains(domains)
|
||
logger.info(f"[AppBlock] 域名简化: {original_count} -> {len(simplified)}")
|
||
|
||
rules = {"rules": [{"type": "HOST", "value": d} for d in simplified]}
|
||
json_str = json.dumps(rules)
|
||
json_b64 = base64.b64encode(json_str.encode("utf-8")).decode("ascii")
|
||
action = "set_rules" if mode == "set" else "append_rules"
|
||
|
||
logger.info(f"[AppBlock] 正在导入 {len(simplified)} 条规则 (mode={mode})")
|
||
output = self.adb_helper.run_cmd([
|
||
"shell", "am", "start", "-W",
|
||
"-n", self.BLOCKLIST_ACTIVITY,
|
||
"-e", "action", action,
|
||
"-e", "rules_json_base64", json_b64,
|
||
])
|
||
logger.info(f"[AppBlock] 导入输出: {output}")
|
||
time.sleep(1)
|
||
return True
|
||
|
||
def clear_blocklist(self):
|
||
"""清除 PCAPDroid blocklist 中的所有规则"""
|
||
logger.info("[AppBlock] 正在清除 blocklist")
|
||
try:
|
||
output = self.adb_helper.run_cmd([
|
||
"shell", "am", "start", "-W",
|
||
"-n", self.BLOCKLIST_ACTIVITY,
|
||
"-e", "action", "clear",
|
||
])
|
||
logger.info(f"[AppBlock] 清除输出: {output}")
|
||
time.sleep(1)
|
||
return True
|
||
except Exception as e:
|
||
logger.warning(f"[AppBlock] 清除 blocklist 失败: {e}")
|
||
return False
|
||
|
||
def import_for_package(self, package_name, mode="set", tag=None):
|
||
"""根据包名（或标签名）查找并导入 blocklist 规则
|
||
|
||
Args:
|
||
package_name: 应用包名
|
||
mode: "set" 或 "append"
|
||
tag: 可选的标签名，优先使用标签名查找
|
||
Returns:
|
||
bool: 是否成功
|
||
"""
|
||
if tag:
|
||
domains, matched = self.find_domains_by_tag(tag)
|
||
if matched:
|
||
logger.info(f"[AppBlock] 通过标签 '{matched}' 找到 {len(domains)} 个域名")
|
||
return self.import_blocklist(domains, mode=mode)
|
||
logger.warning(f"[AppBlock] 标签 '{tag}' 未找到，尝试通过包名查找")
|
||
|
||
domains, app_info = self.find_domains_by_package(package_name)
|
||
if app_info:
|
||
logger.info(
|
||
f"[AppBlock] 包名 '{package_name}' 对应应用: {app_info['app_name']}"
|
||
)
|
||
return self.import_blocklist(domains, mode=mode)
|
||
|
||
logger.warning(f"[AppBlock] 包名 '{package_name}' 在规则库中未找到")
|
||
return False
|
||
|
||
|
||
class PCAPManager:
|
||
"""负责 PCAPdroid 应用的控制、配置与状态监控（ADB 命令版）
|
||
|
||
使用 adb_helper.run_cmd() 执行所有 ADB 命令，确保异常捕获链路完整。
|
||
注意：不使用 adb_helper.shell()，因为底层 ADB.shell() 会对参数做 shlex.quote()，
|
||
导致 am start 的 -e 参数被错误引用。直接通过 run_cmd(["shell", ...]) 绕过。
|
||
"""
|
||
|
||
PKG_NAME = "com.emanuelef.remote_capture"
|
||
CAPTURE_CTRL = f"{PKG_NAME}/{PKG_NAME}.activities.CaptureCtrl"
|
||
|
||
def __init__(self, adb_helper, *, root_capture=True, app_blocker=None):
|
||
"""
|
||
Args:
|
||
adb_helper: ADBHelper 实例，用于执行 ADB 命令
|
||
root_capture: 是否使用 root 模式抓包（开启应用阻塞时必须为 False）
|
||
app_blocker: AppBlocker 实例（可选）
|
||
"""
|
||
self.adb_helper = adb_helper
|
||
self.root_capture = bool(root_capture)
|
||
self.app_blocker = app_blocker
|
||
|
||
# ------------------------------------------------------------------
|
||
# Blocklist 快捷方法
|
||
# ------------------------------------------------------------------
|
||
def import_app_blocklist(self, package_name, mode="set", tag=None):
|
||
"""导入应用阻塞规则（委托给 AppBlocker）"""
|
||
if self.app_blocker:
|
||
return self.app_blocker.import_for_package(package_name, mode=mode, tag=tag)
|
||
logger.warning("[PCAP] AppBlocker 未初始化，跳过导入 blocklist")
|
||
return False
|
||
|
||
def clear_app_blocklist(self):
|
||
"""清除应用阻塞规则（委托给 AppBlocker）"""
|
||
if self.app_blocker:
|
||
return self.app_blocker.clear_blocklist()
|
||
return False
|
||
|
||
# ------------------------------------------------------------------
|
||
# 抓包控制
|
||
# ------------------------------------------------------------------
|
||
def start_capture(self, package_name, pcap_dump_mode="pcap_file", max_retries=3):
|
||
"""开启抓包流程
|
||
|
||
Args:
|
||
package_name: 目标应用包名
|
||
pcap_dump_mode: 抓包数据存储模式，默认 pcap_file
|
||
max_retries: 最大重试次数
|
||
|
||
Returns:
|
||
bool: 是否成功开启抓包
|
||
"""
|
||
print(f"[PCAP] 准备开启抓包: {package_name} (root={self.root_capture})")
|
||
|
||
# 如果已经在运行，先停止
|
||
if self.is_running():
|
||
print("[PCAP] 检测到抓包正在运行，先停止...")
|
||
self.stop_capture()
|
||
time.sleep(2)
|
||
|
||
for attempt in range(max_retries):
|
||
# 通过 CaptureCtrl Activity 启动抓包
|
||
print(f"[PCAP] 执行启动命令 (尝试 {attempt + 1}/{max_retries})")
|
||
output = self.adb_helper.run_cmd([
|
||
"shell", "am", "start", "-n", self.CAPTURE_CTRL,
|
||
"-e", "action", "start",
|
||
"-e", "pcap_dump_mode", pcap_dump_mode,
|
||
"-e", "app_filter", package_name,
|
||
"-e", "root_capture", "true" if self.root_capture else "false",
|
||
])
|
||
logger.info(f"[PCAP] am start 输出: {output}")
|
||
|
||
# 等待抓包启动（PCAPDroid 需要约 15 秒才能在通知栏显示状态）
|
||
time.sleep(15)
|
||
|
||
# 按 Home 键将 PCAPDroid 推到后台
|
||
self.adb_helper.run_cmd(["shell", "input", "keyevent", "3"])
|
||
time.sleep(1)
|
||
|
||
# 验证是否成功启动
|
||
if self.is_running():
|
||
print("[PCAP] 抓包已成功启动")
|
||
return True
|
||
|
||
print(f"[PCAP] 第 {attempt + 1} 次启动未成功，重试...")
|
||
time.sleep(2)
|
||
|
||
print("[PCAP] 抓包启动失败")
|
||
return False
|
||
|
||
def stop_capture(self):
|
||
"""停止抓包"""
|
||
print("[PCAP] 停止 PCAPdroid")
|
||
self.adb_helper.run_cmd([
|
||
"shell", "am", "start", "-n", self.CAPTURE_CTRL,
|
||
"-e", "action", "stop",
|
||
])
|
||
time.sleep(1)
|
||
|
||
def is_running(self):
|
||
"""通过通知栏检查抓包是否正在运行
|
||
|
||
Returns:
|
||
bool: 抓包是否正在运行
|
||
"""
|
||
try:
|
||
output = self.adb_helper.run_cmd([
|
||
"shell", "dumpsys", "notification", "--noredact"
|
||
])
|
||
# 在整个通知输出中搜索 PCAPDroid 的运行通知
|
||
if self.PKG_NAME in output and "Capture running" in output:
|
||
logger.info("[PCAP] 状态检测: 抓包正在运行")
|
||
return True
|
||
logger.info("[PCAP] 状态检测: 抓包未运行")
|
||
except Exception as e:
|
||
logger.warning(f"[PCAP] 检查运行状态失败: {e}")
|
||
return False
|