python+mitmproxy抓包的实现

pip install mitmproxy
# 验证
mitmproxy --version

import ctypes
import winreg


def set_proxy(enable_proxy, proxy_address="http://127.0.0.1:8080"):
    try:
        # 代理服务器地址和端口
        proxy_server = proxy_address

        # 打开注册表键
        key_path = r"Software\Microsoft\Windows\CurrentVersion\Internet Settings"
        key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, key_path, 0, winreg.KEY_SET_VALUE)

        # 设置代理服务器
        if enable_proxy:
            winreg.SetValueEx(key, "ProxyServer", 0, winreg.REG_SZ, proxy_server)
            winreg.SetValueEx(key, "ProxyEnable", 0, winreg.REG_DWORD, 1)
        else:
            # 关闭代理
            winreg.SetValueEx(key, "ProxyEnable", 0, winreg.REG_DWORD, 0)

        # 刷新代理设置
        INTERNET_OPTION_REFRESH = 37
        INTERNET_OPTION_SETTINGS_CHANGED = 39
        internet_set_option = ctypes.windll.Wininet.InternetSetOptionW
        internet_set_option(0, INTERNET_OPTION_REFRESH, 0, 0)
        internet_set_option(0, INTERNET_OPTION_SETTINGS_CHANGED, 0, 0)

        # 关闭注册表键
        winreg.CloseKey(key)
        print("系统代理设置成功！")
    except Exception as e:
        print(f"设置系统代理失败: {e}")


if __name__ == "__main__":
    # 设置代理（启用代理）
    set_proxy(enable_proxy=True, proxy_address="http://127.0.0.1:8080")
    # 设置代理（关闭代理）
    # set_proxy(enable_proxy=False)

import subprocess
import platform


def is_mitmproxy_cert_installed():
    try:
        # 使用 PowerShell 检查证书是否存在
        res = subprocess.check_output(['powershell',
                                       'Get-ChildItem -Path Cert:\CurrentUser\Root | Where-Object {$_.Subject -like "*mitmproxy*"}'])
        if res:
            return True
        return False
    except subprocess.CalledProcessError as e:
        return False


def install_mitmproxy_certificate(cert_path):
    system_platform = platform.system()
    if system_platform == "Windows":
        # Windows系统下使用certutil命令
        try:
            res = subprocess.run(["certutil.exe", "-addstore", "root", cert_path], check=True, capture_output=True,
                                 text=True)
            print(res)
            print("Mitmproxy证书已成功安装到根证书存储中。")
        except subprocess.CalledProcessError as e:
            print(f"安装Mitmproxy证书失败: {e}")


if __name__ == "__main__":
    if is_mitmproxy_cert_installed():
        print("Mitmproxy证书已安装")
    else:
        print("Mitmproxy证书未安装")
        # 替换为实际的证书路径
        certificate_path = r"mitmproxy-ca-cert.cer"
        install_mitmproxy_certificate(certificate_path)

# "certmgr.msc"

import os

import set_proxy

if __name__ == '__main__':
    try:
        set_proxy(enable_proxy=True, proxy_address="http://127.0.0.1:8080")
        os.system("mitmweb")
        # os.system("mitmdump -s .\my_script.py")
    except KeyboardInterrupt:
        set_proxy(enable_proxy=False)

import asyncio
import os

from mitmproxy import options
from mitmproxy.tools.dump import DumpMaster


import set_proxy
import my_script


async def start_mitmproxy():
    opts = options.Options(listen_host='0.0.0.0', listen_port=8080)
    master = DumpMaster(opts)
    master.addons.add(my_script)
    await master.run()


if __name__ == '__main__':
    try:
        set_proxy(enable_proxy=True, proxy_address="http://127.0.0.1:8080")
        asyncio.run(start_mitmproxy())
    except KeyboardInterrupt:
        set_proxy(enable_proxy=False)

import logging
import mitmproxy.http

num = 0
 
def request(flow: mitmproxy.http.HTTPFlow):
    global num
    num = num + 1
    print("We've seen %d flows" % num)

import logging


class Counter:
    def __init__(self):
        self.num = 0

    def request(self, flow):
        self.num = self.num + 1
        logging.info("We've seen %d flows" % self.num)


addons = [Counter()]