#!/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 - 安全登录
AimiliVPN
请输入您的管理账号和安全密码以继续
"""
INDEX_HTML = r"""
AimiliVPN 节点池管理系统
AimiliVPN 节点管理系统
服务加载中...
⭐ 收藏专属管理面板
在这里管理您的收藏节点过滤,以及设置出站连接漂移策略。
仅用收藏是强锁定模式。开启后只会连接收藏节点;如果收藏节点全部不可用,系统不会切换到非收藏节点。
| 状态 |
IP 地址 : 端口 |
延迟 |
物理位置 |
运营主体 / ISP |
IP 类型 |
操作 |
RNVPS (RackNerd) 推荐
超低折扣价格,性价比极高,日常使用实惠方便,海外多机房可选,非常适合普通大众用户。
点击进入官网
搬瓦工 (Bandwagon) 推荐
直连三网顶级专线,经典高带宽 CN2 GIA/9929 优化线路,极致速度且超凡稳定,高端用户首选。
点击进入官网
VPS购买推荐
本地代理出口检测
检测 HTTP/SOCKS5 代理出站连通性与 IP
今日运行日志
"""
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()