移动端数据抓取实战:从零构建mitmproxy与MuMu模拟器的自动化抓包系统
在移动互联网时代,应用数据抓取已成为开发者必备的核心技能之一。无论是进行竞品分析、接口调试,还是构建自动化测试流程,能够精准捕获并解析APP的网络请求都显得尤为重要。不同于传统的抓包工具如Charles或Fiddler,mitmproxy以其开源、可编程的特性,为开发者提供了更灵活的中间人攻击(MITM)解决方案。本文将带领读者从零开始,搭建一套完整的移动端数据抓取系统,涵盖环境配置、证书处理、脚本编写等全流程实战要点。
1. 环境搭建与基础配置
1.1 mitmproxy的安装与验证
mitmproxy作为一款基于Python开发的中间人代理工具,其安装过程简洁明了。对于Python开发者而言,推荐使用pip进行安装:
pip install mitmproxy安装完成后,可通过以下命令验证安装是否成功:
mitmproxy --version若系统提示命令未找到,可能需要将Python的Scripts目录添加到系统环境变量中。mitmproxy实际上包含三个组件:
- mitmproxy:交互式控制台界面
- mitmdump:命令行接口,支持Python脚本扩展
- mitmweb:基于Web的图形界面
1.2 MuMu模拟器的特殊配置
网易出品的MuMu模拟器在抓包场景中表现优异,其网络配置灵活性远超其他安卓模拟器。安装完成后,需特别注意以下配置项:
- 网络模式选择:在设置->属性设置中,确保选择"桥接模式"而非"NAT模式"
- Root权限开启:部分APP会检测Root状态,MuMu需要在设置中显式开启
- 共享文件夹配置:用于证书传输,路径通常为/mnt/shared/
提示:部分国产APP会检测模拟器环境,此时可尝试修改模拟器的build.prop文件来伪装成真实设备
2. HTTPS抓包的核心:证书配置详解
2.1 本地证书安装
mitmproxy运行时会自动生成CA证书,存放于用户目录下的.mitmproxy文件夹中。Windows系统可通过资源管理器直接访问:
C:\Users\[用户名]\.mitmproxy\该目录包含多种格式的证书文件:
- mitmproxy-ca.pem:PEM格式证书
- mitmproxy-ca.p12:PKCS12格式证书
- mitmproxy-ca-cert.cer:DER编码证书
双击mitmproxy-ca.p12文件,按照向导完成证书导入。关键步骤包括:
- 选择"当前用户"存储位置
- 不设置密码(直接留空)
- 选择"将所有证书放入下列存储",并指定"受信任的根证书颁发机构"
2.2 模拟器证书安装
将mitmproxy-ca-cert.pem文件拖入MuMu模拟器的共享文件夹后,按以下步骤操作:
- 打开系统设置->安全->从SD卡安装
- 导航至共享文件夹(通常位于/storage/emulated/0/)
- 选择证书文件并设置任意名称
- 在设置->安全->信任的凭据->用户标签下验证安装
常见问题排查:
- 证书不生效:检查证书是否安装到用户凭据而非系统凭据
- 时间不同步:确保模拟器时间与主机时间误差在5分钟以内
- APP使用证书固定:需配合Xposed框架使用JustTrustMe模块
3. 代理配置与网络调试
3.1 多设备代理方案对比
| 配置方式 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| 模拟器全局代理 | 配置简单 | 影响所有APP | 初步调试 |
| APP单独代理 | 目标明确 | 需要root权限 | 精准抓包 |
| VPN模式 | 绕过部分代理检测 | 配置复杂 | 对抗检测强的APP |
| 透明代理 | 无需客户端配置 | 需要iptables规则 | 路由器级抓包 |
3.2 MuMu模拟器代理设置
在MuMu模拟器中配置代理的完整流程:
- 获取主机IP地址(cmd中执行ipconfig)
- 进入模拟器的WLAN设置,长按当前网络->修改网络
- 选择"显示高级选项",代理选择"手动"
- 输入主机IP和mitmproxy监听端口(默认8080)
- 保存后立即生效,无需重启
验证代理是否生效的快速方法:
adb shell ping 主机IP若网络连通但抓不到包,可能是以下原因:
- 防火墙阻止了mitmproxy端口
- APP使用了WebSocket等非HTTP协议
- 存在证书固定(Certificate Pinning)机制
4. Python脚本自动化实战
4.1 mitmdump基础脚本结构
mitmdump的强大之处在于其Python脚本扩展能力。一个典型的处理脚本包含以下要素:
from mitmproxy import http def request(flow: http.HTTPFlow) -> None: # 请求拦截逻辑 if "target-api" in flow.request.url: print(f"拦截到目标请求: {flow.request.url}") def response(flow: http.HTTPFlow) -> None: # 响应处理逻辑 if flow.response.status_code == 200: print(f"获取到响应: {flow.response.text[:100]}...")启动脚本的命令:
mitmdump -s script.py4.2 实战案例:新闻APP数据抓取
假设我们需要抓取某新闻APP的列表数据,完整脚本示例如下:
import json from mitmproxy import ctx, http class NewsCapture: def __init__(self): self.news_list = [] def response(self, flow: http.HTTPFlow): if "/api/v3/news/list" in flow.request.url: data = json.loads(flow.response.text) for item in data['items']: self.news_list.append({ 'title': item['title'], 'source': item['source'], 'time': item['publish_time'] }) ctx.log.info(f"已捕获{len(self.news_list)}条新闻") def done(self): with open('news.json', 'w', encoding='utf-8') as f: json.dump(self.news_list, f, ensure_ascii=False) addons = [NewsCapture()]该脚本实现了:
- 识别特定API请求
- 提取关键字段并结构化存储
- 运行结束时自动保存为JSON文件
4.3 高级技巧:请求改写与Mock
mitmproxy允许动态修改请求和响应,这在测试场景中极为有用:
def request(flow: http.HTTPFlow): # 修改请求头 flow.request.headers['User-Agent'] = 'Mozilla/5.0 (Custom)' # 重定向请求 if "advertisement" in flow.request.url: flow.request.url = flow.request.url.replace( "production.com", "test-mock.com" ) def response(flow: http.HTTPFlow): # 注入JavaScript if "text/html" in flow.response.headers["content-type"]: html = flow.response.text html = html.replace("</body>", "<script>alert('Injected!')</script></body>") flow.response.text = html5. 数据存储与分析优化
5.1 多格式存储方案
根据数据量和使用场景,可选择不同的存储方式:
| 存储格式 | 写入方式 | 适用场景 |
|---|---|---|
| JSON | json.dump() | 中小规模结构化数据 |
| CSV | csv.writer() | Excel分析 |
| SQLite | sqlite3.connect().cursor() | 复杂查询需求 |
| MongoDB | pymongo.MongoClient()[db][col].insert | 大规模非结构化数据 |
5.2 性能优化技巧
当处理高频率请求时,需注意以下性能要点:
批量写入:避免频繁IO操作,采用缓冲区机制
class BatchWriter: def __init__(self, batch_size=100): self.buffer = [] self.batch_size = batch_size def add(self, item): self.buffer.append(item) if len(self.buffer) >= self.batch_size: self.flush() def flush(self): if self.buffer: # 执行批量写入 self.buffer.clear()异步处理:使用asyncio提升吞吐量
import asyncio async def process_response(flow): await asyncio.sleep(0) # 让出控制权 # 处理逻辑过滤规则:尽早过滤无关请求
def request(flow): if flow.request.host not in TARGET_DOMAINS: return # 快速跳过
6. 反爬对抗策略解析
现代APP常用以下机制阻止抓包,需针对性处理:
SSL Pinning解决方案:
- 使用frida脚本绕过:
frida -U -f package.name -l ssl_pinning_bypass.js - Xposed模块:JustTrustMe
- 修改APK的network_security_config.xml
- 使用frida脚本绕过:
协议混淆处理:
- WebSocket:使用mitmproxy的WebSocket支持
- gRPC:需额外解析HTTP/2帧
- Protobuf:准备.proto文件反序列化
行为检测对抗:
- 模拟真实操作间隔
- 随机化User-Agent
- 保持会话状态
# 随机请求头生成示例 import random USER_AGENTS = [ 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', 'Mozilla/5.0 (iPhone; CPU iPhone OS 13_2_3 like Mac OS X)' ] def request(flow): flow.request.headers['User-Agent'] = random.choice(USER_AGENTS)