diff --git "a/app.py" "b/app.py" new file mode 100644--- /dev/null +++ "b/app.py" @@ -0,0 +1,2286 @@ + +import asyncio +import json +import os +import hashlib +import secrets +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