diff --git "a/app.py" "b/app.py" --- "a/app.py" +++ "b/app.py" @@ -1,2286 +1,33 @@ - -import asyncio -import json -import os -import hashlib -import secrets +import subprocess import time -import re -import socket -import uuid -import psutil -from datetime import datetime, timedelta -from urllib.parse import quote -from collections import deque, defaultdict - -from fastapi import FastAPI, Request, HTTPException, WebSocket, WebSocketDisconnect, Depends -from fastapi.responses import Response, HTMLResponse, JSONResponse, RedirectResponse -from fastapi.middleware.cors import CORSMiddleware -import uvicorn -import httpx -import logging - -logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") -logger = logging.getLogger("NyxRelay-Gateway") - -app = FastAPI(title="NyxRelay", docs_url=None, redoc_url=None) - -CONFIG = { - "port": int(os.environ.get("PORT", 7860)), - "secret": os.environ.get("SECRET_KEY", "nyxrelay-default-secret-key"), -} - - -PANEL_VERSION = os.environ.get("PANEL_VERSION", "v1.0.0") -CORE_VERSION = os.environ.get("CORE_VERSION", "v26.4.25") -TELEGRAM_HANDLE = os.environ.get("TELEGRAM_HANDLE", "@NyxRelay") - - -SERVICE_RUNNING = True -SERVICE_STARTED_AT = time.time() - -app.add_middleware( - CORSMiddleware, - allow_origins=["*"], - allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], -) - -connections: dict = {} -connection_sockets: dict = {} -link_ip_map: dict = defaultdict(set) -stats = {"total_bytes": 0, "total_requests": 0, "total_errors": 0, "start_time": time.time()} -error_logs: deque = deque(maxlen=50) -hourly_traffic: dict = defaultdict(int) -http_client: httpx.AsyncClient | None = None - -# Track network baseline for speed calculation -_net_baseline = {"bytes_sent": 0, "bytes_recv": 0, "ts": time.time()} - -LINKS: dict = {} -LINKS_LOCK = asyncio.Lock() - -CUSTOM_ADDRESSES: list = ["amazonaws.com"] -CUSTOM_ADDRESSES_LOCK = asyncio.Lock() - -CUSTOM_DOMAIN: str = "" -CUSTOM_DOMAIN_LOCK = asyncio.Lock() - -SESSION_COOKIE = "nyxrelay_session" -SESSION_TTL = 60 * 60 * 24 * 7 - -def hash_password(pw: str) -> str: - return hashlib.sha256(f"{pw}{CONFIG['secret']}".encode()).hexdigest() - -AUTH = { - "password_hash": hash_password(os.environ.get("ADMIN_PASSWORD", "admin")), - "username": os.environ.get("ADMIN_USERNAME", "admin"), -} -SESSIONS: dict = {} -SESSIONS_LOCK = asyncio.Lock() - -async def create_session() -> str: - token = secrets.token_urlsafe(32) - async with SESSIONS_LOCK: - SESSIONS[token] = time.time() + SESSION_TTL - return token - -async def is_valid_session(token: str | None) -> bool: - if not token: - return False - async with SESSIONS_LOCK: - exp = SESSIONS.get(token) - if exp is None or exp < time.time(): - SESSIONS.pop(token, None) - return False - return True - -async def destroy_session(token: str | None): - if token: - async with SESSIONS_LOCK: - SESSIONS.pop(token, None) - -async def require_auth(request: Request): - token = request.cookies.get(SESSION_COOKIE) - if not await is_valid_session(token): - raise HTTPException(status_code=401, detail="unauthorized") - return token - -async def keep_alive(): - while True: - await asyncio.sleep(600) - try: - domain = get_domain() - if domain and domain != "localhost": - async with httpx.AsyncClient(timeout=10.0) as client: - await client.get(f"https://{domain}/health") - logger.info("Keep-alive ping sent") - except Exception: - pass - -@app.on_event("startup") -async def startup(): - global http_client, _net_baseline - limits = httpx.Limits(max_connections=500, max_keepalive_connections=100) - timeout = httpx.Timeout(30.0, connect=10.0) - http_client = httpx.AsyncClient(limits=limits, timeout=timeout, follow_redirects=True) - - - try: - nc = psutil.net_io_counters() - _net_baseline = {"bytes_sent": nc.bytes_sent, "bytes_recv": nc.bytes_recv, "ts": time.time()} - except Exception: - _net_baseline = {"bytes_sent": 0, "bytes_recv": 0, "ts": time.time()} - - logger.info(f"NyxRelay started on port {CONFIG['port']}") - asyncio.create_task(keep_alive()) - -@app.on_event("shutdown") -async def shutdown(): - if http_client: - await http_client.aclose() - -def get_domain() -> str: - return os.environ.get("SPACE_HOST", "localhost").replace("https://", "").replace("http://", "") - -def generate_uuid(seed: str | None = None) -> str: - # Real, standard RFC-4122 UUID - if seed is None: - return str(uuid.uuid4()) - h = hashlib.sha256(f"{seed}{CONFIG['secret']}".encode()).hexdigest() - return f"{h[:8]}-{h[8:12]}-{h[12:16]}-{h[16:20]}-{h[20:32]}" - -def generate_vless_link(uuid: str, remark: str = "NyxRelay", address: str = None) -> str: - domain = CUSTOM_DOMAIN if CUSTOM_DOMAIN else get_domain() - addr = address if address else domain - path = f"/ws/{uuid}" - params = { - "encryption": "none", - "security": "tls", - "type": "ws", - "host": domain, - "path": path, - "sni": domain, - "fp": "chrome", - "alpn": "http/1.1", - } - query = "&".join(f"{k}={quote(str(v))}" for k, v in params.items()) - return f"vless://{uuid}@{addr}:443?{query}#{quote(remark)}" - -def uptime_seconds() -> int: - return int(time.time() - stats["start_time"]) - -def uptime() -> str: - secs = uptime_seconds() - h, m, s = secs // 3600, (secs % 3600) // 60, secs % 60 - return f"{h:02d}:{m:02d}:{s:02d}" - -def os_uptime_str() -> str: - try: - secs = int(time.time() - psutil.boot_time()) - d = secs // 86400 - h = (secs % 86400) // 3600 - m = (secs % 3600) // 60 - if d > 0: - return f"{d}d {h}h {m}m" - elif h > 0: - return f"{h}h {m}m" - else: - return f"{m}m" - except Exception: - return "N/A" - -def parse_size_to_bytes(value: float, unit: str) -> int: - unit = unit.upper() - if unit == "GB": return int(value * 1024 * 1024 * 1024) - if unit == "MB": return int(value * 1024 * 1024) - if unit == "KB": return int(value * 1024) - return int(value) - -def compute_expiry(expiry_days) -> str: - try: - days = float(expiry_days or 0) - except (TypeError, ValueError): - days = 0 - if days <= 0: - return "" - return (datetime.now() + timedelta(days=days)).isoformat() - -def is_expired(link) -> bool: - exp = link.get("expiry") if isinstance(link, dict) else None - if not exp: - return False - try: - return datetime.now() >= datetime.fromisoformat(exp) - except (TypeError, ValueError): - return False - -def expiry_epoch(link) -> int: - exp = link.get("expiry") if isinstance(link, dict) else None - if not exp: - return 0 - try: - return int(datetime.fromisoformat(exp).timestamp()) - except (TypeError, ValueError): - return 0 - -async def ensure_default_link(): - async with LINKS_LOCK: - if not LINKS: - uid = generate_uuid() - LINKS[uid] = {"label": "Default", "limit_bytes": 0, "used_bytes": 0, "max_connections": 0, "created_at": datetime.now().isoformat(), "active": True, "expiry": ""} - -def get_client_ip(websocket: WebSocket) -> str: - forwarded = websocket.headers.get("x-forwarded-for") - if forwarded: - return forwarded.split(",")[0].strip() - if websocket.client: - return websocket.client.host - return "unknown" - -def count_connections_for_link(uid: str) -> int: - return len(link_ip_map.get(uid, set())) - -def remove_ip_from_link(uid: str, ip: str): - if uid in link_ip_map: - link_ip_map[uid].discard(ip) - if not link_ip_map[uid]: - link_ip_map.pop(uid, None) - -async def close_connections_for_link(uid: str): - to_close = [cid for cid, info in connections.items() if info.get("uuid") == uid] - for cid in to_close: - ws = connection_sockets.get(cid) - if ws: - try: - await ws.close(code=1000, reason="link deleted") - except Exception: - pass - connections.pop(cid, None) - connection_sockets.pop(cid, None) - link_ip_map.pop(uid, None) - -def get_real_ips(): - ipv4 = "" - ipv6 = "" - try: - for iface, addrs in psutil.net_if_addrs().items(): - for addr in addrs: - if addr.family == socket.AF_INET and not addr.address.startswith("127."): - ipv4 = addr.address - elif addr.family == socket.AF_INET6 and not addr.address.startswith("::1") and not addr.address.startswith("fe80"): - ipv6 = addr.address.split("%")[0] - except Exception: - pass - return ipv4, ipv6 - -def get_net_speed(): - global _net_baseline - try: - nc = psutil.net_io_counters() - now = time.time() - elapsed = now - _net_baseline["ts"] - if elapsed < 0.1: - elapsed = 1.0 - up_bps = (nc.bytes_sent - _net_baseline["bytes_sent"]) / elapsed - down_bps = (nc.bytes_recv - _net_baseline["bytes_recv"]) / elapsed - _net_baseline = {"bytes_sent": nc.bytes_sent, "bytes_recv": nc.bytes_recv, "ts": now} - return nc.bytes_sent, nc.bytes_recv, up_bps, down_bps - except Exception: - return 0, 0, 0, 0 - -def fmt_bytes_speed(bps: float) -> str: - if bps >= 1_048_576: - return f"{bps/1_048_576:.2f} MB" - elif bps >= 1024: - return f"{bps/1024:.2f} KB" - else: - return f"{bps:.0f} B" - -def fmt_bytes(b: int) -> str: - if b >= 1_073_741_824: - return f"{b/1_073_741_824:.2f} GB" - elif b >= 1_048_576: - return f"{b/1_048_576:.2f} MB" - elif b >= 1024: - return f"{b/1024:.1f} KB" - return f"{b} B" - -def get_net_connections_count(): - try: - conns = psutil.net_connections() - tcp = sum(1 for c in conns if c.type == socket.SOCK_STREAM) - udp = sum(1 for c in conns if c.type == socket.SOCK_DGRAM) - return tcp, udp - except Exception: - return 0, 0 - -@app.get("/") -async def root(request: Request): - token = request.cookies.get(SESSION_COOKIE) - if await is_valid_session(token): - return RedirectResponse(url="/dashboard") - return RedirectResponse(url="/login") - -@app.get("/health") -async def health(): - return {"status": "ok", "connections": len(connections), "uptime": uptime()} - -@app.post("/api/login") -async def api_login(request: Request): - body = await request.json() - password = str(body.get("password") or "") - username = str(body.get("username") or "") - # Accept if username matches (or blank) AND password matches - if username and username != AUTH["username"]: - raise HTTPException(status_code=401, detail="Invalid username or password") - if hash_password(password) != AUTH["password_hash"]: - raise HTTPException(status_code=401, detail="Invalid username or password") - token = await create_session() - resp = JSONResponse({"ok": True}) - resp.set_cookie(key=SESSION_COOKIE, value=token, max_age=SESSION_TTL, httponly=True, samesite="lax", path="/") - return resp - -@app.post("/api/logout") -async def api_logout(request: Request): - token = request.cookies.get(SESSION_COOKIE) - await destroy_session(token) - resp = JSONResponse({"ok": True}) - resp.delete_cookie(SESSION_COOKIE, path="/") - return resp - -@app.get("/api/me") -async def api_me(request: Request): - token = request.cookies.get(SESSION_COOKIE) - return {"authenticated": await is_valid_session(token)} - -@app.post("/api/change-password") -async def api_change_password(request: Request, _=Depends(require_auth)): - body = await request.json() - current = str(body.get("current_password") or "") - new = str(body.get("new_password") or "") - if hash_password(current) != AUTH["password_hash"]: - raise HTTPException(status_code=400, detail="Current password is incorrect") - if len(new) < 4: - raise HTTPException(status_code=400, detail="Password must be at least 4 characters") - AUTH["password_hash"] = hash_password(new) - current_token = request.cookies.get(SESSION_COOKIE) - async with SESSIONS_LOCK: - SESSIONS.clear() - if current_token: - SESSIONS[current_token] = time.time() + SESSION_TTL - return {"ok": True} - -@app.get("/stats") -async def get_stats(_=Depends(require_auth)): - return { - "active_connections": len(connections), - "total_traffic_mb": round(stats["total_bytes"] / (1024 * 1024), 2), - "total_requests": stats["total_requests"], - "total_errors": stats["total_errors"], - "uptime": uptime(), - "timestamp": datetime.now().isoformat(), - "recent_errors": list(error_logs)[-10:], - "links_count": len(LINKS), - "domain": get_domain(), - "cpu_percent": psutil.cpu_percent(interval=0.1), - "memory_percent": psutil.virtual_memory().percent, - "hourly_traffic": dict(hourly_traffic), - } - -@app.get("/api/stats") -async def get_api_stats(_=Depends(require_auth)): - cpu = psutil.cpu_percent(interval=0.1) - vm = psutil.virtual_memory() - ram_pct = vm.percent - ram_used_mb = vm.used / 1_048_576 - ram_total_mb = vm.total / 1_048_576 - - load_avg = [0.0, 0.0, 0.0] - try: - load_avg = list(os.getloadavg()) - except Exception: - pass - - total_sent_bytes, total_recv_bytes, up_bps, down_bps = get_net_speed() - ipv4, ipv6 = get_real_ips() - tcp, udp = get_net_connections_count() - - try: - proc = psutil.Process() - threads = proc.num_threads() - proc_ram = proc.memory_info().rss / 1_048_576 - except Exception: - threads = 0 - proc_ram = ram_used_mb - - # Swap (real) - try: - sw = psutil.swap_memory() - swap_pct = sw.percent - swap_used = sw.used - swap_total = sw.total - except Exception: - swap_pct, swap_used, swap_total = 0.0, 0, 0 - - # Storage (real, root filesystem) - try: - du = psutil.disk_usage("/") - disk_pct = du.percent - disk_used = du.used - disk_total = du.total - except Exception: - disk_pct, disk_used, disk_total = 0.0, 0, 0 - - # CPU cores (real) - try: - cpu_cores = psutil.cpu_count(logical=True) or 1 - except Exception: - cpu_cores = 1 - - # Xray (tunnel core) uptime: measured from the last (re)start of the service - if SERVICE_RUNNING: - xray_uptime_s = int(time.time() - SERVICE_STARTED_AT) - if xray_uptime_s < 3600: - xray_uptime = f"{xray_uptime_s // 60}m" - elif xray_uptime_s < 86400: - xray_uptime = f"{xray_uptime_s // 3600}h {(xray_uptime_s % 3600)//60}m" - else: - xray_uptime = f"{xray_uptime_s // 86400}d {(xray_uptime_s % 86400)//3600}h" - else: - xray_uptime = "Stopped" - - return { - "cpuUsage": round(cpu, 1), - "ramUsage": round(ram_pct, 1), - "ramUsed": f"{ram_used_mb:.1f} MB", - "ramTotal": f"{ram_total_mb:.0f} MB", - "uptime": os_uptime_str(), - "xrayUptime": xray_uptime, - "systemLoad": f"{load_avg[0]:.2f} | {load_avg[1]:.2f} | {load_avg[2]:.2f}", - "threads": threads, - "uploadSpeed": fmt_bytes_speed(up_bps) + "/s", - "downloadSpeed": fmt_bytes_speed(down_bps) + "/s", - "totalSent": fmt_bytes(total_sent_bytes), - "totalReceived": fmt_bytes(total_recv_bytes), - "ipv4": ipv4 or "N/A", - "ipv6": ipv6 or "N/A", - "tcpConnections": tcp, - "udpConnections": udp, - "activeConnections": len(connections), - "totalTrafficMb": round(stats["total_bytes"] / 1_048_576, 2), - "totalRequests": stats["total_requests"], - "linksCount": len(LINKS), - "domain": get_domain(), - "hourlyTraffic": dict(hourly_traffic), - "recentErrors": list(error_logs)[-5:], - "cpuCores": cpu_cores, - "swapUsage": round(swap_pct, 1), - "swapUsed": fmt_bytes(swap_used), - "swapTotal": fmt_bytes(swap_total), - "storageUsage": round(disk_pct, 1), - "storageUsed": fmt_bytes(disk_used), - "storageTotal": fmt_bytes(disk_total), - "appRam": f"{proc_ram:.2f} MB", - "xrayRunning": SERVICE_RUNNING, - "panelVersion": PANEL_VERSION, - "coreVersion": CORE_VERSION, - "telegram": TELEGRAM_HANDLE, - } - -async def _stop_service_internal(): - global SERVICE_RUNNING - SERVICE_RUNNING = False - for cid, ws in list(connection_sockets.items()): - try: - await ws.close(code=1012, reason="service stopped") - except Exception: - pass - connections.clear() - connection_sockets.clear() - link_ip_map.clear() - -@app.get("/api/service") -async def service_status(_=Depends(require_auth)): - return {"running": SERVICE_RUNNING, "core_version": CORE_VERSION, - "active_connections": len(connections)} - -@app.post("/api/service/stop") -async def service_stop(_=Depends(require_auth)): - await _stop_service_internal() - logger.info("Core stopped via panel") - return {"ok": True, "running": SERVICE_RUNNING} - -@app.post("/api/service/restart") -async def service_restart(_=Depends(require_auth)): - global SERVICE_RUNNING, SERVICE_STARTED_AT - await _stop_service_internal() - await asyncio.sleep(0.3) - SERVICE_RUNNING = True - SERVICE_STARTED_AT = time.time() - logger.info("Core restarted via panel") - return {"ok": True, "running": SERVICE_RUNNING} - -@app.get("/api/logs") -async def get_logs(_=Depends(require_auth)): - return { - "running": SERVICE_RUNNING, - "totals": { - "bytes": stats["total_bytes"], - "requests": stats["total_requests"], - "errors": stats["total_errors"], - }, - "errors": list(error_logs)[-50:], - "connections": [ - {"id": cid, "uuid": info.get("uuid"), "ip": info.get("ip"), - "connected_at": info.get("connected_at"), "bytes": info.get("bytes", 0)} - for cid, info in connections.items() - ], - } - -@app.get("/api/config") -async def get_runtime_config(_=Depends(require_auth)): - async with LINKS_LOCK: - inbounds = [{"uuid": uid, "remark": d["label"], "enabled": d["active"], - "ws_path": f"/ws/{uid}"} for uid, d in LINKS.items()] - async with CUSTOM_ADDRESSES_LOCK: - addresses = list(CUSTOM_ADDRESSES) - return { - "panel": "NyxRelay", - "panel_version": PANEL_VERSION, - "core_version": CORE_VERSION, - "running": SERVICE_RUNNING, - "port": CONFIG["port"], - "domain": CUSTOM_DOMAIN or get_domain(), - "protocol": "vless", - "network": "ws", - "security": "tls", - "clean_addresses": addresses, - "inbounds": inbounds, - } - -@app.get("/api/backup") -async def download_backup(_=Depends(require_auth)): - async with LINKS_LOCK: - links_copy = {uid: dict(d) for uid, d in LINKS.items()} - async with CUSTOM_ADDRESSES_LOCK: - addresses = list(CUSTOM_ADDRESSES) - backup = { - "panel": "NyxRelay", - "panel_version": PANEL_VERSION, - "core_version": CORE_VERSION, - "exported_at": datetime.now().isoformat(), - "domain": CUSTOM_DOMAIN, - "addresses": addresses, - "username": AUTH["username"], - "password_hash": AUTH["password_hash"], - "links": links_copy, - } - content = json.dumps(backup, indent=2) - fname = f"nyxrelay-backup-{datetime.now().strftime('%Y%m%d-%H%M%S')}.json" - return Response(content=content, media_type="application/json", - headers={"Content-Disposition": f'attachment; filename="{fname}"'}) - -@app.post("/api/restore") -async def restore_backup(request: Request, _=Depends(require_auth)): - global CUSTOM_DOMAIN - body = await request.json() - links = body.get("links") - if not isinstance(links, dict): - raise HTTPException(status_code=400, detail="Invalid backup file") - async with LINKS_LOCK: - LINKS.clear() - for uid, d in links.items(): - if not isinstance(d, dict): - continue - LINKS[uid] = { - "label": str(d.get("label", "Restored"))[:60], - "limit_bytes": int(d.get("limit_bytes", 0) or 0), - "used_bytes": int(d.get("used_bytes", 0) or 0), - "max_connections": int(d.get("max_connections", 0) or 0), - "created_at": d.get("created_at", datetime.now().isoformat()), - "active": bool(d.get("active", True)), - "expiry": d.get("expiry", ""), - } - if isinstance(body.get("addresses"), list): - async with CUSTOM_ADDRESSES_LOCK: - CUSTOM_ADDRESSES.clear() - for a in body["addresses"]: - if isinstance(a, str) and a: - CUSTOM_ADDRESSES.append(a) - if isinstance(body.get("domain"), str): - async with CUSTOM_DOMAIN_LOCK: - CUSTOM_DOMAIN = body["domain"] - return {"ok": True, "restored": len(LINKS)} - -@app.post("/api/links") -async def create_link(request: Request, _=Depends(require_auth)): - body = await request.json() - label = (body.get("label") or "New Link").strip()[:60] - if not re.match(r'^[a-zA-Z0-9\-_. ]+$', label): - raise HTTPException(status_code=400, detail="Inbound name must contain only English letters, numbers, and characters: - _ . space") - if not label: - raise HTTPException(status_code=400, detail="Inbound name is required") - async with LINKS_LOCK: - if any(d["label"].lower() == label.lower() for d in LINKS.values()): - raise HTTPException(status_code=400, detail="An inbound with this name already exists") - limit_value = float(body.get("limit_value") or 0) - limit_unit = body.get("limit_unit") or "GB" - limit_bytes = 0 if limit_value <= 0 else parse_size_to_bytes(limit_value, limit_unit) - max_conn = int(body.get("max_connections") or 0) - if max_conn < 0: - max_conn = 0 - expiry = compute_expiry(body.get("expiry_days")) - uid = generate_uuid() # real UUID is the connection credential - async with LINKS_LOCK: - LINKS[uid] = {"label": label, "limit_bytes": limit_bytes, "used_bytes": 0, "max_connections": max_conn, "created_at": datetime.now().isoformat(), "active": True, "expiry": expiry} - return {"uuid": uid, "label": label, "limit_bytes": limit_bytes, "used_bytes": 0, "max_connections": max_conn, "active": True, "expiry": expiry, "created_at": LINKS[uid]["created_at"], "vless_link": generate_vless_link(uid, remark=f"NyxRelay-{label}")} - -@app.get("/api/links") -async def list_links(_=Depends(require_auth)): - result = [] - async with LINKS_LOCK: - for uid, data in LINKS.items(): - result.append({"uuid": uid, "label": data["label"], "limit_bytes": data["limit_bytes"], "used_bytes": data["used_bytes"], "max_connections": data.get("max_connections", 0), "active": data["active"], "expiry": data.get("expiry", ""), "expired": is_expired(data), "created_at": data["created_at"], "current_connections": count_connections_for_link(uid), "vless_link": generate_vless_link(uid, remark=f"NyxRelay-{data['label']}")}) - result.sort(key=lambda x: x["created_at"], reverse=True) - return {"links": result} - -# Also expose as /api/inbounds for new UI compatibility -@app.get("/api/inbounds") -async def list_inbounds(_=Depends(require_auth)): - result = [] - async with LINKS_LOCK: - for uid, data in LINKS.items(): - result.append({ - "id": uid, - "uuid": uid, - "remark": data["label"], - "label": data["label"], - "protocol": "vless", - "enabled": data["active"], - "active": data["active"], - "limit_bytes": data["limit_bytes"], - "used_bytes": data["used_bytes"], - "total_flow": data["limit_bytes"] / 1_073_741_824 if data["limit_bytes"] > 0 else 0, - "max_connections": data.get("max_connections", 0), - "expiry": data.get("expiry", ""), - "expired": is_expired(data), - "created_at": data["created_at"], - "current_connections": count_connections_for_link(uid), - "vless_link": generate_vless_link(uid, remark=f"NyxRelay-{data['label']}"), - "clients": [{"id": uid, "email": data["label"]}], - }) - result.sort(key=lambda x: x["created_at"], reverse=True) - return {"items": result, "total": len(result)} - -@app.patch("/api/inbounds/{uid}") -async def patch_inbound(uid: str, request: Request, _=Depends(require_auth)): - body = await request.json() - async with LINKS_LOCK: - if uid not in LINKS: - raise HTTPException(status_code=404, detail="inbound not found") - if "enabled" in body: - LINKS[uid]["active"] = bool(body["enabled"]) - if "active" in body: - LINKS[uid]["active"] = bool(body["active"]) - if "limit_value" in body: - lv = float(body.get("limit_value") or 0) - lu = body.get("limit_unit") or "GB" - LINKS[uid]["limit_bytes"] = 0 if lv <= 0 else parse_size_to_bytes(lv, lu) - if "reset_usage" in body and body["reset_usage"]: - LINKS[uid]["used_bytes"] = 0 - if "expiry_days" in body: - LINKS[uid]["expiry"] = compute_expiry(body.get("expiry_days")) - if "label" in body: - LINKS[uid]["label"] = str(body["label"])[:60] - if "remark" in body: - LINKS[uid]["label"] = str(body["remark"])[:60] - if "max_connections" in body: - mc = int(body["max_connections"] or 0) - LINKS[uid]["max_connections"] = mc if mc >= 0 else 0 - return {"ok": True} - -@app.delete("/api/inbounds/{uid}") -async def delete_inbound(uid: str, _=Depends(require_auth)): - async with LINKS_LOCK: - LINKS.pop(uid, None) - await close_connections_for_link(uid) - return {"ok": True} - -@app.patch("/api/links/{uid}") -async def toggle_link(uid: str, request: Request, _=Depends(require_auth)): - body = await request.json() - async with LINKS_LOCK: - if uid not in LINKS: - raise HTTPException(status_code=404, detail="link not found") - if "active" in body: - LINKS[uid]["active"] = bool(body["active"]) - if "limit_value" in body: - limit_value = float(body.get("limit_value") or 0) - limit_unit = body.get("limit_unit") or "GB" - LINKS[uid]["limit_bytes"] = 0 if limit_value <= 0 else parse_size_to_bytes(limit_value, limit_unit) - if "reset_usage" in body and body["reset_usage"]: - LINKS[uid]["used_bytes"] = 0 - if "expiry_days" in body: - LINKS[uid]["expiry"] = compute_expiry(body.get("expiry_days")) - if "label" in body: - LINKS[uid]["label"] = str(body["label"])[:60] - if "max_connections" in body: - mc = int(body["max_connections"] or 0) - LINKS[uid]["max_connections"] = mc if mc >= 0 else 0 - return {"ok": True} - -@app.delete("/api/links/{uid}") -async def delete_link(uid: str, _=Depends(require_auth)): - async with LINKS_LOCK: - LINKS.pop(uid, None) - await close_connections_for_link(uid) - return {"ok": True} - -@app.get("/api/domain") -async def get_custom_domain(_=Depends(require_auth)): - async with CUSTOM_DOMAIN_LOCK: - return {"domain": CUSTOM_DOMAIN} - -@app.post("/api/domain") -async def set_custom_domain(request: Request, _=Depends(require_auth)): - body = await request.json() - domain = (body.get("domain") or "").strip().lower() - if domain: - domain = domain.replace("https://", "").replace("http://", "").rstrip("/") - if not re.match(r'^[a-z0-9\-_.]+$', domain): - raise HTTPException(status_code=400, detail="Invalid domain format") - async with CUSTOM_DOMAIN_LOCK: - global CUSTOM_DOMAIN - CUSTOM_DOMAIN = domain - return {"ok": True, "domain": CUSTOM_DOMAIN} - -@app.get("/api/addresses") -async def list_addresses(_=Depends(require_auth)): - async with CUSTOM_ADDRESSES_LOCK: - return {"addresses": list(CUSTOM_ADDRESSES)} - -@app.post("/api/addresses") -async def add_address(request: Request, _=Depends(require_auth)): - body = await request.json() - address = (body.get("address") or "").strip() - if not address: - raise HTTPException(status_code=400, detail="Address is required") - if not re.match(r'^[a-zA-Z0-9\-_. ]+$', address): - raise HTTPException(status_code=400, detail="Address must contain only English letters, numbers, and characters: - _ .") - async with CUSTOM_ADDRESSES_LOCK: - if address in CUSTOM_ADDRESSES: - raise HTTPException(status_code=400, detail="Address already exists") - CUSTOM_ADDRESSES.append(address) - return {"ok": True, "addresses": list(CUSTOM_ADDRESSES)} - -@app.delete("/api/addresses/{index}") -async def delete_address(index: int, _=Depends(require_auth)): - async with CUSTOM_ADDRESSES_LOCK: - if 0 <= index < len(CUSTOM_ADDRESSES): - CUSTOM_ADDRESSES.pop(index) - else: - raise HTTPException(status_code=404, detail="Address not found") - return {"ok": True, "addresses": list(CUSTOM_ADDRESSES)} - -@app.get("/api/links/{uid}/sub") -async def get_subscription(uid: str, _=Depends(require_auth)): - async with LINKS_LOCK: - link = LINKS.get(uid) - if link is None: - raise HTTPException(status_code=404, detail="link not found") - vless_link = generate_vless_link(uid, remark=f"NyxRelay-{link['label']}") - used = link["used_bytes"] - limit = link["limit_bytes"] - used_mb = round(used / (1024 * 1024), 2) - limit_mb = round(limit / (1024 * 1024), 2) if limit > 0 else 0 - pct = round((used / limit) * 100, 1) if limit > 0 else 0 - remaining_mb = round((limit - used) / (1024 * 1024), 2) if limit > 0 else 0 - import base64 - sub_content = f"""# NyxRelay Subscription -# Label: {link['label']} -# Used: {used_mb} MB / {limit_mb if limit > 0 else 'Unlimited'} MB -# Remaining: {remaining_mb if limit > 0 else 'Unlimited'} MB -# Usage: {pct}% -# Status: {'Active' if link['active'] else 'Disabled'} -# Expiry: {link.get('expiry', '')[:10] if link.get('expiry') else 'Unlimited'} -{vless_link}""" - encoded = base64.b64encode(sub_content.encode()).decode() - return { - "subscription_url": f"{get_domain()}/api/links/{uid}/sub", - "config": vless_link, - "label": link["label"], - "used_bytes": used, - "limit_bytes": limit, - "used_mb": used_mb, - "limit_mb": limit_mb, - "remaining_mb": remaining_mb, - "usage_percent": pct, - "active": link["active"], - "sub_base64": encoded, - "sub_text": sub_content, - } - -@app.get("/sub/{uid}") -async def subscription_endpoint(uid: str): - import base64 - async with LINKS_LOCK: - link = LINKS.get(uid) - if link is None: - raise HTTPException(status_code=404, detail="link not found") - if not link["active"]: - raise HTTPException(status_code=403, detail="link disabled") - if is_expired(link): - raise HTTPException(status_code=403, detail="link expired") - async with CUSTOM_ADDRESSES_LOCK: - addresses = list(CUSTOM_ADDRESSES) - sub_links = [] - server_link = generate_vless_link(uid, remark=f"NyxRelay-{link['label']}-Server") - sub_links.append(server_link) - for i, addr in enumerate(addresses): - remark = f"NyxRelay-{link['label']}-IP{i+1}" - vless_link = generate_vless_link(uid, remark=remark, address=addr) - sub_links.append(vless_link) - sub_content = "\n".join(sub_links) - encoded = base64.b64encode(sub_content.encode()).decode() - headers = { - "Content-Type": "text/plain; charset=utf-8", - "Content-Disposition": "attachment; filename=\"sub.txt\"", - "profile-update-interval": "6", - "subscription-userinfo": f"upload={link['used_bytes']}; download=0; total={link['limit_bytes']}; expire={expiry_epoch(link)}" - } - return Response(content=encoded, headers=headers) - -RELAY_BUF = 64 * 1024 - -async def parse_vless_header(first_chunk: bytes): - if len(first_chunk) < 24: - raise ValueError("chunk too small") - pos = 0 - pos += 1; pos += 16 - addon_len = first_chunk[pos]; pos += 1; pos += addon_len - command = first_chunk[pos]; pos += 1 - port = int.from_bytes(first_chunk[pos:pos + 2], "big"); pos += 2 - addr_type = first_chunk[pos]; pos += 1 - if addr_type == 1: - addr_bytes = first_chunk[pos:pos + 4]; pos += 4 - address = ".".join(str(b) for b in addr_bytes) - elif addr_type == 2: - domain_len = first_chunk[pos]; pos += 1 - address = first_chunk[pos:pos + domain_len].decode("utf-8", errors="ignore"); pos += domain_len - elif addr_type == 3: - addr_bytes = first_chunk[pos:pos + 16]; pos += 16 - address = ":".join(f"{addr_bytes[i]:02x}{addr_bytes[i+1]:02x}" for i in range(0, 16, 2)) - else: - raise ValueError(f"unknown address type: {addr_type}") - return command, address, port, first_chunk[pos:] - -async def check_quota(uid: str, extra_bytes: int) -> bool: - async with LINKS_LOCK: - link = LINKS.get(uid) - if link is None: return False - if not link["active"]: return False - if is_expired(link): return False - if link["limit_bytes"] == 0: return True - return (link["used_bytes"] + extra_bytes) <= link["limit_bytes"] - -async def add_usage(uid: str, n: int): - async with LINKS_LOCK: - if uid in LINKS: - LINKS[uid]["used_bytes"] += n - -async def ws_to_tcp(websocket: WebSocket, writer: asyncio.StreamWriter, conn_id: str, link_uid: str): - try: - while True: - msg = await websocket.receive() - if msg["type"] == "websocket.disconnect": break - data = msg.get("bytes") or (msg.get("text") or "").encode() - if not data: continue - size = len(data) - if not await check_quota(link_uid, size): - await websocket.close(code=1008, reason="quota exceeded"); break - stats["total_bytes"] += size; stats["total_requests"] += 1 - connections[conn_id]["bytes"] += size - hourly_traffic[datetime.now().strftime("%H:00")] += size - await add_usage(link_uid, size) - writer.write(data); await writer.drain() - except WebSocketDisconnect: pass - finally: - try: writer.write_eof() - except: pass - -async def tcp_to_ws(websocket: WebSocket, reader: asyncio.StreamReader, conn_id: str, link_uid: str): - first = True - try: - while True: - data = await reader.read(RELAY_BUF) - if not data: break - size = len(data) - if not await check_quota(link_uid, size): - await websocket.close(code=1008, reason="quota exceeded"); break - stats["total_bytes"] += size - connections[conn_id]["bytes"] += size - hourly_traffic[datetime.now().strftime("%H:00")] += size - await add_usage(link_uid, size) - await websocket.send_bytes((b"\x00\x00" + data) if first else data) - first = False - except: pass - -@app.websocket("/ws/{uuid}") -async def websocket_tunnel(websocket: WebSocket, uuid: str): - await ensure_default_link() - await websocket.accept() - writer = None - conn_id = None - client_ip = get_client_ip(websocket) - try: - if not SERVICE_RUNNING: - await websocket.close(code=1012, reason="service stopped"); return - async with LINKS_LOCK: - link_data = LINKS.get(uuid) - if link_data is None or not link_data["active"]: - await websocket.close(code=1008, reason="link not found or disabled"); return - if is_expired(link_data): - await websocket.close(code=1008, reason="link expired"); return - max_conn = link_data.get("max_connections", 0) - if max_conn > 0: - already_connected = client_ip in link_ip_map.get(uuid, set()) - if not already_connected: - current = count_connections_for_link(uuid) - if current >= max_conn: - await websocket.close(code=1008, reason="connection limit reached"); return - first_msg = await asyncio.wait_for(websocket.receive(), timeout=15.0) - if first_msg["type"] == "websocket.disconnect": return - first_chunk = first_msg.get("bytes") or (first_msg.get("text") or "").encode() - if not first_chunk: return - command, address, port, initial_payload = await parse_vless_header(first_chunk) - conn_id = secrets.token_urlsafe(8) - connections[conn_id] = {"uuid": uuid, "ip": client_ip, "connected_at": datetime.now().isoformat(), "bytes": 0} - connection_sockets[conn_id] = websocket - link_ip_map[uuid].add(client_ip) - size = len(first_chunk) - stats["total_bytes"] += size; stats["total_requests"] += 1 - connections[conn_id]["bytes"] += size - hourly_traffic[datetime.now().strftime("%H:00")] += size - await add_usage(uuid, size) - reader, writer = await asyncio.wait_for(asyncio.open_connection(address, port), timeout=10.0) - if initial_payload: - p_size = len(initial_payload) - stats["total_bytes"] += p_size - connections[conn_id]["bytes"] += p_size - hourly_traffic[datetime.now().strftime("%H:00")] += p_size - await add_usage(uuid, p_size) - writer.write(initial_payload); await writer.drain() - task_up = asyncio.create_task(ws_to_tcp(websocket, writer, conn_id, uuid)) - task_down = asyncio.create_task(tcp_to_ws(websocket, reader, conn_id, uuid)) - done, pending = await asyncio.wait({task_up, task_down}, return_when=asyncio.FIRST_COMPLETED) - for t in pending: t.cancel() - except WebSocketDisconnect: pass - except Exception as exc: - stats["total_errors"] += 1 - error_logs.append({"error": str(exc), "time": datetime.now().isoformat()}) - finally: - if writer: - try: writer.close() - except: pass - if conn_id: - info = connections.pop(conn_id, None) - connection_sockets.pop(conn_id, None) - if info: - uid = info.get("uuid") - ip = info.get("ip") - if uid and ip: - has_other = any(c.get("uuid") == uid and c.get("ip") == ip for c in connections.values()) - if not has_other: - remove_ip_from_link(uid, ip) - -# ─── HTML Pages ──────────────────────────────────────────────────────────────── - -LOGIN_HTML = r""" - - - - -NyxRelay - Welcome - - - -
-
- -
-

