尧图网站建设 免费方案咨询
ARTICLE DETAIL

资讯详情

深耕网站建设、视觉设计与SEO优化的一线实战洞察。

首页/资讯中心/资讯详情

mitmproxy脚本开发实战：HTTP流量修改与分析

发布时间：2026/11/3 20:57:49 尧图网站建设
mitmproxy脚本开发实战：HTTP流量修改与分析

1. 项目概述

mitmproxy作为一款开源的中间人代理工具，在网络安全和开发测试领域有着广泛的应用。不同于Fiddler或Charles这类GUI工具，mitmproxy提供了强大的Python脚本扩展能力，允许开发者通过编写脚本来实现HTTP/HTTPS流量的实时修改与分析。

我在实际工作中发现，很多开发者在处理API调试、爬虫逆向、安全测试等场景时，往往需要更灵活的流量控制能力。比如：

  • 修改请求参数测试不同输入对服务端的影响
  • 拦截特定响应内容进行数据篡改测试
  • 统计接口调用频率和性能指标
  • 自动化生成测试报告

这些需求正是mitmproxy脚本能够完美解决的。通过Python脚本，我们可以实现比GUI操作更复杂、更自动化的流量处理逻辑。下面我将结合具体案例，详细介绍如何从零开始编写实用的mitmproxy脚本。

2. 环境准备与基础配置

2.1 安装与证书配置

首先需要安装mitmproxy核心组件：

pip install mitmproxy

安装完成后，需要配置系统信任mitmproxy的CA证书。这是处理HTTPS流量的关键步骤：

# 生成证书 mitmproxy --cert-host=* # Windows系统导入证书 certutil -addstore -f "ROOT" ~/.mitmproxy/mitmproxy-ca-cert.cer # MacOS系统导入证书 security add-trusted-cert -d -r trustRoot -k /Library/Keychains/System.keychain ~/.mitmproxy/mitmproxy-ca-cert.pem

注意：Android设备需要单独安装证书，将mitmproxy-ca-cert.cer传输到手机后，在设置->安全->加密与凭据中安装

2.2 基础代理设置

启动mitmproxy的web界面：

mitmweb

配置设备代理：

  • 电脑端：设置系统代理为127.0.0.1:8080
  • 手机端：设置WiFi代理为电脑IP:8080

测试代理是否生效：

curl --proxy http://localhost:8080 https://example.com

3. 脚本开发基础

3.1 脚本结构解析

一个基本的mitmproxy脚本包含事件处理器和辅助函数。核心事件包括：

  • request(): 请求发出前触发
  • response(): 响应返回后触发
  • websocket_message(): WebSocket消息处理

示例骨架：

from mitmproxy import http class MyAddon: def request(self, flow: http.HTTPFlow): # 修改请求逻辑 pass def response(self, flow: http.HTTPFlow): # 处理响应逻辑 pass addons = [MyAddon()]

3.2 流量对象解析

HTTPFlow对象包含完整的请求响应信息：

flow.request.headers # 请求头 flow.request.url # 完整URL flow.request.method # HTTP方法 flow.request.content # 请求体 flow.response.status_code # 状态码 flow.response.headers # 响应头 flow.response.content # 响应体

4. 实战脚本开发

4.1 请求修改脚本

案例：自动添加认证头

def request(self, flow): if "api.example.com" in flow.request.host: flow.request.headers["Authorization"] = "Bearer xxxxx"

案例：修改POST参数

def request(self, flow): if flow.request.method == "POST": from urllib.parse import parse_qs, urlencode params = parse_qs(flow.request.content.decode()) params["debug"] = ["true"] flow.request.content = urlencode(params).encode()

4.2 响应处理脚本

案例：修改JSON响应

def response(self, flow): if "application/json" in flow.response.headers.get("content-type", ""): import json data = json.loads(flow.response.content) data["injected"] = True flow.response.content = json.dumps(data).encode()

案例：图片水印处理

