Files
workphone-sdk/sdk/agent/anti_ban/device_guard.py

431 lines
16 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""
设备环境守卫 — Agent 启动时执行全面自检
检测项:
1. Root 隐藏状态Shamiko / MagiskHide
2. Frida 反检测状态(进程名/端口/内存特征)
3. 设备指纹唯一性(上报并校验碰撞)
4. 模拟器检测Build属性/传感器/电池/特征文件)
5. 无障碍服务状态
6. SELinux 状态
7. 关键 App 安装隐藏HMA-OSS 检测)
"""
import hashlib
import logging
import os
import re
import time
from typing import Dict, Any, List, Tuple
logger = logging.getLogger(__name__)
class DeviceGuard:
"""设备环境守卫 — 启动自检 + 持续巡检"""
# 不应被目标 App 看到的包名
HIDDEN_PACKAGES = [
"com.topjohnwu.magisk",
"io.github.vvb2060.magisk",
"me.weishu.kernelsu",
"org.lsposed.manager",
"moe.shizuku.privileged.api",
"eu.chainfire.supersu",
"com.noshufou.android.su",
"com.koushikdutta.superuser",
"com.termux",
]
# 模拟器 Build 特征
EMU_FINGERPRINTS = [
"goldfish", "generic", "vbox", "sdk_gphone",
"Andy", "Droid4X", "nox", "bluestacks",
"genymotion", "ttVM_Hdragon",
]
# Frida 默认特征
FRIDA_INDICATORS = [
"/data/local/tmp/frida-server",
"/data/local/tmp/re.frida.server",
]
def __init__(self, device):
"""
Args:
device: uiautomator2 设备对象
"""
self.d = device
self.report: Dict[str, Any] = {}
# ------------------------------------------------------------------
# 主入口
# ------------------------------------------------------------------
def run_full_check(self) -> Dict[str, Any]:
"""执行全面自检,返回结果报告"""
logger.info("🛡️ 设备环境自检开始...")
start = time.time()
checks: List[Tuple[str, callable]] = [
("root_hidden", self._check_root_hidden),
("frida_stealth", self._check_frida_stealth),
("fingerprint", self._collect_fingerprint),
("emulator", self._check_emulator),
("accessibility", self._check_accessibility),
("selinux", self._check_selinux),
("app_hidden", self._check_app_hidden),
("dangerous_props", self._check_dangerous_props),
]
results = {}
warnings = []
for name, fn in checks:
try:
result = fn()
results[name] = result
if result.get("warning"):
warnings.append(f"{name}: {result['warning']}")
logger.warning(f" ⚠️ {name}: {result['warning']}")
else:
logger.info(f"{name}: OK")
except Exception as e:
results[name] = {"ok": False, "error": str(e)}
warnings.append(f"{name}: {e}")
logger.error(f"{name}: {e}")
elapsed = time.time() - start
self.report = {
"ok": len(warnings) == 0,
"warnings": warnings,
"warning_count": len(warnings),
"checks": results,
"elapsed_ms": int(elapsed * 1000),
"timestamp": int(time.time()),
}
# AB-04量产环境红灯分级。联调态(dev)暴露记为黄,量产态(prod)关键暴露记为红。
self.report["redlight"] = self._classify_redlight(results)
if warnings:
logger.warning(f"🛡️ 自检完成 — {len(warnings)} 项警告 ({elapsed:.1f}s)")
else:
logger.info(f"🛡️ 自检全部通过 ({elapsed:.1f}s)")
return self.report
def _classify_redlight(self, results: Dict[str, Any]) -> Dict[str, Any]:
"""量产环境红灯分级AB-04
env`WP_ENV=prod` 视为量产态,关键暴露 → 红灯(应阻断上线);
否则联调态(dev),同样的暴露只记黄灯(允许联调,但仍提示)。
关键暴露 = Frida 默认特征 / su 可见 / ro.debuggable=1 / adbd running。
"""
env = (os.environ.get("WP_ENV") or os.environ.get("WORKPHONE_ENV") or "dev").lower()
is_prod = env in ("prod", "production", "release")
critical: List[str] = []
frida = results.get("frida_stealth") or {}
if not frida.get("ok", True):
critical.append(f"frida_stealth: {frida.get('warning', '默认特征暴露')}")
root = results.get("root_hidden") or {}
if root.get("warning", "").startswith("su 二进制可见"):
critical.append(f"root_hidden: {root['warning']}")
props = (results.get("dangerous_props") or {}).get("dangerous_props") or {}
if props.get("ro.debuggable"):
critical.append("dangerous_props: ro.debuggable=1")
if props.get("adb_running"):
critical.append("dangerous_props: adbd running")
level = "green"
if critical:
level = "red" if is_prod else "yellow"
return {
"env": env,
"is_prod": is_prod,
"level": level,
"critical_exposures": critical,
"advice": (
"量产红灯:关键暴露须先隐藏(Shamiko/改名改端口/关 adbd/ro.debuggable=0)再上线"
if level == "red" else
("联调态黄灯:暴露项不阻断联调,量产前须清零" if level == "yellow" else "环境干净")
),
}
# ------------------------------------------------------------------
# 1. Root 隐藏检测
# ------------------------------------------------------------------
def _check_root_hidden(self) -> dict:
result = {"ok": True}
# su 二进制是否可被常规方式发现
su_paths = ["/system/bin/su", "/system/xbin/su", "/sbin/su"]
for p in su_paths:
out = self._shell(f"ls {p} 2>/dev/null").strip()
if out and "No such file" not in out:
result["ok"] = False
result["warning"] = f"su 二进制可见: {p}"
return result
# Magisk 目录
magisk_dirs = ["/data/adb/magisk", "/data/adb/modules"]
for d in magisk_dirs:
out = self._shell(f"ls {d} 2>/dev/null").strip()
if out and "No such file" not in out:
# 目录存在但应被 Shamiko 隐藏不让目标App看到
result["magisk_dir_visible"] = True
# 检查 Shamiko 是否安装
shamiko = self._shell("ls /data/adb/modules/shamiko 2>/dev/null").strip()
result["shamiko_installed"] = bool(shamiko and "No such file" not in shamiko)
if not result["shamiko_installed"]:
result["warning"] = "Shamiko 未安装Root 可能暴露"
result["ok"] = False
return result
# ------------------------------------------------------------------
# 2. Frida 反检测
# ------------------------------------------------------------------
def _check_frida_stealth(self) -> dict:
result = {"ok": True}
# 检查默认 frida-server 进程名
ps_out = self._shell("ps -A 2>/dev/null || ps").strip()
if "frida-server" in ps_out or "frida-agent" in ps_out:
result["ok"] = False
result["warning"] = "检测到默认 frida-server 进程名,必须使用 Phantom-Frida"
return result
# 检查默认端口 27042
netstat = self._shell("netstat -tlnp 2>/dev/null || ss -tlnp 2>/dev/null").strip()
if ":27042" in netstat:
result["ok"] = False
result["warning"] = "检测到 Frida 默认端口 27042"
return result
# 检查默认路径
for path in self.FRIDA_INDICATORS:
out = self._shell(f"ls {path} 2>/dev/null").strip()
if out and "No such file" not in out:
result["ok"] = False
result["warning"] = f"Frida 默认路径可见: {path}"
return result
# 检查 /proc/self/maps 中的 frida 字符串(从 shell 模拟目标App视角
maps = self._shell("cat /proc/self/maps 2>/dev/null | head -200").strip()
if "frida" in maps.lower():
result["ok"] = False
result["warning"] = "/proc/self/maps 中发现 frida 特征"
return result
# ------------------------------------------------------------------
# 3. 设备指纹采集
# ------------------------------------------------------------------
def _collect_fingerprint(self) -> dict:
fp_data = {}
props = {
"ro.serialno": "serial",
"ro.build.fingerprint": "build_fp",
"ro.build.display.id": "display_id",
"ro.product.model": "model",
"ro.product.brand": "brand",
"ro.product.device": "device",
"ro.product.board": "board",
"ro.hardware": "hardware",
"ro.boot.serialno": "boot_serial",
"persist.sys.timezone": "timezone",
"ro.build.version.sdk": "sdk_version",
}
for prop, key in props.items():
val = self._shell(f"getprop {prop} 2>/dev/null").strip()
if val:
fp_data[key] = val
# Android ID
android_id = self._shell(
"settings get secure android_id 2>/dev/null"
).strip()
if android_id:
fp_data["android_id"] = android_id
# MAC 地址
mac = self._shell(
"cat /sys/class/net/wlan0/address 2>/dev/null"
).strip()
if mac and mac != "00:00:00:00:00:00":
fp_data["mac"] = mac
# 蓝牙地址
bt = self._shell(
"settings get secure bluetooth_address 2>/dev/null"
).strip()
if bt:
fp_data["bluetooth"] = bt
# 屏幕参数
try:
info = self.d.info
fp_data["screen_w"] = info.get("displayWidth", 0)
fp_data["screen_h"] = info.get("displayHeight", 0)
except Exception:
pass
# IMEI需 root 或特殊权限)
imei = self._shell(
"service call iphonesubinfo 1 2>/dev/null | grep -oP \"'[^']+'\""
).strip()
if imei:
fp_data["imei_raw"] = imei
# 计算指纹哈希
fp_str = "|".join(f"{k}={v}" for k, v in sorted(fp_data.items()))
fp_hash = hashlib.md5(fp_str.encode()).hexdigest()
return {
"ok": True,
"fingerprint_hash": fp_hash,
"dimensions": len(fp_data),
"data": fp_data,
}
# ------------------------------------------------------------------
# 4. 模拟器检测
# ------------------------------------------------------------------
def _check_emulator(self) -> dict:
result = {"ok": True, "is_emulator": False, "signals": []}
# Build 属性
build_fp = self._shell("getprop ro.build.fingerprint 2>/dev/null").strip().lower()
hardware = self._shell("getprop ro.hardware 2>/dev/null").strip().lower()
product = self._shell("getprop ro.product.name 2>/dev/null").strip().lower()
for emu in self.EMU_FINGERPRINTS:
if emu.lower() in build_fp or emu.lower() in hardware or emu.lower() in product:
result["signals"].append(f"Build特征: {emu}")
# 特征文件
emu_files = [
"/dev/qemu_pipe", "/dev/goldfish_pipe",
"/system/lib/libc_malloc_debug_qemu.so",
"/sys/qemu_trace", "/dev/socket/qemud",
]
for f in emu_files:
out = self._shell(f"ls {f} 2>/dev/null").strip()
if out and "No such file" not in out:
result["signals"].append(f"特征文件: {f}")
# 传感器数量(真机通常 > 10模拟器 ≤ 3
sensors = self._shell(
"dumpsys sensorservice 2>/dev/null | grep -c 'Sensor' || echo 0"
).strip()
try:
sensor_count = int(sensors)
result["sensor_count"] = sensor_count
if sensor_count < 5:
result["signals"].append(f"传感器过少: {sensor_count}")
except ValueError:
pass
# 电池状态(模拟器通常电池 API 异常)
battery_status = self._shell(
"dumpsys battery 2>/dev/null | grep 'status' | head -1"
).strip()
if "status: 1" in battery_status:
result["signals"].append("电池状态异常(status=1=UNKNOWN)")
if result["signals"]:
result["is_emulator"] = True
result["ok"] = False
result["warning"] = f"疑似模拟器: {', '.join(result['signals'][:3])}"
return result
# ------------------------------------------------------------------
# 5. 无障碍服务检测
# ------------------------------------------------------------------
def _check_accessibility(self) -> dict:
out = self._shell(
"settings get secure enabled_accessibility_services 2>/dev/null"
).strip()
services = [s for s in out.split(":") if s.strip()] if out and out != "null" else []
return {
"ok": True,
"active_services": services,
"count": len(services),
}
# ------------------------------------------------------------------
# 6. SELinux 状态
# ------------------------------------------------------------------
def _check_selinux(self) -> dict:
out = self._shell("getenforce 2>/dev/null").strip().lower()
enforcing = out == "enforcing"
result = {"ok": True, "mode": out}
if not enforcing:
result["warning"] = f"SELinux 非 Enforcing 模式: {out}"
result["ok"] = False
return result
# ------------------------------------------------------------------
# 7. App 隐藏检测HMA-OSS
# ------------------------------------------------------------------
def _check_app_hidden(self) -> dict:
visible = []
for pkg in self.HIDDEN_PACKAGES:
out = self._shell(f"pm list packages {pkg} 2>/dev/null").strip()
if f"package:{pkg}" in out:
visible.append(pkg)
result = {"ok": len(visible) == 0, "visible_sensitive_apps": visible}
if visible:
result["warning"] = f"敏感 App 未隐藏: {', '.join(visible)}"
return result
# ------------------------------------------------------------------
# 8. 危险属性检测
# ------------------------------------------------------------------
def _check_dangerous_props(self) -> dict:
"""检测会暴露自动化/Root 的系统属性"""
dangerous = {}
# ro.debuggable
debuggable = self._shell("getprop ro.debuggable 2>/dev/null").strip()
if debuggable == "1":
dangerous["ro.debuggable"] = "1 (应为 0)"
# ro.secure
secure = self._shell("getprop ro.secure 2>/dev/null").strip()
if secure == "0":
dangerous["ro.secure"] = "0 (应为 1)"
# init.svc.adbd
adbd = self._shell("getprop init.svc.adbd 2>/dev/null").strip()
if adbd == "running":
dangerous["adb_running"] = True
result = {"ok": len(dangerous) == 0, "dangerous_props": dangerous}
if dangerous:
result["warning"] = f"危险属性: {list(dangerous.keys())}"
return result
# ------------------------------------------------------------------
# 工具
# ------------------------------------------------------------------
def _shell(self, cmd: str) -> str:
try:
timeout = float(os.environ.get("WP_GUARD_SHELL_TIMEOUT", "4") or "4")
return self.d.shell(cmd, timeout=timeout).output or ""
except Exception:
return ""