说明
先安装依赖,脚本运行后访问 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()