Python 中非命令行启动 mitmproxy 的多种实现方案
在 Python 应用中集成 mitmproxy 代理服务
在移动应用测试或接口分析过程中，抓包是常见的需求。mitmproxy 作为一个支持 HTTPS、HTTP/2 和 WebSocket 的中间人代理工具，提供了强大的流量拦截与修改能力。除了通过命令行启动外，我们还可以将其嵌入到 Python 程序中，实现更灵活的控制。
核心启动方式对比
- 阻塞式启动：适用于独立脚本任务，主程序需等待代理结束。
- 非阻塞式启动：适合 GUI 或后台服务场景，可动态启停代理进程。
方法一：使用 main.mitmdump 直接调用（阻塞）
利用 mitmproxy 内部的 mitmdump 函数启动代理，配置参数以列表形式传入。
from mitmproxy.tools.main import mitmdump
class RequestLogger:
def request(self, flow):
print(f"请求地址: {flow.request.pretty_url}")
def response(self, flow):
if flow.response.status_code == 200:
print(f"响应长度: {len(flow.response.content)}")
if __name__ == "__main__":
# 启动代理并加载当前脚本作为插件
mitmdump([
'-s', __file__,
'--listen-host', '0.0.0.0',
'-p', '8888',
'--set', 'block_global=false'
])
该方式会阻塞主线程，无法继续执行后续代码。
方法二：基于 DumpMaster 与异步事件循环（阻塞）
手动创建 DumpMaster 实例，结合 asyncio 运行代理主循环。
import asyncio
from mitmproxy.options import Options
from mitmproxy.tools.dump import DumpMaster
class TrafficCounter:
def __init__(self):
self.count = 0
def response(self, flow):
self.count += 1
print(f"已捕获 {self.count} 个响应")
async def run_proxy(host='127.0.0.1', port=8080):
opts = Options(listen_host=host, listen_port=port)
master = DumpMaster(opts)
master.addons.add(TrafficCounter())
try:
await master.run()
except KeyboardInterrupt:
master.shutdown()
if __name__ == "__main__":
asyncio.run(run_proxy('0.0.0.0', 8888))
方法三（推荐）：多进程封装异步代理（非阻塞）
为解决阻塞问题，将代理运行于独立进程中，主程序可通过句柄控制其生命周期。
from multiprocessing import Process
import asyncio
from mitmproxy.options import Options
from mitmproxy.tools.dump import DumpMaster
class ProxyController:
def __init__(self, host='127.0.0.1', port=8088):
self.host = host
self.port = port
self.process = None
async def _start(self):
options = Options(listen_host=self.host, listen_port=self.port)
master = DumpMaster(options)
master.addons.add(TrafficCounter()) # 添加自定义插件
try:
await master.run()
except Exception as e:
print(f"代理异常退出: {e}")
finally:
master.shutdown()
def _run_in_event_loop(self):
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(self._start())
loop.close()
def start(self):
"""启动代理进程"""
if self.process is None:
self.process = Process(target=self._run_in_event_loop, daemon=True)
self.process.start()
print(f"代理服务已在 {self.host}:{self.port} 启动")
def stop(self):
"""终止代理进程"""
if self.process and self.process.is_alive():
self.process.terminate()
self.process.join(timeout=3)
if self.process.is_alive():
self.process.kill() # 强制结束
self.process = None
print("代理已关闭")
使用示例：
if __name__ == "__main__":
proxy = ProxyController('0.0.0.0', 8888)
proxy.start()
import time
time.sleep(30) # 模拟运行30秒
proxy.stop()
此方案稳定且易于集成，推荐用于生产环境。
方法四：QProcess 启动（PyQt/PySide 兼容）
在图形界面应用中，可使用 Qt 的 QProcess 来管理外部代理进程。
from PyQt5.QtCore import QProcess, QObject
class QtProxyManager(QObject):
def __init__(self, parent=None):
super().__init__(parent)
self.process = QProcess(self)
self.process.stateChanged.connect(self.on_state_change)
def start_proxy(self):
args = [
'--listen-host', '0.0.0.0',
'-p', '8888',
'--set', 'block_global=false',
'-s', './addons.py' # 外部插件文件
]
self.process.start('mitmdump', args)
def on_state_change(self, state):
states = {0: "未运行", 1: "启动中", 2: "运行中"}
print(f"代理状态: {states.get(state, '未知')}")
def stop_proxy(self):
if self.process.state() != QProcess.NotRunning:
self.process.terminate()
self.process.waitForFinished(3000)
注意：打包时需确保 mitmdump 可执行文件存在于 PATH 或同级目录。
注意事项
- 权限问题：监听非本地地址需设置
block_global=false，否则移动端无法连接。 - 端口释放：强制终止进程后可能出现端口占用，建议合理处理退出逻辑。
daemon=True设置守护进程可避免主程序退出时残留子进程。
