GithubBackup/aimili-vpngate
1
0
Fork 0
mirror of https://github.com/baoweise-bot/aimili-vpngate.git synced 2026-09-06 16:16:53 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: enable full IPv6/dual-stack support for networking, proxy binding, and web management interface

Browse Source
This commit is contained in:
baoweise-bot
2026-05-30 14:47:37 +08:00
parent 46db7a1836
commit 28c32c59d2
4 changed files with 251 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
install.sh
+78 -33
View File
@@ -207,7 +207,7 @@ def generate_random_suffix():
def load_ui_cfg():
import json
path = "/opt/aimilivpn/vpngate_data/ui_auth.json"
cfg = {"host": "0.0.0.0", "port": 8787, "secret_path": "EJsW2EeBo9lY", "password": ""}
cfg = {"host": "::", "port": 8787, "secret_path": "EJsW2EeBo9lY", "password": ""}
if os.path.exists(path):
try:
with open(path, "r", encoding="utf-8") as f:
@@ -295,31 +295,35 @@ def get_public_ip():
except Exception:
pass
import urllib.request
try:
req = urllib.request.Request("https://api.ipify.org", headers={"User-Agent": "curl/7.68.0"})
with urllib.request.urlopen(req, timeout=1.5) as r:
ip = r.read().decode().strip()
if ip:
try:
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
f.write(ip)
except Exception:
pass
return ip
except Exception:
pass
# Try dual-stack first, then IPv6-only, then IPv4-only
for api_url in ["https://api64.ipify.org", "https://api6.ipify.org", "https://api.ipify.org"]:
try:
req = urllib.request.Request(api_url, headers={"User-Agent": "curl/7.68.0"})
with urllib.request.urlopen(req, timeout=2) as r:
ip = r.read().decode().strip()
if ip:
try:
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w", encoding="utf-8") as f:
f.write(ip)
except Exception:
pass
return ip
except Exception:
pass
return "您的服务器公网IP"
def check_port_listening(port):
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(0.2)
try:
s.connect(("127.0.0.1", port))
s.close()
return True
except Exception:
return False
for host, family in [("127.0.0.1", socket.AF_INET), ("::1", socket.AF_INET6)]:
try:
s = socket.socket(family, socket.SOCK_STREAM)
s.settimeout(0.2)
s.connect((host, port))
s.close()
return True
except Exception:
pass
return False
def get_service_pid(service_name="aimilivpn.service"):
try:
@@ -413,7 +417,15 @@ def print_status():
print_line(format_line(f"管理后台 (Port {ui_port})", backend_status))
print_line(format_line("连接核心 (OpenVPN)", openvpn_status))
login_ip = "127.0.0.1" if cfg.get("host") == "127.0.0.1" else get_public_ip()
host_cfg = cfg.get("host", "::")
if host_cfg in ("127.0.0.1", "localhost"):
login_ip = "127.0.0.1"
elif host_cfg == "::1":
login_ip = "[::1]"
elif host_cfg == "::":
login_ip = get_public_ip()
else:
login_ip = f"[{host_cfg}]" if ":" in host_cfg else host_cfg
print_line(format_line("网页登录地址", f"{yellow}http://{login_ip}:{ui_port}/{secret_path}/{reset}"))
print_line(format_line("网页管理账号", cfg.get("username", "未配置")))
curr_pwd = cfg.get("password", "")
@@ -441,9 +453,26 @@ def print_status():
else:
print_line(format_line("节点状态", "无活动连接"))
print_line()
local_proxy = state.get("local_proxy", "http://127.0.0.1:7928")
import urllib.parse
try:
parsed = urllib.parse.urlsplit(local_proxy)
proxy_host = parsed.hostname or "127.0.0.1"
proxy_port = parsed.port or 7928
except Exception:
proxy_host = "127.0.0.1"
proxy_port = 7928
if proxy_host == "::":
socks_addr = "127.0.0.1"
elif ":" in proxy_host:
socks_addr = f"[{proxy_host}]"
else:
socks_addr = proxy_host
print_line("【使用方法】")
print_line(f" export http_proxy=socks5://127.0.0.1:7928")
print_line(f" export https_proxy=socks5://127.0.0.1:7928")
print_line(f" export http_proxy=socks5://{socks_addr}:{proxy_port}")
print_line(f" export https_proxy=socks5://{socks_addr}:{proxy_port}")
print_line("=======================================================")
def run_service_cmd(cmd):
@@ -593,13 +622,19 @@ def configure_web():
if key == '1':
print("\033[H\033[J", end="")
print("选择网页登录绑定地址：")
print(" 1. 仅允许本地登录 (127.0.0.1 - 更安全)")
print(" 2. 允许公网IP登录 (0.0.0.0 - 方便远程)")
sel = input("请选择 (1 或 2, 默认2): ").strip()
print(" 1. 仅允许本地 IPv4 登录 (127.0.0.1 - 更安全)")
print(" 2. 允许 IPv4 公网登录 (0.0.0.0)")
print(" 3. 允许 IPv4 & IPv6 双栈公网登录 (:: - 推荐)")
print(" 4. 仅允许本地 IPv6 登录 (::1)")
sel = input("请选择 (1/2/3/4, 默认3): ").strip()
if sel == '1':
cfg['host'] = "127.0.0.1"
else:
elif sel == '2':
cfg['host'] = "0.0.0.0"
elif sel == '4':
cfg['host'] = "::1"
else:
cfg['host'] = "::"
save_ui_cfg(cfg)
print(f"绑定地址已更新为: {cfg['host']}")
ask_restart()
@@ -611,7 +646,10 @@ def configure_web():
save_ui_cfg(cfg)
print("安全登录后缀已随机重置成功！")
print(f"您的全新安全登录后缀为: {new_path}")
print(f"新的访问路径为: http://{cfg['host']}:{cfg['port']}/{new_path}/")
display_host = cfg['host']
if ":" in display_host:
display_host = f"[{display_host}]"
print(f"新的访问路径为: http://{display_host}:{cfg['port']}/{new_path}/")
ask_restart()
break
elif key == '3' or key == 'q' or key == '\x03':
@@ -970,7 +1008,7 @@ while True:
python3 -c "
import json
cfg = {
'host': '0.0.0.0',
'host': '::',
'port': int('$UI_PORT'),
'secret_path': '$SECRET_PATH',
'username': '$UI_USERNAME',
@@ -1041,13 +1079,20 @@ echo -e "正在获取 VPS 公网 IP..."
PUBLIC_IP=$(curl -s --max-time 3 https://api.ipify.org || curl -s --max-time 3 https://ifconfig.me || curl -s --max-time 3 icanhazip.com || echo "您的服务器公网IP")
echo -n "$PUBLIC_IP" > "${INSTALL_DIR}/vpngate_data/public_ip.txt"
# Get VPS public IPv6
echo -e "正在获取 VPS 公网 IPv6..."
PUBLIC_IPV6=$(curl -6 -s --max-time 3 https://api.ipify.org || curl -6 -s --max-time 3 https://ifconfig.me || curl -6 -s --max-time 3 icanhazip.com || echo "")
echo -e "\n${GREEN}==========================================================${PLAIN}"
echo -e "${GREEN} AimiliVPN 源码一键部署已完成！${PLAIN}"
echo -e "${GREEN}==========================================================${PLAIN}"
echo -e " * 网页控制面板: ${BLUE}http://${PUBLIC_IP}:${UI_PORT}/${SECRET_PATH}/${PLAIN}"
if [ -n "$PUBLIC_IPV6" ]; then
echo -e " * 网页控制面板(IPv6): ${BLUE}http://[${PUBLIC_IPV6}]:${UI_PORT}/${SECRET_PATH}/${PLAIN}"
fi
echo -e " * 网页管理账号: ${YELLOW}${USERNAME}${PLAIN}"
echo -e " * 网页管理密码: ${YELLOW}${PASSWORD}${PLAIN}"
echo -e " * HTTP/SOCKS5 代理端口: ${BLUE}http://127.0.0.1:7928/${PLAIN}"
echo -e " * HTTP/SOCKS5 代理端口: ${BLUE}http://127.0.0.1:7928/${PLAIN} 或 ${BLUE}http://[::1]:7928/${PLAIN}"
echo -e " --------------------------------------------------------"
echo -e " * 快速状态指令: ${YELLOW}ml status${PLAIN} 或 ${YELLOW}ml${PLAIN}"
echo -e " * 查看实时日志: ${YELLOW}ml logs${PLAIN}"
proxy_server.py
+47 -6
View File
@@ -28,6 +28,11 @@ def resolve_dns_over_tun0(host: str, dns_server: str = "8.8.8.8", timeout: float
return host
except OSError:
pass
try:
socket.inet_pton(socket.AF_INET6, host)
return host
except OSError:
pass
import random
tx_id = random.getrandbits(16).to_bytes(2, "big")
@@ -261,18 +266,54 @@ def proxy_client(client: socket.socket, address: tuple[str, int]) -> None:
pass
def start_proxy_server(host: str, port: int) -> None:
is_ipv6 = ":" in host or host == ""
af = socket.AF_INET6 if is_ipv6 else socket.AF_INET
try:
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server = socket.socket(af, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
if is_ipv6:
try:
server.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0)
except OSError:
pass
server.bind((host, port))
server.listen(256)
print(f"HTTP/SOCKS5 proxy listening on {host}:{port}", flush=True)
except Exception as e:
import vpn_utils
diag = vpn_utils.diagnose_local_obstructions(port)
diag_msg = diag[1] if diag else str(e)
print(f"[ERROR] Failed to start HTTP/SOCKS5 proxy on {host}:{port}: {diag_msg}", flush=True)
return
if is_ipv6 and host == "::":
print(f"[警告] 绑定 IPv6 {host}:{port} 失败 ({e})，正在尝试回退至 IPv4 0.0.0.0 ...", flush=True)
try:
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(("0.0.0.0", port))
server.listen(256)
print(f"HTTP/SOCKS5 proxy listening on 0.0.0.0:{port} (仅 IPv4)", flush=True)
except Exception as ex:
import vpn_utils
diag = vpn_utils.diagnose_local_obstructions(port, host="0.0.0.0")
diag_msg = diag[1] if diag else str(ex)
print(f"[ERROR] Failed to start HTTP/SOCKS5 proxy on 0.0.0.0:{port}: {diag_msg}", flush=True)
return
elif is_ipv6 and host == "::1":
print(f"[警告] 绑定 IPv6 {host}:{port} 失败 ({e})，正在尝试回退至 IPv4 127.0.0.1 ...", flush=True)
try:
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(("127.0.0.1", port))
server.listen(256)
print(f"HTTP/SOCKS5 proxy listening on 127.0.0.1:{port} (仅 IPv4)", flush=True)
except Exception as ex:
import vpn_utils
diag = vpn_utils.diagnose_local_obstructions(port, host="127.0.0.1")
diag_msg = diag[1] if diag else str(ex)
print(f"[ERROR] Failed to start HTTP/SOCKS5 proxy on 127.0.0.1:{port}: {diag_msg}", flush=True)
return
else:
import vpn_utils
diag = vpn_utils.diagnose_local_obstructions(port, host=host)
diag_msg = diag[1] if diag else str(e)
print(f"[ERROR] Failed to start HTTP/SOCKS5 proxy on {host}:{port}: {diag_msg}", flush=True)
return
while True:
try:
vpn_utils.py
+40 -18
View File
@@ -194,7 +194,9 @@ def get_physical_interface() -> str | None:
def tcp_latency_ms(host: str, port: int, dev: str | None = None) -> int:
started = time.time()
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# Auto-detect address family based on host address
af = socket.AF_INET6 if ":" in host else socket.AF_INET
s = socket.socket(af, socket.SOCK_STREAM)
try:
s.settimeout(5)
if dev:
@@ -265,21 +267,29 @@ def ping_latency_ms(host: str, port: int, fallback_ping: int = 0) -> int:
def check_and_fix_dns() -> None:
"""
Checks if DNS resolution is broken in WSL.
Checks if DNS resolution is broken.
If names fail but direct IP connections work, appends public DNS nameservers to /etc/resolv.conf.
Supports both IPv4 and IPv6 network environments.
"""
try:
socket.gethostbyname("www.vpngate.net")
socket.getaddrinfo("www.vpngate.net", 443)
return
except socket.gaierror:
except (socket.gaierror, OSError):
pass
network_ok = False
for ip in ["8.8.8.8", "1.1.1.1"]:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
# Test IPv4 DNS servers first, then IPv6
dns_targets = [
("8.8.8.8", 53, socket.AF_INET),
("1.1.1.1", 53, socket.AF_INET),
("2001:4860:4860::8888", 53, socket.AF_INET6),
("2606:4700:4700::1111", 53, socket.AF_INET6),
]
for ip, port, af in dns_targets:
s = socket.socket(af, socket.SOCK_DGRAM)
try:
s.settimeout(2)
s.connect((ip, 53))
s.connect((ip, port))
network_ok = True
break
except Exception:
@@ -435,7 +445,7 @@ def diagnose_api_failure(api_url: str = "https://www.vpngate.net/api/iphone/") -
dns_ok = False
for test_domain in ["api.ipify.org", "dns.google", "one.one.one.one"]:
try:
socket.gethostbyname(test_domain)
socket.getaddrinfo(test_domain, 443)
dns_ok = True
break
except Exception:
@@ -443,10 +453,12 @@ def diagnose_api_failure(api_url: str = "https://www.vpngate.net/api/iphone/") -
# 2. 检查是否能解析 API 域名
api_dns_ok = False
api_ip = None
api_addr = None # (af, ip) tuple
try:
api_ip = socket.gethostbyname(domain)
api_dns_ok = True
results = socket.getaddrinfo(domain, port, 0, socket.SOCK_STREAM)
if results:
api_dns_ok = True
api_addr = (results[0][0], results[0][4][0]) # (address_family, ip)
except Exception:
pass
@@ -458,7 +470,8 @@ def diagnose_api_failure(api_url: str = "https://www.vpngate.net/api/iphone/") -
# 3. 检查 TCP 连接 API 域名
api_conn_ok = False
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
api_af, api_ip = api_addr
s = socket.socket(api_af, socket.SOCK_STREAM)
s.settimeout(4)
try:
s.connect((api_ip, port))
@@ -473,8 +486,15 @@ def diagnose_api_failure(api_url: str = "https://www.vpngate.net/api/iphone/") -
if not api_conn_ok:
ext_conn_ok = False
for test_ip, test_port in [("8.8.8.8", 53), ("1.1.1.1", 53)]:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# Test both IPv4 and IPv6 external connectivity
ext_targets = [
("8.8.8.8", 53, socket.AF_INET),
("1.1.1.1", 53, socket.AF_INET),
("2001:4860:4860::8888", 53, socket.AF_INET6),
("2606:4700:4700::1111", 53, socket.AF_INET6),
]
for test_ip, test_port, af in ext_targets:
s = socket.socket(af, socket.SOCK_STREAM)
s.settimeout(3)
try:
s.connect((test_ip, test_port))
@@ -490,7 +510,7 @@ def diagnose_api_failure(api_url: str = "https://www.vpngate.net/api/iphone/") -
if ext_conn_ok:
return 1008, f"[ERR_API_IP_BLOCKED_OR_DOWN] 连接 API 服务器失败。原因: 外部网络连接通畅，但无法建立到 {domain} ({api_ip}:{port}) 的连接，可能是由于官方 IP 遭 GFW/防火墙 IP 阻断封锁或官方服务器宕机。"
else:
return 1009, "[ERR_VPS_OUTBOUND_BLOCKED] VPS 完全断网。原因: 任何外部测试连接均失败，请检查 VPS 网卡和宿主机连接。"
return 1009, "[ERR_VPS_OUTBOUND_BLOCKED] VPS 完全断网。原因: 任何外部测试连接均失败（IPv4 和 IPv6 均不可达），请检查 VPS 网卡和宿主机连接。"
return 1010, f"[ERR_API_TLS_INTERFERENCE] HTTPS/TLS 握手被干扰。原因: 可以建立 TCP 连接但请求超时，通常是由于防火墙通过 SNI 阻断了 TLS 握手流。"
@@ -524,13 +544,15 @@ def diagnose_openvpn_failure(log_tail: list[str]) -> tuple[int, str]:
return 2010, "[ERR_OVPN_UNKNOWN] OpenVPN 其他运行时异常。原因: 连接握手期间发生其他协议错误，详细信息请查看日志尾部。"
def diagnose_local_obstructions(proxy_port: int = 7928) -> tuple[int, str] | None:
def diagnose_local_obstructions(proxy_port: int = 7928, host: str = "127.0.0.1") -> tuple[int, str] | None:
import sys
# 1. 检查端口是否被占用
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
is_ipv6 = ":" in host or host == ""
af = socket.AF_INET6 if is_ipv6 else socket.AF_INET
s = socket.socket(af, socket.SOCK_STREAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
s.bind(("127.0.0.1", proxy_port))
s.bind((host, proxy_port))
except OSError as e:
if e.errno == 98 or e.errno == 10048 or "already in use" in str(e).lower():
return 3005, f"[ERR_PORT_IN_USE] 本地代理端口 {proxy_port} 被占用。原因: 其他进程已抢占该端口，导致本系统代理网关启动失败。请运行 'lsof -i :{proxy_port}' 检查占用进程。"
vpngate_manager.py
+86 -20
View File
@@ -23,14 +23,57 @@ import concurrent.futures
import sys
import uuid
# Force socket to resolve IPv4 only to avoid slow AAAA (IPv6) DNS resolution timeouts (e.g. in WSL)
# Prefer IPv4 resolution to avoid slow AAAA DNS timeouts (e.g. in WSL),
# but fall back to system default (IPv6) if IPv4 resolution fails.
# This ensures pure-IPv6 VPS (with NAT64/clatd) can still function.
_orig_getaddrinfo = socket.getaddrinfo
def _ipv4_getaddrinfo(host, port, family=0, type=0, proto=0, flags=0):
if family == 0:
family = socket.AF_INET
if isinstance(host, str) and ":" in host:
return _orig_getaddrinfo(host, port, socket.AF_INET6, type, proto, flags)
# Try IPv4 first for speed; fall back to system default (allows IPv6/NAT64)
try:
results = _orig_getaddrinfo(host, port, socket.AF_INET, type, proto, flags)
if results:
return results
except socket.gaierror:
pass
return _orig_getaddrinfo(host, port, 0, type, proto, flags)
return _orig_getaddrinfo(host, port, family, type, proto, flags)
socket.getaddrinfo = _ipv4_getaddrinfo
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
@@ -43,9 +86,9 @@ OPENVPN_TEST_TIMEOUT_SECONDS = int(os.environ.get("OPENVPN_TEST_TIMEOUT_SECONDS"
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_HOST = os.environ.get("LOCAL_PROXY_HOST", "::")
LOCAL_PROXY_PORT = int(os.environ.get("LOCAL_PROXY_PORT", "7928"))
UI_HOST = os.environ.get("UI_HOST", "0.0.0.0")
UI_HOST = os.environ.get("UI_HOST", "::")
UI_PORT = int(os.environ.get("UI_PORT", "8787"))
INVALID_BACKOFF_SECONDS = int(os.environ.get("INVALID_BACKOFF_SECONDS", str(30 * 60)))
@@ -122,7 +165,7 @@ def load_ui_config() -> dict[str, Any]:
"username": "",
"secret_path": "EJsW2EeBo9lY",
"password": "",
"host": "0.0.0.0",
"host": "::",
"port": 8787
}
updated = False
@@ -214,7 +257,8 @@ def get_state() -> dict[str, Any]:
state.setdefault("target_valid_nodes", TARGET_VALID_NODES)
state.setdefault("fetch_interval_seconds", FETCH_INTERVAL_SECONDS)
state.setdefault("check_interval_seconds", CHECK_INTERVAL_SECONDS)
state.setdefault("local_proxy", f"http://{LOCAL_PROXY_HOST}:{LOCAL_PROXY_PORT}")
_proxy_display = f"[{LOCAL_PROXY_HOST}]" if ":" in LOCAL_PROXY_HOST else LOCAL_PROXY_HOST
state.setdefault("local_proxy", f"http://{_proxy_display}:{LOCAL_PROXY_PORT}")
state.setdefault("last_fetch_status", "not_started")
state.setdefault("last_check_message", "")
state.setdefault("blacklisted_nodes", 0)
@@ -933,7 +977,8 @@ def connect_node(node_id: str) -> str:
for item in nodes:
item["active"] = item.get("id") == node_id
if item["active"]:
item["probe_message"] = f"Active node. HTTP proxy: http://{LOCAL_PROXY_HOST}:{LOCAL_PROXY_PORT}"
_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, nodes)
set_state(last_check_message="正在测试本地代理出站联通性与出口 IP...")
@@ -2281,7 +2326,7 @@ INDEX_HTML = r"""<!doctype html>
<svg xmlns="http://www.w3.org/2000/svg" class="stat-icon" fill="none" viewBox="0 0 24 24" stroke="currentColor" stroke-width="2" style="color: var(--primary);"><path stroke-linecap="round" stroke-linejoin="round" d="M8.111 16.404a5.5 5.5 0 017.778 0M12 20h.01m-7.08-7.071a10.5 10.5 0 0114.14 0M1.414 8.05a16 16 0 0121.172 0" /></svg>
</div>
<div>
<h3 style="margin: 0 0 4px 0; font-size: 16px; font-weight: 600; color: var(--text-primary);">本地代理出口检测 (Port 7928)</h3>
<h3 style="margin: 0 0 4px 0; font-size: 16px; font-weight: 600; color: var(--text-primary);">本地代理出口检测</h3>
<p style="margin: 0; font-size: 13px; color: var(--text-secondary);">
测试本地 HTTP/SOCKS5 代理是否成功通过当前 VPN 节点出站，并获取实际出口公网 IP 和延迟。
</p>
@@ -2696,7 +2741,8 @@ function render(){
const statusMessage = state.last_check_message || "";
const activeNodeInfo = activeNode ? `<span class="badge available" style="margin-left:8px; padding:2px 8px;">${esc(translateCountry(activeNode.country))} (${activeNode.id})</span>` : `<span class="badge unavailable" style="margin-left:8px; padding:2px 8px;">无</span>`;
$("status").innerHTML=`<span class="status-dot"></span>HTTP 代理本地接口：http://127.0.0.1:7928 | 活动节点：${activeNodeInfo} | 状态：${statusMessage}`;
const localProxy = state.local_proxy || `http://127.0.0.1:${state.proxy_port || 7928}`;
$("status").innerHTML=`<span class="status-dot"></span>HTTP 代理本地接口：${localProxy} | 活动节点：${activeNodeInfo} | 状态：${statusMessage}`;
// Update proxy test status card based on background checks
const pBadge = $("proxy_status_badge");
@@ -3318,12 +3364,17 @@ setInterval(async () => {
def check_proxy_health() -> dict[str, Any]:
# 1. 检测代理服务端口是否在监听
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
is_ipv6 = ":" in LOCAL_PROXY_HOST
af = socket.AF_INET6 if is_ipv6 else socket.AF_INET
s = socket.socket(af, socket.SOCK_STREAM)
s.settimeout(1.5)
try:
s.connect(("127.0.0.1", LOCAL_PROXY_PORT))
connect_host = LOCAL_PROXY_HOST
if connect_host in ("::", "0.0.0.0", ""):
connect_host = "::1" if is_ipv6 else "127.0.0.1"
s.connect((connect_host, LOCAL_PROXY_PORT))
except Exception as e:
diag = vpn_utils.diagnose_local_obstructions(LOCAL_PROXY_PORT)
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,
@@ -3345,10 +3396,20 @@ def check_proxy_health() -> dict[str, Any]:
# 3. 使用 curl 通过本地 SOCKS5 代理接口测试 IP 与实际延迟
def _curl_check_ip(url: str) -> dict[str, Any] | None:
proxy_host = LOCAL_PROXY_HOST
if proxy_host == "::":
proxy_url = f"socks5h://[::1]:{LOCAL_PROXY_PORT}"
elif proxy_host == "0.0.0.0":
proxy_url = f"socks5h://127.0.0.1:{LOCAL_PROXY_PORT}"
elif ":" in proxy_host:
proxy_url = f"socks5h://[{proxy_host}]:{LOCAL_PROXY_PORT}"
else:
proxy_url = f"socks5h://{proxy_host}:{LOCAL_PROXY_PORT}"
cmd = [
"curl", "-4", "-s",
"curl", "-s",
"-w", "\n%{time_total} %{http_code}",
"-x", f"socks5h://127.0.0.1:{LOCAL_PROXY_PORT}",
"-x", proxy_url,
url,
"--max-time", "5"
]
@@ -3376,7 +3437,7 @@ def check_proxy_health() -> dict[str, Any]:
if result:
return result
diag = vpn_utils.diagnose_local_obstructions(LOCAL_PROXY_PORT)
diag = vpn_utils.diagnose_local_obstructions(LOCAL_PROXY_PORT, host=LOCAL_PROXY_HOST)
if diag:
return {"ok": False, "error": f"出口连接测试失败 | 本机诊断结果: {diag[1]}"}
@@ -3404,7 +3465,7 @@ def background_proxy_checker() -> None:
else:
error_msg = res.get("error", "未知错误")
if active_openvpn_node_id:
print(f"[警告] 7928 端口本地代理当前不可用！原因: {error_msg}", flush=True)
print(f"[警告] {LOCAL_PROXY_PORT} 端口本地代理当前不可用！原因: {error_msg}", flush=True)
log_to_json("WARNING", "Proxy", f"代理不可用: {error_msg}")
set_state(
proxy_ok=False,
@@ -3840,7 +3901,7 @@ def main() -> None:
"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}:{LOCAL_PROXY_PORT}",
"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_check_message": "服务已启动，正在初始化网络并获取候选 VPN 节点...",
@@ -3854,11 +3915,16 @@ def main() -> None:
# 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 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s = socket.socket(af, socket.SOCK_STREAM)
try:
s.settimeout(0.5)
s.connect((LOCAL_PROXY_HOST, LOCAL_PROXY_PORT))
connect_host = LOCAL_PROXY_HOST
if connect_host in ("::", "0.0.0.0", ""):
connect_host = "::1" if is_ipv6 else "127.0.0.1"
s.connect((connect_host, LOCAL_PROXY_PORT))
gateway_ready = True
break
except Exception:
@@ -3884,7 +3950,7 @@ def main() -> None:
print(f"UI: http://{ui_host}:{ui_port}/", flush=True)
print(f"Proxy: http://{LOCAL_PROXY_HOST}:{LOCAL_PROXY_PORT}", flush=True)
ThreadingHTTPServer((ui_host, ui_port), Handler).serve_forever()
DualStackHTTPServer((ui_host, ui_port), Handler).serve_forever()
if __name__ == "__main__":
main()