#!/usr/bin/env python3 from __future__ import annotations import base64 import csv import json import os import queue import re import select import shlex import signal import socket import subprocess import threading import time import urllib.parse import urllib.request from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any import concurrent.futures import sys import uuid class DualStackHTTPServer(ThreadingHTTPServer): def __init__(self, server_address, RequestHandlerClass, bind_and_activate=True): host, port = server_address if ":" in host or host == "": self.address_family = socket.AF_INET6 else: self.address_family = socket.AF_INET try: super().__init__(server_address, RequestHandlerClass, bind_and_activate) except OSError as e: if self.address_family == socket.AF_INET6: fallback_host = "0.0.0.0" if host in ("::", "") else "127.0.0.1" print(f"[警告] 绑定 Web 管理后台 IPv6 {host}:{port} 失败 ({e}),正在尝试回退至 IPv4 {fallback_host} ...", flush=True) # 关闭第一次失败时可能已创建的 socket try: self.socket.close() except Exception: pass self.address_family = socket.AF_INET super().__init__((fallback_host, port), RequestHandlerClass, bind_and_activate) else: raise e def server_bind(self): if self.address_family == socket.AF_INET6: try: self.socket.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0) except OSError: pass super().server_bind() import vpn_utils import proxy_server import snapshot_utils def env_int(name: str, default: int, min_value: int | None = None, max_value: int | None = None) -> int: raw = os.environ.get(name) try: value = int(raw) if raw not in (None, "") else default except (TypeError, ValueError): print(f"[配置警告] 环境变量 {name}={raw!r} 不是有效整数,使用默认值 {default}", flush=True) value = default if min_value is not None and value < min_value: print(f"[配置警告] 环境变量 {name}={value} 小于允许值 {min_value},使用默认值 {default}", flush=True) return default if max_value is not None and value > max_value: print(f"[配置警告] 环境变量 {name}={value} 大于允许值 {max_value},使用默认值 {default}", flush=True) return default return value def bounded_int(value: Any, default: int, min_value: int | None = None, max_value: int | None = None) -> int: try: parsed = int(value) except (TypeError, ValueError): return default if min_value is not None and parsed < min_value: return default if max_value is not None and parsed > max_value: return default return parsed API_HTTPS_URL = os.environ.get("VPNGATE_API_HTTPS_URL", "https://www.vpngate.net/api/iphone/").strip() API_HTTP_URL = os.environ.get("VPNGATE_API_HTTP_URL", "http://www.vpngate.net/api/iphone/").strip() MIRROR_HTTPS_URL = os.environ.get( "VPNGATE_MIRROR_HTTPS_URL", "https://baoweise-bot.github.io/aimili-vpngate/vpngate.csv", ).strip() MIRROR_HTTP_URL = os.environ.get( "VPNGATE_MIRROR_HTTP_URL", "http://baoweise-bot.github.io/aimili-vpngate/vpngate.csv", ).strip() # Kept as the primary URL for diagnostics and backwards-compatible state output. API_URL = API_HTTPS_URL FETCH_INTERVAL_SECONDS = env_int("FETCH_INTERVAL_SECONDS", 1260, 1) CHECK_INTERVAL_SECONDS = env_int("CHECK_INTERVAL_SECONDS", 1260, 1) TARGET_VALID_NODES = env_int("TARGET_VALID_NODES", 3, 1) MAX_SCAN_ROWS = env_int("MAX_SCAN_ROWS", 300, 1) API_FETCH_TIMEOUT_SECONDS = env_int("API_FETCH_TIMEOUT_SECONDS", 10, 1, 60) OPENVPN_TEST_TIMEOUT_SECONDS = env_int("OPENVPN_TEST_TIMEOUT_SECONDS", 35, 1) MANUAL_TEST_NODE_LIMIT = env_int("MANUAL_TEST_NODE_LIMIT", 5, 1, 20) INITIAL_CONNECT_TEST_LIMIT = env_int("INITIAL_CONNECT_TEST_LIMIT", 10, 1, 50) NODE_PROBE_WORKERS = env_int("NODE_PROBE_WORKERS", 5, 1, 20) PROXY_FAILURE_THRESHOLD = env_int("PROXY_FAILURE_THRESHOLD", 3, 1, 10) SWITCH_PREFLIGHT_MAX_AGE_SECONDS = env_int("SWITCH_PREFLIGHT_MAX_AGE_SECONDS", 180, 0, 3600) OPENVPN_CMD = os.environ.get("OPENVPN_CMD", "openvpn") OPENVPN_AUTH_USER = os.environ.get("OPENVPN_AUTH_USER", "vpn") OPENVPN_AUTH_PASS = os.environ.get("OPENVPN_AUTH_PASS", "vpn") LOCAL_PROXY_HOST = os.environ.get("LOCAL_PROXY_HOST", "127.0.0.1") LOCAL_PROXY_PORT = env_int("LOCAL_PROXY_PORT", 7928, 1, 65535) UI_HOST = os.environ.get("UI_HOST", "::") UI_PORT = env_int("UI_PORT", 8787, 1, 65535) INVALID_BACKOFF_SECONDS = env_int("INVALID_BACKOFF_SECONDS", 30 * 60, 1) DEPLOYMENT_MODE = os.environ.get("DEPLOYMENT_MODE", "source").strip().lower() if DEPLOYMENT_MODE not in {"source", "docker"}: DEPLOYMENT_MODE = "source" DEPLOYMENT_MODE_LABEL = "Docker 容器" if DEPLOYMENT_MODE == "docker" else "Python 源码" UPDATE_COMMAND = ( "docker compose pull && docker compose up -d" if DEPLOYMENT_MODE == "docker" else "ml update" ) ROOT_DIR = Path(sys.executable).resolve().parent if globals().get("__compiled__") else Path(__file__).resolve().parent DEFAULT_APP_VERSION = "2.1.0" try: _version_text = (ROOT_DIR / "VERSION").read_text(encoding="utf-8").strip() except OSError: _version_text = DEFAULT_APP_VERSION APP_VERSION = _version_text if re.fullmatch(r"\d+\.\d+(?:\.\d+)?", _version_text) else DEFAULT_APP_VERSION APP_VERSION_SHORT = ".".join(APP_VERSION.split(".")[:2]) APP_VERSION_LABEL = f"V{APP_VERSION_SHORT} 正式版" GITHUB_REPOSITORY = "baoweise-bot/aimili-vpngate" GITHUB_REPOSITORY_URL = f"https://github.com/{GITHUB_REPOSITORY}" GITHUB_MAIN_BRANCH_URL = f"{GITHUB_REPOSITORY_URL}/tree/main" GITHUB_LATEST_RELEASE_API = f"https://api.github.com/repos/{GITHUB_REPOSITORY}/releases/latest" DATA_DIR = Path(os.environ["VPNGATE_DATA_DIR"]).resolve() if os.environ.get("VPNGATE_DATA_DIR") else ROOT_DIR / "vpngate_data" CONFIG_DIR = DATA_DIR / "configs" NODES_FILE = DATA_DIR / "nodes.json" STATE_FILE = DATA_DIR / "state.json" AUTH_FILE = DATA_DIR / "vpngate_auth.txt" UPSTREAM_PROXY_AUTH_FILE = DATA_DIR / "upstream_proxy_auth.txt" BLACKLIST_FILE = DATA_DIR / "blacklist.json" API_CACHE_FILE = DATA_DIR / "api_snapshot.csv" API_CACHE_META_FILE = DATA_DIR / "api_snapshot.meta.json" BUNDLED_SNAPSHOT_FILE = ROOT_DIR / "mirror" / "vpngate.csv" lock = threading.RLock() maintenance_lock = threading.Lock() connection_attempt_lock = threading.Lock() background_refill_lock = threading.Lock() background_refill_cancel_event = threading.Event() background_refill_thread: threading.Thread | None = None active_sessions: dict[str, float] = {} active_openvpn_process: subprocess.Popen[str] | None = None pending_openvpn_process: subprocess.Popen[str] | None = None active_connection_cancel_event: threading.Event | None = None connection_epoch = 0 active_openvpn_node_id = "" is_connecting = False last_active_ping_time = 0.0 last_active_latency = 0 consecutive_proxy_failures = 0 last_proxy_failure_node_id = "" last_collector_heartbeat = 0.0 last_checker_heartbeat = 0.0 last_pinger_heartbeat = 0.0 server_start_time = time.time() class ConnectionCancelled(RuntimeError): pass def ensure_dirs() -> None: DATA_DIR.mkdir(exist_ok=True, parents=True) CONFIG_DIR.mkdir(exist_ok=True, parents=True) if not AUTH_FILE.exists(): AUTH_FILE.write_text(f"{OPENVPN_AUTH_USER}\n{OPENVPN_AUTH_PASS}\n", encoding="utf-8") try: AUTH_FILE.chmod(0o600) except OSError: pass def upstream_proxy_auth_file() -> str | None: username, password = vpn_utils.get_upstream_proxy_auth() if username is None: return None try: DATA_DIR.mkdir(exist_ok=True, parents=True) UPSTREAM_PROXY_AUTH_FILE.write_text(f"{username}\n{password or ''}\n", encoding="utf-8") try: UPSTREAM_PROXY_AUTH_FILE.chmod(0o600) except OSError: pass return str(UPSTREAM_PROXY_AUTH_FILE) except Exception as exc: print(f"[上游代理认证] 写入认证文件失败: {exc}", flush=True) return None def write_json(path: Path, data: Any) -> None: with lock: tmp = path.with_suffix(path.suffix + ".tmp") tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") tmp.replace(path) def read_json(path: Path, default: Any) -> Any: with lock: try: return json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return default import hashlib import random def generate_random_password() -> str: import string chars = string.ascii_letters + string.digits while True: pwd = "".join(random.choices(chars, k=12)) # Ensure it contains at least one lowercase, one uppercase, and one digit has_lower = any(c.islower() for c in pwd) has_upper = any(c.isupper() for c in pwd) has_digit = any(c.isdigit() for c in pwd) if has_lower and has_upper and has_digit: return pwd def generate_random_username() -> str: import string chars = string.ascii_letters + string.digits while True: uname = "".join(random.choices(chars, k=12)) # Ensure it starts with a letter and contains at least one lowercase, one uppercase, and one digit if uname[0].isalpha(): has_lower = any(c.islower() for c in uname) has_upper = any(c.isupper() for c in uname) has_digit = any(c.isdigit() for c in uname) if has_lower and has_upper and has_digit: return uname def normalize_discovery_countries(value: Any) -> list[str]: if not isinstance(value, (list, tuple, set)): return [] normalized: list[str] = [] seen: set[str] = set() for item in value: code = str(item or "").strip().upper() if not re.fullmatch(r"[A-Z]{2}", code) or code in seen: continue normalized.append(code) seen.add(code) if len(normalized) >= 250: break return normalized def load_ui_config() -> dict[str, Any]: with lock: auth_file = DATA_DIR / "ui_auth.json" config = { "username": "", "secret_path": "EJsW2EeBo9lY", "password": "", "host": UI_HOST, "port": UI_PORT, "proxy_port": LOCAL_PROXY_PORT, "routing_mode": "auto", "force_country": "", "routing_ip_type": "all", "connection_enabled": True, "fixed_node_id": "", "favorite_node_ids": [], "fav_fail_fallback": False, "discovery_countries": [], } updated = False if auth_file.exists(): try: data = json.loads(auth_file.read_text(encoding="utf-8")) for key, val in data.items(): config[key] = val for key in ["host", "port", "proxy_port", "routing_mode", "force_country", "routing_ip_type", "connection_enabled", "fixed_node_id", "favorite_node_ids", "fav_fail_fallback", "discovery_countries"]: if key not in data: updated = True except Exception: pass if not config.get("username"): config["username"] = generate_random_username() updated = True if not config.get("password"): config["password"] = generate_random_password() updated = True normalized_port = bounded_int(config.get("port"), UI_PORT, 1, 65535) if normalized_port != config.get("port"): config["port"] = normalized_port updated = True normalized_proxy_port = bounded_int(config.get("proxy_port"), LOCAL_PROXY_PORT, 1024, 65535) if normalized_proxy_port == normalized_port: fallback_proxy_port = LOCAL_PROXY_PORT if LOCAL_PROXY_PORT != normalized_port else 7928 if fallback_proxy_port == normalized_port: fallback_proxy_port = 7929 normalized_proxy_port = fallback_proxy_port if normalized_proxy_port != config.get("proxy_port"): config["proxy_port"] = normalized_proxy_port updated = True normalized_discovery_countries = normalize_discovery_countries(config.get("discovery_countries")) if normalized_discovery_countries != config.get("discovery_countries"): config["discovery_countries"] = normalized_discovery_countries updated = True if not auth_file.exists() or updated: try: DATA_DIR.mkdir(exist_ok=True, parents=True) write_json(auth_file, config) except Exception: pass return config def persist_discovery_countries(value: Any) -> list[str]: if not isinstance(value, list): raise ValueError("国家筛选范围必须是国家代码列表") countries = normalize_discovery_countries(value) ui_cfg = load_ui_config() ui_cfg["discovery_countries"] = countries auth_file = DATA_DIR / "ui_auth.json" with lock: DATA_DIR.mkdir(exist_ok=True, parents=True) write_json(auth_file, ui_cfg) return countries # 初始化时优先从 ui_auth.json 加载保存的代理出站端口和网页端口配置以覆盖环境变量 try: _init_cfg = load_ui_config() if "proxy_port" in _init_cfg: LOCAL_PROXY_PORT = bounded_int(_init_cfg["proxy_port"], LOCAL_PROXY_PORT, 1024, 65535) if "port" in _init_cfg: UI_PORT = bounded_int(_init_cfg["port"], UI_PORT, 1, 65535) if "host" in _init_cfg: UI_HOST = _init_cfg["host"] except Exception: pass def get_session_token(password: str, username: str = "admin") -> str: salt = "aimilivpn_secure_salt_2026" return hashlib.sha256((username + ":" + password + salt).encode("utf-8")).hexdigest() _last_cleanup_time = 0.0 def cleanup_old_logs(logs_dir: Path) -> None: global _last_cleanup_time now = time.time() with lock: if now - _last_cleanup_time < 3600: return _last_cleanup_time = now try: three_days_sec = 3 * 24 * 60 * 60 for path in logs_dir.glob("*.json"): match = re.match(r"^(\d{4}-\d{2}-\d{2})\.json$", path.name) if match: date_str = match.group(1) try: file_time = time.mktime(time.strptime(date_str, "%Y-%m-%d")) today_str = time.strftime("%Y-%m-%d", time.localtime()) today_time = time.mktime(time.strptime(today_str, "%Y-%m-%d")) if today_time - file_time >= three_days_sec: with lock: path.unlink() print(f"[清理] 已删除3天前的旧日志文件: {path.name}", flush=True) except Exception: if now - path.stat().st_mtime > three_days_sec: with lock: path.unlink() except Exception as e: print(f"[清理错误] 清理旧日志失败: {e}", flush=True) def log_to_json(level: str, module: str, message: str) -> None: try: logs_dir = DATA_DIR / "logs" logs_dir.mkdir(exist_ok=True, parents=True) date_str = time.strftime("%Y-%m-%d", time.localtime()) log_file = logs_dir / f"{date_str}.json" entry = { "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), "level": level, "module": module, "message": message } with lock: with open(log_file, "a", encoding="utf-8") as f: f.write(json.dumps(entry, ensure_ascii=False) + "\n") cleanup_old_logs(logs_dir) except Exception as e: print(f"[Log Error] Failed to write JSON log: {e}", flush=True) def set_state(**updates: Any) -> None: # Keep the read-modify-write transaction atomic across background threads. with lock: state = get_state() state.update(updates) write_json(STATE_FILE, state) def read_nodes() -> list[dict[str, Any]]: raw = read_json(NODES_FILE, []) if not isinstance(raw, list): return [] return [item for item in raw if isinstance(item, dict)] def get_state() -> dict[str, Any]: global active_openvpn_node_id, is_connecting state = read_json(STATE_FILE, {}) state.pop("password", None) state["active_openvpn_node_id"] = active_openvpn_node_id state["is_connecting"] = is_connecting state["maintenance_running"] = maintenance_lock.locked() state.setdefault("api_url", API_URL) state.setdefault("mirror_url", MIRROR_HTTPS_URL) state.setdefault("last_fetch_source", "") state.setdefault("target_valid_nodes", TARGET_VALID_NODES) state.setdefault("fetch_interval_seconds", FETCH_INTERVAL_SECONDS) state.setdefault("check_interval_seconds", CHECK_INTERVAL_SECONDS) _proxy_display = f"[{LOCAL_PROXY_HOST}]" if ":" in LOCAL_PROXY_HOST else LOCAL_PROXY_HOST state["local_proxy"] = f"http://{_proxy_display}:{LOCAL_PROXY_PORT}" state.setdefault("last_fetch_status", "not_started") state.setdefault("last_check_message", "") state.setdefault("pending_node_id", "") state.setdefault("blacklisted_nodes", 0) state["app_version"] = APP_VERSION state["app_version_label"] = APP_VERSION_LABEL state["deployment_mode"] = DEPLOYMENT_MODE state["deployment_mode_label"] = DEPLOYMENT_MODE_LABEL # Pre-populate settings inputs in UI ui_cfg = load_ui_config() state["username"] = ui_cfg.get("username", "admin") state["port"] = ui_cfg.get("port", 8787) state["secret_path"] = ui_cfg.get("secret_path", "EJsW2EeBo9lY") state["password_set"] = bool(ui_cfg.get("password")) state["proxy_port"] = ui_cfg.get("proxy_port", 7928) state["routing_mode"] = ui_cfg.get("routing_mode", "auto") state["force_country"] = ui_cfg.get("force_country", "") state["routing_ip_type"] = ui_cfg.get("routing_ip_type", "all") state["connection_enabled"] = ui_cfg.get("connection_enabled", True) state["fixed_node_id"] = ui_cfg.get("fixed_node_id", "") state["favorite_node_ids"] = ui_cfg.get("favorite_node_ids", []) state["discovery_countries"] = normalize_discovery_countries(ui_cfg.get("discovery_countries")) state["fav_fail_fallback"] = False return state def safe_name(value: str) -> str: value = re.sub(r"[^A-Za-z0-9_.-]+", "_", value.strip()) return value.strip("._") or "node" def clear_active_connection_state(message: str) -> None: stop_active_openvpn() with lock: nodes = read_nodes() for item in nodes: item["active"] = False write_json(NODES_FILE, nodes) set_state( active_openvpn_node_id="", is_connecting=False, pending_node_id="", active_node_latency="无活动连接", proxy_ok=False, proxy_ip="-", proxy_latency_ms=0, proxy_error=message, last_check_message=message, ) def parse_int(value: Any) -> int: try: return int(value) except (TypeError, ValueError): return 0 def proxy_basic_auth_header(username: str, password: str) -> str: token = base64.b64encode(f"{username}:{password}".encode("utf-8")).decode("ascii") return f"Proxy-Authorization: Basic {token}\r\n" def recv_exact_from_socket(sock: socket.socket, size: int) -> bytes: data = b"" while len(data) < size: chunk = sock.recv(size - len(data)) if not chunk: raise RuntimeError("Unexpected EOF while reading proxy response") data += chunk return data def read_http_response_head(sock: socket.socket, limit: int = 65536) -> bytes: data = b"" while b"\r\n\r\n" not in data: chunk = sock.recv(4096) if not chunk: break data += chunk if len(data) > limit: raise RuntimeError("Proxy response header too large") if b"\r\n\r\n" not in data: raise RuntimeError("Incomplete HTTP proxy response header") return data def socks5_address_bytes(host: str) -> tuple[int, bytes]: try: return 1, socket.inet_aton(host) except OSError: pass try: return 4, socket.inet_pton(socket.AF_INET6, host) except OSError: pass host_bytes = host.encode("idna") if len(host_bytes) > 255: raise RuntimeError("SOCKS5 target host name is too long") return 3, bytes([len(host_bytes)]) + host_bytes def read_socks5_connect_reply(sock: socket.socket) -> None: header = recv_exact_from_socket(sock, 4) if header[0] != 5: raise RuntimeError("Invalid SOCKS5 reply version") atyp = header[3] if atyp == 1: recv_exact_from_socket(sock, 4) elif atyp == 3: domain_len = recv_exact_from_socket(sock, 1)[0] recv_exact_from_socket(sock, domain_len) elif atyp == 4: recv_exact_from_socket(sock, 16) else: raise RuntimeError(f"Invalid SOCKS5 reply address type: {atyp}") recv_exact_from_socket(sock, 2) if header[1] != 0: raise RuntimeError(f"SOCKS5 connection request rejected, code={header[1]}") def format_host_port(host: str, port: int) -> str: return f"[{host}]:{port}" if ":" in host and not host.startswith("[") else f"{host}:{port}" def fetch_api_text_via_proxy(url: str, ptype: str, phost: str, pport: int, use_ssl_verify: bool = True) -> str: import socket import ssl import urllib.parse parsed = urllib.parse.urlsplit(url) domain = parsed.hostname or "www.vpngate.net" port = parsed.port or (443 if parsed.scheme == "https" else 80) is_https = parsed.scheme == "https" path = parsed.path or "/" if parsed.query: path += "?" + parsed.query is_ipv6 = ":" in phost af = socket.AF_INET6 if is_ipv6 else socket.AF_INET s = None try: s = socket.socket(af, socket.SOCK_STREAM) s.settimeout(API_FETCH_TIMEOUT_SECONDS) s.connect((phost, pport)) proxy_user, proxy_pass = vpn_utils.get_upstream_proxy_auth() if ptype == "socks": # SOCKS5 Handshake if proxy_user is not None: s.sendall(b"\x05\x02\x00\x02") else: s.sendall(b"\x05\x01\x00") resp = recv_exact_from_socket(s, 2) if len(resp) < 2 or resp[0] != 5: raise RuntimeError("SOCKS5 authentication failed or unsupported") if resp[1] == 2: if proxy_user is None: raise RuntimeError("SOCKS5 proxy requires username/password authentication") user_bytes = proxy_user.encode("utf-8") pass_bytes = (proxy_pass or "").encode("utf-8") if len(user_bytes) > 255 or len(pass_bytes) > 255: raise RuntimeError("SOCKS5 proxy credentials are too long") s.sendall(b"\x01" + bytes([len(user_bytes)]) + user_bytes + bytes([len(pass_bytes)]) + pass_bytes) auth_resp = recv_exact_from_socket(s, 2) if len(auth_resp) < 2 or auth_resp[1] != 0: raise RuntimeError("SOCKS5 username/password authentication failed") elif resp[1] != 0: raise RuntimeError("SOCKS5 authentication method unsupported") # SOCKS5 Connect atyp, addr_bytes = socks5_address_bytes(domain) req = b"\x05\x01\x00" + bytes([atyp]) + addr_bytes + port.to_bytes(2, 'big') s.sendall(req) read_socks5_connect_reply(s) # If HTTPS, wrap socket with SSL if is_https: ctx = ssl.create_default_context() if use_ssl_verify else ssl._create_unverified_context() s = ctx.wrap_socket(s, server_hostname=domain) else: # http proxy if is_https: # HTTP CONNECT tunnel authority = format_host_port(domain, port) auth_header = proxy_basic_auth_header(proxy_user, proxy_pass or "") if proxy_user is not None else "" req_str = f"CONNECT {authority} HTTP/1.1\r\nHost: {authority}\r\nUser-Agent: Mozilla/5.0 vpngate-openvpn-manager/2.0\r\n{auth_header}Proxy-Connection: Keep-Alive\r\n\r\n" s.sendall(req_str.encode('ascii')) resp = read_http_response_head(s) status_line = resp.split(b"\r\n", 1)[0].decode("utf-8", errors="replace") status_parts = status_line.split() status_code = int(status_parts[1]) if len(status_parts) >= 2 and status_parts[1].isdigit() else 0 if status_code != 200: raise RuntimeError(f"HTTP CONNECT tunnel failed: {status_line}") # Wrap socket with SSL ctx = ssl.create_default_context() if use_ssl_verify else ssl._create_unverified_context() s = ctx.wrap_socket(s, server_hostname=domain) else: # Direct HTTP request through proxy: request URI must be absolute pass # Send HTTP GET request if ptype == "http" and not is_https: request_uri = url else: request_uri = path req_headers = ( f"GET {request_uri} HTTP/1.1\r\n" f"Host: {domain}\r\n" f"User-Agent: Mozilla/5.0 vpngate-openvpn-manager/2.0\r\n" f"Accept: text/plain,*/*\r\n" f"{proxy_basic_auth_header(proxy_user, proxy_pass or '') if ptype == 'http' and not is_https and proxy_user is not None else ''}" f"Connection: close\r\n\r\n" ) s.sendall(req_headers.encode('utf-8')) # Read response response_data = b"" while True: chunk = s.recv(4096) if not chunk: break response_data += chunk if len(response_data) > snapshot_utils.MAX_SNAPSHOT_BYTES + 65536: raise RuntimeError("API response exceeds the maximum allowed size") finally: if s is not None: try: s.close() except Exception: pass # Parse HTTP response header_end = response_data.find(b"\r\n\r\n") if header_end == -1: raise RuntimeError("Invalid HTTP response format") headers_part = response_data[:header_end].decode('utf-8', errors='replace') body_part = response_data[header_end+4:] # Check for HTTP status code lines = headers_part.splitlines() if not lines: raise RuntimeError("Empty response headers") status_line = lines[0] status_parts = status_line.split() if len(status_parts) >= 2: try: status_code = int(status_parts[1]) if status_code != 200: raise RuntimeError(f"HTTP Server returned status {status_code}: {status_line}") except ValueError: pass # Handle chunked transfer encoding is_chunked = False for line in lines[1:]: if ":" in line: k, v = line.split(":", 1) if k.strip().lower() == "transfer-encoding" and "chunked" in v.lower(): is_chunked = True break if is_chunked: decoded = b"" idx = 0 while idx < len(body_part): c_end = body_part.find(b"\r\n", idx) if c_end == -1: break chunk_size_str = body_part[idx:c_end].split(b";")[0].strip() try: chunk_size = int(chunk_size_str, 16) except ValueError: break if chunk_size == 0: break idx = c_end + 2 decoded += body_part[idx : idx + chunk_size] idx += chunk_size + 2 body_part = decoded return body_part.decode('utf-8', errors='replace') def fetch_api_text(url: str | None = None, use_ssl_verify: bool = True) -> str: if url is None: url = API_URL ptype, phost, pport = vpn_utils.get_upstream_proxy() if ptype and phost and pport: try: print(f"[fetch_api_text] 监测到上游代理 ({ptype}://{phost}:{pport}),尝试通过代理获取 API...", flush=True) return fetch_api_text_via_proxy(url, ptype, phost, pport, use_ssl_verify) except Exception as e: print(f"[fetch_api_text] 通过代理获取 API 失败: {e},尝试使用直连/默认系统代理...", flush=True) log_to_json("WARNING", "Main", f"使用代理 {ptype}://{phost}:{pport} 获取 API 失败: {e}") request = urllib.request.Request( url, headers={ "User-Agent": f"Mozilla/5.0 AimiliVPN/{APP_VERSION}", "Accept": "text/plain,*/*", }, ) def read_limited(response: Any) -> bytes: chunks: list[bytes] = [] total = 0 while True: chunk = response.read(65536) if not chunk: break total += len(chunk) if total > snapshot_utils.MAX_SNAPSHOT_BYTES: raise RuntimeError("API response exceeds the maximum allowed size") chunks.append(chunk) return b"".join(chunks) if url.startswith("https://") and not use_ssl_verify: import ssl ctx = ssl._create_unverified_context() with urllib.request.urlopen(request, timeout=API_FETCH_TIMEOUT_SECONDS, context=ctx) as response: return read_limited(response).decode("utf-8", errors="replace") else: with urllib.request.urlopen(request, timeout=API_FETCH_TIMEOUT_SECONDS) as response: return read_limited(response).decode("utf-8", errors="replace") def parse_release_version(value: Any) -> tuple[int, int, int]: match = re.search(r"(?i)(?:^|[^a-z0-9])v?(\d+)(?:\.(\d+))?(?:\.(\d+))?", str(value or "").strip()) if not match: raise ValueError("GitHub Release 版本号格式无效") return tuple(int(part or 0) for part in match.groups()) def check_latest_release() -> dict[str, Any]: payload = json.loads(fetch_api_text(GITHUB_LATEST_RELEASE_API, True)) if not isinstance(payload, dict): raise ValueError("GitHub Release API 返回格式无效") if payload.get("draft") or payload.get("prerelease"): raise ValueError("GitHub 最新版本不是正式版") latest_tag = str(payload.get("tag_name") or "").strip() latest_version = parse_release_version(latest_tag) current_version = parse_release_version(APP_VERSION) release_url = f"{GITHUB_REPOSITORY_URL}/releases/tag/{urllib.parse.quote(latest_tag, safe='')}" return { "ok": True, "current_version": APP_VERSION, "current_version_label": APP_VERSION_LABEL, "latest_version": ".".join(str(part) for part in latest_version), "latest_tag": latest_tag, "latest_name": str(payload.get("name") or latest_tag), "published_at": str(payload.get("published_at") or ""), "update_available": latest_version > current_version, "release_url": release_url, "main_branch_url": GITHUB_MAIN_BRANCH_URL, "deployment_mode": DEPLOYMENT_MODE, "deployment_mode_label": DEPLOYMENT_MODE_LABEL, "update_command": UPDATE_COMMAND, } def is_certificate_verification_error(exc: BaseException) -> bool: import ssl current: BaseException | None = exc seen: set[int] = set() while current is not None and id(current) not in seen: seen.add(id(current)) if isinstance(current, ssl.SSLCertVerificationError): return True reason = getattr(current, "reason", None) cause = getattr(current, "__cause__", None) current = reason if isinstance(reason, BaseException) else cause return False def parse_vpngate_rows(text: str) -> list[dict[str, str]]: return snapshot_utils.parse_and_validate_snapshot(text, max_rows=MAX_SCAN_ROWS) def decode_config(encoded: str) -> str: return snapshot_utils.decode_config(encoded) def load_blacklist() -> dict[str, dict[str, Any]]: now = time.time() raw = read_json(BLACKLIST_FILE, {}) if not isinstance(raw, dict): return {} cleaned: dict[str, dict[str, Any]] = {} changed = False for key, entry in raw.items(): if not isinstance(entry, dict): changed = True continue until = float(entry.get("until", 0) or 0) if until and until > now: cleaned[str(key)] = entry else: changed = True if changed: write_json(BLACKLIST_FILE, cleaned) return cleaned def mark_blacklisted(node: dict[str, Any], message: str) -> None: node_id = str(node.get("id") or "").strip() if not node_id: return blacklist = load_blacklist() now = time.time() blacklist[node_id] = { "id": node_id, "ip": node.get("ip") or node.get("remote_host") or "", "country": node.get("country", ""), "reason": message, "marked_at": now, "until": now + INVALID_BACKOFF_SECONDS, } write_json(BLACKLIST_FILE, blacklist) def row_to_node(row: dict[str, str], config_text: str) -> dict[str, Any]: ip = row.get("IP", "") country_short = row.get("CountryShort", "") remote_host, remote_port, proto = vpn_utils.parse_remote(config_text, ip) node_id = safe_name("_".join([country_short or "XX", ip or remote_host, str(remote_port), proto])) config_path = CONFIG_DIR / f"{node_id}.ovpn" country_long = row.get("CountryLong", "") country_zh = vpn_utils.COUNTRY_TRANSLATIONS.get(country_long, vpn_utils.COUNTRY_TRANSLATIONS.get(country_long.strip(), country_long)) return { "id": node_id, "country": country_zh, "country_short": country_short, "host_name": row.get("HostName", ""), "ip": ip, "score": parse_int(row.get("Score")), "ping": parse_int(row.get("Ping")), "speed": parse_int(row.get("Speed")), "sessions": parse_int(row.get("NumVpnSessions")), "owner": "", "asn": "", "as_name": "", "location": "", "ip_type": "", "quality": "", "latency_ms": 0, "config_file": str(config_path), "config_text": config_text, "proto": proto, "remote_host": remote_host, "remote_port": remote_port, "fetched_at": time.time(), "probe_status": "not_checked", "probe_message": "", "probed_at": 0, } def api_network_sources() -> list[tuple[str, str]]: configured = [ ("official_https", API_HTTPS_URL), ("official_http", API_HTTP_URL), ("github_pages_https", MIRROR_HTTPS_URL), ("github_pages_http", MIRROR_HTTP_URL), ] sources: list[tuple[str, str]] = [] seen: set[str] = set() for label, url in configured: normalized = str(url or "").strip() if not normalized or normalized in seen: continue if not normalized.startswith(("https://", "http://")): print(f"[配置警告] 忽略不支持的节点源 URL: {normalized}", flush=True) continue seen.add(normalized) sources.append((label, normalized)) return sources def read_snapshot_file(path: Path) -> str: size = path.stat().st_size if size <= 0 or size > snapshot_utils.MAX_SNAPSHOT_BYTES: raise ValueError(f"本地快照大小无效: {size}") return path.read_bytes().decode("utf-8", errors="strict") def cache_api_snapshot(text: str, source: str) -> None: validated_rows = snapshot_utils.parse_and_validate_snapshot(text, max_rows=MAX_SCAN_ROWS) encoded = text.encode("utf-8") with lock: DATA_DIR.mkdir(exist_ok=True, parents=True) tmp = API_CACHE_FILE.with_suffix(API_CACHE_FILE.suffix + ".tmp") tmp.write_bytes(encoded) tmp.replace(API_CACHE_FILE) write_json( API_CACHE_META_FILE, { "source": source, "cached_at": time.time(), "row_count": len(validated_rows), "byte_count": len(encoded), "sha256": hashlib.sha256(encoded).hexdigest(), }, ) def rows_to_candidates( rows: list[dict[str, str]], blacklist: dict[str, dict[str, Any]], ) -> list[dict[str, Any]]: candidates: list[dict[str, Any]] = [] seen_ips: set[str] = set() for row in rows[:MAX_SCAN_ROWS]: ip = row.get("IP", "") if not ip or ip in seen_ips: continue try: config_text = decode_config(row.get("OpenVPN_ConfigData_Base64", "")) snapshot_utils.validate_openvpn_config(config_text) node = row_to_node(row, config_text) except Exception as row_exc: print(f"[fetch_candidates] 跳过损坏或不安全的节点配置记录: {row_exc}", flush=True) log_to_json("WARNING", "Main", f"跳过损坏或不安全的节点配置记录: {row_exc}") continue entry = blacklist.get(node["id"]) if entry and float(entry.get("until", 0) or 0) > time.time(): continue candidates.append(node) seen_ips.add(ip) return candidates def filter_candidates_by_discovery_countries( candidates: list[dict[str, Any]], country_codes: Any, ) -> list[dict[str, Any]]: selected = set(normalize_discovery_countries(country_codes)) if not selected: return candidates return [ candidate for candidate in candidates if str(candidate.get("country_short") or "").strip().upper() in selected ] def fetch_candidates() -> list[dict[str, Any]]: blacklist = load_blacklist() discovery_countries = normalize_discovery_countries( load_ui_config().get("discovery_countries") ) last_err: Exception | None = None log_to_json("INFO", "Main", "开始按官方、GitHub Pages、本地缓存顺序拉取节点列表...") for source_name, url in api_network_sources(): try: msg = f"尝试节点源 {source_name}: {url}" print(f"[fetch_candidates] {msg}", flush=True) log_to_json("INFO", "Main", msg) api_text = fetch_api_text(url, True) rows = parse_vpngate_rows(api_text) candidates = rows_to_candidates(rows, blacklist) if not candidates: raise ValueError("节点源通过格式校验,但没有未被屏蔽的候选节点") # Plain HTTP remains available for older machines, but never replaces # the last snapshot obtained through an authenticated HTTPS channel. if url.startswith("https://"): cache_api_snapshot(api_text, source_name) filtered_candidates = filter_candidates_by_discovery_countries( candidates, discovery_countries, ) scope_message = ( f"按国家范围 {', '.join(discovery_countries)} 筛选后保留 {len(filtered_candidates)} 个" if discovery_countries else f"保留全部 {len(filtered_candidates)} 个" ) set_state( last_fetch_at=time.time(), last_fetch_status="ok", last_fetch_source=source_name, last_fetch_message=( f"从 {source_name} 成功获取 {len(candidates)} 个候选节点,{scope_message}。" ), blacklisted_nodes=len(blacklist), ) log_to_json( "INFO", "Main", f"节点源 {source_name} 获取成功,共 {len(candidates)} 个候选节点,{scope_message}", ) return filtered_candidates except Exception as e: last_err = e print(f"[fetch_candidates] 节点源 {source_name} 失败: {e}", flush=True) log_to_json("WARNING", "Main", f"节点源 {source_name} 失败: {e}") local_sources = [("local_cache", API_CACHE_FILE)] if BUNDLED_SNAPSHOT_FILE != API_CACHE_FILE: local_sources.append(("bundled_initial", BUNDLED_SNAPSHOT_FILE)) for source_name, path in local_sources: try: if not path.exists(): continue api_text = read_snapshot_file(path) rows = parse_vpngate_rows(api_text) candidates = rows_to_candidates(rows, blacklist) if not candidates: raise ValueError("本地快照没有未被屏蔽的候选节点") if source_name == "bundled_initial" and not API_CACHE_FILE.exists(): cache_api_snapshot(api_text, source_name) filtered_candidates = filter_candidates_by_discovery_countries( candidates, discovery_countries, ) scope_message = ( f"按国家范围 {', '.join(discovery_countries)} 筛选后保留 {len(filtered_candidates)} 个" if discovery_countries else f"保留全部 {len(filtered_candidates)} 个" ) set_state( last_fetch_at=time.time(), last_fetch_status="cached", last_fetch_source=source_name, last_fetch_message=( f"网络节点源不可用,已载入 {source_name} 的 {len(candidates)} 个候选节点," f"{scope_message}。" ), blacklisted_nodes=len(blacklist), ) log_to_json( "WARNING", "Main", f"网络节点源不可用,使用 {source_name},共 {len(candidates)} 个候选节点,{scope_message}", ) return filtered_candidates except Exception as e: last_err = e print(f"[fetch_candidates] 本地节点源 {source_name} 失败: {e}", flush=True) log_to_json("WARNING", "Main", f"本地节点源 {source_name} 失败: {e}") err_code, diag_msg = vpn_utils.diagnose_api_failure(API_URL) full_err_msg = f"所有节点源和本地缓存均失败: {last_err} | 诊断结果: {diag_msg}" print(f"[错误代码 {err_code}] {full_err_msg}", flush=True) log_to_json("ERROR", "Main", f"[错误代码 {err_code}] {full_err_msg}") set_state( last_fetch_status="error", last_fetch_error_code=err_code, last_fetch_source="", last_fetch_message=diag_msg, ) if last_err: raise RuntimeError(diag_msg) from last_err raise RuntimeError(diag_msg) def cached_nodes() -> list[dict[str, Any]]: return read_nodes() _openvpn_version = None def split_openvpn_command() -> list[str]: try: return shlex.split(OPENVPN_CMD, posix=(os.name != "nt")) or ["openvpn"] except ValueError as exc: raise RuntimeError(f"OPENVPN_CMD 配置无法解析: {exc}") from exc def get_openvpn_version() -> float: global _openvpn_version if _openvpn_version is not None: return _openvpn_version try: cmd = split_openvpn_command() res = subprocess.run(cmd + ["--version"], capture_output=True, text=True, timeout=2) match = re.search(r"OpenVPN\s+(\d+\.\d+)", res.stdout or res.stderr) if match: _openvpn_version = float(match.group(1)) return _openvpn_version except Exception: pass _openvpn_version = 2.4 return _openvpn_version def openvpn_command(config_file: str, route_nopull: bool, dev: str = "tun0") -> list[str]: command = split_openvpn_command() command.extend( [ "--config", config_file, "--dev", dev, "--dev-type", "tun", "--pull-filter", "ignore", "route-ipv6", "--pull-filter", "ignore", "ifconfig-ipv6", "--route-delay", "2", "--connect-retry-max", "1", "--connect-timeout", "15", "--auth-user-pass", str(AUTH_FILE), "--auth-nocache", ] ) version = get_openvpn_version() if version >= 2.5: command.extend(["--data-ciphers", "AES-128-CBC:AES-256-GCM:AES-128-GCM:CHACHA20-POLY1305"]) else: command.extend(["--ncp-ciphers", "AES-128-CBC:AES-256-GCM:AES-128-GCM:CHACHA20-POLY1305"]) command.extend(["--verb", "3"]) if os.path.exists("/etc/ssl/certs"): command.extend(["--capath", "/etc/ssl/certs"]) try: content = Path(config_file).read_text(encoding="utf-8", errors="replace") if vpn_utils.is_config_tcp(content): ptype, host, port = vpn_utils.get_upstream_proxy() auth_file = upstream_proxy_auth_file() if ptype == "socks" and host and port: command.extend(["--socks-proxy", host, str(port)]) if auth_file: command.append(auth_file) elif ptype == "http" and host and port: command.extend(["--http-proxy", host, str(port)]) if auth_file: command.append(auth_file) except Exception: pass if route_nopull: command.append("--route-nopull") return command def stop_process(process: subprocess.Popen[str] | None) -> None: if process is None or process.poll() is not None: return try: process.terminate() except OSError: return try: process.wait(timeout=8) except subprocess.TimeoutExpired: try: process.kill() except OSError: pass def begin_connection_attempt() -> tuple[int, threading.Event]: global connection_epoch, active_connection_cancel_event, is_connecting if not connection_attempt_lock.acquire(blocking=False): raise RuntimeError("当前已有连接切换任务正在运行,请稍后再试") cancel_event = threading.Event() with lock: if is_connecting: connection_attempt_lock.release() raise RuntimeError("当前已有连接或节点检测任务正在运行,请稍后再试") connection_epoch += 1 token = connection_epoch active_connection_cancel_event = cancel_event is_connecting = True return token, cancel_event def connection_attempt_is_current(token: int, cancel_event: threading.Event) -> bool: with lock: return token == connection_epoch and not cancel_event.is_set() def finish_connection_attempt(token: int, cancel_event: threading.Event) -> None: global active_connection_cancel_event, is_connecting with lock: if active_connection_cancel_event is cancel_event: active_connection_cancel_event = None if token == connection_epoch: is_connecting = False connection_attempt_lock.release() def cancel_pending_connection_attempt() -> None: global connection_epoch, pending_openvpn_process, is_connecting pending = None with lock: if active_connection_cancel_event is None: return connection_epoch += 1 active_connection_cancel_event.set() pending = pending_openvpn_process pending_openvpn_process = None is_connecting = False stop_process(pending) def kill_existing_openvpn_processes() -> None: if not sys.platform.startswith("linux"): return try: own_markers = [ str(DATA_DIR), str(CONFIG_DIR), str(AUTH_FILE), str(UPSTREAM_PROXY_AUTH_FILE), ] killed_pids: list[int] = [] proc_root = Path("/proc") if not proc_root.exists(): return for proc_dir in proc_root.iterdir(): if not proc_dir.name.isdigit(): continue pid = int(proc_dir.name) if pid == os.getpid(): continue try: raw = (proc_dir / "cmdline").read_bytes() except OSError: continue if not raw: continue args = [part.decode("utf-8", errors="replace") for part in raw.split(b"\0") if part] if not args: continue cmdline = " ".join(args) executable = Path(args[0]).name.lower() if "openvpn" not in executable and "openvpn" not in cmdline.lower(): continue if any(marker and marker in cmdline for marker in own_markers): try: os.kill(pid, signal.SIGTERM) killed_pids.append(pid) except ProcessLookupError: pass except PermissionError: print(f"[Cleanup] No permission to terminate OpenVPN PID {pid}", flush=True) if killed_pids: time.sleep(0.5) for pid in killed_pids: try: raw = (proc_root / str(pid) / "cmdline").read_bytes() cmdline = " ".join(part.decode("utf-8", errors="replace") for part in raw.split(b"\0") if part) if any(marker and marker in cmdline for marker in own_markers): os.kill(pid, signal.SIGKILL) except ProcessLookupError: pass except (OSError, PermissionError): pass print(f"[Cleanup] Terminated AimiliVPN OpenVPN processes: {killed_pids}", flush=True) except Exception as e: print(f"[Cleanup Error] Failed to kill existing OpenVPN processes: {e}", flush=True) def update_handshake_status(line_lower: str) -> None: status_map = { "resolving": ("解析域名", "正在解析服务器域名与 IP 地址..."), "udp link local": ("物理连接", "已创建本地套接字,开始尝试发送数据包..."), "tcp link local": ("物理连接", "已创建本地套接字,开始尝试发送数据包..."), "tls: initial packet": ("证书握手", "已成功发送首包,正在与远程服务器建立 TLS 安全通道..."), "verify ok": ("证书校验", "服务器证书校验成功,正在进行身份验证..."), "peer connection initiated": ("协商加密", "控制通道已建立,已初始化与服务器的加密对等连接..."), "push_request": ("请求配置", "正在向服务器发送 PUSH_REQUEST 请求配置参数与 IP 分配..."), "push_reply": ("应用配置", "已接收服务器 PUSH_REPLY,获取到 IP 分配,正在准备配置网卡..."), "tun/tap device": ("创建网卡", "正在创建虚拟通道并打开 TUN 虚拟网卡设备..."), "do_ifconfig": ("网卡配置", "正在为虚拟网卡配置 IP 地址及相关网络属性..."), } for key, (short_status, detailed_desc) in status_map.items(): if key in line_lower: set_state(active_node_latency=short_status, last_check_message=detailed_desc) break def run_openvpn_until_ready( config_file: str, keep_alive: bool, route_nopull: bool, timeout: int | None = None, dev: str = "tun0", cancel_event: threading.Event | None = None, track_pending: bool = False, ) -> tuple[bool, str, subprocess.Popen[str] | None]: global pending_openvpn_process limit = timeout if timeout is not None else OPENVPN_TEST_TIMEOUT_SECONDS try: process = subprocess.Popen( openvpn_command(config_file, route_nopull, dev), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, encoding="utf-8", errors="replace", cwd=str(ROOT_DIR), ) except FileNotFoundError: return False, "[错误代码 2001] [ERR_OVPN_CMD_NOT_FOUND] 未找到 openvpn 命令。原因: 系统未安装 openvpn,或 PATH 环境变量不正确。", None except OSError as exc: return False, f"[错误代码 2002] [ERR_OVPN_START_FAILED] openvpn 启动失败: {exc}。原因: 系统权限不足或配置冲突。", None if track_pending: with lock: if cancel_event is not None and cancel_event.is_set(): stop_process(process) return False, "连接操作已取消", None pending_openvpn_process = process lines: queue.Queue[str | None] = queue.Queue() startup_done = [False] openvpn_logs: list[str] = [] def reader() -> None: assert process.stdout is not None for line in process.stdout: line_str = line.rstrip() if not startup_done[0]: openvpn_logs.append(line_str) lines.put(line_str) else: if keep_alive: print(f"[OpenVPN] {line_str}", flush=True) level = "INFO" line_lower = line_str.lower() if "error" in line_lower or "failed" in line_lower or "cannot" in line_lower or "fatal" in line_lower or "permission denied" in line_lower: level = "ERROR" elif "warning" in line_lower or "warn" in line_lower or "deprecated" in line_lower: level = "WARNING" log_to_json(level, "VPN", f"[OpenVPN] {line_str}") if not startup_done[0]: lines.put(None) threading.Thread(target=reader, daemon=True).start() started = time.time() tail: list[str] = [] ok = False cancelled = False message = "OpenVPN did not complete initialization." while time.time() - started < limit: if cancel_event is not None and cancel_event.is_set(): cancelled = True message = "连接操作已取消" break try: line = lines.get(timeout=0.5) except queue.Empty: if process.poll() is not None: break continue if line is None: break if line: tail.append(line) tail = tail[-50:] if keep_alive: print(f"[OpenVPN] {line}", flush=True) lower = line.lower() if keep_alive: update_handshake_status(lower) if "initialization sequence completed" in lower: if cancel_event is not None and cancel_event.is_set(): cancelled = True message = "连接操作已取消" else: ok = True message = f"OpenVPN connected in {int((time.time() - started) * 1000)} ms." break if "auth_failed" in lower or "authentication failed" in lower: message = "AUTH_FAILED" break if "cannot ioctl" in lower or "fatal error" in lower: message = line[-220:] break else: message = f"OpenVPN timeout after {limit}s." # Bulk write accumulated startup logs for line_str in openvpn_logs: level = "INFO" line_lower = line_str.lower() if "error" in line_lower or "failed" in line_lower or "cannot" in line_lower or "fatal" in line_lower or "permission denied" in line_lower: level = "ERROR" elif "warning" in line_lower or "warn" in line_lower or "deprecated" in line_lower: level = "WARNING" log_to_json(level, "VPN", f"[OpenVPN] {line_str}") if not ok and not cancelled: err_code, diag_msg = vpn_utils.diagnose_openvpn_failure(tail) message = f"[错误代码 {err_code}] {diag_msg} (原始日志尾部: {tail[-1][-100:] if tail else '无'})" startup_done[0] = True if not keep_alive or not ok: stop_process(process) process = None if track_pending: with lock: if pending_openvpn_process is process or pending_openvpn_process is not None and pending_openvpn_process.poll() is not None: pending_openvpn_process = None return ok, message, process def setup_policy_routing(interface: str = "tun0") -> bool: try: subprocess.run(["ip", "rule", "del", "table", "100"], capture_output=True, timeout=2) except Exception: pass try: subprocess.run(["ip", "route", "flush", "table", "100"], capture_output=True, timeout=2) except Exception: pass success = False for attempt in range(1, 4): try: subprocess.run(["ip", "route", "add", "default", "dev", interface, "table", "100"], check=True, timeout=2) subprocess.run(["ip", "rule", "add", "oif", interface, "table", "100"], check=True, timeout=2) # 配置反向路径过滤 rp_filter 为 loose 模式 (2),防止回包被内核静默丢弃 for proc_path in ["all", "default", interface]: try: subprocess.run(["sysctl", "-w", f"net.ipv4.conf.{proc_path}.rp_filter=2"], capture_output=True, timeout=2) except Exception: pass print(f"[policy_routing] Enabled policy routing for interface {interface} (attempt {attempt} success)", flush=True) success = True break except Exception as e: print(f"[policy_routing] Attempt {attempt} failed to enable policy routing: {e}", flush=True) time.sleep(1) if not success: print("[路由配置失败] [错误代码 3003] [ERR_ROUTE_TABLE_ADD_FAILED] 策略路由配置失败。原因: 无法向路由表 100 添加默认路由,这可能会导致通过 VPN 接口的出站路由无法正常解析。请检查系统是否支持策略路由、iproute2 工具是否完整,以及是否具有 root 权限。", flush=True) log_to_json("ERROR", "Routing", "[错误代码 3003] [ERR_ROUTE_TABLE_ADD_FAILED] 策略路由配置失败。原因: 无法向路由表 100 添加默认路由") return success def cleanup_policy_routing() -> None: try: subprocess.run(["ip", "rule", "del", "table", "100"], capture_output=True, timeout=2) subprocess.run(["ip", "route", "flush", "table", "100"], capture_output=True, timeout=2) print("[policy_routing] Cleared policy routing table 100", flush=True) except Exception: pass def stop_active_openvpn() -> None: global active_openvpn_process, active_openvpn_node_id with lock: cleanup_policy_routing() config_to_delete = None if active_openvpn_node_id: nodes = read_nodes() node = next((item for item in nodes if item.get("id") == active_openvpn_node_id), None) if node: config_to_delete = node.get("config_file") stop_process(active_openvpn_process) active_openvpn_process = None active_openvpn_node_id = "" if config_to_delete: try: path = Path(config_to_delete) if path.exists(): path.unlink() except Exception: pass def active_openvpn_running() -> bool: return active_openvpn_process is not None and active_openvpn_process.poll() is None def sort_all_nodes(nodes: list[dict[str, Any]]) -> list[dict[str, Any]]: available_nodes = sorted( [n for n in nodes if n.get("probe_status") == "available" or n.get("active")], key=lambda n: ( 0 if n.get("ip_type") in ("residential", "mobile") else 1, parse_int(n.get("latency_ms")) or 999999, -parse_int(n.get("score")) ) ) untested_nodes = sorted( [n for n in nodes if n.get("probe_status") in ("not_checked", "testing") and not n.get("active")], key=lambda n: (-parse_int(n.get("score")), parse_int(n.get("ping"))) ) unavailable_nodes = sorted( [n for n in nodes if n.get("probe_status") == "unavailable" and not n.get("active")], key=lambda n: (-parse_int(n.get("score")), -float(n.get("probed_at", 0))) ) return available_nodes + untested_nodes + unavailable_nodes def apply_routing_filters( nodes: list[dict[str, Any]], ui_cfg: dict[str, Any], include_unknown_ip_type: bool = False, ) -> list[dict[str, Any]]: candidates = list(nodes) routing_mode = ui_cfg.get("routing_mode", "auto") target_country = ui_cfg.get("force_country", "") if routing_mode == "fixed_region" and target_country: candidates = [ n for n in candidates if country_matches(n.get("country"), target_country) ] elif routing_mode == "favorites": fav_ids = set(ui_cfg.get("favorite_node_ids", [])) candidates = [n for n in candidates if n.get("id") in fav_ids] routing_ip_type = ui_cfg.get("routing_ip_type", "all") if routing_ip_type == "residential": candidates = [ n for n in candidates if n.get("ip_type") in ("residential", "mobile") or (include_unknown_ip_type and not n.get("ip_type")) ] elif routing_ip_type == "hosting": candidates = [ n for n in candidates if n.get("ip_type") == "hosting" or (include_unknown_ip_type and not n.get("ip_type")) ] return candidates def normalized_country_name(country: Any) -> str: value = str(country or "").strip() return vpn_utils.COUNTRY_TRANSLATIONS.get(value, value) def country_matches(node_country: Any, target_country: Any) -> bool: return bool(target_country) and normalized_country_name(node_country) == normalized_country_name(target_country) def probe_priority_key(node: dict[str, Any]) -> tuple[int, int, int, int]: ping = parse_int(node.get("ping")) or 999999 return ( ping, -parse_int(node.get("score")), -parse_int(node.get("speed")), parse_int(node.get("sessions")), ) def current_fixed_node_id(ui_cfg: dict[str, Any]) -> str: if active_openvpn_node_id: return active_openvpn_node_id nodes = read_nodes() active_node = next((n for n in nodes if n.get("active") and n.get("id")), None) if active_node: return str(active_node.get("id") or "") return str(ui_cfg.get("fixed_node_id") or "").strip() def validate_node_allowed_by_routing(node: dict[str, Any], ui_cfg: dict[str, Any]) -> None: routing_mode = ui_cfg.get("routing_mode", "auto") node_id = str(node.get("id") or "") if routing_mode == "fixed_region": target_country = ui_cfg.get("force_country", "") if target_country and not country_matches(node.get("country"), target_country): raise RuntimeError(f"当前已锁定国家【{target_country}】,不能连接其他国家节点") elif routing_mode == "favorites": fav_ids = set(ui_cfg.get("favorite_node_ids", [])) if node_id not in fav_ids: raise RuntimeError("当前处于仅用收藏模式,不能连接未收藏节点") routing_ip_type = ui_cfg.get("routing_ip_type", "all") node_ip_type = node.get("ip_type") if routing_ip_type == "residential" and node_ip_type not in ("residential", "mobile"): raise RuntimeError("当前已锁定住宅 IP 出站,不能连接非住宅节点") if routing_ip_type == "hosting" and node_ip_type != "hosting": raise RuntimeError("当前已锁定机房 IP 出站,不能连接非机房节点") def enforce_active_node_allowed_by_routing(ui_cfg: dict[str, Any], reason: str = "路由规则已更新") -> str | None: active_id = active_openvpn_node_id if not active_id: return None nodes = read_nodes() active_node = next((item for item in nodes if item.get("id") == active_id), None) if not active_node: clear_active_connection_state(f"{reason},当前活动节点已不在节点列表中,已断开连接") return "当前活动节点已不在节点列表中,已断开连接" try: validate_node_allowed_by_routing(active_node, ui_cfg) return None except Exception as exc: msg = f"{reason},当前活动节点 {active_id} 不符合新规则,已断开连接: {exc}" print(f"[路由规则] {msg}", flush=True) log_to_json("WARNING", "Routing", msg) stop_active_openvpn() with lock: nodes = read_nodes() for item in nodes: item["active"] = False write_json(NODES_FILE, nodes) set_state( active_openvpn_node_id="", active_node_latency="无活动连接", proxy_ok=False, proxy_ip="-", proxy_latency_ms=0, proxy_error=msg, last_check_message=msg, ) if ui_cfg.get("connection_enabled", True) and ui_cfg.get("routing_mode") != "fixed_ip": threading.Thread(target=auto_switch_node, daemon=True).start() return msg def reconnect_fixed_node_if_needed(ui_cfg: dict[str, Any]) -> bool: global is_connecting if ui_cfg.get("routing_mode") != "fixed_ip" or active_openvpn_running(): return False target_id = current_fixed_node_id(ui_cfg) if not target_id: return False nodes = read_nodes() if not any(n.get("id") == target_id for n in nodes): return False print(f"[维护线程] 固定 IP 模式下 OpenVPN 未运行,正在重新拉起同一节点: {target_id}", flush=True) previous_connecting = is_connecting is_connecting = False try: connect_node(target_id) return active_openvpn_running() except Exception as e: print(f"[维护线程] 重新拉起固定节点 {target_id} 失败: {e}", flush=True) return False finally: is_connecting = previous_connecting active_test_indexes = set() test_indexes_lock = threading.Lock() def get_free_test_index() -> int: with test_indexes_lock: for idx in range(2, 100): if idx not in active_test_indexes: active_test_indexes.add(idx) return idx raise RuntimeError("没有可用的 OpenVPN 测试网卡编号,请稍后重试") def release_test_index(idx: int) -> None: with test_indexes_lock: active_test_indexes.discard(idx) def test_config_path(node_id: str) -> Path: safe_id = safe_name(node_id) return CONFIG_DIR / f".test_{safe_id}_{uuid.uuid4().hex}.ovpn" def test_node_by_id(node_id: str) -> dict[str, Any]: with lock: nodes = read_nodes() node = next((item for item in nodes if item.get("id") == node_id), None) if not node: raise ValueError(f"Node not found: {node_id}") config_text = node.get("config_text") or "" h = str(node.get("remote_host") or node.get("ip")) p = parse_int(node.get("remote_port")) fallback_ping = parse_int(node.get("ping")) temp_path = test_config_path(node_id) try: CONFIG_DIR.mkdir(exist_ok=True, parents=True) temp_path.write_text(config_text, encoding="utf-8") except Exception as e: raise RuntimeError(f"Failed to write temp config file: {e}") latency = vpn_utils.ping_latency_ms(h, p, fallback_ping) idx = None try: idx = get_free_test_index() ok, message, _ = run_openvpn_until_ready(str(temp_path), keep_alive=False, route_nopull=True, timeout=12, dev=f"tun{idx}") finally: if idx is not None: release_test_index(idx) try: if temp_path.exists(): temp_path.unlink() except Exception: pass temp_node = { "id": node_id, "ip": h, "remote_host": h, "remote_port": p, "owner": "", "asn": "", "as_name": "", "location": "", "ip_type": "", "quality": "", } if ok: vpn_utils.enrich_ip_info([temp_node]) with lock: nodes = read_nodes() node = next((item for item in nodes if item.get("id") == node_id), None) if node: node["latency_ms"] = latency node["probe_status"] = "available" if ok else "unavailable" node["probe_message"] = message node["probed_at"] = time.time() if ok: node["owner"] = temp_node["owner"] node["asn"] = temp_node["asn"] node["as_name"] = temp_node["as_name"] node["location"] = temp_node["location"] node["ip_type"] = temp_node["ip_type"] node["quality"] = temp_node["quality"] sorted_nodes = sort_all_nodes(nodes) write_json(NODES_FILE, sorted_nodes) res = next((item for item in sorted_nodes if item.get("id") == node_id), node) return res else: return {} def is_systemic_probe_failure(message: Any) -> bool: normalized = str(message or "").lower() return any( token in normalized for token in ( "err_ovpn_cmd_not_found", "err_ovpn_permission_denied", "err_ovpn_tun_not_available", "no such file or directory", "cannot open tun/tap dev", "cannot allocate tun", ) ) def test_multiple_nodes(node_ids: list[str], target_available: int | None = None) -> list[dict[str, Any]]: with lock: nodes = read_nodes() to_test = [n for n in nodes if n.get("id") in node_ids] def test_worker(args: tuple[int, dict[str, Any]]) -> dict[str, Any]: idx, n_info = args node_id = n_info["id"] config_text = n_info.get("config_text") or "" h = str(n_info.get("remote_host") or n_info.get("ip")) p = parse_int(n_info.get("remote_port")) fallback_ping = parse_int(n_info.get("ping")) temp_path = test_config_path(node_id) try: CONFIG_DIR.mkdir(exist_ok=True, parents=True) temp_path.write_text(config_text, encoding="utf-8") except Exception as e: return { "id": node_id, "latency_ms": 0, "probe_status": "unavailable", "probe_message": f"Failed to write configuration: {e}", "probed_at": time.time(), "owner": "", "asn": "", "as_name": "", "location": "", "ip_type": "", "quality": "", } latency = vpn_utils.ping_latency_ms(h, p, fallback_ping) tun_idx = None try: tun_idx = get_free_test_index() dev_name = f"tun{tun_idx}" ok, message, _ = run_openvpn_until_ready(str(temp_path), keep_alive=False, route_nopull=True, timeout=12, dev=dev_name) finally: if tun_idx is not None: release_test_index(tun_idx) try: if temp_path.exists(): temp_path.unlink() except Exception: pass temp_node = { "id": node_id, "ip": n_info.get("ip") or h, "remote_host": h, "remote_port": p, "latency_ms": latency, "probe_status": "available" if ok else "unavailable", "probe_message": message, "probed_at": time.time(), "owner": "", "asn": "", "as_name": "", "location": "", "ip_type": "", "quality": "", } return temp_node updated_nodes_map: dict[str, dict[str, Any]] = {} available_count = 0 systemic_failure = "" max_workers = min(NODE_PROBE_WORKERS, max(1, len(to_test))) with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: for batch_start in range(0, len(to_test), max_workers): if systemic_failure or (target_available is not None and available_count >= target_available): break batch = to_test[batch_start : batch_start + max_workers] batch_ids = {str(n.get("id") or "") for n in batch} with lock: current_nodes = read_nodes() now = time.time() for current in current_nodes: if current.get("id") in batch_ids and not current.get("active"): current["probe_status"] = "testing" current["probe_message"] = "正在检测节点连通性..." current["probed_at"] = now write_json(NODES_FILE, sort_all_nodes(current_nodes)) futures = { executor.submit(test_worker, (batch_start + idx, node)): node["id"] for idx, node in enumerate(batch) } for future in concurrent.futures.as_completed(futures): nid = futures[future] try: result = future.result() except Exception as exc: result = { "id": nid, "probe_status": "unavailable", "probe_message": f"Test exception: {exc}", "latency_ms": 0, } updated_nodes_map[nid] = result if result.get("probe_status") == "available": available_count += 1 if is_systemic_probe_failure(result.get("probe_message")): systemic_failure = str(result.get("probe_message") or "") with lock: current_nodes = read_nodes() for current in current_nodes: if current.get("id") == nid: current.update(result) break write_json(NODES_FILE, sort_all_nodes(current_nodes)) if systemic_failure: message = f"检测到系统级 OpenVPN 故障,已停止剩余节点探测: {systemic_failure}" print(f"[节点检测] {message}", flush=True) log_to_json("ERROR", "VPN", message) set_state(last_check_message=message) break # 批量查询并丰富可用节点的地理及 ISP 信息,防止并发时被定位 API 接口限流 successful_nodes = [res for res in updated_nodes_map.values() if res.get("probe_status") == "available"] if successful_nodes: try: vpn_utils.enrich_ip_info(successful_nodes) except Exception as ee: print(f"[test_multiple_nodes] 批量富化 IP 失败: {ee}", flush=True) with lock: current_nodes = read_nodes() for n in current_nodes: nid = n.get("id") if nid in updated_nodes_map: n.update(updated_nodes_map[nid]) sorted_nodes = sort_all_nodes(current_nodes) write_json(NODES_FILE, sorted_nodes) return list(updated_nodes_map.values()) def cancel_background_refill() -> None: background_refill_cancel_event.set() def schedule_background_refill() -> bool: global background_refill_thread with background_refill_lock: if background_refill_thread is not None and background_refill_thread.is_alive(): return False background_refill_cancel_event.clear() def refill_worker() -> None: global background_refill_thread try: for delay in (60, 120, 300): if background_refill_cancel_event.wait(delay): return ui_cfg = load_ui_config() if not ui_cfg.get("connection_enabled", True): return try: maintain_valid_nodes(force=False) except Exception as exc: log_to_json("WARNING", "Main", f"后台节点补齐失败: {exc}") if active_openvpn_running(): return finally: with background_refill_lock: if background_refill_thread is threading.current_thread(): background_refill_thread = None background_refill_thread = threading.Thread( target=refill_worker, name="vpngate-node-refill", daemon=True, ) background_refill_thread.start() return True def auto_switch_node(attempt: int = 0) -> None: if attempt >= 3: print("[自动切换] 连续切换失败已达 3 次,停止切换以防止主线程死锁,将在后台重新加载节点...", flush=True) return ui_cfg = load_ui_config() connection_enabled = ui_cfg.get("connection_enabled", True) if not connection_enabled: print("[自动切换] 连接已禁用,不进行自动切换。", flush=True) return routing_mode = ui_cfg.get("routing_mode", "auto") target_country = ui_cfg.get("force_country", "") if routing_mode == "fixed_ip": print("[自动切换] 当前处于固定 IP 模式,不进行自动连接或切换。", flush=True) return # Find the next best available node with lock: nodes = read_nodes() candidates = [ n for n in nodes if n.get("probe_status") == "available" and not n.get("active") ] candidates = apply_routing_filters(candidates, ui_cfg) candidates.sort(key=lambda n: (parse_int(n.get("latency_ms")) or 999999, -parse_int(n.get("score")))) if candidates: next_node = candidates[0] msg = f"当前连接已失效或代理连通性检测失败,正在自动切换至最佳备用节点: {next_node['id']}" print(f"[自动切换] {msg}", flush=True) log_to_json("INFO", "VPN", msg) try: connect_node(next_node["id"]) except Exception as e: err_msg = f"切换到备用节点 {next_node['id']} 失败: {e},将尝试下一个..." print(f"[自动切换] {err_msg}", flush=True) log_to_json("WARNING", "VPN", err_msg) auto_switch_node(attempt + 1) else: msg = "没有可用的备选节点,将自动断开并清理当前连接状态,同时在后台异步获取新节点..." if routing_mode == "fixed_region" and target_country: msg = f"没有可用的【{target_country}】备选节点,已断开连接,将在后台持续尝试获取新节点..." print(f"[自动切换] {msg}", flush=True) log_to_json("WARNING", "VPN", msg) stop_active_openvpn() with lock: nodes = read_nodes() for item in nodes: item["active"] = False write_json(NODES_FILE, nodes) set_state(active_openvpn_node_id="", last_check_message=msg) if schedule_background_refill(): log_to_json("INFO", "Main", "已启动唯一后台节点补齐任务") def recover_after_manual_connect_failure(previous_node_id: str) -> None: if active_openvpn_running(): return if previous_node_id: try: log_to_json("WARNING", "VPN", f"手动切换失败,正在恢复原节点: {previous_node_id}") connect_node(previous_node_id) return except Exception as exc: log_to_json("ERROR", "VPN", f"恢复原节点 {previous_node_id} 失败: {exc}") ui_cfg = load_ui_config() if ui_cfg.get("connection_enabled", True) and ui_cfg.get("routing_mode") != "fixed_ip": auto_switch_node() def connect_node(node_id: str) -> str: global active_openvpn_process, active_openvpn_node_id global last_active_ping_time, last_active_latency global consecutive_proxy_failures, last_proxy_failure_node_id node_id = str(node_id or "").strip() if not node_id: raise ValueError("Node id is required") token, cancel_event = begin_connection_attempt() stopped_existing = False previous_node_id = "" try: set_state( is_connecting=True, pending_node_id=node_id, active_node_latency="正在连接", last_check_message=f"正在初始化连接配置: {node_id}", ) log_to_json("INFO", "VPN", f"开始连接节点: {node_id}") nodes = read_nodes() node = next((item for item in nodes if item.get("id") == node_id), None) if not node: raise ValueError(f"Node not found: {node_id}") with lock: if active_openvpn_running(): previous_node_id = active_openvpn_node_id ui_cfg = load_ui_config() validate_node_allowed_by_routing(node, ui_cfg) ui_cfg["connection_enabled"] = True auth_file = DATA_DIR / "ui_auth.json" with lock: DATA_DIR.mkdir(exist_ok=True, parents=True) write_json(auth_file, ui_cfg) set_state(active_node_latency="写入配置", last_check_message="正在写入 OpenVPN 节点配置文件...") config_path = Path(node["config_file"]) try: CONFIG_DIR.mkdir(exist_ok=True, parents=True) config_path.write_text(node.get("config_text") or "", encoding="utf-8") except Exception as e: raise RuntimeError(f"Failed to write configuration: {e}") probed_at = float(node.get("probed_at", 0) or 0) should_preflight = ( SWITCH_PREFLIGHT_MAX_AGE_SECONDS > 0 and bool(previous_node_id) and previous_node_id != node_id and time.time() - probed_at > SWITCH_PREFLIGHT_MAX_AGE_SECONDS ) if should_preflight: set_state(active_node_latency="切换预检", last_check_message="正在保持当前连接并预检目标节点...") test_index = None try: test_index = get_free_test_index() preflight_ok, preflight_message, _ = run_openvpn_until_ready( str(config_path), keep_alive=False, route_nopull=True, timeout=12, dev=f"tun{test_index}", cancel_event=cancel_event, track_pending=True, ) finally: if test_index is not None: release_test_index(test_index) if not connection_attempt_is_current(token, cancel_event): raise ConnectionCancelled("连接操作已取消") if not preflight_ok: with lock: current_nodes = read_nodes() failed_node = next((item for item in current_nodes if item.get("id") == node_id), None) if failed_node: failed_node["probe_status"] = "unavailable" failed_node["probe_message"] = preflight_message failed_node["probed_at"] = time.time() write_json(NODES_FILE, sort_all_nodes(current_nodes)) raise RuntimeError(f"目标节点预检失败,已保留当前连接: {preflight_message}") if not connection_attempt_is_current(token, cancel_event): raise ConnectionCancelled("连接操作已取消") set_state(active_node_latency="清理连接", last_check_message="目标节点可用,正在关闭旧的 VPN 连接及网卡...") stop_active_openvpn() stopped_existing = True set_state(active_node_latency="启动核心", last_check_message="正在启动 OpenVPN Core 核心服务并建立连接...") ok, message, process = run_openvpn_until_ready( str(config_path), keep_alive=True, route_nopull=True, cancel_event=cancel_event, track_pending=True, ) if not connection_attempt_is_current(token, cancel_event): stop_process(process) raise ConnectionCancelled("连接操作已取消") if not ok or process is None: try: if config_path.exists(): config_path.unlink() except Exception: pass node["probe_status"] = "unavailable" node["probe_message"] = message for item in nodes: item["active"] = False write_json(NODES_FILE, sort_all_nodes(nodes)) log_to_json("ERROR", "VPN", f"连接节点 {node_id} 失败: {message}") print(f"[连接核心失败] 无法与 VPN 节点 {node_id} 建立隧道连接!详情: {message}", flush=True) raise RuntimeError(message) with lock: if not connection_attempt_is_current(token, cancel_event): stop_process(process) raise ConnectionCancelled("连接操作已取消") active_openvpn_process = process active_openvpn_node_id = node_id set_state(active_node_latency="配置路由", last_check_message="正在配置策略路由规则与流量转发...") routing_ready = setup_policy_routing("tun0") if not connection_attempt_is_current(token, cancel_event): raise ConnectionCancelled("连接操作已取消") last_active_ping_time = time.time() last_active_latency = 0 set_state(active_node_latency="测试延迟", last_check_message="正在直连测试代理出口延迟与可用性...") try: ip = node.get("ip") or node.get("remote_host") port = parse_int(node.get("remote_port")) fallback = parse_int(node.get("ping")) latency = vpn_utils.ping_latency_ms(ip, port, fallback) if latency > 0: last_active_latency = latency except Exception: pass set_state(last_check_message="正在测试本地代理出站联通性与出口 IP...") res = check_proxy_health() if not connection_attempt_is_current(token, cancel_event): raise ConnectionCancelled("连接操作已取消") if not res["ok"]: route_note = ";策略路由配置失败" if not routing_ready else "" raise RuntimeError(f"VPN 隧道已建立但代理出口不可用{route_note}: {res.get('error', '未知错误')}") latest_ui_cfg = load_ui_config() validate_node_allowed_by_routing(node, latest_ui_cfg) latest_ui_cfg["connection_enabled"] = True if latest_ui_cfg.get("routing_mode") == "fixed_ip": latest_ui_cfg["fixed_node_id"] = node_id latency_str = f"{last_active_latency} ms" if last_active_latency > 0 else "检测超时" with lock: if not connection_attempt_is_current(token, cancel_event): raise ConnectionCancelled("连接操作已取消") current_nodes = read_nodes() for item in current_nodes: item["active"] = item.get("id") == node_id if item["active"]: item["probe_status"] = "available" item["probed_at"] = time.time() _ph = f"[{LOCAL_PROXY_HOST}]" if ":" in LOCAL_PROXY_HOST else LOCAL_PROXY_HOST item["probe_message"] = f"Active node. HTTP proxy: http://{_ph}:{LOCAL_PROXY_PORT}" write_json(NODES_FILE, sort_all_nodes(current_nodes)) write_json(auth_file, latest_ui_cfg) consecutive_proxy_failures = 0 last_proxy_failure_node_id = node_id set_state( active_openvpn_node_id=node_id, is_connecting=False, pending_node_id="", last_check_message=f"Connected {node_id}", active_node_latency=latency_str, proxy_ok=True, proxy_ip=res["ip"], proxy_latency_ms=res["latency_ms"], proxy_error="", ) log_to_json("INFO", "VPN", f"节点 {node_id} 连接成功,出口网卡 tun0 已启用") cancel_background_refill() return f"Connected {node_id}" except ConnectionCancelled: if stopped_existing: stop_active_openvpn() raise except Exception as exc: if stopped_existing or (active_openvpn_node_id == node_id and not active_openvpn_running()): with lock: current_nodes = read_nodes() failed_node = next((item for item in current_nodes if item.get("id") == node_id), None) if failed_node: failed_node["probe_status"] = "unavailable" failed_node["probe_message"] = str(exc) failed_node["probed_at"] = time.time() write_json(NODES_FILE, sort_all_nodes(current_nodes)) clear_active_connection_state(f"连接失败: {exc}") else: set_state(is_connecting=False, pending_node_id="", last_check_message=f"连接失败: {exc}") raise finally: finish_connection_attempt(token, cancel_event) set_state(pending_node_id="") def maintain_valid_nodes(force: bool = False) -> str: global active_openvpn_process, active_openvpn_node_id, is_connecting ensure_dirs() if not maintenance_lock.acquire(blocking=False): msg = "节点维护任务正在运行,请稍后再试" set_state(last_check_message=msg) return msg with lock: if is_connecting: maintenance_lock.release() msg = "当前已有连接或节点测试任务正在运行,请稍后再试" set_state(last_check_message=msg) return msg is_connecting = True try: # A forced refresh must not tear down a healthy tunnel. It only forces # the node-pool maintenance path below. if not active_openvpn_running(): ui_cfg = load_ui_config() routing_mode = ui_cfg.get("routing_mode", "auto") connection_enabled = ui_cfg.get("connection_enabled", True) if connection_enabled: if routing_mode == "fixed_ip": reconnect_fixed_node_if_needed(ui_cfg) else: has_active_id = False with lock: if active_openvpn_node_id: has_active_id = True stop_active_openvpn() if has_active_id: print("[维护线程] 检测到当前 OpenVPN 进程已意外退出,准备自动切换节点", flush=True) is_connecting = False auto_switch_node() is_connecting = True try: set_state(is_connecting=True, last_check_message="正在拉取最新的免费 VPN 节点列表...") candidates = fetch_candidates() except Exception as exc: vpn_utils.check_and_fix_dns() diag_msg = str(exc) if not any(token in diag_msg for token in ["[ERR_", "错误代码"]): err_code, raw_diag = vpn_utils.diagnose_api_failure(API_URL) diag_msg = f"[错误代码 {err_code}] 获取节点失败: {exc} | 诊断结果: {raw_diag}" set_state(last_fetch_at=time.time(), last_fetch_status="error", last_fetch_message=diag_msg) candidates = [] if not candidates: return "没有拉取到新节点" with lock: current_nodes = read_nodes() current_by_id = { str(n.get("id")): n for n in current_nodes if n.get("id") } active_node = None if active_openvpn_node_id: active_node = next((n for n in current_nodes if n.get("id") == active_openvpn_node_id), None) merged: list[dict[str, Any]] = [] seen_ids: set[str] = set() if active_node: merged.append(active_node) seen_ids.add(active_node["id"]) for cand in candidates: if cand["id"] not in seen_ids: previous = current_by_id.get(str(cand["id"])) if previous: for key in [ "probe_status", "probe_message", "latency_ms", "probed_at", "owner", "asn", "as_name", "location", "ip_type", "quality", ]: if previous.get(key) not in (None, ""): cand[key] = previous.get(key) merged.append(cand) seen_ids.add(cand["id"]) if len(merged) > 1000: merged = merged[:1000] for n in merged: config_path = Path(n["config_file"]) if not config_path.exists(): try: config_path.write_text(n["config_text"], encoding="utf-8") except Exception: pass write_json(NODES_FILE, merged) initial_tested_ids: set[str] = set() fast_results: list[dict[str, Any]] = [] systemic_probe_failure = "" ui_cfg = load_ui_config() should_fast_connect = ( ui_cfg.get("connection_enabled", True) and ui_cfg.get("routing_mode", "auto") != "fixed_ip" and not active_openvpn_running() ) if should_fast_connect: with lock: current_nodes = read_nodes() fast_candidates = [ n for n in current_nodes if not n.get("active") and n.get("probe_status") != "unavailable" ] fast_candidates = apply_routing_filters(fast_candidates, ui_cfg, include_unknown_ip_type=True) fast_candidates.sort(key=probe_priority_key) fast_test_ids = [ n["id"] for n in fast_candidates if n.get("id") ][:INITIAL_CONNECT_TEST_LIMIT] if fast_test_ids: msg = f"首次快速连接模式:优先测试 {len(fast_test_ids)} 个高优先级节点,发现可用节点后立即连接" print(f"[快速首连] {msg}", flush=True) log_to_json("INFO", "Main", msg) set_state(is_connecting=True, last_check_message=msg) fast_results = test_multiple_nodes(fast_test_ids, target_available=TARGET_VALID_NODES) systemic_probe_failure = next( ( str(result.get("probe_message") or "") for result in fast_results if is_systemic_probe_failure(result.get("probe_message")) ), "", ) initial_tested_ids = { str(result.get("id") or "") for result in fast_results if result.get("id") } with lock: fast_nodes = read_nodes() available_candidates = [ n for n in fast_nodes if n.get("probe_status") == "available" and not n.get("active") ] available_candidates = apply_routing_filters(available_candidates, ui_cfg) if available_candidates: is_connecting = False set_state(is_connecting=False, last_check_message="快速首连已找到可用节点,正在建立连接...") auto_switch_node() if active_openvpn_running(): valid_nodes_count = len([n for n in read_nodes() if n.get("probe_status") == "available"]) message = f"Fetched {len(candidates)} nodes. Fast-tested {len(fast_results)} nodes and connected." set_state( last_check_at=time.time(), last_check_message=message, active_openvpn_node_id=active_openvpn_node_id, valid_nodes=valid_nodes_count, ) return message is_connecting = True tested_results: list[dict[str, Any]] = [] if systemic_probe_failure: msg = f"已跳过本轮剩余节点检测,系统级故障需要先处理: {systemic_probe_failure}" print(f"[周期检测] {msg}", flush=True) log_to_json("ERROR", "VPN", msg) set_state(last_check_message=msg) else: # Test remaining non-active nodes from the list with lock: current_nodes = read_nodes() to_test = [ n for n in current_nodes if not n.get("active") and n.get("id") not in initial_tested_ids ] to_test = apply_routing_filters(to_test, ui_cfg, include_unknown_ip_type=True) to_test.sort(key=probe_priority_key) to_test_ids = [n["id"] for n in to_test] msg = f"开始对列表中所有候选节点进行周期连通性与延迟测试,待检测节点共 {len(to_test_ids)} 个" print(f"[周期检测] {msg}", flush=True) log_to_json("INFO", "Main", msg) set_state(is_connecting=True, last_check_message="正在并发检测所有节点可用性...") tested_results = test_multiple_nodes(to_test_ids, target_available=TARGET_VALID_NODES) is_connecting = False with lock: merged = read_nodes() # Identify available, unavailable, and active nodes available_nodes = [n["id"] for n in merged if n.get("probe_status") == "available"] unavailable_nodes = [n["id"] for n in merged if n.get("probe_status") == "unavailable"] active_node = next((n["id"] for n in merged if n.get("active")), "无") status_report = ( f"周期节点检测完成。实时同步状态: 获取到候选节点共 {len(merged)} 个。 " f"其中【可用节点】{len(available_nodes)} 个: {available_nodes[:15]}...; " f"【不可用节点】{len(unavailable_nodes)} 个; " f"当前【正在正常运行的活动连接节点】为: {active_node}。" ) print(f"[周期检测] {status_report}", flush=True) log_to_json("INFO", "Main", status_report) if active_node != "无" and not active_openvpn_running(): warn_msg = f"[诊断警告] 活动节点 {active_node} 被标记为活动状态,但 OpenVPN 进程实际并未正常运行!" print(warn_msg, flush=True) log_to_json("WARNING", "Main", warn_msg) if not active_openvpn_running(): ui_cfg = load_ui_config() connection_enabled = ui_cfg.get("connection_enabled", True) if connection_enabled: routing_mode = ui_cfg.get("routing_mode", "auto") if routing_mode != "fixed_ip": available_candidates = [n for n in merged if n.get("probe_status") == "available"] available_candidates = apply_routing_filters(available_candidates, ui_cfg) if available_candidates: auto_switch_node() valid_nodes_count = len([n for n in merged if n.get("probe_status") == "available"]) total_tested = len(fast_results) + len(tested_results) message = f"Fetched {len(candidates)} nodes. Tested {total_tested} prioritized non-active nodes." set_state( last_check_at=time.time(), last_check_message=message, active_openvpn_node_id=active_openvpn_node_id, valid_nodes=valid_nodes_count, ) return message except Exception as e: raise e finally: is_connecting = False maintenance_lock.release() def collector_loop() -> None: global last_collector_heartbeat while True: last_collector_heartbeat = time.time() success = False try: print("[守护线程] 开始执行节点拉取与可用性检测周期任务...", flush=True) log_to_json("INFO", "Main", "开始执行节点拉取与可用性检测周期任务...") res = maintain_valid_nodes(force=False) if "没有拉取到新节点" not in res: success = True log_to_json("INFO", "Main", f"周期同步与检测任务完成,结果: {res}") except Exception as exc: err_msg = f"周期节点同步任务执行异常: {exc}" print(f"[错误] {err_msg}", flush=True) log_to_json("ERROR", "Main", err_msg) set_state(last_check_at=time.time(), last_check_message=f"check error: {exc}") if not active_openvpn_running() and not success: sleep_time = 30 else: sleep_time = CHECK_INTERVAL_SECONDS time.sleep(sleep_time) LOGIN_HTML = r""" AimiliVPN - 安全登录
""" INDEX_HTML = r""" AimiliVPN 节点池管理系统

