diff --git a/install.sh b/install.sh index 6465731..78b46ab 100644 --- a/install.sh +++ b/install.sh @@ -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}" diff --git a/proxy_server.py b/proxy_server.py index e54a34c..f3aba03 100644 --- a/proxy_server.py +++ b/proxy_server.py @@ -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: diff --git a/vpn_utils.py b/vpn_utils.py index b16e5a6..56bd06c 100644 --- a/vpn_utils.py +++ b/vpn_utils.py @@ -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}' 检查占用进程。" diff --git a/vpngate_manager.py b/vpngate_manager.py index d688471..e817cbb 100644 --- a/vpngate_manager.py +++ b/vpngate_manager.py @@ -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"""
-

本地代理出口检测 (Port 7928)

+

本地代理出口检测

测试本地 HTTP/SOCKS5 代理是否成功通过当前 VPN 节点出站，并获取实际出口公网 IP 和延迟。

@@ -2696,7 +2741,8 @@ function render(){ const statusMessage = state.last_check_message || ""; const activeNodeInfo = activeNode ? `${esc(translateCountry(activeNode.country))} (${activeNode.id})` : `无`; - $("status").innerHTML=`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=`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() \ No newline at end of file