商城接口自動化：Mitmproxy代理錄制實戰(zhàn)指南

一、環(huán)境準(zhǔn)備與配置

1. 安裝與啟動

# 安裝mitmproxy（Python 3.6+）
pip install mitmproxy

# 啟動錄制代理（端口8080）
mitmdump -w mall_recording.mitm -p 8080

# 查看可用接口
mitmproxy --help

2. 客戶端代理配置

二、基礎(chǔ)流量錄制

1. 簡單錄制模式

# 錄制所有流量到文件
mitmdump -w mall_recording.mitm -p 8080

# 播放錄制的流量
mitmdump -n -r mall_recording.mitm -p 8081

2. 過濾指定域名

# 只錄制商城相關(guān)域名
mitmdump -w mall_recording.mitm -p 8080 --filter "~d mall.com ~d api.mall.com"

三、自動化錄制腳本

1. 智能錄制腳本

# smart_recorder.py
from mitmproxy import http
import json

class Recorder:
    def __init__(self):
        self.api_count = 0
        self.flows = []

    def request(self, flow: http.HTTPFlow):
        if "api.mall.com" in flow.request.host:
            flow.metadata["start_time"] = time.time()
            self.api_count += 1
            print(f"捕獲API請求: {flow.request.method} {flow.request.path}")

    def response(self, flow: http.HTTPFlow):
        if "api.mall.com" in flow.request.host:
            latency = time.time() - flow.metadata["start_time"]
            status = flow.response.status_code
            print(f"API響應(yīng): {status} 延遲: {latency:.2f}s")
            self.flows.append({
                "url": flow.request.url,
                "method": flow.request.method,
                "request": flow.request.text,
                "response": flow.response.text,
                "status": status,
                "latency": latency
            })

addons = [Recorder()]

2. 啟動帶腳本的錄制

mitmdump -s smart_recorder.py -w mall_recording.mitm -p 8080

四、流量分析與轉(zhuǎn)換

1. 生成自動化測試用例

# generate_testcases.py
import json
from mitmproxy import io
from mitmproxy.exceptions import FlowReadException

def convert_to_pytest(flow):
    template = f"""
def test_{flow.request.path.replace('/', '_')}():
    url = "{flow.request.url}"
    headers = {dict(flow.request.headers)}
    data = {flow.request.text if flow.request.text else None}
    
    response = requests.{flow.request.method.lower()}(
        url,
        headers=headers,
        json=data if flow.request.method in ["POST", "PUT"] else None
    )
    
    assert response.status_code == {flow.response.status_code}
    assert response.json() == {flow.response.text}
"""
    return template

with open("mall_recording.mitm", "rb") as f:
    flows = []
    reader = io.FlowReader(f)
    try:
        for flow in reader.stream():
            if "api.mall.com" in flow.request.host:
                flows.append(flow)
    except FlowReadException as e:
        print(f"流量文件損壞: {e}")

with open("test_api_auto.py", "w") as f:
    f.write("import requests\n\n")
    for flow in flows[:10]:  # 取前10個生成用例
        f.write(convert_to_pytest(flow))

五、高級錄制技巧

1. 接口自動分類

# api_classifier.py
API_CATEGORIES = {
    "user": ["/login", "/register", "/profile"],
    "product": ["/products", "/search", "/detail"],
    "order": ["/cart", "/checkout", "/payment"]
}

def request(self, flow: http.HTTPFlow):
    for category, paths in API_CATEGORIES.items():
        if any(p in flow.request.path for p in paths):
            flow.metadata["category"] = category
            break
    else:
        flow.metadata["category"] = "other"

2. 敏感數(shù)據(jù)脫敏

# data_scrubber.py
def response(self, flow: http.HTTPFlow):
    if "user" in flow.metadata.get("category", ""):
        import json
        data = json.loads(flow.response.text)
        if "phone" in data:
            data["phone"] = "***" + data["phone"][-4:]
        if "address" in data:
            data["address"] = "<REDACTED>"
        flow.response.text = json.dumps(data)

六、自動化測試集成

1. 結(jié)合Pytest框架

# conftest.py
import pytest
from mitmproxy.tools.main import mitmdump
import threading

@pytest.fixture(scope="session")
def proxy():
    def run():
        mitmdump(["-s", "recorder.py", "-w", "test_flows.mitm"])
    
    t = threading.Thread(target=run, daemon=True)
    t.start()
    yield
    # 測試結(jié)束后停止代理

@pytest.fixture
def recorded_flows():
    from mitmproxy import io
    flows = []
    with open("test_flows.mitm", "rb") as f:
        reader = io.FlowReader(f)
        try:
            for flow in reader.stream():
                flows.append(flow)
        except:
            pass
    return flows

2. 流量斷言測試

# test_replay.py
def test_order_flow(recorded_flows):
    order_flows = [f for f in recorded_flows if "/order/" in f.request.path]
    
    for flow in order_flows:
        response = requests.request(
            flow.request.method,
            flow.request.url,
            headers=dict(flow.request.headers),
            data=flow.request.content
        )
        
        assert response.status_code == flow.response.status_code
        if flow.response.headers.get("Content-Type") == "application/json":
            assert response.json() == flow.response.json()

七、CI/CD集成方案

1. GitHub Actions配置

name: API Test with Mitmproxy

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v2
    - name: Set up Python
      uses: actions/setup-python@v2
    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install mitmproxy pytest requests
    - name: Run tests
      run: |
        mitmdump -s recorder.py -w test_flows.mitm &
        pytest tests/ -v
    - name: Upload artifacts
      uses: actions/upload-artifact@v2
      with:
        name: test-flows
        path: test_flows.mitm

八、最佳實踐建議

1. 錄制策略：按業(yè)務(wù)場景分開錄制（登錄流程、下單流程等）每個場景錄制3-5次獲取典型用例
2. 數(shù)據(jù)管理：
3. 自動化增強(qiáng)：Response:"""
4. 異常檢測：

九、典型問題解決方案

1. HTTPS證書問題

# 啟動時跳過證書驗證
mitmdump --ssl-insecure -w flows.mitm

# 手動信任證書（手機(jī)端）
adb push ~/.mitmproxy/mitmproxy-ca-cert.cer /sdcard/
# 然后在設(shè)置中安裝證書

2. 錄制文件過大

# 按大小自動分割文件
from mitmproxy import ctx

class Rotator:
    def __init__(self):
        self.count = 0
        self.max_size = 100 * 1024 * 1024  # 100MB
    
    def running(self):
        if os.path.getsize(ctx.options.flow_detail) > self.max_size:
            self.count += 1
            new_file = f"recording_{self.count}.mitm"
            os.rename(ctx.options.flow_detail, new_file)

3. 移動端請求捕獲不全

# 強(qiáng)制攔截所有流量
def request(flow):
    if flow.request.host not in ["api.mall.com", "static.mall.com"]:
        print(f"攔截非商城流量: {flow.request.host}")
        flow.kill()

通過這套實戰(zhàn)方案，您可以快速實現(xiàn)：

1. 商城接口的自動化錄制
2. 流量到測試用例的自動轉(zhuǎn)換
3. 異常接口的智能檢測
4. 與CI系統(tǒng)的無縫集成
5. 可追溯的接口文檔生成

建議從核心交易鏈路開始實施，逐步擴(kuò)大覆蓋范圍，最終建立完整的接口自動化測試體系。