使用 mitmproxy 作为代理记录 http/https 请求/响应 详情

README.md

本文档使用 docker image mitmproxy/mitmproxy:9.0.1 进行演示

开发自定义 log add-on

根据 mitmproxy 的文档，我们可以编写自定义的 Addon 进行 http 的日志采集和外发，详见 这里

启动容器

docker run --rm -it \
  -v /data/mitmproxy-test/:/home/mitmproxy/.mitmproxy \  # 将 mitmproxy 的默认工作路径映射到宿主机固定目录, 这样子可以确保始终使用同一个 ca 证书
  -v /data/mitmproxy-test/log/:/data/log/ \  # 可选, 挂载一个目录到容器, 以便将测试的 http log 持久化保存
  -v /data/mitmproxy-test/http-log-addon.py:/data/http-log-addon.py \  # 将宿主机上自定义的 py 脚本挂载到容器指定路径
  -p 8181:8080 \  # 将容器内的 8080 映射到宿主机的 8181
  mitmproxy/mitmproxy:9.0.1 \  # 本例中使用的镜像名称
  mitmproxy \  # 运行自定义的程序, 下边都是要传给它的参数
  -s /data/http-log-addon.py \  # 自定义的 py 脚本路径, 注意上边通过 -v 挂载到容器内
  --mode socks5 \  # 使用 socks5 协议而非默认的 http 协议
  --ssl-insecure \  # 忽略目标服务器的证书错误
  --listen-host 0.0.0.0 --listen-port 8080

客户端配置信任自定义根证书

首先配置自己的浏览器（或者其他软件）使用 socks5 协议（以上方启动命令为准）的代理。

比如设置为: socks5://1.2.3.4:8181

然后，在浏览器中访问 http://mitm.it, 即可看到证书配置引导页面, 大致如下:

然后根据页面引导进行证书配置即可, 详细的方法还可参考 mitmproxy 的文档

测试代理

此时, 我们可以使用 curl 进行测试, 命令如下:

curl -v https://www.baidu.com/test-404.html -x 'socks5://1.2.3.4:8181' 

运行截图如下:

采集到的请求响应日志见这里

http-log-addon.py

# Events 列表: https://docs.mitmproxy.org/stable/api/events.html
# Flow 结构体定义: https://docs.mitmproxy.org/stable/api/mitmproxy/http.html#HTTPFlow

import json


class HTTPLog(object):
    def __init__(self):
        """
        用来表示一条 http log
        """

        # 自行根据需要来调整下述字段
        self.method: str = ""
        self.url: str = ""
        self.request_headers = {}
        self.request_body: str = ""

        # 可以自行根据定义来扩展下述字段
        self.status_code: int = 0
        self.response_headers = {}
        self.response_body: str = ""

    def update_by_request(self, request):
        """
        Request 定义: https://docs.mitmproxy.org/stable/api/mitmproxy/http.html#Request
        """
        self.method = request.method
        self.url = request.url
        # https://docs.mitmproxy.org/stable/api/mitmproxy/http.html#Headers.items
        self.request_headers = list(request.headers.items(multi=True))
        try:
            self.request_body = request.text
        except ValueError:
            pass

    def update_by_response(self, response):
        """
        Response 定义: https://docs.mitmproxy.org/stable/api/mitmproxy/http.html#Response
        """
        self.status_code = response.status_code
        # https://docs.mitmproxy.org/stable/api/mitmproxy/http.html#Headers.items
        self.response_headers = list(response.headers.items(multi=True))
        try:
            self.response_body = response.text
        except ValueError:
            pass

    def update_by_http_flow(self, http_flow):
        self.update_by_request(http_flow.request)
        self.update_by_response(http_flow.response)

    def to_json(self):
        data = {
            "method": self.method,
            "url": self.url,
            "request_headers": self.request_headers,
            "request_body": self.request_body
        }
        if self.status_code != 0:  # has response
            data.update({
                "has_response": True,
                "status_code": self.status_code,
                "response_headers": self.response_headers,
                "response_body": self.response_body
            })
        else:
            data.update({
                "has_response": False
            })
        return data


class APTPLogger(object):
    """
    自定义的 mitmproxy Addon, 用于记录请求响应
    """

    def __init__(self):
        # 容器里的路径
        self.log_file = open("/home/mitmproxy/.mitmproxy/test-http.log", "w+b")

    def response(self, flow):
        """
        详见: https://docs.mitmproxy.org/stable/api/events.html#HTTPEvents.response
        """

        log = HTTPLog()
        log.update_by_http_flow(flow)
        self.write_to(log)

    def error(self, flow):
        """
        详见: https://docs.mitmproxy.org/stable/api/events.html#HTTPEvents.error
        使用该 event 记录 request, 无 response
        """
        log = HTTPLog()
        log.update_by_request(flow.request)
        # no response when error happened
        self.write_to(log)

    def write_to(self, log_entry: HTTPLog):
        """
        FIXME
        这里只是作为一个简单的演示
        可以自行将 log 发往想要的地方
        """
        data = log_entry.to_json()
        result = json.dumps(data, ensure_ascii=False, )
        self.log_file.write(result.encode("utf-8", errors="ignore"))
        self.log_file.write(b"\n")
        self.log_file.flush()


addons = [APTPLogger()]

log-sample.json

{
  "method": "GET",
  "url": "https://180.101.50.242/test-404.html",
  "request_headers": [
    [
      "Host",
      "www.baidu.com"
    ],
    [
      "User-Agent",
      "curl/7.79.1"
    ],
    [
      "Accept",
      "*/*"
    ]
  ],
  "request_body": "",
  "has_response": true,
  "status_code": 404,
  "response_headers": [
    [
      "Content-Length",
      "211"
    ],
    [
      "Content-Type",
      "text/html; charset=iso-8859-1"
    ],
    [
      "Date",
      "Thu, 20 Apr 2023 11:08:42 GMT"
    ],
    [
      "Server",
      "Apache"
    ]
  ],
  "response_body": "<!DOCTYPE HTML PUBLIC \"-//IETF//DTD HTML 2.0//EN\">\n<html><head>\n<title>404 Not Found</title>\n</head><body>\n<h1>Not Found</h1>\n<p>The requested URL /test-404.html was not found on this server.</p>\n</body></html>\n"
}