Files
wxdown-service/ui/startup.py
T
jock 3bf0b1aaa2 fix: 改善启动失败反馈、Ctrl+C 立退与 Windows 兼容性
- main: Ctrl+C 时直接 SIGKILL 所有 daemon 子进程,跳过 mitmproxy graceful shutdown
  与 mitmdump 重启循环里的 time.sleep(3),实现秒退
- main: 启动失败时透传子进程真实错误行(端口/CA 缺失/文件权限等),告别"请切换端口后重试"
  一刀切提示;Windows 下等用户按回车再退出,防止 PowerShell/双击启动窗口闪退
- mitm: start() 返回 (proxy_address, error_line);mitmdump 重启循环处理 KeyboardInterrupt
- watcher: start() 返回 (ws_address, queue, error_line);将 wss 证书生成失败纳入快速失败识别
- utils: check_system_proxy 修复 KeyError — Windows 下 getproxies() 可能返回仅含 'no' 等
  非 http/https key 的字典,旧写法 proxy_obj.keys() < {'http', 'https'} 是真子集判断,
  会漏掉这些场景继续访问 proxy_obj['http'] 引发崩溃
- ui/startup: 证书安装命令里 %userprofile% 在 PowerShell 下不会展开导致
  certutil 报 ERROR_PATH_NOT_FOUND;改为 Python 侧展开为绝对路径,cmd/PowerShell/bash/zsh
  均可直接粘贴执行,同时用双引号包住路径兼容带空格的用户名
2026-04-20 13:53:45 +08:00

81 lines
3.3 KiB
Python

import multiprocessing
from pathlib import Path
from rich.layout import Layout
from ui.layout import make_layout
from ui.header_panel import HeaderPanel
from ui.service_panel import ServicePanel
from ui.status_panel import StatusPanel
from rich.live import Live
import time
import utils
import cert
import platform
import threading
from logger import logger
def startup_ui_loop(watcher_output_queue: multiprocessing.Queue, mitm_proxy_address = None, ws_address = None):
layout = make_layout()
layout['header'].update(HeaderPanel())
layout['service'].update(ServicePanel([
{'name': 'mitmproxy', 'address': mitm_proxy_address},
{'name': 'websocket', 'address': ws_address},
]))
layout['status'].update(StatusPanel())
threading.Thread(target=log_watcher_output, args=(watcher_output_queue,layout), daemon=True).start()
with Live(layout, refresh_per_second=4, screen=False, transient=True):
while True:
time.sleep(1)
# 检查证书是否安装
try:
if not cert.is_certificate_installed('mitmproxy'):
# 用 Python 先把 ~ / %userprofile% 展开为绝对路径,避免
# PowerShell 不认 %userprofile%、导致 certutil 报
# ERROR_PATH_NOT_FOUND 的经典坑。这条命令在 cmd / PowerShell /
# Windows Terminal / bash / zsh 下都能直接粘贴执行。
mitm_dir = Path.home() / '.mitmproxy'
if platform.system() == 'Windows':
cer_path = mitm_dir / 'mitmproxy-ca-cert.cer'
cmd = f'certutil -addstore root "{cer_path}"'
elif platform.system() == 'Darwin':
pem_path = mitm_dir / 'mitmproxy-ca-cert.pem'
cmd = (f'sudo security add-trusted-cert -d -p ssl -p basic '
f'-k /Library/Keychains/System.keychain "{pem_path}"')
else:
cmd = '(当前系统暂不支持自动给出安装命令)'
layout['status'].update(
StatusPanel(is_success=False, reason="系统中未检测到 mitmproxy 的证书,请手动安装。",
details=f"执行以下命令安装证书:\n[bold green]{cmd}[/]"))
continue
except Exception as e:
logger.warning(e)
layout['status'].update(
StatusPanel(is_success=False, reason="系统检测 mitmproxy 证书时异常。",
details="请将日志文件发送给开发者"))
continue
# 检查代理是否正确
success, reason, details = utils.check_system_proxy(mitm_proxy_address)
layout['status'].update(StatusPanel(is_success=success, ws_address=ws_address, reason=reason, details=details))
continue
def log_watcher_output(watcher_output_queue: multiprocessing.Queue, layout: Layout):
clients = 0
credentials = 0
while True:
line = watcher_output_queue.get()
parts = line.split(':')
if parts[0] == 'clients':
clients = int(parts[1])
if parts[0] == 'credentials':
credentials = int(parts[1])
layout['header'].update(HeaderPanel(clients, credentials))