mitmproxy抓包

1. mitmproxy 插件机制

核心概念： mitmproxy 通过 addons 列表加载插件，插件是一个普通的 Python 类，通过定义特定方法名来 hook 不同的事件。

addons = [
    MitmLogger(),  # mitmproxy 启动时会扫描这个列表，实例化并注册
]

常用的 hook 方法：

方法名	触发时机
`request(self, flow)`	客户端请求到达代理时
`response(self, flow)`	服务器响应返回时
`error(self, flow)`	连接出错时
`tcp_message(self, flow)`	TCP 层消息

mitmproxy 不需要你继承某个基类或注册装饰器，只要方法名对了就会被自动调用（鸭子类型）。

2. flow 对象结构

flow: mitmproxy.http.HTTPFlow 是核心数据结构

flow.request.url          # 完整 URL
flow.request.method       # GET/POST/PUT...
flow.request.headers      # 请求头（类字典对象）
flow.request.content      # 请求体（bytes 类型）

flow.response.status_code # HTTP 状态码
flow.response.headers     # 响应头
flow.response.content     # 响应体（bytes）
flow.response.text        # 响应体（自动解码为 str）

注意 content 是 bytes，需要 .decode('utf-8') 才能当字符串用。

3. 为什么不用 logging 模块

# 不用这个：
logging.basicConfig(...)  # 会被 mitmproxy 内部覆盖

# 改用直接写文件：
def write_log(msg: str):
    with open(LOG_FILE, 'a', encoding='utf-8') as f:
        f.write(line)

原因： mitmproxy 启动时会重新配置 Python 的 logging 系统（设置自己的 handler 和 formatter），把你通过 basicConfig 设置的 FileHandler 覆盖掉。直接用 open() 写文件绕开了这个问题。

4. os.makedirs 的 exist_ok 参数

os.makedirs(LOG_DIR, exist_ok=True)

exist_ok=True：目录已存在时不报错
如果不加这个参数，目录存在会抛 FileExistsError

5. f-string 中嵌套引号

LOG_FILE = os.path.join(LOG_DIR, f'mitm_{datetime.now().strftime("%Y%m%d_%H%M%S")}.log')

f-string 外层用单引号 '，内层 strftime 的格式字符串用双引号 "，避免冲突。

6. JSON 美化输出的 try 嵌套

body = flow.request.content.decode('utf-8')  # 先尝试解码
try:
    body = json.dumps(json.loads(body), ensure_ascii=False, indent=2)  # 再尝试格式化
except (json.JSONDecodeError, ValueError):
    pass  # 不是 JSON 就保持原样

这是一个"尽力而为"的模式：能格式化就格式化，不能就原样输出。ensure_ascii=False 保证中文不会被转成 \uXXXX。

7. Content-Type 判断逻辑

if any(t in content_type for t in ['text', 'json', 'javascript', 'xml']):

只对文本类型的响应记录 body，图片、视频等二进制内容只记录长度。any() + 生成器表达式是 Python 中判断"多个条件满足其一"的惯用写法。

8. 防止日志爆炸

body[:2000]  # 只取前 2000 字符

网页响应可能很大（几百 KB 的 HTML），不限制的话日志文件会迅速膨胀。

9. 运行方式

mitmdump -s mitmLogger.py      # 无界面，纯命令行
mitmproxy -s mitmLogger.py     # 带 TUI 界面
mitmweb -s mitmLogger.py       # 带 Web 界面

-s 参数指定加载的脚本。mitmdump 最轻量，适合后台运行。

10.完整代码

# -*- coding:utf-8 -*-
# mitmproxy 插件：将抓到的请求和响应信息输出到日志文件
import os
import json
from datetime import datetime
import mitmproxy.http

# 日志配置
LOG_DIR = './logs'
os.makedirs(LOG_DIR, exist_ok=True)
LOG_FILE = os.path.join(LOG_DIR, f'mitm_{datetime.now().strftime("%Y%m%d_%H%M%S")}.log')

def write_log(msg: str):
    """直接写文件，避免 mitmproxy 覆盖 logging 配置的问题"""
    timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    line = f"[{timestamp}] {msg}\n"
    with open(LOG_FILE, 'a', encoding='utf-8') as f:
        f.write(line)

class MitmLogger:
    """将 mitmproxy 抓到的请求/响应记录到日志文件"""
    def __init__(self):
        # 可在此配置过滤规则，只记录包含指定关键词的 URL
        self.url_filter = None  # 设为 None 表示记录所有请求
        write_log(f"MitmLogger 已启动，日志文件: {LOG_FILE}")
    def request(self, flow: mitmproxy.http.HTTPFlow):
        """记录请求信息"""
        url = flow.request.url
        if self.url_filter and self.url_filter not in url:
            return
        write_log("=" * 80)
        write_log(f"[REQUEST] {flow.request.method} {url}")
        write_log(f"  Headers:")
        for key, value in flow.request.headers.items():
            write_log(f"    {key}: {value}")
        if flow.request.content:
            try:
                body = flow.request.content.decode('utf-8')
                try:
                    body = json.dumps(json.loads(body), ensure_ascii=False, indent=2)
                except (json.JSONDecodeError, ValueError):
                    pass
                write_log(f"  Body:\n{body[:2000]}")
            except UnicodeDecodeError:
                write_log(f"  Body: [二进制数据, 长度={len(flow.request.content)} bytes]")

    def response(self, flow: mitmproxy.http.HTTPFlow):
        """记录响应信息"""
        url = flow.request.url
        if self.url_filter and self.url_filter not in url:
            return

        write_log(f"[RESPONSE] {flow.response.status_code} {url}")
        write_log(f"  Headers:")
        for key, value in flow.response.headers.items():
            write_log(f"    {key}: {value}")

        content_type = flow.response.headers.get('content-type', '')
        if flow.response.content:
            if any(t in content_type for t in ['text', 'json', 'javascript', 'xml']):
                try:
                    body = flow.response.content.decode('utf-8')
                    try:
                        body = json.dumps(json.loads(body), ensure_ascii=False, indent=2)
                    except (json.JSONDecodeError, ValueError):
                        pass
                    write_log(f"  Body (前2000字符):\n{body[:2000]}")
                except UnicodeDecodeError:
                    write_log(f"  Body: [解码失败, 长度={len(flow.response.content)} bytes]")
            else:
                write_log(f"  Body: [非文本类型: {content_type}, 长度={len(flow.response.content)} bytes]")
        write_log("")


addons = [
    MitmLogger(),
]