Mitmproxy 原理深度解析

一、核心架构设计

1. 整体架构图

graph TD
    A[客户端] -->|1. 连接| B[Mitmproxy]
    B -->|2. 解密| C[SSL/TLS]
    C -->|3. 解析| D[协议层]
    D -->|4. 修改| E[脚本系统]
    E -->|5. 转发| F[目标服务器]
    F -->|6. 响应| B
    B -->|7. 加密| A

2. 核心组件

二、中间人代理原理

1. TLS/SSL解密流程

# 简化的证书生成逻辑
def generate_cert(original_cert):
    ca_key = load_ca_private_key()  # 加载CA私钥
    fake_cert = Certificate(
        issuer=ca_key.public_key(),
        subject=original_cert.subject,
        serial=random_serial(),
        validity=original_cert.validity
    )
    fake_cert.sign(ca_key, "sha256")
    return fake_cert

2. 连接劫持过程

1. 客户端发送CONNECT请求
2. Mitmproxy返回200 Connection Established
3. 客户端发起TLS握手
4. Mitmproxy动态生成服务器证书
5. 建立双向TLS连接（客户端⇄Mitmproxy⇄服务器）

三、事件处理机制

1. 关键事件钩子

2. 事件处理示例

from mitmproxy import http

def request(flow: http.HTTPFlow) -> None:
    # 请求阶段修改
    if "example.com" in flow.request.host:
        flow.request.headers["X-Proxy"] = "mitmproxy"
        
def response(flow: http.HTTPFlow) -> None:
    # 响应阶段修改
    if "json" in flow.response.headers.get("content-type", ""):
        flow.response.text = flow.response.text.replace(
            "original", 
            "modified"
        )

四、协议支持实现

1. HTTP/1.x处理

class HTTP1Layer:
    def __init__(self, connection):
        self.conn = connection
    
    def read_request(self):
        # 解析请求行和头部
        method, path, version = parse_request_line(self.conn.rfile)
        headers = parse_headers(self.conn.rfile)
        return Request(method, path, version, headers)
    
    def write_response(self, response):
        # 构造响应报文
        self.conn.wfile.write(f"HTTP/1.1 {response.status_code}\r\n")
        for k, v in response.headers.items():
            self.conn.wfile.write(f"{k}: {v}\r\n")
        self.conn.wfile.write("\r\n")
        self.conn.wfile.write(response.body)

2. HTTP/2特性支持

五、脚本系统原理

1. 脚本执行流程

sequenceDiagram
    participant C as Client
    participant M as Mitmproxy
    participant S as Server
    
    C->>M: 发起请求
    M->>Script: 触发request事件
    Script-->>M: 修改请求
    M->>S: 转发修改后请求
    S->>M: 返回响应
    M->>Script: 触发response事件
    Script-->>M: 修改响应
    M->>C: 返回修改后响应

2. 脚本加载机制

def load_scripts(script_files):
    scripts = []
    for path in script_files:
        # 创建独立命名空间
        namespace = {
            "__file__": os.path.abspath(path),
            "mitmproxy": mitmproxy_imports
        }
        with open(path) as f:
            code = compile(f.read(), path, "exec")
            exec(code, namespace)
        scripts.append(namespace)
    return scripts

六、性能优化设计

1. 连接池管理

class ConnectionPool:
    def __init__(self):
        self.pool = {}
    
    def get_connection(self, host, port):
        key = (host, port)
        if key not in self.pool or self.pool[key].closed:
            self.pool[key] = create_connection(host, port)
        return self.pool[key]

2. 内存优化策略

七、安全机制详解

1. 证书信任链验证

def verify_certificate(cert):
    # 检查证书是否由mitmproxy CA签发
    if not cert.issuer == mitmproxy_ca.subject:
        raise CertificateError("Untrusted issuer")
    
    # 检查主机名匹配
    if not hostname_matches(cert, flow.request.host):
        raise CertificateError("Hostname mismatch")
    
    # 检查有效期
    if cert.not_valid_after < datetime.now():
        raise CertificateError("Expired certificate")

2. 防中间人攻击保护

def check_http_security(headers):
    # 检测HSTS头
    if "strict-transport-security" in headers:
        log.warning("HSTS detected, MITM may fail")
    
    # 检测CSP头
    if "content-security-policy" in headers:
        log.warning("CSP may block modified content")

八、扩展开发接口

1. Addon开发示例

from mitmproxy import addonmanager

class MyAddon:
    def load(self, loader):
        loader.add_option(
            name="my_option", 
            typespec=str,
            default="",
            help="Custom option"
        )
    
    def request(self, flow):
        if ctx.options.my_option:
            flow.request.headers["X-My-Option"] = ctx.options.my_option

addons = [MyAddon()]

2. 自定义协议支持

class MyProtocolLayer:
    def __init__(self, config):
        self.config = config
    
    def __call__(self, data):
        # 解析自定义协议
        if is_my_protocol(data):
            return self.handle(data)
        return data
    
    def handle(self, data):
        # 处理协议逻辑
        return modified_data

九、与传统代理对比

十、内部工作流程示例

1. 完整请求处理时序

sequenceDiagram
    participant C as Client
    participant P as ProxyCore
    participant S as Script
    participant D as Destination
    
    C->>P: TCP握手
    P->>C: 200 Connection Established
    C->>P: TLS握手（含SNI）
    P->>P: 动态生成证书
    P->>C: 返回伪造证书
    C->>P: 加密请求
    P->>P: TLS解密
    P->>S: 触发request事件
    S->>P: 返回修改后请求
    P->>D: 转发请求
    D->>P: 返回响应
    P->>S: 触发response事件
    S->>P: 返回修改后响应
    P->>P: TLS加密
    P->>C: 返回加密响应

2. 关键数据结构

class HTTPFlow:
    """表示完整的HTTP事务"""
    def __init__(self):
        self.request: Request     # 请求对象
        self.response: Response   # 响应对象
        self.error: Error         # 错误信息
        self.metadata: dict       # 自定义元数据

class Request:
    """HTTP请求"""
    def __init__(self):
        self.method: str
        self.url: str
        self.headers: Headers
        self.content: bytes
        self.timestamp_start: float

class Response:
    """HTTP响应"""
    def __init__(self):
        self.status_code: int
        self.headers: Headers
        self.content: bytes
        self.timestamp_end: float

Mitmproxy的独特优势在于：

1. 深度可编程性：通过Python脚本实现任意流量修改逻辑
2. 协议透明性：完整解析各层协议（包括加密流量）
3. 双向拦截能力：可同时修改请求和响应
4. 跨平台支持：无缝支持桌面和移动端设备
5. 丰富的扩展接口：支持开发复杂插件系统

理解这些原理有助于：

• 更高效地编写拦截脚本
• 诊断代理连接问题
• 开发定制化中间件
• 优化大规模流量处理性能
• 构建自动化测试平台