使用 mitmproxy 自定义脚本过滤请求并直接生成 JMX

下面是一个完整的解决方案，通过自定义 mitmproxy Python 脚本捕获 HTTP 流量，过滤特定请求，并直接生成 JMeter 可用的 JMX 文件，无需经过 HAR 转换中间步骤。

完整脚本代码

将以下脚本保存为 mitmproxy_jmx_exporter.py：

python

复制

from mitmproxy import http, ctx
from xml.etree.ElementTree import Element, SubElement, tostring
from xml.dom import minidom
import urllib.parse
import datetimeclass JMXGenerator:def __init__(self):# 初始化JMX文档结构self.jmx = Element('jmeterTestPlan', version="1.2", properties="5.0", jmeter="5.4.1")hash_tree = SubElement(self.jmx, 'hashTree')# 创建测试计划test_plan = SubElement(hash_tree, 'TestPlan', {'guiclass': 'TestPlanGui','testclass': 'TestPlan','testname': 'Mitmproxy Generated Test','enabled': 'true'})SubElement(test_plan, 'boolProp', {'name': 'TestPlan.user_define_classpath'}).text = 'false'# 创建线程组thread_group_hash = SubElement(hash_tree, 'hashTree')self.thread_group = SubElement(thread_group_hash, 'ThreadGroup', {'guiclass': 'ThreadGroupGui','testclass': 'ThreadGroup','testname': 'Thread Group','enabled': 'true'})# 线程组配置SubElement(self.thread_group, 'intProp', {'name': 'ThreadGroup.num_threads'}).text = '1'SubElement(self.thread_group, 'intProp', {'name': 'ThreadGroup.ramp_time'}).text = '1'SubElement(self.thread_group, 'boolProp', {'name': 'ThreadGroup.scheduler'}).text = 'false'# 存储采样器的hashTreeself.samplers_hash = SubElement(thread_group_hash, 'hashTree')# 请求计数器self.request_count = 0self.filtered_count = 0def add_request(self, flow: http.HTTPFlow):"""添加HTTP请求到JMX"""url = flow.request.url# 过滤条件 - 修改这里实现你的过滤逻辑if self._should_filter(flow):self.filtered_count += 1returnself.request_count += 1# 创建HTTP采样器sampler = SubElement(self.samplers_hash, 'HTTPSamplerProxy', {'guiclass': 'HttpTestSampleGui','testclass': 'HTTPSamplerProxy','testname': f"{flow.request.method} {flow.request.host}",'enabled': 'true'})# 配置HTTP请求SubElement(sampler, 'stringProp', {'name': 'HTTPSampler.domain'}).text = flow.request.hostSubElement(sampler, 'stringProp', {'name': 'HTTPSampler.port'}).text = str(flow.request.port or 443 if flow.request.scheme == 'https' else 80)SubElement(sampler, 'stringProp', {'name': 'HTTPSampler.protocol'}).text = flow.request.schemeSubElement(sampler, 'stringProp', {'name': 'HTTPSampler.path'}).text = urllib.parse.quote(flow.request.path)SubElement(sampler, 'stringProp', {'name': 'HTTPSampler.method'}).text = flow.request.method# 添加请求头if flow.request.headers:headers = SubElement(self.samplers_hash, 'HeaderManager', {'guiclass': 'HeaderPanel','testclass': 'HeaderManager','testname': 'HTTP Header Manager','enabled': 'true'})collection_prop = SubElement(headers, 'collectionProp', {'name': 'HeaderManager.headers'})for name, value in flow.request.headers.items():if name.lower() in ['host', 'content-length']:  # 跳过这些头continueelement_prop = SubElement(collection_prop, 'elementProp', {'name': name,'elementType': 'Header'})SubElement(element_prop, 'stringProp', {'name': 'Header.name'}).text = nameSubElement(element_prop, 'stringProp', {'name': 'Header.value'}).text = valueSubElement(self.samplers_hash, 'hashTree')# 添加请求体if flow.request.content:SubElement(sampler, 'boolProp', {'name': 'HTTPSampler.postBodyRaw'}).text = 'true'SubElement(sampler, 'elementProp', {'name': 'HTTPsampler.Arguments','elementType': 'Arguments'})SubElement(sampler, 'stringProp', {'name': 'Argument.value'}).text = flow.request.content.decode('utf-8', errors='replace')SubElement(self.samplers_hash, 'hashTree')def _should_filter(self, flow: http.HTTPFlow) -> bool:"""定义过滤逻辑"""# 示例1: 排除静态资源if flow.request.path.endswith(('.js', '.css', '.png', '.jpg', '.gif', '.ico', '.woff', '.woff2')):return True# 示例2: 排除特定域名excluded_domains = ['google-analytics.com', 'googletagmanager.com']if any(domain in flow.request.host for domain in excluded_domains):return True# 示例3: 只包含特定API路径# if not flow.request.path.startswith('/api/'):#     return Truereturn Falsedef save_jmx(self, filename: str):"""保存JMX文件"""xml_str = tostring(self.jmx, encoding='utf-8')pretty_xml = minidom.parseString(xml_str).toprettyxml(indent="  ", encoding='utf-8')with open(filename, 'wb') as f:f.write(pretty_xml)ctx.log.info(f"Saved JMX file with {self.request_count} requests (filtered {self.filtered_count} requests)")# 全局JMX生成器实例
jmx_generator = JMXGenerator()def request(flow: http.HTTPFlow):"""处理每个请求"""jmx_generator.add_request(flow)def done():"""mitmproxy结束时保存JMX文件"""jmx_generator.save_jmx('mitmproxy_generated.jmx')# 启动命令: mitmproxy -s mitmproxy_jmx_exporter.py

