揭秘抓包利器：Python和Mitmproxy让您轻松实现接口请求抓取与分析！

原文 https://zhuanlan.zhihu.com/p/648011072?utm_psn=1791741010897940480

在接口测试和自动化领域，我们经常需要抓取网络请求，分析接口参数和响应，以便更好地理解接口的行为和性能。借助 Mitmproxy 这个强大的抓包工具，我们能够轻松地监控和分析接口请求，为接口自动化测试提供更多的可能性。

什么是 Mitmproxy？

Mitmproxy 是一个开源的中间人代理工具，它能够截取和修改 HTTP 和 HTTPS 流量。它提供了命令行界面和 Web 界面，能够帮助我们实时查看和分析网络请求。

准备工作：安装 Mitmproxy

在开始之前，我们首先需要安装 Mitmproxy。可以通过 pip 命令来安装 Mitmproxy：

1	pip install mitmproxy

编写 Python 脚本：抓取请求并保存到 CSV

使用 Mitmproxy 来拦截接口请求，并将请求记录保存到 CSV 文件中。

# filename: request_recorder.py
import csv
from mitmproxy import http

class RequestRecorder:
    def __init__(self):
        self.records = []
    
    def request(self, flow: http.HTTPFlow):
        if "api.xxx.com" in flow.request.url:
            # 获取请求的 URL、方法、请求头和请求参数
            url = flow.request.url
            method = flow.request.method
            headers = dict(flow.request.headers)
            params = dict(flow.request.query or {})     
            record = {
                "URL": url,
                "Method": method,
                "Headers": headers,
                "Params": params,
            }
            self.records.append(record)
    
    def done(self):
        # 保存记录到 CSV 文件
        with open("api_requests.csv", "w", newline="", encoding="utf-8") as csvfile:
            fieldnames = ["URL", "Method", "Headers", "Params", "Response"]
            writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
            writer.writeheader()
            for record in self.records:
                writer.writerow(record)

addons = [
    RequestRecorder()
]

# filename: response_recorder.py
import csv
from mitmproxy import http
import datetime
import os

class ResponseRecorder:
    def __init__(self):
        self.records = []
        self.init_datetime = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
        os.makedirs(self.init_datetime)

    
    def response(self, flow: http.HTTPFlow):
        
        if not "api.xxx.com" in flow.request.url:
            # 获取请求的 URL、方法、请求头和请求参数
            req_datetime = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
            req_url = flow.request.url
            req_method = flow.request.method
            req_headers = dict(flow.request.headers)
            req_params = dict(flow.request.query or {})     
            req_text = flow.request.get_text()
            res_headers = dict(flow.response.headers)
            res_text = flow.response.get_text()
            record = {
                "Req_datetime": req_datetime,
                "Req_url": req_url,
                "Req_method": req_method,
                "Req_headers": req_headers,
                "Req_params": req_params,
                "Req_text": req_text,
                "Res_headers": res_headers,
                "Res_text": res_text,
            }
            print(f"{record['Req_url']} {record['Req_method']}")
            self.records.append(record)

        with open(f"./{self.init_datetime}/res_{req_datetime}.txt", "w", newline="", encoding="utf-8") as file:
            file.write(res_text)

    
    def done(self):
        # 保存记录到 CSV 文件
        with open(f"api_requests{self.init_datetime}.csv", "w", newline="", encoding="utf-8") as csvfile:
            fieldnames = ["Req_datetime", "Req_url", "Req_method", "Req_headers", "Req_params", "Req_text", "Res_headers", "Res_text"]
            writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
            writer.writeheader()
            for record in self.records:
                record['Res_text'] = ''
                writer.writerow(record)

        with open(f"api_requests{self.init_datetime}.txt", "w", newline="", encoding="utf-8") as file:
            fieldnames = ["Req_datetime", "Req_url", "Req_method", "Req_headers", "Req_params", "Req_text", "Res_headers", "Res_text"]
            for record in self.records:
                s = f'==={fieldnames[0]}===\r\n{record[fieldnames[0]]}\r\n==={fieldnames[1]}===\r\n{record[fieldnames[1]]}\r\n==={fieldnames[2]}===\r\n{record[fieldnames[2]]}\r\n==={fieldnames[3]}===\r\n{record[fieldnames[3]]}\r\n==={fieldnames[4]}===\r\n{record[fieldnames[4]]}\r\n==={fieldnames[5]}===\r\n{record[fieldnames[5]]}\r\n==={fieldnames[6]}===\r\n{record[fieldnames[6]]}\r\n==={fieldnames[7]}===\r\n{record[fieldnames[7]]}\r\n===END===\r\n\r\n'
                
                file.write(s)


addons = [
    ResponseRecorder()
]

启动 Mitmproxy 并运行脚本

在终端中运行以下命令启动 Mitmproxy，并加载我们编写的脚本 request_recorder.py：

1 2	mitmdump -s request_recorder.py mitmweb -s response_recorder.py --ssl-insecure --listen-host 0.0.0.0 -p 8088 --web-host 0.0.0.0 --web-port 8089 --set block_global=false

设置浏览器代理

打开浏览器，并将代理地址设置为 http://localhost:8080。

访问接口

在浏览器中访问示例接口 https://api.xxx.com/data。Mitmproxy 将拦截这个请求并将其保存到 CSV 文件.

查看保存的数据

在脚本运行完成后，当前目录下会生成一个名为 api_requests.csv 的 CSV 文件:

英雄本色的博客