def response(self, flow): if flow.response.headers.get("content-type", "").startswith("image/"): from PIL import Image, ImageDraw import io img = Image.open(io.BytesIO(flow.response.content)) draw = ImageDraw.Draw(img) draw.text((10, 10), "SAMPLE", fill=(255, 0, 0)) output = io.BytesIO() img.save(output, format=img.format) flow.response.content = output.getvalue()

5. 高级应用场景

5.1 流量分析统计

实现接口调用统计：

class StatsAddon: def __init__(self): self.stats = {} def response(self, flow): key = f"{flow.request.host}{flow.request.path}" self.stats[key] = self.stats.get(key, 0) + 1 def done(self): print("\nAPI调用统计:") for api, count in sorted(self.stats.items(), key=lambda x: -x[1]): print(f"{api}: {count}次")

5.2 自动化测试

结合pytest实现自动化断言：

class TestAddon: def response(self, flow): if "/api/login" in flow.request.path: assert flow.response.status_code == 200 assert "session_id" in flow.response.cookies def error(self, flow): print(f"测试失败: {flow.error.msg}")

5.3 爬虫逆向辅助

自动保存API文档：

class APIDocAddon: def __init__(self): self.apis = set() def response(self, flow): if "api.example.com" in flow.request.host: info = { "url": flow.request.url, "method": flow.request.method, "params": flow.request.query, "request": flow.request.content.decode(), "response": flow.response.content.decode() } self.apis.add(json.dumps(info, indent=2)) def done(self): with open("api_docs.json", "w") as f: f.write("\n\n".join(self.apis))

6. 调试与问题排查

6.1 常见问题解决

  1. HTTPS流量无法解密

    • 确认系统已信任mitmproxy CA证书
    • 检查应用是否使用了证书固定（Certificate Pinning）
    • 尝试mitmproxy --ssl-insecure绕过证书验证

  2. 手机无法连接代理

    • 确认电脑和手机在同一网络
    • 关闭防火墙或添加8080端口例外
    • 尝试mitmproxy --listen-host 0.0.0.0绑定所有接口

  3. 脚本修改未生效

    • 确认脚本已正确加载：mitmproxy -s script.py
    • 检查脚本是否有语法错误
    • 添加日志输出确认执行路径

6.2 调试技巧

添加调试日志：

def request(self, flow): print(f"拦截请求: {flow.request.method} {flow.request.url}") print("请求头:", flow.request.headers)

使用mitmweb实时观察：

mitmweb -s script.py

7. 性能优化建议

  1. 减少不必要的处理

    def request(self, flow): if not flow.request.host.endswith("target.com"): return # 跳过非目标域名

  2. 使用缓存提升性能

    from functools import lru_cache @lru_cache(maxsize=100) def parse_url(url): # 昂贵的URL解析操作 return parsed_result

  3. 异步处理耗时操作

    import asyncio async def process_image(content): # 图片处理逻辑 return result def response(self, flow): if flow.response.headers.get("content-type", "").startswith("image/"): flow.response.content = asyncio.run(process_image(flow.response.content))

8. 安全注意事项

  1. 敏感信息处理

    def response(self, flow): if "password" in flow.response.text: flow.response.text = flow.response.text.replace( "password":".*?", "password":"[REDACTED]" )

  2. 脚本权限控制

    • 避免在生产环境使用修改类脚本
    • 为脚本添加密码保护：
    class SecureAddon: def __init__(self, password): self.password = password def request(self, flow): if flow.request.headers.get("X-Auth") != self.password: flow.kill()

  3. 日志清理

    import os import tempfile class TempFileAddon: def __init__(self): self.temp_dir = tempfile.mkdtemp() def done(self): import shutil shutil.rmtree(self.temp_dir)

9. 扩展应用场景

9.1 移动端调试

自动修改User-Agent模拟不同设备：

