https 拦截器

说明

先安装依赖,脚本运行后访问 http://mitm.it 安装证书。

pip install mitmproxy pywin32

脚本

import atexit
import sys
import winreg
import subprocess
import asyncio
import os
from datetime import datetime
from mitmproxy import http, options
from mitmproxy.tools.dump import DumpMaster

# 配置参数
PROXY_PORT = 28181
RESPONSE_CONFIG = {
    "url1": '{"status":"0"}'.encode('utf-8'),
    "url2": '{"code":1}'.encode('utf-8'),
}
PROXY_BYPASS = '<local>;localhost;127.*;10.*;172.16.*;172.17.*;172.18.*;172.19.*;172.20.*;172.21.*;172.22.*;172.23.*;172.24.*;172.25.*;172.26.*;172.27.*;172.28.*;172.29.*;172.30.*;172.31.*;192.168.*'

class Logger:
    @staticmethod
    def log(action: str, details: str = ""):
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        log_entry = f"[{timestamp}] {action}: {details}\n"
        with open("proxy.log", "a", encoding="utf-8") as f:
            f.write(log_entry)
        print(log_entry.strip())

class WindowsProxyManager:
    def __init__(self):
        self.original_settings = {}
        self.reg_path = r"Software\Microsoft\Windows\CurrentVersion\Internet Settings"

    def set_proxy(self, enable: bool):
        try:
            with winreg.OpenKey(winreg.HKEY_CURRENT_USER, self.reg_path, 0, winreg.KEY_ALL_ACCESS) as key:
                if enable:
                    self._backup_settings(key)
                    self._apply_proxy(key)
                else:
                    self._restore_settings(key)  # 确保方法存在
        except PermissionError:
            Logger.log("错误", "需要管理员权限")
            sys.exit(1)
        except Exception as e:
            Logger.log("代理错误", str(e))
            sys.exit(1)
        self._set_winhttp_proxy(enable)

    def _backup_settings(self, key):
        """备份原始注册表设置"""
        self.original_settings = {
            "ProxyServer": "",
            "ProxyEnable": 0,
            "ProxyOverride": ""
        }
        # try:
        #     self.original_settings = {
        #         "ProxyServer": winreg.QueryValueEx(key, "ProxyServer")[0],
        #         "ProxyEnable": winreg.QueryValueEx(key, "ProxyEnable")[0],
        #         "ProxyOverride": winreg.QueryValueEx(key, "ProxyOverride")[0]
        #     }
        #     Logger.log("系统代理", "备份成功")
        # except FileNotFoundError:
        #     Logger.log("警告", "初始代理设置不存在")
        #     self.original_settings = {
        #         "ProxyServer": "",
        #         "ProxyEnable": 0,
        #         "ProxyOverride": ""
        #     }

    def _apply_proxy(self, key):
        """应用新代理设置"""
        winreg.SetValueEx(key, "ProxyServer", 0, winreg.REG_SZ, f"127.0.0.1:{PROXY_PORT}")
        winreg.SetValueEx(key, "ProxyEnable", 0, winreg.REG_DWORD, 1)
        winreg.SetValueEx(key, "ProxyOverride", 0, winreg.REG_SZ, PROXY_BYPASS)
        self._refresh_system_settings()
        Logger.log("系统代理", "已启用新配置")

    def _restore_settings(self, key):
        """恢复原始设置(新增此方法)"""
        winreg.SetValueEx(
            key, 
            "ProxyServer", 
            0, 
            winreg.REG_SZ, 
            self.original_settings.get("ProxyServer", "")
        )
        winreg.SetValueEx(
            key, 
            "ProxyEnable", 
            0, 
            winreg.REG_DWORD, 
            self.original_settings.get("ProxyEnable", 0)
        )
        winreg.SetValueEx(
            key, 
            "ProxyOverride", 
            0, 
            winreg.REG_SZ, 
            self.original_settings.get("ProxyOverride", "")
        )
        self._refresh_system_settings()
        Logger.log("系统代理", "已恢复原始配置")

    def _refresh_system_settings(self):
        """刷新系统设置(修复here-string格式)"""
        try:
            # 使用精确格式化的 PowerShell 命令
            Logger.log("当前的 系统 代理服务器设置")
            ps_command = r'''
Add-Type -TypeDefinition @" 
using System.Runtime.InteropServices;
public class WinInet {
    [DllImport("wininet.dll")]
    public static extern bool InternetSetOption(int hInternet, int dwOption, int lpBuffer, int dwBufferLength);
}
"@
[WinInet]::InternetSetOption(0, 39, 0, 0) >$null
[WinInet]::InternetSetOption(0, 37, 0, 0) >$null
Get-ItemProperty -Path "HKCU:\\Software\\Microsoft\Windows\\CurrentVersion\\Internet Settings" | Select-Object ProxyServer, ProxyEnable, ProxyOverride
'''
            subprocess.run(
                ["powershell", "-Command", ps_command.strip()],
                check=True,
                timeout=10
            )
            Logger.log("系统刷新", "成功通知系统更新代理设置")
        except subprocess.CalledProcessError as e:
            Logger.log("刷新失败", f"错误代码: {e.returncode}")
        except subprocess.TimeoutExpired:
            Logger.log("刷新超时", "系统未及时响应")

    def _set_winhttp_proxy(self, enable: bool):
        """配置系统级代理"""
        try:
            cmd = (f'netsh winhttp set proxy 127.0.0.1:{PROXY_PORT} bypass-list="{PROXY_BYPASS}"' 
                   if enable else 'netsh winhttp reset proxy')
            
            result = subprocess.run(
                cmd,
                shell=True,
                check=True,
                capture_output=True,
                text=True
            )
            Logger.log(result.stdout.strip())
        except subprocess.CalledProcessError as e:
            Logger.log("WinHTTP错误", e.stderr.strip())

