不止于抓包：用mitmdump+Python脚本实现App请求自动修改与数据清洗
不止于抓包：用mitmdump+Python脚本实现App请求自动修改与数据清洗
在移动应用开发和逆向工程领域，抓包工具早已成为开发者工具箱中的标配。但大多数开发者仅停留在"观察"阶段——查看请求和响应数据，分析API调用流程。这就像只学会了使用显微镜观察细胞，却不知道如何操作基因。mitmproxy系列工具的真正威力在于其可编程性，特别是mitmdump与Python脚本的结合，能将被动观察转变为主动干预，实现请求自动修改、数据实时清洗等高级功能。
想象一下这样的场景：你需要批量测试App在不同User-Agent下的兼容性，传统方式需要手动修改设备或反复安装不同版本App。而通过mitmdump脚本，只需20行Python代码就能实现请求头的自动轮换。又或者，你需要采集某电商App的商品数据，但响应中是加密的JSON，通过response()钩子可以实时解密并存储到数据库。这些正是本文要探讨的进阶技巧。
1. 环境配置与核心概念
1.1 mitmproxy生态工具选型
mitmproxy实际上包含三个组件，各有适用场景：
| 工具名称 | 交互方式 | 适用场景 | Python脚本支持 |
|---|---|---|---|
| mitmproxy | 命令行TUI | 交互式调试、实时监控 | 是 |
| mitmweb | Web界面 | 团队协作、可视化分析 | 是 |
| mitmdump | 无界面 | 自动化任务、持续运行 | 是 |
对于自动化场景，mitmdump是唯一选择。它能在后台静默运行，通过Python脚本处理所有流量。安装非常简单：
pip install mitmproxy # 验证安装 mitmdump --version1.2 代理配置与证书安装
要让设备流量经过mitmdump，需要完成以下步骤：
- 设置代理：确保测试设备和运行mitmdump的电脑在同一局域网，在设备WiFi设置中手动配置代理，指向电脑IP和默认端口8080
- 安装CA证书：
- 在设备浏览器访问
http://mitm.it - 下载对应操作系统/设备的证书
- 在系统设置中完成证书安装（Android需要设置为系统信任证书）
- 在设备浏览器访问
注意：iOS和Android高版本对用户安装证书有额外限制，可能需要额外配置才能拦截HTTPS流量
2. 脚本基础架构与核心API
2.1 脚本基本结构
一个完整的mitmdump脚本通常包含两个核心方法：
def request(flow): """处理所有请求对象""" # 可访问flow.request的所有属性 pass def response(flow): """处理所有响应对象""" # 可访问flow.response的所有属性 pass2.2 Flow对象关键属性
request和response方法接收的flow参数包含完整的事务信息：
请求对象(flow.request)常用属性：
url: 完整请求URLmethod: HTTP方法(GET/POST等)headers: 请求头字典cookies: Cookies字典host: 请求主机名path: URL路径部分query: URL查询参数(MultiDict)text: 请求体文本(适用于POST等)
响应对象(flow.response)常用属性：
status_code: HTTP状态码headers: 响应头字典cookies: Cookies字典text: 响应体文本content: 响应体二进制内容
3. 实战：请求干预技术
3.1 动态修改请求头
修改User-Agent是最常见的用例之一，以下脚本实现按规则轮换UA：
USER_AGENTS = [ "Mozilla/5.0 (iPhone; CPU iPhone OS 15_0 like Mac OS X)", "Mozilla/5.0 (Linux; Android 12; SM-S901U)", "Mozilla/5.0 (Windows NT 10.0; Win64; x64)" ] def request(flow): import random ua = random.choice(USER_AGENTS) flow.request.headers["User-Agent"] = ua ctx.log.info(f"设置User-Agent: {ua}")更复杂的场景可以基于请求特征动态选择UA：
def request(flow): if "api.example.com" in flow.request.host: flow.request.headers["X-Client-Version"] = "2.3.1" flow.request.headers["Authorization"] = "Bearer fake_token"3.2 请求重定向与URL改写
实现类似API网关的功能，将特定请求转发到其他端点：
API_MAPPING = { "/old-api/": "/new-api/", "/v1/users/": "/v2/members/" } def request(flow): for old, new in API_MAPPING.items(): if old in flow.request.path: new_url = flow.request.url.replace(old, new) flow.request.url = new_url ctx.log.info(f"重定向请求 {old} → {new}") break3.3 请求体实时修改
对于POST请求，可以动态修改提交的表单或JSON数据：
def request(flow): if flow.request.method == "POST" and "application/json" in flow.request.headers.get("content-type", ""): try: import json data = json.loads(flow.request.text) # 修改特定字段 if "user_id" in data: data["user_id"] = "anonymous_" + data["user_id"][-4:] flow.request.text = json.dumps(data) except Exception as e: ctx.log.error(f"JSON处理错误: {e}")4. 响应处理与数据清洗
4.1 实时数据解密
许多App会对响应数据加密，可以在脚本中实现自动解密：
def decrypt_data(encrypted): # 实现实际解密逻辑 return encrypted[::-1] # 示例：简单反转字符串 def response(flow): if flow.response.headers.get("X-Encrypted") == "true": try: original = flow.response.text decrypted = decrypt_data(original) flow.response.text = decrypted flow.response.headers["X-Encrypted"] = "false" except Exception as e: ctx.log.error(f"解密失败: {e}")4.2 结构化数据提取与存储
将API响应直接保存到数据库：
import sqlite3 from urllib.parse import urlparse # 初始化数据库 conn = sqlite3.connect("app_data.db") cursor = conn.cursor() cursor.execute("""CREATE TABLE IF NOT EXISTS api_responses (id INTEGER PRIMARY KEY AUTOINCREMENT, url TEXT, timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, data TEXT)""") def response(flow): if "api.target.com" in flow.request.host and flow.response.status_code == 200: try: path = urlparse(flow.request.url).path cursor.execute("INSERT INTO api_responses (url, data) VALUES (?, ?)", (path, flow.response.text)) conn.commit() ctx.log.info(f"保存数据: {path}") except Exception as e: conn.rollback() ctx.log.error(f"数据库错误: {e}")4.3 敏感数据脱敏
在数据采集过程中实现实时隐私保护：
import re def anonymize(text): # 手机号脱敏 text = re.sub(r'(1[3-9]\d)\d{4}(\d{4})', r'\1****\2', text) # 身份证脱敏 text = re.sub(r'([1-9]\d{5})\d{8}(\d{4})', r'\1********\2', text) return text def response(flow): if "application/json" in flow.response.headers.get("content-type", ""): try: flow.response.text = anonymize(flow.response.text) except Exception as e: ctx.log.warn(f"脱敏处理失败: {e}")5. 高级技巧与性能优化
5.1 脚本模块化与复用
随着脚本复杂度增加，建议采用模块化组织：
mitm_scripts/ ├── __init__.py ├── core.py # 基础功能 ├── processors/ # 各种处理器 │ ├── header_modifier.py │ ├── data_extractor.py │ └── api_mocker.py └── utils/ # 工具函数 ├── logger.py └── db.py然后通过主脚本集成：
from mitm_scripts.processors.header_modifier import modify_headers from mitm_scripts.processors.data_extractor import extract_data def request(flow): modify_headers(flow) def response(flow): extract_data(flow)5.2 流量过滤与性能调优
默认情况下mitmdump会处理所有流量，可以通过过滤器提高性能：
mitmdump -s script.py --ignore-hosts '^(?!.*\.target\.com).*$'或者在脚本中实现过滤：
TARGET_DOMAINS = ["api.target.com", "static.target.com"] def request(flow): if flow.request.host not in TARGET_DOMAINS: return # 跳过非目标域名 # 处理逻辑...5.3 异步处理与批量操作
对于耗时操作（如网络请求或复杂计算），使用异步避免阻塞：
import asyncio from mitmproxy import ctx async def process_data(data): # 模拟耗时操作 await asyncio.sleep(1) return data.upper() def response(flow): if flow.response.text: # 创建任务但不等待完成 task = ctx.master.loop.create_task( process_data(flow.response.text) ) task.add_done_callback( lambda t: ctx.log.info(f"处理结果: {t.result()}") )6. 调试与错误处理
6.1 日志分级输出
mitmproxy提供多级日志输出：
from mitmproxy import ctx def request(flow): ctx.log.debug("详细调试信息") # 默认不显示 ctx.log.info("常规信息") # 白色 ctx.log.warn("警告信息") # 黄色 ctx.log.error("错误信息") # 红色启动时指定日志级别：
mitmdump -s script.py --set console_eventlog_verbosity=debug6.2 异常捕获与恢复
确保脚本健壮性：
def response(flow): try: # 可能出错的操作 process_response(flow.response) except ValueError as e: ctx.log.error(f"值错误: {e}") flow.response.status_code = 500 flow.response.text = "处理失败" except Exception as e: ctx.log.error(f"未知错误: {e}") # 保持原始响应不变6.3 脚本热重载
开发过程中可以启用脚本自动重载：
mitmdump -s script.py --scripts-refresh-interval 1这样修改脚本后保存，mitmdump会自动重新加载而无需重启。