使用说明

1. 安装依赖：

   bash

   复制

   pip install mitmproxy xmltodict

2. 运行脚本：

   bash

   复制

   mitmproxy -s mitmproxy_jmx_exporter.py

   或使用非交互模式：

   bash

   复制

   mitmdump -s mitmproxy_jmx_exporter.py

3. 配置代理：

   • 设置设备或浏览器使用 mitmproxy 作为代理（默认端口 8080）

   • 安装 mitmproxy 的 CA 证书以捕获 HTTPS 流量

自定义过滤规则

修改脚本中的 _should_filter 方法实现你的过滤逻辑：

python

复制

def _should_filter(self, flow: http.HTTPFlow) -> bool:"""自定义过滤逻辑"""# 示例: 只记录特定API路径if not flow.request.path.startswith('/api/v1/'):return True# 示例: 排除OPTIONS方法if flow.request.method == 'OPTIONS':return True# 示例: 只包含特定内容类型content_type = flow.request.headers.get('Content-Type', '')if 'application/json' not in content_type:return Truereturn False

生成的 JMX 文件包含

1. 完整的测试计划结构

2. 线程组配置

3. 每个 HTTP 请求采样器

4. 请求方法、URL、头信息

5. 请求体内容

6. 自动过滤掉的静态资源和不需要的请求

高级功能扩展

1. 添加思考时间：

   python

   复制

   # 在add_request方法中添加
   if self.request_count > 0:timer = SubElement(self.samplers_hash, 'ConstantTimer', {'guiclass': 'ConstantTimerGui','testclass': 'ConstantTimer','testname': 'Think Time','enabled': 'true'})SubElement(timer, 'stringProp', {'name': 'ConstantTimer.delay'}).text = '1000'  # 1秒SubElement(self.samplers_hash, 'hashTree')

2. 自动参数化：

   python

   复制

   # 自动替换session ID等为变量
   if 'session_id' in flow.request.path:flow.request.path = flow.request.path.replace(session_id, '${SESSION_ID}')

3. 添加断言：

   python

   复制

   # 在add_request方法末尾添加
   assertion = SubElement(self.samplers_hash, 'ResponseAssertion', {'guiclass': 'AssertionGui','testclass': 'ResponseAssertion','testname': 'Response Assertion','enabled': 'true'
   })
   SubElement(assertion, 'stringProp', {'name': 'Assertion.test_field'}).text = 'Assertion.response_code'
   SubElement(assertion, 'intProp', {'name': 'Assertion.test_type'}).text = '2'  # 2表示相等
   SubElement(assertion, 'stringProp', {'name': 'Assertion.test_string'}).text = '200'

这个脚本提供了高度可定制的解决方案，可以直接从实时流量生成准备就绪的 JMeter 测试计划，特别适合复杂 API 的测试场景创建。