class RequestInterceptor:
    async def request(self, flow: http.HTTPFlow):
        # 遍历所有配置的URL
        for url in RESPONSE_CONFIG:
            if url in flow.request.pretty_url:
                Logger.log("拦截请求", f"{flow.request.url} -> 返回预设响应")
                flow.response = http.Response.make(
                    200,
                    RESPONSE_CONFIG[url],
                    {"Content-Type": "application/json; charset=utf-8"}
                )
                return  # 匹配到第一个符合的URL即返回

async def proxy_main():
    """修复后的异步主函数"""
    proxy_mgr = WindowsProxyManager()
    
    # if input("是否自动配置系统代理?(y/n) ").lower() != 'y':
    #     print("请手动设置代理: 127.0.0.1:28181")
    #     return

    proxy_mgr.set_proxy(True)
    atexit.register(proxy_mgr.set_proxy, False)

    # 初始化 mitmproxy 实例
    opts = options.Options(
        listen_port=PROXY_PORT,
        ssl_insecure=True,
        http2=True
    )
    master = DumpMaster(opts)
    master.addons.add(RequestInterceptor())

    try:
        print(f"\n{' MITM代理已启动 ':=^40}")
        print(f"| 监听端口: {PROXY_PORT:^24} |")
        print(f"{'':=^40}\n")
        
        # 使用官方推荐的运行方式
        await master.run()
    except asyncio.CancelledError:
        Logger.log("状态", "正在关闭代理服务...")
    finally:
        # 正确关闭资源
        if hasattr(master, "server"):
            master.server.close()
            Logger.log("状态", "网络端口已释放")

def main():
    """主入口函数"""
    if os.name == 'nt':
        from ctypes import windll
        if not windll.shell32.IsUserAnAdmin():
            windll.shell32.ShellExecuteW(
                None, "runas", sys.executable, " ".join(sys.argv), None, 1
            )
            return

    Logger.log("启动", f"进程ID: {os.getpid()}")
    
    try:
        asyncio.run(proxy_main())
    except KeyboardInterrupt:
        Logger.log("状态", "用户主动终止操作")
    finally:
        Logger.log("关闭", "服务已停止")

if __name__ == "__main__":
    main()
最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。

推荐阅读更多精彩内容