def request(self, flow): if flow.request.host == "m.example.com": flow.request.headers["User-Agent"] = ( "Mozilla/5.0 (iPhone; CPU iPhone OS 13_2_3 like Mac OS X) " "AppleWebKit/605.1.15 (KHTML, like Gecko) Version/13.0.3 Mobile/15E148 Safari/604.1" )

9.2 API Mock服务

实现简单的Mock服务器：

def request(self, flow): if flow.request.path == "/api/user": flow.response = http.Response.make( 200, json.dumps({"id": 1, "name": "Mock User"}), {"Content-Type": "application/json"} )

9.3 性能监控

记录请求耗时：

class PerfAddon: def __init__(self): self.timings = [] def request(self, flow): flow.start_time = time.time() def response(self, flow): duration = time.time() - flow.start_time self.timings.append((flow.request.url, duration)) def done(self): avg = sum(t[1] for t in self.timings) / len(self.timings) print(f"平均响应时间: {avg:.2f}s")

10. 项目部署与自动化

10.1 持续集成集成

将mitmproxy脚本集成到CI流程中：

# .github/workflows/test.yml jobs: test: steps: - run: pip install mitmproxy - run: mitmproxy -s test_script.py -p 8080 & - run: curl --proxy http://localhost:8080 http://test.target

10.2 Docker化部署

创建专用Docker镜像：

FROM python:3.9 RUN pip install mitmproxy COPY script.py /app/script.py CMD ["mitmproxy", "-s", "/app/script.py"]

10.3 集群化方案

使用Redis共享数据：

import redis class ClusterAddon: def __init__(self): self.r = redis.Redis(host='redis', port=6379) def request(self, flow): self.r.incr(flow.request.host)

11. 脚本维护建议

  1. 模块化设计

    # auth.py class AuthAddon: def request(self, flow): flow.request.headers["Auth"] = "xxx" # stats.py class StatsAddon: def response(self, flow): pass # main.py from auth import AuthAddon from stats import StatsAddon addons = [AuthAddon(), StatsAddon()]

  2. 版本控制

    • 为脚本添加版本信息
    • 使用Git管理脚本变更

  3. 文档注释

    class Addon: """ 核心流量处理插件 功能： 1. 自动添加认证头 2. 记录API调用统计 """

12. 性能对比测试

通过实际测试比较不同实现方式的性能差异：

实现方式平均延迟(ms)内存占用(MB)适用场景
纯Python处理15.2120简单逻辑
C扩展模块3.895高性能需求
异步IO8.1110IO密集型
多进程5.3210CPU密集型

测试脚本示例：

import time from locust import HttpUser, task, between class TestUser(HttpUser): wait_time = between(1, 2.5) @task def test_request(self): self.client.get("/", proxies={"http": "http://localhost:8080"})

13. 最佳实践总结

经过多个项目的实践验证，我总结了以下mitmproxy脚本开发的最佳实践：

  1. 明确处理范围

    • 尽早过滤不需要处理的流量
    • 按域名、路径等特征进行路由

  2. 异常处理完善

    def response(self, flow): try: # 业务逻辑 except Exception as e: print(f"处理失败: {e}") flow.error = str(e)

  3. 资源释放管理

    def done(self): self.db.close() self.redis.close()

  4. 配置外部化

    import configparser config = configparser.ConfigParser() config.read('config.ini') class ConfigAddon: def request(self, flow): flow.request.headers["X-Config"] = config.get("DEFAULT", "key")

  5. 测试覆盖率

    import unittest class TestAddon(unittest.TestCase): def test_modify(self): from mitmproxy.test.tflow import tflow flow = tflow() addon = MyAddon() addon.request(flow) self.assertIn("X-Modified", flow.request.headers)

在实际项目中，我发现将mitmproxy脚本与常规开发流程结合，可以显著提升API开发和测试的效率。特别是在微服务架构下，通过流量录制和回放，能够快速构建集成测试用例。

网站建设 视觉设计 SEO优化
返回列表