AimiliVPN 节点管理系统

Telegram
状态 IP 地址 : 端口 延迟 物理位置 运营主体 / ISP IP 类型 操作
VPS购买推荐
""" def check_proxy_health() -> dict[str, Any]: # 1. 检测代理服务端口是否在监听 is_ipv6 = ":" in LOCAL_PROXY_HOST af = socket.AF_INET6 if is_ipv6 else socket.AF_INET s = None try: s = socket.socket(af, socket.SOCK_STREAM) s.settimeout(1.5) connect_host = LOCAL_PROXY_HOST if connect_host in ("::", "0.0.0.0", ""): connect_host = "::1" if is_ipv6 else "127.0.0.1" try: s.connect((connect_host, LOCAL_PROXY_PORT)) except Exception as e: if connect_host == "::1": s.close() s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(1.5) s.connect(("127.0.0.1", LOCAL_PROXY_PORT)) else: raise e except Exception as e: diag = vpn_utils.diagnose_local_obstructions(LOCAL_PROXY_PORT, host=LOCAL_PROXY_HOST) diag_msg = diag[1] if diag else f"端口 {LOCAL_PROXY_PORT} 连接失败,原因: {e}" return { "ok": False, "error": f"代理服务未运行 ({diag_msg})" } finally: if s is not None: try: s.close() except Exception: pass # 2. 检测虚拟网卡 tun0 是否存在 (Linux 下) tun_path = Path("/sys/class/net/tun0") if sys.platform.startswith("linux") and not tun_path.exists(): return { "ok": False, "error": "[错误代码 3004] [ERR_ROUTE_DEV_NOT_FOUND] VPN 虚拟网卡 (tun0) 未启用,请确保当前已成功连接 VPN 节点" } # 3. 使用 curl 通过本地 SOCKS5 代理接口测试 IP 与实际延迟 def _curl_check_ip(url: str) -> dict[str, Any] | None: proxy_hosts = [] if LOCAL_PROXY_HOST == "::": proxy_hosts = ["[::1]", "127.0.0.1"] elif LOCAL_PROXY_HOST == "0.0.0.0": proxy_hosts = ["127.0.0.1"] elif ":" in LOCAL_PROXY_HOST: proxy_hosts = [f"[{LOCAL_PROXY_HOST}]", "127.0.0.1"] else: proxy_hosts = [LOCAL_PROXY_HOST] for p_host in proxy_hosts: proxy_url = f"socks5h://{p_host}:{LOCAL_PROXY_PORT}" proxy_user, proxy_pass = proxy_server.get_proxy_credentials() cmd = [ "curl", "-s", "-w", "\n%{time_total} %{http_code}", "-x", proxy_url, url, "--max-time", "5" ] if proxy_user is not None and proxy_pass is not None: cmd.extend(["--proxy-user", f"{proxy_user}:{proxy_pass}"]) try: res = subprocess.run(cmd, capture_output=True, text=True, timeout=6) if res.returncode == 0: lines = res.stdout.strip().splitlines() if len(lines) >= 2: ip = lines[0].strip() time_info = lines[1].strip().split() if len(time_info) == 2: total_time_str, http_code = time_info if http_code == "200" and ip: latency_ms = int(float(total_time_str) * 1000) return {"ok": True, "ip": ip, "latency_ms": latency_ms} except Exception: pass return None try: result = _curl_check_ip("http://ip.sb") if result: return result result = _curl_check_ip("http://api.ipify.org") if result: return result # 此时外网测试失败,检测本地代理端口是否依然能连通。若仍能连通,直接抛出出口测试失败,不调用占用诊断 port_still_listening = False test_sock = None try: test_sock = socket.socket(af, socket.SOCK_STREAM) test_sock.settimeout(1.0) connect_host = LOCAL_PROXY_HOST if connect_host in ("::", "0.0.0.0", ""): connect_host = "::1" if is_ipv6 else "127.0.0.1" try: test_sock.connect((connect_host, LOCAL_PROXY_PORT)) port_still_listening = True except Exception: if connect_host == "::1": test_sock.close() test_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) test_sock.settimeout(1.0) test_sock.connect(("127.0.0.1", LOCAL_PROXY_PORT)) port_still_listening = True except Exception: pass finally: if test_sock is not None: try: test_sock.close() except Exception: pass if not port_still_listening: diag = vpn_utils.diagnose_local_obstructions(LOCAL_PROXY_PORT, host=LOCAL_PROXY_HOST) if diag: return {"ok": False, "error": f"出口连接测试失败 | 本机诊断结果: {diag[1]}"} return {"ok": False, "error": "出口连接测试失败 (ip.sb 和 api.ipify.org 均无法连通,可能是节点已失效或 VPS 防火墙限制了 UDP/TCP 出站端口)"} except Exception as e: return {"ok": False, "error": f"出口连接测试异常: {e}"} def reset_proxy_failure_counter(node_id: str = "") -> None: global consecutive_proxy_failures, last_proxy_failure_node_id with lock: consecutive_proxy_failures = 0 last_proxy_failure_node_id = node_id def record_proxy_failure(node_id: str) -> int: global consecutive_proxy_failures, last_proxy_failure_node_id with lock: if node_id != last_proxy_failure_node_id: consecutive_proxy_failures = 0 last_proxy_failure_node_id = node_id consecutive_proxy_failures += 1 return consecutive_proxy_failures def background_proxy_checker() -> None: global last_checker_heartbeat, is_connecting time.sleep(30) while True: last_checker_heartbeat = time.time() try: if is_connecting: time.sleep(5) continue checked_node_id = active_openvpn_node_id res = check_proxy_health() if checked_node_id != active_openvpn_node_id: continue if res["ok"]: reset_proxy_failure_counter(checked_node_id) set_state( proxy_ok=True, proxy_ip=res["ip"], proxy_latency_ms=res["latency_ms"], proxy_error="" ) log_to_json("INFO", "Proxy", f"代理可用,IP: {res['ip']}, 延迟: {res['latency_ms']} ms") else: error_msg = res.get("error", "未知错误") failure_count = record_proxy_failure(checked_node_id) if checked_node_id else 0 process_exited = bool(checked_node_id) and not active_openvpn_running() should_recover = process_exited or failure_count >= PROXY_FAILURE_THRESHOLD if checked_node_id: print(f"[警告] {LOCAL_PROXY_PORT} 端口本地代理当前不可用!原因: {error_msg}", flush=True) log_to_json( "WARNING", "Proxy", f"代理不可用 ({failure_count}/{PROXY_FAILURE_THRESHOLD}): {error_msg}", ) display_error = error_msg if checked_node_id and not process_exited and not should_recover: display_error = f"{error_msg}(连续失败 {failure_count}/{PROXY_FAILURE_THRESHOLD},暂不切换)" set_state( proxy_ok=False, proxy_ip="-", proxy_latency_ms=0, proxy_error=display_error, ) # A dead OpenVPN process is recovered immediately. Transient # external probe failures must cross the configured threshold. if checked_node_id and should_recover: reset_proxy_failure_counter(checked_node_id) ui_cfg = load_ui_config() routing_mode = ui_cfg.get("routing_mode", "auto") if routing_mode != "fixed_ip": with lock: nodes = read_nodes() active_node = next((n for n in nodes if n.get("id") == checked_node_id), None) if active_node: mark_blacklisted(active_node, f"代理连通性检测失败: {error_msg}") active_node["probe_status"] = "unavailable" write_json(NODES_FILE, nodes) auto_switch_node() else: print(f"[代理守护线程] 固定 IP 模式下代理不可用,正在尝试重启连接同一节点: {checked_node_id}", flush=True) try: connect_node(checked_node_id) except Exception as e: print(f"[代理守护线程] 重启固定节点失败: {e}", flush=True) except Exception as e: print(f"[错误] 代理后台检测发生异常: {e}", flush=True) log_to_json("ERROR", "Proxy", f"检测守护线程发生异常: {e}") time.sleep(30) def active_node_pinger() -> None: global last_pinger_heartbeat while True: last_pinger_heartbeat = time.time() try: if active_openvpn_running() and active_openvpn_node_id: nodes = read_nodes() node = next((n for n in nodes if n.get("id") == active_openvpn_node_id), None) if node: ip = node.get("ip") or node.get("remote_host") port = parse_int(node.get("remote_port")) fallback = parse_int(node.get("ping")) if ip: latency = vpn_utils.ping_latency_ms(ip, port, fallback) if latency > 0: set_state(active_node_latency=f"{latency} ms") else: set_state(active_node_latency="检测超时") else: set_state(active_node_latency="检测超时") else: set_state(active_node_latency="检测超时") elif is_connecting: set_state(active_node_latency="测试中...") else: set_state(active_node_latency="无活动连接") except Exception as e: print(f"[ERROR] active_node_pinger error: {e}", flush=True) time.sleep(10) class Handler(BaseHTTPRequestHandler): def get_secret_path(self) -> str: ui_cfg = load_ui_config() return ui_cfg.get("secret_path", "EJsW2EeBo9lY") def is_authorized(self) -> bool: ui_cfg = load_ui_config() pwd = ui_cfg.get("password") if not pwd: print("[Auth] 管理后台密码为空,已拒绝访问。请检查 ui_auth.json。", flush=True) return False cookie_header = self.headers.get("Cookie", "") cookies = {} if cookie_header: for item in cookie_header.split(";"): item = item.strip() if "=" in item: k, v = item.split("=", 1) cookies[k.strip()] = v.strip() session_token = cookies.get("session") if not session_token: return False with lock: exp_time = active_sessions.get(session_token) if exp_time is not None and exp_time > time.time(): return True return False def validate_path(self) -> str: secret_path = self.get_secret_path() request_path = urllib.parse.urlsplit(self.path).path if not secret_path: return request_path if request_path == f"/{secret_path}": self.send_response(HTTPStatus.FOUND) self.send_header("Location", f"/{secret_path}/") self.end_headers() return "" prefix = f"/{secret_path}/" if request_path.startswith(prefix): return "/" + request_path[len(prefix):] self.send_response(HTTPStatus.NOT_FOUND) self.end_headers() return "" def log_message(self, format: str, *args: Any) -> None: print(f"[{self.log_date_time_string()}] {format % args}", flush=True) def send_bytes(self, body: bytes, content_type: str, status: HTTPStatus = HTTPStatus.OK) -> None: self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) def send_json(self, data: Any, status: HTTPStatus = HTTPStatus.OK) -> None: self.send_bytes(json.dumps(data, ensure_ascii=False).encode("utf-8"), "application/json; charset=utf-8", status) def read_request_body(self, max_bytes: int = 65536) -> bytes: length = parse_int(self.headers.get("Content-Length")) if length < 0: raise ValueError("Content-Length 无效") if length > max_bytes: raise ValueError(f"请求体过大,最大允许 {max_bytes} 字节") return self.rfile.read(length) if length > 0 else b"" def read_json_body(self, max_bytes: int = 65536) -> dict[str, Any]: body = self.read_request_body(max_bytes) if not body: return {} data = json.loads(body.decode("utf-8")) if not isinstance(data, dict): raise ValueError("请求 JSON 必须是对象") return data def do_GET(self) -> None: effective_path = self.validate_path() if effective_path == "": return if not self.is_authorized(): if effective_path in ("/", "/index.html"): self.send_bytes(LOGIN_HTML.encode("utf-8"), "text/html; charset=utf-8") return else: self.send_json({"error": "Unauthorized"}, HTTPStatus.UNAUTHORIZED) return if effective_path in ("/", "/index.html"): self.send_bytes(INDEX_HTML.encode("utf-8"), "text/html; charset=utf-8") elif effective_path == "/api/nodes": global last_active_ping_time, last_active_latency, active_openvpn_node_id nodes = read_nodes() active_node = next((n for n in nodes if active_openvpn_node_id and n.get("id") == active_openvpn_node_id), None) for n in nodes: n["active"] = (active_openvpn_node_id and n.get("id") == active_openvpn_node_id) if active_node: ip = active_node.get("ip") or active_node.get("remote_host") if ip: now = time.time() if now - last_active_ping_time > 15.0: last_active_ping_time = now def bg_ping(ip_addr: str, port: int, fallback: int) -> None: global last_active_latency try: latency = vpn_utils.ping_latency_ms(ip_addr, port, fallback) if latency > 0: last_active_latency = latency except Exception: pass threading.Thread( target=bg_ping, args=(ip, parse_int(active_node.get("remote_port")), parse_int(active_node.get("ping"))), daemon=True ).start() if last_active_latency > 0: active_node["latency_ms"] = last_active_latency stripped_nodes = [] for n in nodes: stripped = n.copy() if "config_text" in stripped: del stripped["config_text"] stripped_nodes.append(stripped) self.send_json({"nodes": stripped_nodes, "state": get_state()}) elif effective_path == "/api/check_update": try: self.send_json(check_latest_release()) except Exception as exc: self.send_json( {"ok": False, "error": f"无法检查 GitHub 正式版更新: {exc}"}, HTTPStatus.BAD_GATEWAY, ) elif effective_path.startswith("/configs/"): filename = urllib.parse.unquote(effective_path.removeprefix("/configs/")) with lock: nodes = read_nodes() node = next((n for n in nodes if Path(n.get("config_file", "")).name == filename), None) if node and node.get("config_text"): self.send_bytes(node["config_text"].encode("utf-8"), "application/x-openvpn-profile") else: self.send_json({"error": "not found"}, HTTPStatus.NOT_FOUND) elif effective_path == "/api/gateway_status": web_ui_status = { "name": "Web 管理服务", "status": "running", "details": f"监听地址: {load_ui_config().get('host', UI_HOST)}:{load_ui_config().get('port', UI_PORT)}", "error": "" } proxy_ok = False proxy_err = "" is_ipv6 = ":" in LOCAL_PROXY_HOST af = socket.AF_INET6 if is_ipv6 else socket.AF_INET s = None try: s = socket.socket(af, socket.SOCK_STREAM) s.settimeout(0.5) connect_host = LOCAL_PROXY_HOST if connect_host in ("::", "0.0.0.0", ""): connect_host = "::1" if is_ipv6 else "127.0.0.1" try: s.connect((connect_host, LOCAL_PROXY_PORT)) proxy_ok = True except Exception: if connect_host == "::1": s.close() s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(0.5) s.connect(("127.0.0.1", LOCAL_PROXY_PORT)) proxy_ok = True else: raise except Exception as e: diag = vpn_utils.diagnose_local_obstructions(LOCAL_PROXY_PORT, host=LOCAL_PROXY_HOST) proxy_err = diag[1] if diag else f"本地代理网关无法连通: {e}" finally: if s is not None: try: s.close() except Exception: pass proxy_gateway_status = { "name": "本地代理网关", "status": "running" if proxy_ok else "stopped", "details": f"监听地址: {LOCAL_PROXY_HOST}:{LOCAL_PROXY_PORT}", "error": proxy_err } ovpn_ok = active_openvpn_running() ovpn_err = "" ovpn_details = "未连接" if ovpn_ok: ovpn_details = f"已连接节点: {active_openvpn_node_id}" if sys.platform.startswith("linux"): if not Path("/sys/class/net/tun0").exists(): ovpn_err = "[警告] 虚拟网卡 (tun0) 未启用,可能存在策略路由配置问题。" else: if active_openvpn_node_id: ovpn_err = "连接已中断或 OpenVPN 核心程序异常退出。" ovpn_details = f"尝试连接节点 {active_openvpn_node_id} 失败" openvpn_status = { "name": "OpenVPN 核心连接", "status": "running" if ovpn_ok else "stopped", "details": ovpn_details, "error": ovpn_err } now = time.time() server_uptime = now - server_start_time collector_ok = (last_collector_heartbeat > 0.0 and now - last_collector_heartbeat < (CHECK_INTERVAL_SECONDS * 1.5)) or (server_uptime < 15.0) collector_status = { "name": "节点同步守护线程", "status": "running" if collector_ok else "stopped", "details": f"上次心跳: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(last_collector_heartbeat)) if last_collector_heartbeat > 0 else '等待启动'}", "error": "" if collector_ok else "线程可能已异常终止,导致无法在后台拉取和测速新节点。" } checker_ok = (last_checker_heartbeat > 0.0 and now - last_checker_heartbeat < 90.0) or (server_uptime < 35.0) checker_status = { "name": "出口检测守护线程", "status": "running" if checker_ok else "stopped", "details": f"上次心跳: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(last_checker_heartbeat)) if last_checker_heartbeat > 0 else '等待启动'}", "error": "" if checker_ok else "线程可能已挂起或终止,导致无法实时获取代理出口状态。" } pinger_ok = (last_pinger_heartbeat > 0.0 and now - last_pinger_heartbeat < 30.0) or (server_uptime < 15.0) pinger_status = { "name": "延迟测速守护线程", "status": "running" if pinger_ok else "stopped", "details": f"上次心跳: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(last_pinger_heartbeat)) if last_pinger_heartbeat > 0 else '等待启动'}", "error": "" if pinger_ok else "线程可能已中止,无法实时刷新活动节点的 Ping 延迟。" } self.send_json({ "ok": True, "services": [ web_ui_status, proxy_gateway_status, openvpn_status, collector_status, checker_status, pinger_status ] }) elif effective_path == "/api/logs": logs_dir = DATA_DIR / "logs" date_str = time.strftime("%Y-%m-%d", time.localtime()) log_file = logs_dir / f"{date_str}.json" entries = [] if log_file.exists(): try: with lock: with open(log_file, "r", encoding="utf-8") as f: for line in f: line = line.strip() if line: try: entries.append(json.loads(line)) except Exception: pass except Exception as e: print(f"[API Logs] Error reading log file: {e}", flush=True) self.send_json({"logs": entries}) else: self.send_json({"error": "not found"}, HTTPStatus.NOT_FOUND) def do_POST(self) -> None: global is_connecting effective_path = self.validate_path() if effective_path == "": return if effective_path == "/api/login": try: payload = self.read_json_body() input_pwd = str(payload.get("password") or "") input_uname = str(payload.get("username") or "") ui_cfg = load_ui_config() expected_pwd = ui_cfg.get("password", "") expected_uname = ui_cfg.get("username", "admin") if expected_pwd and input_pwd == expected_pwd and input_uname == expected_uname: token = uuid.uuid4().hex with lock: active_sessions[token] = time.time() + 30 * 24 * 3600 body = json.dumps({"ok": True}).encode("utf-8") self.send_response(HTTPStatus.OK) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") secret_path = self.get_secret_path() cookie_path = f"/{secret_path}/" if secret_path else "/" self.send_header("Set-Cookie", f"session={token}; Path={cookie_path}; HttpOnly; SameSite=Lax; Max-Age=2592000") self.end_headers() self.wfile.write(body) else: self.send_json({"ok": False, "error": "用户名或密码不正确,请重新输入"}, HTTPStatus.FORBIDDEN) except Exception as exc: self.send_json({"ok": False, "error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) return if effective_path == "/api/logout": try: cookie_header = self.headers.get("Cookie", "") cookies = {} if cookie_header: for item in cookie_header.split(";"): item = item.strip() if "=" in item: k, v = item.split("=", 1) cookies[k.strip()] = v.strip() session_token = cookies.get("session") if session_token: with lock: active_sessions.pop(session_token, None) secret_path = self.get_secret_path() cookie_path = f"/{secret_path}/" if secret_path else "/" body = json.dumps({"ok": True}).encode("utf-8") self.send_response(HTTPStatus.OK) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.send_header("Set-Cookie", f"session=; Path={cookie_path}; HttpOnly; SameSite=Lax; Max-Age=0; Expires=Thu, 01 Jan 1970 00:00:00 GMT") self.end_headers() self.wfile.write(body) except Exception as exc: self.send_json({"ok": False, "error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) return if not self.is_authorized(): self.send_json({"error": "Unauthorized"}, HTTPStatus.UNAUTHORIZED) return if effective_path == "/api/update_credentials": try: payload = self.read_json_body() new_username = str(payload.get("username") or "").strip() new_password = str(payload.get("password") or "").strip() new_port = payload.get("port") new_suffix = str(payload.get("secret_path") or "").strip() ui_cfg = load_ui_config() if not new_username or (not new_password and not ui_cfg.get("password")): self.send_json({"ok": False, "error": "用户名不能为空;首次设置时密码不能为空"}, HTTPStatus.BAD_REQUEST) return try: new_port_int = int(new_port) if not (1 <= new_port_int <= 65535): raise ValueError() except (TypeError, ValueError): self.send_json({"ok": False, "error": "网页管理端口范围必须是 1 至 65535"}, HTTPStatus.BAD_REQUEST) return if not new_suffix or not re.match(r"^[A-Za-z0-9]+$", new_suffix): self.send_json({"ok": False, "error": "安全后缀仅能由英文字母和数字组成"}, HTTPStatus.BAD_REQUEST) return expected_username = ui_cfg.get("username", "") expected_password = ui_cfg.get("password", "") expected_port = ui_cfg.get("port", 8787) expected_suffix = ui_cfg.get("secret_path", "EJsW2EeBo9lY") ui_cfg["username"] = new_username if new_password: ui_cfg["password"] = new_password ui_cfg["port"] = new_port_int ui_cfg["secret_path"] = new_suffix auth_file = DATA_DIR / "ui_auth.json" reauth_required = new_username != expected_username or (new_password and new_password != expected_password) with lock: DATA_DIR.mkdir(exist_ok=True, parents=True) write_json(auth_file, ui_cfg) if reauth_required: active_sessions.clear() restart_needed = (new_port_int != expected_port or new_suffix != expected_suffix) if restart_needed: self.send_json({"ok": True, "restart_needed": True, "reauth_required": reauth_required, "message": "配置更新成功,网页管理端口或路径已变更,将在 2 秒内重启..."}) def restart_server(): time.sleep(2) print("[系统] 管理后台安全配置更新,进程即将退出以触发自动重启...", flush=True) os._exit(0) threading.Thread(target=restart_server, daemon=True).start() else: self.send_json({"ok": True, "restart_needed": False, "reauth_required": reauth_required, "message": "账号密码配置更新成功,已即时生效!"}) except Exception as exc: self.send_json({"ok": False, "error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) return elif effective_path == "/api/update_settings": try: payload = self.read_json_body() new_proxy_port = payload.get("proxy_port") routing_mode = str(payload.get("routing_mode") or "auto").strip() force_country = str(payload.get("force_country") or "").strip() routing_ip_type = str(payload.get("routing_ip_type") or "all").strip() try: new_proxy_port_int = int(new_proxy_port) if not (1024 <= new_proxy_port_int <= 65535): raise ValueError() except (TypeError, ValueError): self.send_json({"ok": False, "error": "代理出站端口范围必须是 1024 至 65535"}, HTTPStatus.BAD_REQUEST) return if routing_mode not in ("auto", "fixed_ip", "fixed_region", "favorites"): self.send_json({"ok": False, "error": "无效的路由配置模式"}, HTTPStatus.BAD_REQUEST) return if routing_mode == "fixed_region" and not force_country: self.send_json({"ok": False, "error": "启用固定地区前,请先选择一个要锁定的国家"}, HTTPStatus.BAD_REQUEST) return if routing_ip_type not in ("all", "residential", "hosting"): self.send_json({"ok": False, "error": "无效的IP出站类型过滤"}, HTTPStatus.BAD_REQUEST) return ui_cfg = load_ui_config() expected_proxy_port = ui_cfg.get("proxy_port", 7928) fixed_node_id = current_fixed_node_id(ui_cfg) if routing_mode == "fixed_ip" else "" if new_proxy_port_int == ui_cfg.get("port", 8787): self.send_json({"ok": False, "error": "代理出站端口不能与网页管理端口相同"}, HTTPStatus.BAD_REQUEST) return if routing_mode == "fixed_ip" and not fixed_node_id: self.send_json({"ok": False, "error": "启用固定 IP 前,请先连接一个要锁定的节点"}, HTTPStatus.BAD_REQUEST) return ui_cfg["proxy_port"] = new_proxy_port_int ui_cfg["routing_mode"] = routing_mode ui_cfg["force_country"] = force_country ui_cfg["routing_ip_type"] = routing_ip_type if routing_mode == "favorites": ui_cfg["fav_fail_fallback"] = False if routing_mode == "fixed_ip": ui_cfg["fixed_node_id"] = fixed_node_id auth_file = DATA_DIR / "ui_auth.json" with lock: DATA_DIR.mkdir(exist_ok=True, parents=True) write_json(auth_file, ui_cfg) policy_message = enforce_active_node_allowed_by_routing(ui_cfg, "路由设置已更新") restart_needed = (new_proxy_port_int != expected_proxy_port) if restart_needed: self.send_json({"ok": True, "restart_needed": True, "message": "配置更新成功,代理出站端口变更,将在 2 秒内重启..."}) def restart_server(): time.sleep(2) print("[系统] 代理出站端口变更,进程即将退出以触发自动重启...", flush=True) os._exit(0) threading.Thread(target=restart_server, daemon=True).start() else: message = policy_message or "配置更新成功,已即时生效!" self.send_json({"ok": True, "restart_needed": False, "message": message}) except Exception as exc: self.send_json({"ok": False, "error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) return elif effective_path == "/api/update_routing": try: payload = self.read_json_body() routing_mode = str(payload.get("routing_mode") or "auto").strip() force_country = str(payload.get("force_country") or "").strip() routing_ip_type = str(payload.get("routing_ip_type") or "all").strip() fav_fail_fallback = False if routing_mode not in ("auto", "fixed_ip", "fixed_region", "favorites"): self.send_json({"ok": False, "error": "无效的路由配置模式"}, HTTPStatus.BAD_REQUEST) return if routing_mode == "fixed_region" and not force_country: self.send_json({"ok": False, "error": "启用固定地区前,请先选择一个要锁定的国家"}, HTTPStatus.BAD_REQUEST) return if routing_ip_type not in ("all", "residential", "hosting"): self.send_json({"ok": False, "error": "无效的IP出站类型过滤"}, HTTPStatus.BAD_REQUEST) return ui_cfg = load_ui_config() fixed_node_id = current_fixed_node_id(ui_cfg) if routing_mode == "fixed_ip" else "" if routing_mode == "fixed_ip" and not fixed_node_id: self.send_json({"ok": False, "error": "启用固定 IP 前,请先连接一个要锁定的节点"}, HTTPStatus.BAD_REQUEST) return ui_cfg["routing_mode"] = routing_mode ui_cfg["force_country"] = force_country ui_cfg["routing_ip_type"] = routing_ip_type ui_cfg["fav_fail_fallback"] = fav_fail_fallback if routing_mode == "fixed_ip": ui_cfg["fixed_node_id"] = fixed_node_id ui_cfg.pop("enable_force_country", None) auth_file = DATA_DIR / "ui_auth.json" with lock: DATA_DIR.mkdir(exist_ok=True, parents=True) write_json(auth_file, ui_cfg) policy_message = enforce_active_node_allowed_by_routing(ui_cfg, "出站路由配置已更新") self.send_json({"ok": True, "message": policy_message or "出站路由配置更新成功,已即时生效!"}) except Exception as exc: self.send_json({"ok": False, "error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) return elif effective_path == "/api/toggle_favorite": try: payload = self.read_json_body() node_id = str(payload.get("id") or "").strip() if not node_id: self.send_json({"ok": False, "error": "节点 ID 不能为空"}, HTTPStatus.BAD_REQUEST) return ui_cfg = load_ui_config() fav_ids = ui_cfg.get("favorite_node_ids", []) if not isinstance(fav_ids, list): fav_ids = [] if node_id in fav_ids: fav_ids.remove(node_id) else: fav_ids.append(node_id) ui_cfg["favorite_node_ids"] = fav_ids auth_file = DATA_DIR / "ui_auth.json" with lock: DATA_DIR.mkdir(exist_ok=True, parents=True) write_json(auth_file, ui_cfg) policy_message = None if ui_cfg.get("routing_mode") == "favorites": policy_message = enforce_active_node_allowed_by_routing(ui_cfg, "收藏列表已更新") self.send_json({"ok": True, "favorite_node_ids": fav_ids, "message": policy_message or ""}) except Exception as exc: self.send_json({"ok": False, "error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) return if effective_path == "/api/check": try: self.send_json({"ok": True, "message": maintain_valid_nodes(force=True)}) except Exception as exc: self.send_json({"ok": False, "error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) elif effective_path == "/api/refresh_nodes": try: payload = self.read_json_body() if "discovery_countries" in payload: discovery_countries = persist_discovery_countries( payload.get("discovery_countries") ) else: discovery_countries = normalize_discovery_countries( load_ui_config().get("discovery_countries") ) if maintenance_lock.locked(): self.send_json({ "ok": True, "message": "节点维护任务正在运行,国家范围已保存并将在下一轮生效", "running": True, "discovery_countries": discovery_countries, }) else: threading.Thread(target=maintain_valid_nodes, args=(False,), daemon=True).start() self.send_json({ "ok": True, "message": "已在后台启动节点更新流程", "running": False, "discovery_countries": discovery_countries, }) except ValueError as exc: self.send_json({"ok": False, "error": str(exc)}, HTTPStatus.BAD_REQUEST) except Exception as exc: self.send_json({"ok": False, "error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) elif effective_path == "/api/test_nodes": try: payload = self.read_json_body(max_bytes=262144) node_ids = payload.get("ids", []) if not isinstance(node_ids, list): self.send_json({"ok": False, "error": "节点 ID 列表无效"}, HTTPStatus.BAD_REQUEST) return node_ids = [str(node_id or "").strip() for node_id in node_ids] node_ids = [node_id for node_id in node_ids if node_id] if len(node_ids) > MANUAL_TEST_NODE_LIMIT: self.send_json({"ok": False, "error": f"单次最多测试 {MANUAL_TEST_NODE_LIMIT} 个节点"}, HTTPStatus.BAD_REQUEST) return if not maintenance_lock.acquire(blocking=False): self.send_json({"ok": False, "error": "当前已有连接或节点维护任务正在运行,请稍后再试"}, HTTPStatus.CONFLICT) return with lock: if is_connecting: maintenance_lock.release() self.send_json({"ok": False, "error": "当前已有连接或节点维护任务正在运行,请稍后再试"}, HTTPStatus.CONFLICT) return is_connecting = True try: set_state(is_connecting=True, last_check_message="正在手动测试节点可用性...") tested_nodes = test_multiple_nodes(node_ids) self.send_json({"ok": True, "nodes": tested_nodes}) finally: with lock: is_connecting = False set_state(is_connecting=False) maintenance_lock.release() except Exception as exc: self.send_json({"ok": False, "error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) elif effective_path == "/api/disconnect": try: cancel_background_refill() cancel_pending_connection_attempt() ui_cfg = load_ui_config() ui_cfg["connection_enabled"] = False auth_file = DATA_DIR / "ui_auth.json" with lock: DATA_DIR.mkdir(exist_ok=True, parents=True) write_json(auth_file, ui_cfg) stop_active_openvpn() with lock: nodes = read_nodes() for item in nodes: item["active"] = False write_json(NODES_FILE, nodes) global last_active_ping_time, last_active_latency last_active_ping_time = 0.0 last_active_latency = 0 global consecutive_proxy_failures, last_proxy_failure_node_id consecutive_proxy_failures = 0 last_proxy_failure_node_id = "" set_state( active_openvpn_node_id="", pending_node_id="", last_check_message="手动断开连接", active_node_latency="无活动连接", proxy_ok=False, proxy_ip="-", proxy_latency_ms=0, proxy_error="连接已手动断开", ) self.send_json({"ok": True}) except Exception as exc: self.send_json({"ok": False, "error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) elif effective_path == "/api/connect": previous_node_id = active_openvpn_node_id if active_openvpn_running() else "" try: payload = self.read_json_body() self.send_json({"ok": True, "message": connect_node(str(payload.get("id") or ""))}) except ConnectionCancelled as exc: self.send_json({"ok": False, "cancelled": True, "error": str(exc)}, HTTPStatus.CONFLICT) except RuntimeError as exc: if str(exc).startswith("当前已有"): self.send_json({"ok": False, "error": str(exc)}, HTTPStatus.CONFLICT) return threading.Thread( target=recover_after_manual_connect_failure, args=(previous_node_id,), daemon=True, ).start() self.send_json({"ok": False, "error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) except Exception as exc: threading.Thread( target=recover_after_manual_connect_failure, args=(previous_node_id,), daemon=True, ).start() self.send_json({"ok": False, "error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) elif effective_path == "/api/test_node": try: payload = self.read_json_body() node_id = str(payload.get("id") or "") if not node_id.strip(): self.send_json({"ok": False, "error": "节点 ID 不能为空"}, HTTPStatus.BAD_REQUEST) return if not maintenance_lock.acquire(blocking=False): self.send_json({"ok": False, "error": "当前已有连接或节点维护任务正在运行,请稍后再试"}, HTTPStatus.CONFLICT) return with lock: if is_connecting: maintenance_lock.release() self.send_json({"ok": False, "error": "当前已有连接或节点维护任务正在运行,请稍后再试"}, HTTPStatus.CONFLICT) return is_connecting = True try: set_state(is_connecting=True, last_check_message="正在手动测试节点可用性...") updated_node = test_node_by_id(node_id) self.send_json({"ok": True, "node": updated_node}) finally: with lock: is_connecting = False set_state(is_connecting=False) maintenance_lock.release() except Exception as exc: self.send_json({"ok": False, "error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) elif effective_path == "/api/test_proxy": try: self.read_request_body() result = check_proxy_health() if result["ok"]: set_state( proxy_ok=True, proxy_ip=result["ip"], proxy_latency_ms=result["latency_ms"], proxy_error="" ) else: set_state( proxy_ok=False, proxy_ip="-", proxy_latency_ms=0, proxy_error=result.get("error", "未知错误") ) self.send_json(result) except Exception as exc: self.send_json({"ok": False, "error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) else: self.send_json({"error": "not found"}, HTTPStatus.NOT_FOUND) class Tee: def __init__(self, file_path: str): Path(file_path).parent.mkdir(exist_ok=True, parents=True) self.file = open(file_path, "a", encoding="utf-8") self.stdout = sys.stdout def write(self, data: str) -> None: self.stdout.write(data) self.file.write(data) self.file.flush() def flush(self) -> None: self.stdout.flush() self.file.flush() def isatty(self) -> bool: return self.stdout.isatty() def __getattr__(self, attr: str) -> Any: return getattr(self.stdout, attr) def main() -> None: ensure_dirs() kill_existing_openvpn_processes() log_file = DATA_DIR / "vpngate.log" tee = Tee(str(log_file)) sys.stdout = tee sys.stderr = tee write_json( STATE_FILE, { "api_url": API_URL, "mirror_url": MIRROR_HTTPS_URL, "target_valid_nodes": TARGET_VALID_NODES, "fetch_interval_seconds": FETCH_INTERVAL_SECONDS, "check_interval_seconds": CHECK_INTERVAL_SECONDS, "local_proxy": f"http://{'[' + LOCAL_PROXY_HOST + ']' if ':' in LOCAL_PROXY_HOST else LOCAL_PROXY_HOST}:{LOCAL_PROXY_PORT}", "active_openvpn_node_id": "", "last_fetch_status": "starting", "last_fetch_source": "", "last_check_message": "服务已启动,正在初始化网络并获取候选 VPN 节点...", "is_connecting": True, "pending_node_id": "", "active_node_latency": "正在准备", "blacklisted_nodes": 0, }, ) threading.Thread(target=proxy_server.start_proxy_server, args=(LOCAL_PROXY_HOST, LOCAL_PROXY_PORT), daemon=True).start() # Wait for the gateway to officially start print("[网关] 正在启动代理网关...", flush=True) gateway_ready = False is_ipv6 = ":" in LOCAL_PROXY_HOST af = socket.AF_INET6 if is_ipv6 else socket.AF_INET for _ in range(30): s = None try: s = socket.socket(af, socket.SOCK_STREAM) s.settimeout(0.5) connect_host = LOCAL_PROXY_HOST if connect_host in ("::", "0.0.0.0", ""): connect_host = "::1" if is_ipv6 else "127.0.0.1" try: s.connect((connect_host, LOCAL_PROXY_PORT)) gateway_ready = True break except Exception: if connect_host == "::1": try: s.close() s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.settimeout(0.5) s.connect(("127.0.0.1", LOCAL_PROXY_PORT)) gateway_ready = True break except Exception: pass raise except Exception: time.sleep(0.5) finally: if s is not None: try: s.close() except Exception: pass if gateway_ready: print("[网关] 代理网关已成功启动监听,启动同步与检测脚本...", flush=True) else: print("[警告] 代理网关启动超时,继续执行脚本...", flush=True) threading.Thread(target=collector_loop, daemon=True).start() threading.Thread(target=background_proxy_checker, daemon=True).start() threading.Thread(target=active_node_pinger, daemon=True).start() ui_cfg = load_ui_config() ui_host = ui_cfg.get("host", UI_HOST) ui_port = bounded_int(ui_cfg.get("port"), UI_PORT, 1, 65535) print(f"UI: http://{ui_host}:{ui_port}/", flush=True) print(f"Proxy: http://{LOCAL_PROXY_HOST}:{LOCAL_PROXY_PORT}", flush=True) DualStackHTTPServer((ui_host, ui_port), Handler).serve_forever() if __name__ == "__main__": main()