#!/usr/bin/env python3 """ Simple HTTP Web Server for VPN Proxy Control Provides a web UI to manage sing-box subscriptions """ import http.server import json import os import platform import socketserver import urllib.request import urllib.error import uuid from urllib.parse import parse_qs from pathlib import Path PORT = 3456 APP_NAME = "VPN-Proxy-Control by Dokril" APP_DIR = Path(__file__).parent BASE_DIR = APP_DIR.parent WEB_DIR = APP_DIR DATA_DIR = BASE_DIR / "data" CONFIG_FILE = DATA_DIR / "client.json" HWID_FILE = DATA_DIR / "hwid" SUBSCRIPTION_FILE = DATA_DIR / "subscription.json" def get_hwid() -> str: """Get or generate hardware ID""" DATA_DIR.mkdir(parents=True, exist_ok=True) if HWID_FILE.exists(): return HWID_FILE.read_text().strip() # Generate new random HWID hwid = uuid.uuid4().hex[:16] HWID_FILE.write_text(hwid) return hwid def get_system_info() -> dict: """Get system information for headers""" system = platform.system().lower() # windows, linux, darwin version = platform.release() # 10, 5.15.0, 22.0.0 return { "os": system, "version": version } def save_subscription(url: str, selected_server: str = None): """Save subscription URL and selected server to file""" DATA_DIR.mkdir(parents=True, exist_ok=True) data = { "url": url, "selectedServer": selected_server } SUBSCRIPTION_FILE.write_text(json.dumps(data, ensure_ascii=False, indent=2)) def load_subscription() -> dict: """Load subscription from file""" if SUBSCRIPTION_FILE.exists(): try: return json.loads(SUBSCRIPTION_FILE.read_text()) except json.JSONDecodeError: pass return None 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://") # Remove scheme url_no_scheme = url[8:] # Split by fragment (#tag) if '#' in url_no_scheme: url_part, tag = url_no_scheme.split('#', 1) tag = unquote(tag) else: url_part = url_no_scheme tag = "reality" # Split by query (?) if '?' in url_part: uuid_host_port, query_string = url_part.split('?', 1) else: raise ValueError("Missing query parameters") # Parse UUID@host:port if '@' not in uuid_host_port: raise ValueError("Missing @ separator") uuid_str, host_port = uuid_host_port.split('@', 1) if ':' not in host_port: raise ValueError("Missing port") host, port_str = host_port.rsplit(':', 1) port = int(port_str) # Parse query parameters params = {} for param in query_string.split('&'): if '=' in param: key, value = param.split('=', 1) params[key] = unquote(value) # Extract required parameters pbk = params.get('pbk', '') sid = params.get('sid', '') sni = params.get('sni', host) fp = params.get('fp', 'chrome') flow = params.get('flow', '') if not pbk or not sid: raise ValueError("Missing required parameters: pbk or sid") return { 'uuid': uuid_str, 'server': host, 'server_port': port, 'tag': tag, 'public_key': pbk, 'short_id': sid, 'server_name': sni, 'fingerprint': fp, 'flow': flow } def generate_vless_config(vless_params: dict) -> dict: """Generate sing-box configuration from VLESS parameters""" config = { "dns": { "independent_cache": True }, "log": { "level": "debug", "disabled": True, "timestamp": True }, "route": { "final": vless_params['tag'], "auto_detect_interface": True }, "inbounds": [ { "tag": "mixed-in", "type": "mixed", "sniff": True, "users": [], "listen": "0.0.0.0", "listen_port": 8082, "set_system_proxy": False } ], "outbounds": [ { "type": "vless", "tag": vless_params['tag'], "server": vless_params['server'], "server_port": vless_params['server_port'], "flow": vless_params['flow'], "tls": { "enabled": True, "server_name": vless_params['server_name'], "reality": { "enabled": True, "public_key": vless_params['public_key'], "short_id": vless_params['short_id'] }, "utls": { "enabled": True, "fingerprint": vless_params['fingerprint'] } }, "uuid": vless_params['uuid'] }, { "tag": "direct", "type": "direct" } ] } return config class ProxyControlHandler(http.server.BaseHTTPRequestHandler): """HTTP Request Handler for Proxy Control""" def log_message(self, format, *args): """Override to add timestamp prefix""" print(f"[WebUI] {args[0]}") def send_json(self, data: dict, status: int = 200): """Send JSON response""" self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(json.dumps(data, ensure_ascii=False).encode("utf-8")) def send_html(self, content: bytes): """Send HTML response""" self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.end_headers() self.wfile.write(content) def do_GET(self): """Handle GET requests""" if self.path == "/" or self.path == "/index.html": self.serve_index() elif self.path == "/status": self.get_status() elif self.path == "/subscription": self.get_subscription() elif self.path.startswith("/static/"): self.serve_static() else: self.send_error(404) def do_POST(self): """Handle POST requests""" if self.path == "/apply": self.apply_config() elif self.path == "/fetch-subscription": self.fetch_subscription() elif self.path == "/apply-subscription": self.apply_subscription() else: self.send_error(404) def serve_index(self): """Serve main HTML page""" index_path = WEB_DIR / "index.html" if index_path.exists(): self.send_html(index_path.read_bytes()) else: self.send_error(404, "index.html not found") def serve_static(self): """Serve static files""" file_path = WEB_DIR / self.path[8:] # Remove /static/ if file_path.exists() and file_path.is_file(): content_type = "text/css" if str(file_path).endswith(".css") else "application/javascript" self.send_response(200) self.send_header("Content-Type", content_type) self.end_headers() self.wfile.write(file_path.read_bytes()) else: self.send_error(404) def get_status(self): """Get current proxy status""" config_exists = CONFIG_FILE.exists() current_tag = None current_server = None if config_exists: try: config = json.loads(CONFIG_FILE.read_text()) for outbound in config.get("outbounds", []): if outbound.get("type") == "vless": current_tag = outbound.get("tag", "unknown") current_server = outbound.get("server", "unknown") break except Exception: pass self.send_json({ "active": config_exists, "tag": current_tag, "server": current_server }) def get_subscription(self): """Get saved subscription info""" sub = load_subscription() if sub: self.send_json({ "saved": True, "url": sub.get("url"), "selectedServer": sub.get("selectedServer") }) else: self.send_json({"saved": False}) def apply_config(self): """Apply new config from VLESS URL""" try: content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length).decode("utf-8") data = json.loads(body) url = data.get("url", "").strip() if not url: self.send_json({"success": False, "error": "URL не указан"}, 400) return if not url.startswith("vless://"): self.send_json({"success": False, "error": "Неверный формат. Поддерживаются только vless:// ссылки"}, 400) return # Parse VLESS URL try: vless_params = parse_vless_url(url) except ValueError as e: self.send_json({"success": False, "error": f"Ошибка парсинга URL: {str(e)}"}, 400) return # Generate config config = generate_vless_config(vless_params) # Ensure data directory exists DATA_DIR.mkdir(parents=True, exist_ok=True) # Write config file CONFIG_FILE.write_text(json.dumps(config, indent=2, ensure_ascii=False)) # Trigger reload via internal control port try: urllib.request.urlopen("http://localhost:9090/reload", timeout=5) except Exception as e: print(f"[WebUI] Warning: reload request failed: {e}") self.send_json({ "success": True, "message": f"Конфигурация '{vless_params['tag']}' успешно применена!" }) except json.JSONDecodeError: self.send_json({"success": False, "error": "Неверный JSON"}, 400) except Exception as e: self.send_json({"success": False, "error": str(e)}, 500) def fetch_subscription(self): """Fetch servers list from subscription URL""" try: content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length).decode("utf-8") data = json.loads(body) url = data.get("url", "").strip() if not url: self.send_json({"success": False, "error": "URL подписки не указан"}, 400) return # Fetch subscription config sys_info = get_system_info() req = urllib.request.Request( url, headers={ "User-Agent": "singbox", "x-hwid": get_hwid(), "x-device-os": sys_info["os"], "x-ver-os": sys_info["version"], "x-device-model": APP_NAME } ) try: with urllib.request.urlopen(req, timeout=15) as response: config_text = response.read().decode("utf-8") config = json.loads(config_text) except urllib.error.HTTPError as e: self.send_json({"success": False, "error": f"Ошибка HTTP: {e.code}"}, 400) return except urllib.error.URLError as e: self.send_json({"success": False, "error": f"Ошибка подключения: {e.reason}"}, 400) return # Extract outbound servers outbounds = config.get("outbounds", []) servers = [] for outbound in outbounds: if outbound.get("type") in ["vless", "vmess", "trojan", "shadowsocks", "hysteria2"]: servers.append({ "tag": outbound.get("tag", "unknown"), "type": outbound.get("type"), "server": outbound.get("server", "unknown"), "port": outbound.get("server_port", 443) }) if not servers: self.send_json({"success": False, "error": "Серверы не найдены в подписке"}, 400) return self.send_json({ "success": True, "servers": servers, "config": config }) except json.JSONDecodeError: self.send_json({"success": False, "error": "Неверный JSON в ответе"}, 400) except Exception as e: self.send_json({"success": False, "error": str(e)}, 500) def apply_subscription(self): """Apply config from subscription with selected server""" try: content_length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_length).decode("utf-8") data = json.loads(body) config = data.get("config") selected_tag = data.get("selectedServer") sub_url = data.get("subUrl") # URL подписки для сохранения if not config: self.send_json({"success": False, "error": "Конфигурация не указана"}, 400) return if not selected_tag: self.send_json({"success": False, "error": "Сервер не выбран"}, 400) return # Modify config to use only selected server outbounds = config.get("outbounds", []) new_outbounds = [] selected_outbound = None for outbound in outbounds: if outbound.get("tag") == selected_tag: selected_outbound = outbound elif outbound.get("type") in ["direct", "block", "dns"]: new_outbounds.append(outbound) elif outbound.get("type") == "selector": # Skip selector, we'll add selected server directly pass if not selected_outbound: self.send_json({"success": False, "error": f"Сервер '{selected_tag}' не найден"}, 400) return # Add selected server as main outbound 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 } # Simplify DNS configuration to match client.json format config["dns"] = { "independent_cache": True } # Remove platform-specific and experimental fields from root config 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 config["inbounds"] = [ { "tag": "mixed-in", "type": "mixed", "sniff": True, "users": [], "listen": "0.0.0.0", "listen_port": 8082, "set_system_proxy": False } ] config["outbounds"] = new_outbounds config["route"] = routes # Ensure data directory exists DATA_DIR.mkdir(parents=True, exist_ok=True) # Write config file CONFIG_FILE.write_text(json.dumps(config, indent=2, ensure_ascii=False)) # Save subscription URL for persistence if sub_url: save_subscription(sub_url, selected_tag) # Trigger reload via internal control port try: urllib.request.urlopen("http://localhost:9090/reload", timeout=5) except Exception as e: print(f"[WebUI] Warning: reload request failed: {e}") self.send_json({ "success": True, "message": f"Сервер '{selected_tag}' успешно применён!" }) except json.JSONDecodeError: self.send_json({"success": False, "error": "Неверный JSON"}, 400) except Exception as e: self.send_json({"success": False, "error": str(e)}, 500) def main(): """Start the web server""" with socketserver.TCPServer(("", 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() if __name__ == "__main__": main()