-
-
+
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
📭
-
Серверы не загружены
-
-
-
-
-
-
-
-
-
-
-
-
-
Данные для подключения
-
- HTTP / HTTPS
- ...
-
-
- SOCKS5
- ...
-
-
-
-
-
+
+
+
+
+
+
+
+
+
+
+
+
+ VPN_CLIENT
+
+
Secure Shell v4.2
+
+
+
+
+
+ Status
+
+
+
+ Traffic_Used
+ -- / --
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ Available_Endpoints
+ Click row to initiate link
+
+
+
+
+
+
+
+
+
+
+
Tunnel_Intelligence
+
+
+
+
+
+
+
+
+ NO_SIGNAL
+
+
+ Link not initialized
+
+
+
+
+
+
+
+ Speed
+
+
+
-- Mb/s
+
+
+
+
+
+
+
+
+
+
+ Connection_Logs
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/web/server.py b/web/server.py
index 0143eba..d9ac191 100644
--- a/web/server.py
+++ b/web/server.py
@@ -12,7 +12,9 @@ import socketserver
import urllib.request
import urllib.error
import uuid
-from urllib.parse import parse_qs
+import socket
+import time
+from urllib.parse import parse_qs, unquote
from pathlib import Path
PORT = 3456
@@ -50,12 +52,13 @@ def get_system_info() -> dict:
}
-def save_subscription(url: str, selected_server: str = None):
- """Save subscription URL and selected server to file"""
+def save_subscription(url: str, selected_server: str = None, user_info: dict = None):
+ """Save subscription URL, selected server and user info to file"""
DATA_DIR.mkdir(parents=True, exist_ok=True)
data = {
"url": url,
- "selectedServer": selected_server
+ "selectedServer": selected_server,
+ "userInfo": user_info
}
SUBSCRIPTION_FILE.write_text(json.dumps(data, ensure_ascii=False, indent=2))
@@ -70,10 +73,113 @@ def load_subscription() -> dict:
return None
+def measure_tcp_latency(host: str, port: int, timeout: float = 2.0) -> int:
+ """Measure TCP latency to a host:port in milliseconds"""
+ start_time = time.time()
+ try:
+ with socket.create_connection((host, port), timeout=timeout):
+ latency = (time.time() - start_time) * 1000
+ return int(latency)
+ except Exception:
+ return -1
+
+
+def measure_proxy_performance(enable_speed_test: bool = False) -> dict:
+ """Measure proxy latency, speed and public IP via local proxy"""
+ proxy_url = "http://127.0.0.1:8082"
+ proxies = {"http": proxy_url, "https": proxy_url}
+
+ # 1. Measure Latency (Ping)
+ latency = "Timeout"
+ try:
+ start_time = time.time()
+ # Use a reliable endpoint for ping
+ opener = urllib.request.build_opener(urllib.request.ProxyHandler(proxies))
+ req = urllib.request.Request("http://www.gstatic.com/generate_204", headers={"User-Agent": "singbox-test"})
+ with opener.open(req, timeout=5) as response:
+ lat_ms = int((time.time() - start_time) * 1000)
+ latency = f"{lat_ms}ms"
+ except Exception as e:
+ latency = "Error"
+
+ # 2. Get Public IP (IPv4)
+ ip = "Unknown"
+ try:
+ opener = urllib.request.build_opener(urllib.request.ProxyHandler(proxies))
+ # Use v4.ident.me to force IPv4
+ req = urllib.request.Request("http://v4.ident.me", headers={"User-Agent": "curl/7.68.0"})
+ with opener.open(req, timeout=5) as response:
+ ip = response.read().decode('utf-8').strip()
+ except Exception:
+ # Fallback to ipify if ident.me fails or returns garbage
+ try:
+ req = urllib.request.Request("http://api.ipify.org", headers={"User-Agent": "curl/7.68.0"})
+ with opener.open(req, timeout=5) as response:
+ ip = response.read().decode('utf-8').strip()
+ except Exception:
+ pass
+
+ # 3. Measure Download Speed
+ speed_mbps = 0.0
+ if enable_speed_test:
+ test_files = [
+ # Tele2 Speedtest (Usually very reliable and fast)
+ ("https://speedtest.selectel.ru/100MB", 100),
+ # ThinkBroadband (Reliable backup)
+ ("https://speedtest.selectel.ru/1GB", 1000)
+ ]
+
+ for url, size_mb in test_files:
+ try:
+ print(f"[WebUI] Testing speed with: {url}")
+ start_time = time.time()
+ opener = urllib.request.build_opener(urllib.request.ProxyHandler(proxies))
+ # Set a longer timeout for speed tests
+ with opener.open(url, timeout=30) as response:
+ downloaded = 0
+ # Larger chunk size for better throughput measurement
+ chunk_size = 1024 * 256 # 256KB chunks
+
+ # Download for at least 2 seconds or up to 25MB for accurate measurement
+ min_test_duration = 2.0 # seconds
+ max_download_bytes = 25 * 1024 * 1024 # 25MB
+
+ while True:
+ chunk = response.read(chunk_size)
+ if not chunk:
+ break
+ downloaded += len(chunk)
+
+ elapsed = time.time() - start_time
+ # Stop if we've downloaded enough AND tested for minimum duration
+ if downloaded >= max_download_bytes or (elapsed >= min_test_duration and downloaded >= 2 * 1024 * 1024):
+ break
+
+ duration = time.time() - start_time
+ if duration > 0.1 and downloaded > 0:
+ # Calculate speed in Mbps (megabits per second)
+ # downloaded bytes * 8 bits/byte / 1,000,000 / seconds
+ speed_mbps = round((downloaded * 8) / (1000 * 1000) / duration, 1)
+ print(f"[WebUI] Speed test: downloaded {downloaded / (1024*1024):.1f}MB in {duration:.1f}s = {speed_mbps} Mbps")
+ break # Stop if successful
+ except Exception as e:
+ print(f"[WebUI] Speed test failed for {url}: {e}")
+ continue
+
+ result = {
+ "latency": latency,
+ "ip": ip
+ }
+
+ if enable_speed_test:
+ # If speed is still 0.0 but we tried, return Error or 0.0
+ result["speed"] = f"{speed_mbps} Mbps"
+
+ return result
+
+
def parse_vless_url(url: str) -> dict:
"""Parse VLESS URL and extract connection parameters"""
- from urllib.parse import urlparse, parse_qs, unquote
-
if not url.startswith("vless://"):
raise ValueError("URL must start with vless://")
@@ -194,6 +300,10 @@ def generate_vless_config(vless_params: dict) -> dict:
return config
+class ThreadingHTTPServer(socketserver.ThreadingTCPServer):
+ allow_reuse_address = True
+
+
class ProxyControlHandler(http.server.BaseHTTPRequestHandler):
"""HTTP Request Handler for Proxy Control"""
@@ -224,6 +334,8 @@ class ProxyControlHandler(http.server.BaseHTTPRequestHandler):
self.get_status()
elif self.path == "/subscription":
self.get_subscription()
+ elif self.path.startswith("/test-connection"):
+ self.test_connection()
elif self.path.startswith("/static/"):
self.serve_static()
else:
@@ -237,6 +349,8 @@ class ProxyControlHandler(http.server.BaseHTTPRequestHandler):
self.fetch_subscription()
elif self.path == "/apply-subscription":
self.apply_subscription()
+ elif self.path == "/ping-target":
+ self.ping_target()
else:
self.send_error(404)
@@ -290,10 +404,43 @@ class ProxyControlHandler(http.server.BaseHTTPRequestHandler):
self.send_json({
"saved": True,
"url": sub.get("url"),
- "selectedServer": sub.get("selectedServer")
+ "selectedServer": sub.get("selectedServer"),
+ "userInfo": sub.get("userInfo")
})
else:
self.send_json({"saved": False})
+
+ def test_connection(self):
+ """Test active proxy connection"""
+ query_components = {}
+ if '?' in self.path:
+ _, query = self.path.split('?', 1)
+ query_components = parse_qs(query)
+
+ enable_speed = query_components.get('speed', ['false'])[0].lower() == 'true'
+
+ result = measure_proxy_performance(enable_speed_test=enable_speed)
+ self.send_json(result)
+
+ def ping_target(self):
+ """Ping a specific target"""
+ try:
+ content_length = int(self.headers.get("Content-Length", 0))
+ body = self.rfile.read(content_length).decode("utf-8")
+ data = json.loads(body)
+
+ server = data.get("server")
+ port = int(data.get("port", 443))
+
+ if not server:
+ self.send_json({"error": "No server specified"}, 400)
+ return
+
+ latency = measure_tcp_latency(server, port)
+ self.send_json({"latency": latency})
+
+ except Exception as e:
+ self.send_json({"error": str(e)}, 500)
def apply_config(self):
"""Apply new config from VLESS URL"""
@@ -372,6 +519,20 @@ class ProxyControlHandler(http.server.BaseHTTPRequestHandler):
with urllib.request.urlopen(req, timeout=15) as response:
config_text = response.read().decode("utf-8")
config = json.loads(config_text)
+
+ # Parse User Info header
+ user_info_header = response.headers.get("subscription-userinfo", "")
+ user_info = {}
+ if user_info_header:
+ parts = user_info_header.split(';')
+ for part in parts:
+ if '=' in part:
+ key, value = part.strip().split('=', 1)
+ try:
+ user_info[key] = int(value)
+ except ValueError:
+ pass
+
except urllib.error.HTTPError as e:
self.send_json({"success": False, "error": f"Ошибка HTTP: {e.code}"}, 400)
return
@@ -389,7 +550,7 @@ class ProxyControlHandler(http.server.BaseHTTPRequestHandler):
"tag": outbound.get("tag", "unknown"),
"type": outbound.get("type"),
"server": outbound.get("server", "unknown"),
- "port": outbound.get("server_port", 443)
+ "server_port": outbound.get("server_port", 443)
})
if not servers:
@@ -399,7 +560,8 @@ class ProxyControlHandler(http.server.BaseHTTPRequestHandler):
self.send_json({
"success": True,
"servers": servers,
- "config": config
+ "config": config,
+ "userInfo": user_info
})
except json.JSONDecodeError:
@@ -417,6 +579,7 @@ class ProxyControlHandler(http.server.BaseHTTPRequestHandler):
config = data.get("config")
selected_tag = data.get("selectedServer")
sub_url = data.get("subUrl") # URL подписки для сохранения
+ user_info = data.get("userInfo")
if not config:
self.send_json({"success": False, "error": "Конфигурация не указана"}, 400)
@@ -448,7 +611,6 @@ class ProxyControlHandler(http.server.BaseHTTPRequestHandler):
new_outbounds.insert(0, selected_outbound)
# Update route - remove incompatible fields and set only final
- # Some subscription configs have route.rules with "action" field which is not supported
routes = {
"final": selected_tag,
"auto_detect_interface": True
@@ -463,8 +625,7 @@ class ProxyControlHandler(http.server.BaseHTTPRequestHandler):
config.pop("platform", None)
config.pop("experimental", None)
- # Replace TUN inbounds with mixed proxy (TUN requires privileges in Docker)
- # Use mixed proxy on 127.0.0.1:2412 instead
+ # Replace TUN inbounds with mixed proxy
config["inbounds"] = [
{
"tag": "mixed-in",
@@ -488,7 +649,7 @@ class ProxyControlHandler(http.server.BaseHTTPRequestHandler):
# Save subscription URL for persistence
if sub_url:
- save_subscription(sub_url, selected_tag)
+ save_subscription(sub_url, selected_tag, user_info)
# Trigger reload via internal control port
try:
@@ -509,7 +670,8 @@ class ProxyControlHandler(http.server.BaseHTTPRequestHandler):
def main():
"""Start the web server"""
- with socketserver.TCPServer(("", PORT), ProxyControlHandler) as httpd:
+ # Use ThreadingTCPServer for concurrent requests
+ with ThreadingHTTPServer(("", PORT), ProxyControlHandler) as httpd:
print(f"[WebUI] Server started on port {PORT}")
print(f"[WebUI] Open http://localhost:{PORT} in your browser")
httpd.serve_forever()