Mitmproxy在商城系统中的实战应用指南

一、环境准备与配置

1. 安装与启动

# 安装mitmproxy
pip install mitmproxy

# 启动带Web界面的代理（端口8080）
mitmweb -p 8080 --web-host 0.0.0.0

# 生成CA证书（首次使用需要）
mkdir -p ~/.mitmproxy
mitmproxy --create-ca-cert

2. 客户端代理配置

二、核心业务流量分析

1. 用户登录流程抓包

# login_analyzer.py
from mitmproxy import http
import json

def request(flow: http.HTTPFlow):
    if "/api/login" in flow.request.path:
        print(f"登录请求: {flow.request.method} {flow.request.url}")
        print(f"请求头: {flow.request.headers}")
        try:
            body = json.loads(flow.request.text)
            print(f"登录账号: {body.get('username')}")
        except:
            pass

def response(flow: http.HTTPFlow):
    if "/api/login" in flow.request.path:
        print(f"登录响应状态: {flow.response.status_code}")
        if flow.response.status_code == 200:
            print(f"SessionID: {flow.response.cookies.get('sessionid')}")

2. 商品搜索分析

# search_monitor.py
from urllib.parse import parse_qs

def request(flow: http.HTTPFlow):
    if "/api/search" in flow.request.path:
        params = parse_qs(flow.request.query)
        print(f"搜索关键词: {params.get('q', ['无'])[0]}")
        print(f"筛选条件: {params.get('filters', ['无'])[0]}")

三、请求/响应修改实战

1. 自动添加优惠券

# auto_coupon.py
def request(flow: http.HTTPFlow):
    if "/api/cart/checkout" in flow.request.path:
        import json
        try:
            data = json.loads(flow.request.text)
            if "coupon_code" not in data:
                data["coupon_code"] = "WELCOME2023"
                flow.request.text = json.dumps(data)
                print("已自动添加优惠券")
        except:
            pass

2. 修改商品价格测试

# price_modifier.py
def response(flow: http.HTTPFlow):
    if "/api/product/detail" in flow.request.path:
        import json
        try:
            data = json.loads(flow.response.text)
            original_price = data["price"]
            data["price"] = 0.01  # 修改为测试价格
            flow.response.text = json.dumps(data)
            print(f"价格已从{original_price}修改为0.01")
        except:
            pass

四、性能测试辅助

1. 接口延迟统计

# latency_stats.py
import time

def request(flow: http.HTTPFlow):
    flow.start_time = time.time()

def response(flow: http.HTTPFlow):
    latency = (time.time() - flow.start_time) * 1000
    print(f"{flow.request.path} 延迟: {latency:.2f}ms")
    if latency > 500:
        print("⚠️ 高延迟警告")

2. 流量录制与回放

# 录制测试流量（保存到mall_flow.mitm）
mitmdump -w mall_flow.mitm -p 8080

# 回放流量（测试服务器性能）
mitmdump -n -r mall_flow.mitm -p 8081

五、安全测试案例

1. 越权访问测试

# auth_bypass.py
def request(flow: http.HTTPFlow):
    if "/api/admin/" in flow.request.path:
        # 尝试绕过权限检查
        flow.request.headers["X-Admin-Token"] = "bypass_token"
        print("尝试注入管理员token")

2. XSS漏洞检测

# xss_detector.py
def response(flow: http.HTTPFlow):
    if "text/html" in flow.response.headers.get("content-type", ""):
        if "<script>" in flow.response.text:
            print(f"⚠️ 发现可能的XSS漏洞: {flow.request.url}")

六、自动化测试集成

1. 与Pytest结合

# test_payment.py
import pytest
import requests

@pytest.fixture
def proxy():
    from mitmproxy.tools.main import mitmdump
    import threading
    
    def run():
        mitmdump(["-s", "mock_payment.py", "-p", "8080"])
    
    t = threading.Thread(target=run, daemon=True)
    t.start()
    yield
    # 测试结束自动停止

def test_payment_success(proxy):
    proxies = {"http": "http://localhost:8080"}
    resp = requests.post(
        "http://mall.com/api/pay",
        json={"order_id": "TEST123", "amount": 100},
        proxies=proxies,
        verify=False
    )
    assert resp.json()["status"] == "success"

2. 性能测试脚本

# load_test.py
from locust import HttpUser, task, between

class MallUser(HttpUser):
    wait_time = between(1, 3)
    
    def on_start(self):
        self.proxies = {"http": "http://localhost:8080"}
    
    @task
    def browse_products(self):
        self.client.get(
            "/api/products",
            proxies=self.proxies,
            verify=False
        )
    
    @task
    def checkout(self):
        self.client.post(
            "/api/checkout",
            json={"items": [{"id": 1, "qty": 2}]},
            proxies=self.proxies,
            verify=False
        )

七、移动端专项测试

1. 小程序流量捕获

# weapp_filter.py
def request(flow: http.HTTPFlow):
    # 微信小程序特有header
    if "wx-miniprogram" in flow.request.headers.get("User-Agent", ""):
        print(f"小程序请求: {flow.request.path}")
        # 修改小程序环境检测
        flow.request.headers["X-WX-Source"] = "developer"

2. APP接口Mock

# app_mock.py
def response(flow: http.HTTPFlow):
    if flow.request.host == "api.mall.com":
        if "/app/v3/home" in flow.request.path:
            flow.response = http.Response.make(
                200,
                open("mock_data/home.json").read(),
                {"Content-Type": "application/json"}
            )

八、高级调试技巧

1. 断点调试

# debug_flow.py
from mitmproxy import ctx

def request(flow: http.HTTPFlow):
    if "checkout" in flow.request.path:
        ctx.log.info("进入调试模式，请求暂停...")
        from pdb import set_trace; set_trace()  # 交互式调试
        ctx.log.info("继续执行请求")

2. 流量对比分析

# diff_analyzer.py
import difflib

def response(flow: http.HTTPFlow):
    if flow.request.path == "/api/products":
        current = flow.response.text
        cached = get_cached_response()
        diff = difflib.unified_diff(
            cached.splitlines(),
            current.splitlines(),
            fromfile='cached',
            tofile='current'
        )
        print('\n'.join(diff))

九、最佳实践建议

1. 环境隔离：为不同测试目的创建独立配置
2. 敏感数据处理：在保存流量时自动脱敏
3. 脚本模块化：按功能拆分脚本文件
4. 自动化集成：与CI/CD流程结合

十、典型问题解决方案

1. HTTPS抓包失败

# 确保客户端已安装CA证书
# 启动时添加--ssl-insecure参数
mitmproxy -p 8080 --ssl-insecure

# 检查证书有效期
openssl x509 -in ~/.mitmproxy/mitmproxy-ca-cert.pem -noout -dates

2. 移动端无法连接

# 确保代理IP正确（电脑和手机同一网络）
import socket
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(("8.8.8.8", 80))
print("电脑IP:", s.getsockname()[0])

3. 性能问题排查

# 监控mitmproxy自身资源使用
def running():
    import psutil, time
    while True:
        mem = psutil.Process().memory_info().rss / 1024 / 1024
        print(f"内存占用: {mem:.2f}MB")
        time.sleep(5)

通过Mitmproxy在商城系统中可以实现：

• 实时监控API调用情况
• 快速验证前后端数据交互
• 模拟异常场景测试系统容错
• 性能瓶颈定位与分析
• 安全漏洞检测

建议从核心交易链路开始（登录->浏览->下单->支付），逐步扩展到全功能测试，最终形成完整的自动化测试体系。