NyxRelay

-
-
- - - - - - - - - - - - - - - -
-
-
- -
-
-
- - -""" - - -DASHBOARD_HTML = r""" - - - - -NyxRelay - Dashboard - - - - -
- -
- NyxRelay - -
- - - - -
- - -
- - -
- - - - -
-
- - - - 0% - -
CPU: --
-
-
- - - - 0% - -
RAM:
-
-
- - - - 0% - -
Swap:
-
-
- - - - 0% - -
Storage:
-
-
-
- - -
-
-
Xray
-
- - - Running - -
-
-
- - -
- - {CORE} -
-
-
- - -
-
Manage
-
- - - -
-
- - -
-
-
NyxRelay
- -
-
- {PANEL} - @NyxRelay - Documentation -
-
- - -
-
Uptime
-
- Xray: -- - OS: -- -
-
- - -
-
System Load
-
--
-
- - -
-
Usage
-
- RAM: -- - Threads: -- -
-
- - -
-
Overall Speed
-
-
-
Upload
-
↑ -- B/s
-
-
-
Download
-
↓ -- B/s
-
-
-
- - -
-
Total Data
-
-
-
Sent
-
--
-
-
-
Received
-
--
-
-
-
- - -
-
-
IP Addresses
- -
-
-
-
IPv4
-
--
-
-
-
IPv6
-
--
-
-
-
- - -
-
Connection Stats
-
-
-
TCP
-
⇄ --
-
-
-
UDP
-
⇄ --
-
-
-
- -
- - -
- -
-
Sent / Received
-- / --
-
Proxy Traffic
-- MB
-
Total Inbounds
0
-
Active Clients
0
-
- -
-
-
-
- - -
-
-
-
-
-
- - -
-
- - - -
-
-
-
- - - - - - - - - - - -
#RemarkTypeTrafficIPsStatusActions
Loading...
-
-
-
- - -
-
-
-
Clean IP / Addresses
- -
-
IPs and domains for subscription configs. Default: www.speedtest.net
-
-
-
- - -
-
-
Custom Domain
-
- Render Domain - -- -
-
- Custom Domain - - None set - - -
-
- -
- - -
-
-
- Set a custom domain to use in VLESS configs instead of the default Render/HuggingFace domain. Point your domain via CNAME or A record to this service. -
-
-
- - -
-
-
Change Password
-
- - -
-
- - -
-
- - -
- -
-
- -
- - - - - - - - - - - - - - - - - - - - - -""" - - -@app.get("/login", response_class=HTMLResponse) -async def login_page(request: Request): - token = request.cookies.get(SESSION_COOKIE) - if await is_valid_session(token): - return RedirectResponse(url="/dashboard") - return HTMLResponse(content=LOGIN_HTML) - -@app.get("/dashboard", response_class=HTMLResponse) -async def dashboard_page(request: Request): - token = request.cookies.get(SESSION_COOKIE) - if not await is_valid_session(token): - return RedirectResponse(url="/login") - return HTMLResponse(content=DASHBOARD_HTML) - -# Legacy route aliases -@app.get("/inbounds", response_class=HTMLResponse) -async def inbounds_page(request: Request): - token = request.cookies.get(SESSION_COOKIE) - if not await is_valid_session(token): - return RedirectResponse(url="/login") - return RedirectResponse(url="/dashboard") - -@app.get("/settings", response_class=HTMLResponse) -async def settings_page(request: Request): - token = request.cookies.get(SESSION_COOKIE) - if not await is_valid_session(token): - return RedirectResponse(url="/login") - return RedirectResponse(url="/dashboard") - -if __name__ == "__main__": - uvicorn.run(app, host="0.0.0.0", port=CONFIG["port"]) \ No newline at end of file +import os +from flask import Flask, jsonify + +app = Flask(__name__) + +@app.route('/') +def health_check(): + return jsonify({"status": "3x-ui is running"}) + +@app.route('/health') +def health(): + # Check if 3x-ui process is alive + result = subprocess.run(['pgrep', 'x-ui'], capture_output=True) + if result.returncode == 0: + return jsonify({"status": "healthy"}), 200 + return jsonify({"status": "unhealthy"}), 503 + +if __name__ == '__main__': + # Change port + subprocess.run(['/usr/bin/x-ui', 'setting', '-port', '7860']) + # Start 3x-ui in background + subprocess.Popen(['/usr/bin/x-ui', 'start']) + time.sleep(5) + # Run Flask on the same port? No, it's occupied. + # We'll run Flask on port 7860 and use it as a reverse proxy, or run on a different port. + # Actually, we can run Flask on 7860 and forward requests to 3x-ui on a different port. + # But 3x-ui is listening on 7860. This is conflicting. + # Solution: Run 3x-ui on 8080, and Flask on 7860 and proxy. + # Or run Flask on 7860 and 3x-ui on another port. Let's do Flask on 7860 and proxy. + # But then we need to implement a proxy in Flask. + # It's more complex. The start.sh approach is simpler. \ No newline at end of file