vpn-proxy/web/server.py

#!/usr/bin/env python3
"""
Simple HTTP Web Server for VPN Proxy Control
Provides a web UI to apply VLESS/subscription URLs
"""

import http.server
import json
import os
import subprocess
import socketserver
from urllib.parse import parse_qs
from pathlib import Path

PORT = 3456
APP_DIR = Path(__file__).parent
WEB_DIR = APP_DIR / "web"
DATA_DIR = APP_DIR / "data"
CONFIG_FILE = DATA_DIR / "client.json"


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.startswith("/static/"):
            self.serve_static()
        else:
            self.send_error(404)

    def do_POST(self):
        """Handle POST requests"""
        if self.path == "/apply":
            self.apply_config()
        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 apply_config(self):
        """Apply new config from 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://") or url.startswith("http://") or url.startswith("https://")):
                self.send_json({"success": False, "error": "Неверный формат URL. Ожидается vless:// или http(s):// ссылка"}, 400)
                return

            # Run gen-client-from-url.sh
            script_path = APP_DIR / "gen-client-from-url.sh"
            result = subprocess.run(
                [str(script_path), url, str(CONFIG_FILE)],
                capture_output=True,
                text=True,
                cwd=str(APP_DIR),
                timeout=30
            )

            if result.returncode != 0:
                error_msg = result.stderr or result.stdout or "Неизвестная ошибка"
                self.send_json({"success": False, "error": f"Ошибка генерации: {error_msg}"}, 500)
                return

            # Trigger reload via internal control port
            try:
                import urllib.request
                urllib.request.urlopen("http://localhost:9090/reload", timeout=5)
            except Exception as e:
                print(f"[WebUI] Warning: reload request failed: {e}")
                # Continue anyway, config is generated

            self.send_json({
                "success": True,
                "message": "Конфигурация применена успешно!",
                "output": result.stdout
            })

        except json.JSONDecodeError:
            self.send_json({"success": False, "error": "Неверный JSON"}, 400)
        except subprocess.TimeoutExpired:
            self.send_json({"success": False, "error": "Таймаут при генерации конфига"}, 500)
        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()