用mitmproxy录制流量，快速构建自动化测试用例

摘要：其实，在实际功能测试过程中，可以将相关的操作产生的流量，录制并保存下来，从而作为自动化测试的输入数据。这里我们介绍一下基于mitmproxy的流量录制。

在项目测试过程中通常需要构建业务的自动化测试用例，但是，不规范的接口文档或接口文档的缺失对测试人员来说是构建自动化测试用例的一大阻碍。

大多数测试人员通常通过F12进行抓包，然后导入Postman等进行接口的调试与开发。

其实，在实际功能测试过程中，可以将相关的操作产生的流量，录制并保存下来，从而作为自动化测试的输入数据。这里我们介绍一下基于mitmproxy的流量录制。

1. 支持录制http/https请求流量

2. 支持保存到SQL数据库中

1. Python

2. mitmproxy

3. Sqlite

import osclass Config(object):"""配置文件"""# 过滤配置# Host设置HOST = ["isgp.hik-partner.com"] # 需要抓取的域名列表APIS = # 需要抓取的API列表# 日志相关配置# 日志和数据存储上级目录路径PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)))# 日志路径LOG_PATH = os.path.join(PATH, "logs")if not os.path.exists(LOG_PATH):os.makedirs(LOG_PATH)LOG_NAME = os.path.join(LOG_PATH, "mitmproxy_records.log")# filehandler设置开关FILE_HANDLER_ENABLED = True# StreamHandler设置开关STREAM_HANDLER_ENABLED = False# 日志级别# """# CRITICAL = 15# ERROR = 14# WARNING = 13# INFO = 11# DEBUG = 10# """LOG_LEVEL = 11 # 日志级别# 数据库相关配置# 数据存储路径DATA_PATH = os.path.join(PATH, "data")if not os.path.exists(DATA_PATH):os.makedirs(DATA_PATH)DATA_NAME = os.path.join(DATA_PATH, "mitmproxy_records.db")TABLE_NAME = "mitmproxy_records"# 数据库启用开关SQLITE_ENABLED = True # 启用sqlite数据库

主要代码：

#!/usr/bin/python# -*- coding:UTF-8 -*-import jsonimport osimport reimport timeimport uuidfrom datetime import datetime, timezonefrom difflib import SequenceMatcherfrom dotenv import load_dotenvfrom mitmproxy import httpfrom config import Configfrom db.SQLiteUtils import SQLiteUtilsfrom utils.logger import Logfrom utils.utils import generate_name_current_time, write_msg_into_logload_dotenv # 加载环境变量mitm_log = Log('flowproxy')class RecordHandler(object):def __init__(self):self.record_id = str(uuid.uuid1) # 初始化记录idmitm_log.info(f"记录id:{self.record_id}")self.result = # 初始化记录结果self.id_result_map = {} # 初始化请求结果映射self.count = 1 # 初始化计数器self.file_path = os.path.join(Config.PATH, "logs",f"records_{generate_name_current_time}.txt") # 初始化日志路径if Config.SQLITE_ENABLED:self.sqlite_handler = SQLiteUtils # 初始化sqlite实例else:mitm_log.debug("SQLite数据库写入未启用")write_msg_into_log(self.file_path, f"mitmproxy记录开始;记录id:{self.record_id}") # 写入log@classmethoddef calculate_similarity(cls, uri):"""计算APIS中的api与uri的相似度:param uri: 请求的uri:return:"""for api in Config.APIS:match = SequenceMatcher(None, uri, api).ratioif match > 0.6:mitm_log.debug(f"匹配成功:{api}：{match}")return Truereturn False@classmethoddef filter_flow(cls, flow: http.HTTPFlow):"""过滤流量:param flow: 流量对象:return:"""if flow.request.path.endswith(('.ico', '.css', '.js', '.png', 'jpg', 'gif', 'svg', 'ttf', 'woff', 'woff2', 'eot', 'otf')):return Falseif flow.request.pretty_host not in Config.HOST:return Falseif ('hpp' or 'hcc' or 'ccf') not in flow.request.url:return Falseif Config.APIS and not cls.calculate_similarity(flow.request.path):return Falsereturn Truedef request(self, flow: http.HTTPFlow):"""处理请求:param flow: 流量对象:return:"""if not self.filter_flow(flow):returnrequest_id = flow.id # 请求idrequest = flow.requestself.result = [self.record_id, request.path, request.method,request.headers.get('Content-Type','application/json;charset=UTF-8'),request.get_text]self.id_result_map[request_id] = self.resultdef response(self, flow: http.HTTPFlow):"""处理响应:param flow: 流量对象:return:"""if not self.filter_flow(flow):returnresponse_id = flow.idresponse = flow.responseself.result = self.id_result_map[response_id]if flow.request.path != self.result[1]:returnself.result.extend([response.status_code, response.get_text])log_data = {str(self.count): {"uri": self.result[1],"method": self.result[2],"headers": self.result[3], # 请求头"body": self.result[4], # 请求体"response_status": self.result[5], # 响应状态码"response_text": self.result[6], # 响应体}}# 写入文件write_msg_into_log(self.file_path,f"{json.dumps(log_data, indent=4, ensure_ascii=False)}")# 替换文件中的自动生产的id信息if re.findall(r"/[0-9a-z]{30,}", self.result[1]):self.result[1] = re.sub(r"/[0-9a-z]{30,}", "/*", self.result[1])# 写入SQLite数据库if Config.SQLITE_ENABLED: # 写入数据库try:self.sqlite_handler.insert_data("mitmproxy_records", self.result)except Exception as e:mitm_log.error(f"{e}")mitm_log.error("SQLite数据库写入失败")self.count += 1addons = [RecordHandler]

1. 证书安装

需要在C:\\Users\\{用户名}\\.mitmproxy目录下双击cer文件，安装证书，选择本地计算机，选择"受信任的根证书颁发机构"，然后点击确定即可。

2. 开启代理