diff --git a/README.md b/README.md index c14dbe2..c0fd64c 100644 --- a/README.md +++ b/README.md @@ -35,6 +35,12 @@ aggregator.py ──→ FastAPI сервер - **ShortID ротация**: автоматическая смена shortId на серверах каждые N часов - **MOTD**: объявления из motd.txt или settings - **Health scoring**: взвешенная оценка здоровья сервера (CPU 30%, RAM 25%, Disk 20%, Net 15%, I/O 10%) +- **Server health ping**: фоновый пинг всех серверов каждые 30с — dead сервера автоматически исключаются из подписки, трафика, ротации shortid и админки +- **Auto-recovery**: при возвращении сервера в сеть он снова включается в работу без перезагрузки +- **Rate limiting**: не более 10 попыток входа в админку за 60с с одного IP +- **DA polling backoff**: при ошибках DonationAlerts API интервал удваивается до 1ч, при 401 — сразу 1ч ожидания +- **CSP + XSS protection**: Content-Security-Policy на всех HTML-страницах, экранирование user-контролируемых полей +- **Graceful shutdown**: все фоновые задачи корректно завершаются при остановке - **Trojan support**: создание клиентов на Trojan инбаундах (password вместо uuid) ## wdtt (Zern-BlackOut) Compatibility @@ -213,10 +219,13 @@ curl -X POST http://localhost:8000/admin/api/propagate/{server_name} -b "admin_t ## DonationAlerts -Polling-based (every ~2 seconds). Amount mapping from `settings.conf → tiers → prices`: +Polling-based с adaptive backoff. Amount mapping from `settings.conf → tiers → prices`: - Any configured amount → corresponding tier + days - Donor identified by `id` or `username` from donation message - `tariff_end_at` extends dynamically (MAX of current end, now + days) +- При ошибках API интервал удваивается (1 → 2 → 4 → ... → 3600с) +- При 401 Unauthorized — сразу 1ч ожидания (токен невалиден) +- После успешного ответа интервал сбрасывается к базовому ## Happ.su / WDTT Support @@ -228,10 +237,14 @@ Subscription response includes compatible headers: ## Tech -- **FastAPI** + uvicorn +- **FastAPI** + uvicorn (bind на `127.0.0.1` — за reverse proxy) - **SQLite** (users only) - **py3xui** — 3x-UI API client -- **httpx** — async HTTP for sub links +- **httpx** — async HTTP for sub links + server health pings - **qrcode** — QR generation - **Plus Jakarta Sans / JetBrains Mono** — UI typography - **Glassmorphism** — design system +- **CSP + Security headers** — Content-Security-Policy, X-Frame-Options, X-Content-Type-Options на всех страницах +- **CORS** — ограничен до домена из конфига +- **Rate limiting** — in-memory на `/admin/login` +- **Server health** — фоновый пинг каждые 30с, автоисключение dead серверов diff --git a/aggregator.py b/aggregator.py index 52d8fa0..31a653e 100644 --- a/aggregator.py +++ b/aggregator.py @@ -15,7 +15,9 @@ import hashlib import secrets import sqlite3 import time +import html from datetime import datetime, timedelta +from collections import defaultdict from qrcode.image.styledpil import StyledPilImage from qrcode.image.styles.moduledrawers import SquareModuleDrawer from typing import List, Dict, Optional, Tuple @@ -47,12 +49,63 @@ servers = [] settings = {} last_donation_id = 0 +_global_lock = asyncio.Lock() +_donation_lock = asyncio.Lock() + +_ratelimit_attempts: Dict[str, list] = defaultdict(list) +RATELIMIT_MAX = 10 +RATELIMIT_WINDOW = 60 + +_background_tasks: List[asyncio.Task] = [] + # In-memory cache _links_cache: Dict[str, Tuple[List[str], float]] = {} _traffic_cache: Dict[str, Tuple[Dict, float]] = {} CACHE_TTL_TRAFFIC = 120 CACHE_TTL_LINKS = 86400 * 7 +# Server health tracking +_server_health: Dict[str, bool] = {} +HEALTH_PING_INTERVAL = 30 + +async def _ping_server(srv: dict) -> bool: + url = srv.get("status_url", "") or srv.get("subscription_url", "") + if not url: + return True + try: + async with httpx.AsyncClient(timeout=5.0, verify=False) as client: + resp = await client.get(url) + return resp.status_code < 500 + except: + return False + +async def _server_health_loop(): + while True: + for srv in servers: + name = srv.get("name", "") + alive = await _ping_server(srv) + prev = _server_health.get(name) + if alive != prev: + _server_health[name] = alive + logger.info(f"Server {name} is now {'alive' if alive else 'dead'}") + await asyncio.sleep(HEALTH_PING_INTERVAL) + +def is_server_alive(name: str) -> bool: + return _server_health.get(name, True) + +def h(text) -> str: + if text is None: + return "" + return html.escape(str(text), quote=True) + +def is_ratelimited(ip: str) -> bool: + now = time.time() + _ratelimit_attempts[ip] = [t for t in _ratelimit_attempts[ip] if now - t < RATELIMIT_WINDOW] + if len(_ratelimit_attempts[ip]) >= RATELIMIT_MAX: + return True + _ratelimit_attempts[ip].append(now) + return False + def _get_cached(key: str, cache: dict, ttl: float): entry = cache.get(key) if entry and time.time() - entry[1] < ttl: @@ -83,20 +136,18 @@ def clear_cache(sub_id: str = None): _traffic_cache.clear() def get_wdtt_servers() -> List[dict]: - """Возвращает серверы с is_wdtt: true""" - return [s for s in servers if s.get("is_wdtt") and s.get("is_active")] + return [s for s in servers if s.get("is_wdtt") and s.get("is_active") and is_server_alive(s.get("name", ""))] async def _wdtt_api_call(server: dict, method: str, path: str, body: dict = None) -> dict: - """Вызов REST API wdtt-server""" api_key = server.get("wdtt_api_key", "") host = server.get("wdtt_host", "") if not api_key or not host: return {"error": "wdtt server not configured"} - + port = server.get("wdtt_api_port", "8080") url = f"http://{host}:{port}{path}" headers = {"X-API-Key": api_key, "Content-Type": "application/json"} - + try: async with httpx.AsyncClient(timeout=10.0, verify=False) as client: if method == "POST": @@ -105,7 +156,7 @@ async def _wdtt_api_call(server: dict, method: str, path: str, body: dict = None resp = await client.request("DELETE", url, json=body, headers=headers) else: resp = await client.get(url, headers=headers) - + if resp.status_code in (200, 201): return resp.json() else: @@ -119,25 +170,23 @@ async def _wdtt_api_call(server: dict, method: str, path: str, body: dict = None return {"error": str(e)[:100]} async def sync_wdtt_password(user: dict): - """Создаёт или обновляет wdtt-пароль для премиум пользователя""" wdtt_servers = get_wdtt_servers() if not wdtt_servers: return - + tier = user.get("tier", "free") remaining_days = get_remaining_days(user) - + conn = get_db() try: db_user = conn.execute("SELECT * FROM users WHERE id = ?", (user["id"],)).fetchone() if not db_user: return db_user = dict(db_user) - + expires_at = db_user.get("tariff_end_at", 0) current_password = db_user.get("wdtt_password") - - # Если не премиум — удаляем пароль + if tier != "paid" or remaining_days <= 0: if current_password: for srv in wdtt_servers: @@ -148,13 +197,11 @@ async def sync_wdtt_password(user: dict): conn.commit() logger.info(f"WDTT password removed for user {user['username']}") return - - # Берём первый wdtt-сервер + server = wdtt_servers[0] server_id = servers.index(server) - + if current_password: - # Обновляем expiry result = await _wdtt_api_call(server, "POST", "/api/passwords/update", {"password": current_password, "expires_at": expires_at}) if result.get("status") == "ok": @@ -165,7 +212,6 @@ async def sync_wdtt_password(user: dict): else: logger.warning(f"Failed to update WDTT password for {user['username']}: {result.get('error')}") else: - # Генерируем новый пароль new_password = secrets.token_urlsafe(24) result = await _wdtt_api_call(server, "POST", "/api/passwords/add", {"password": new_password, "expires_at": expires_at}) @@ -182,7 +228,6 @@ async def sync_wdtt_password(user: dict): conn.close() async def remove_wdtt_password(user: dict): - """Удаляет wdtt-пароль пользователя""" conn = get_db() try: db_user = conn.execute("SELECT * FROM users WHERE id = ?", (user["id"],)).fetchone() @@ -192,11 +237,11 @@ async def remove_wdtt_password(user: dict): password = db_user.get("wdtt_password") if not password: return - + for srv in get_wdtt_servers(): await _wdtt_api_call(srv, "DELETE", "/api/passwords/remove", {"password": password}) - + conn.execute("UPDATE users SET wdtt_password = NULL, wdtt_expires_at = 0, wdtt_server_id = NULL WHERE id = ?", (user["id"],)) conn.commit() @@ -294,13 +339,13 @@ def render_template(name: str, **kwargs) -> str: path = os.path.join(WEB_DIR, name) try: with open(path, "r", encoding="utf-8") as f: - html = f.read() + html_content = f.read() except FileNotFoundError: logger.error(f"Template not found: {path}") return "Template error" for k, v in kwargs.items(): - html = html.replace(f"{{%{k}%}}", str(v)) - return html + html_content = html_content.replace(f"{{%{k}%}}", str(v)) + return html_content def get_flag_emoji(country_code: str) -> str: if not country_code or len(country_code) < 2: @@ -336,7 +381,6 @@ def get_motd() -> str: return f.read().strip() def get_random_hashes(count: int = 4) -> List[str]: - """Читает до `count` случайных хешей из hashes.txt""" if not os.path.exists(HASHES_FILE): return [] with open(HASHES_FILE, encoding="utf-8") as f: @@ -416,8 +460,10 @@ def get_traffic_stats(sub_id: str) -> dict: total_up = 0 total_down = 0 by_server = {} - + for srv in servers: + if not is_server_alive(srv.get("name", "")): + continue srv_up = 0 srv_down = 0 for inbound in srv.get("inbounds", []): @@ -425,14 +471,14 @@ def get_traffic_stats(sub_id: str) -> dict: api_user = inbound.get("api_user") api_pass = inbound.get("api_pass") inbound_id = inbound.get("id") - + if not all([api_host, api_user, api_pass, inbound_id]): continue - + try: api = Api(host=api_host, username=api_user, password=api_pass, use_tls_verify=False) api.login() - + inbounds = api.inbound.get_list() for ib in inbounds: if ib.id == inbound_id and ib.client_stats: @@ -441,56 +487,61 @@ def get_traffic_stats(sub_id: str) -> dict: srv_up += client.up or 0 srv_down += client.down or 0 except: - pass - + continue + if srv_up or srv_down: by_server[srv["name"]] = {"up": srv_up, "down": srv_down} total_up += srv_up total_down += srv_down - + return {"total_up": total_up, "total_down": total_down, "by_server": by_server} def generate_sub_id(length: int = 16) -> str: return ''.join(secrets.choice('abcdefghijklmnopqrstuvwxyz0123456789') for _ in range(length)) async def fetch_vless_links(url: str) -> List[str]: - async with httpx.AsyncClient(verify=False, timeout=10.0, follow_redirects=True) as client: - try: - resp = await client.get(url) - if resp.status_code == 404 or resp.status_code == 400: - logger.warning(f"Sub not found on server: {url}") + try: + async with httpx.AsyncClient(verify=False, timeout=10.0, follow_redirects=True) as client: + try: + resp = await client.get(url) + if resp.status_code == 404 or resp.status_code == 400: + logger.warning(f"Sub not found on server: {url}") + return [] + if resp.status_code == 200: + content = resp.text.strip() + try: + decoded = base64.b64decode(content).decode('utf-8') + links = re.findall(r'([a-z]+://[^\s\n]+)', decoded) + if links: + return links + except: + pass + return re.findall(r'([a-z]+://[^\s\n]+)', content) + except Exception as e: + logger.warning(f"Server unreachable {url}: {e}") return [] - if resp.status_code == 200: - content = resp.text.strip() - try: - decoded = base64.b64decode(content).decode('utf-8') - links = re.findall(r'([a-z]+://[^\s\n]+)', decoded) - if links: - return links - except: - pass - return re.findall(r'([a-z]+://[^\s\n]+)', content) - except Exception as e: - logger.error(f"Fetch error: {e}") + except Exception as e: + logger.warning(f"Fetch error for {url}: {e}") return [] def find_user_on_servers(sub_id: str) -> Optional[str]: - """Ищет юзера на 3x-UI серверах по subId, возвращает username""" logger.info(f"Looking for sub_id: {sub_id}") for srv in servers: + if not is_server_alive(srv.get("name", "")): + continue for inbound in srv.get("inbounds", []): api_host = inbound.get("api_host") api_user = inbound.get("api_user") api_pass = inbound.get("api_pass") inbound_id = inbound.get("id") - + if not all([api_host, api_user, api_pass, inbound_id]): continue - + try: api = Api(host=api_host, username=api_user, password=api_pass, use_tls_verify=False) api.login() - + inbounds = api.inbound.get_list() for ib in inbounds: if ib.id == inbound_id and ib.client_stats: @@ -506,15 +557,15 @@ def find_user_on_servers(sub_id: str) -> Optional[str]: logger.info(f"Migrated user: {username} (found on {srv['name']}/{inbound['name']})") return username except Exception as e: - logger.error(f"Error checking {srv['name']}/{inbound['name']}: {e}") - + logger.warning(f"Skipping unavailable server {srv['name']}/{inbound['name']}: {e}") + return None def get_user_tier(sub_id: str) -> Tuple[Optional[dict], str]: conn = get_db() try: user = conn.execute("SELECT * FROM users WHERE subscription_id = ?", (sub_id,)).fetchone() - + if not user: username = find_user_on_servers(sub_id) if username: @@ -528,17 +579,17 @@ VALUES (?, ?, 'free', 0, 0, 0, 0, 1) logger.info(f"User migrated: {username}") except sqlite3.IntegrityError: pass - + if not user: return None, "free" - + user = dict(user) - + if user['tier'] == 'paid' and get_remaining_days(user) <= 0: conn.execute("UPDATE users SET tier = 'free' WHERE id = ?", (user['id'],)) conn.commit() user['tier'] = 'free' - + return user, user['tier'] finally: conn.close() @@ -548,6 +599,8 @@ def get_servers_for_tier(tier: str) -> List[dict]: for srv in servers: if not srv.get("is_active"): continue + if not is_server_alive(srv.get("name", "")): + continue result.append(srv) return result @@ -558,7 +611,7 @@ def deduplicate_inbounds(servers_list: List[dict], tier: str) -> List[Tuple[dict for inbound in srv.get("inbounds", []): if tier == "free" and not inbound.get("is_free", True): continue - + key = (srv["name"], inbound.get("name", "")) if key not in seen: seen.add(key) @@ -571,37 +624,222 @@ async def lifespan(app): if ap_cfg.get("enabled", False): interval = max(10, ap_cfg.get("interval_minutes", 60)) * 60 logger.info(f"Auto-propagate enabled, interval={interval}s") - loop = asyncio.get_event_loop() - async def _run(): + async def _run_auto_propagate(): while True: for srv in servers: if not srv.get("is_active", True): continue + if not is_server_alive(srv.get("name", "")): + continue try: - result = await loop.run_in_executor(None, propagate_server_sync, srv["name"]) + result = await asyncio.get_event_loop().run_in_executor(None, propagate_server_sync, srv["name"]) if result.get("added", 0) or result.get("failed", 0): logger.info(f"Auto-propagate {srv['name']}: +{result.get('added',0)} added, {result.get('failed',0)} failed, {result.get('skipped',0)} skipped") except: logger.exception(f"Auto-propagate error on {srv.get('name','?')}") await asyncio.sleep(interval) - task = asyncio.create_task(_run()) + task = asyncio.create_task(_run_auto_propagate()) + _background_tasks.append(task) + + async def _run_rotate_shortids(): + while True: + try: + rotation_hours = settings.get("shortid_rotation_hours", 11) + await asyncio.sleep(rotation_hours * 3600) + for srv in servers: + if not is_server_alive(srv.get("name", "")): + continue + for inbound in srv.get("inbounds", []): + api_host = inbound.get("api_host") + api_user = inbound.get("api_user") + api_pass = inbound.get("api_pass") + if not all([api_host, api_user, api_pass]): + continue + try: + api = Api(host=api_host, username=api_user, password=api_pass, use_tls_verify=False) + api.login() + new_shortid = secrets.token_hex(4) + inbounds = api.inbound.get_inbounds() + for ib in inbounds: + if ib.id == inbound.get("id"): + ib.shortid = new_shortid + api.inbound.update_inbound(ib) + logger.info(f"ShortID rotated for {srv['name']}/{inbound['name']}: {new_shortid}") + break + except Exception as e: + logger.error(f"ShortID rotation error {srv['name']}/{inbound['name']}: {e}") + except Exception as e: + logger.error(f"ShortID rotation error: {e}") + await asyncio.sleep(3600) + + task2 = asyncio.create_task(_run_rotate_shortids()) + _background_tasks.append(task2) + + async def _run_poll_donationalerts(): + global last_donation_id + backoff = 1 + max_backoff = 3600 + while True: + try: + da_config = settings.get("payments", {}).get("donationalerts", {}) + if not da_config.get("enabled"): + await asyncio.sleep(300) + continue + + api_token = da_config.get("api_token", "") + if not api_token: + logger.warning("DonationAlerts API token not configured") + await asyncio.sleep(300) + continue + + interval = da_config.get("check_interval_minutes", 5) + sleep_time = max(interval * 60, backoff) + await asyncio.sleep(sleep_time) + + async with httpx.AsyncClient(timeout=30.0) as client: + resp = await client.get( + "https://www.donationalerts.com/api/v1/alerts/donations", + headers={"Authorization": f"Bearer {api_token}"} + ) + + if resp.status_code == 401: + logger.error(f"DA API returned 401 Unauthorized — token is invalid. Backing off for {max_backoff}s") + backoff = max_backoff + continue + elif resp.status_code != 200: + logger.error(f"DA API error: HTTP {resp.status_code}, backing off") + backoff = min(backoff * 2, max_backoff) + continue + + backoff = 1 + + data = resp.json() + donations = data.get("data", []) + + for donation in reversed(donations): + donation_id = donation.get("id") + async with _donation_lock: + if donation_id <= last_donation_id: + continue + last_donation_id = donation_id + + amount = donation.get("amount", 0) + username = donation.get("username", "") + message = donation.get("message", "") + + tiers_config = settings.get("tiers", {}) + + tier = None + days = 0 + + for tier_name, tier_data in tiers_config.items(): + prices = tier_data.get("prices", {}) + tier_days = tier_data.get("days", {}) + for price_key, price_val in prices.items(): + if amount == price_val: + tier = tier_name + days = tier_days.get(price_key, 30) + break + if tier: + break + + if not tier: + logger.info(f"DA: ignoring amount {amount} RUB (not in config)") + continue + + user = None + message_parts = message.split() if message else [] + + conn = get_db() + try: + for part in message_parts: + if part.isdigit(): + user = conn.execute("SELECT * FROM users WHERE id = ?", (int(part),)).fetchone() + if not user: + user = conn.execute("SELECT * FROM users WHERE username = ? COLLATE NOCASE", (part,)).fetchone() + if user: + break + + if not user and username: + user = conn.execute("SELECT * FROM users WHERE username = ? COLLATE NOCASE", (username,)).fetchone() + + if user: + now_ts = int(time.time()) + current_end = user.get("tariff_end_at", 0) + if current_end > now_ts: + new_end = current_end + days * 86400 + else: + new_end = now_ts + days * 86400 + + conn.execute(""" + UPDATE users SET + tier = ?, + tariff_days_bought = tariff_days_bought + ?, + tariff_end_at = ?, + total_paid_rubles = total_paid_rubles + ? + WHERE id = ? + """, (tier, days, new_end, amount, user["id"])) + conn.commit() + + sub_id = user["subscription_id"] + update_3xui_expiry(sub_id, days) + + updated = conn.execute("SELECT * FROM users WHERE id = ?", (user["id"],)).fetchone() + if updated: + asyncio.create_task(sync_wdtt_password(dict(updated))) + + logger.info(f"VPN payment: {username} paid {amount} RUB, tier={tier}, +{days} days") + finally: + conn.close() + + except Exception as e: + logger.error(f"DonationAlerts polling error: {e}") + await asyncio.sleep(300) + + task3 = asyncio.create_task(_run_poll_donationalerts()) + _background_tasks.append(task3) + + task4 = asyncio.create_task(_server_health_loop()) + _background_tasks.append(task4) + yield - if ap_cfg.get("enabled", False): - task.cancel() + + for t in _background_tasks: + t.cancel() + await asyncio.gather(*_background_tasks, return_exceptions=True) app = FastAPI(title="ZernProxy Manager", docs_url=None, redoc_url=None, lifespan=lifespan) -app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) + +allowed_host = settings.get("general", {}).get("host", "") +app.add_middleware( + CORSMiddleware, + allow_origins=[f"https://{allowed_host}", f"http://{allowed_host}"] if allowed_host else [], + allow_credentials=True, + allow_methods=["GET", "POST"], + allow_headers=["Content-Type", "Authorization", "X-Webhook-Secret"], +) app.mount("/web", StaticFiles(directory=WEB_DIR), name="web") ADMIN_USER = settings.get("admin", {}).get("username", "admin") ADMIN_PASS = settings.get("admin", {}).get("password", "") +@app.middleware("http") +async def add_security_headers(request: Request, call_next): + response = await call_next(request) + ctype = response.headers.get("content-type", "") + if "text/html" in ctype: + response.headers["Content-Security-Policy"] = "default-src 'self'; img-src 'self' data:; style-src 'self' 'unsafe-inline'; script-src 'self' 'unsafe-inline'; font-src 'self' data:;" + response.headers["X-Content-Type-Options"] = "nosniff" + response.headers["X-Frame-Options"] = "DENY" + return response + def check_admin(request: Request) -> bool: if not ADMIN_PASS: return True token = request.cookies.get("admin_token", "") - if token == ADMIN_PASS: - return True + if token: + if token == ADMIN_PASS: + return True auth = request.headers.get("Authorization", "") if auth.startswith("Basic "): try: @@ -615,94 +853,103 @@ def check_admin(request: Request) -> bool: @app.get("/admin/login") async def admin_login_page(): - return HTMLResponse(content=render_template("admin_login.html", title="Вход", error_msg="")) + return HTMLResponse(content=render_template("admin_login.html", title=h(settings.get("general", {}).get("title", "ZernProxy")), error_msg="")) @app.post("/admin/login") async def admin_login(request: Request): + ip = request.client.host if request.client else "unknown" + if is_ratelimited(ip): + return HTMLResponse( + content=render_template("admin_login.html", title=h(settings.get("general", {}).get("title", "ZernProxy")), + error_msg='
Слишком много попыток. Подождите.
'), + status_code=429) + data = await request.form() if data.get("username") == ADMIN_USER and data.get("password") == ADMIN_PASS: resp = RedirectResponse(url="/admin/users", status_code=303) - resp.set_cookie(key="admin_token", value=ADMIN_PASS, max_age=86400 * 7, httponly=True) + resp.set_cookie(key="admin_token", value=ADMIN_PASS, max_age=86400 * 7, httponly=True, secure=True, samesite="lax") return resp - return HTMLResponse(content=render_template("admin_login.html", title="Вход", error_msg='
Неверный пароль
'), status_code=401) + return HTMLResponse( + content=render_template("admin_login.html", title=h(settings.get("general", {}).get("title", "ZernProxy")), + error_msg='
Неверный пароль
'), + status_code=401) @app.get("/sub/{subscription_id}") async def get_subscription(request: Request, subscription_id: str, format: str = Query("base64", pattern="^(json|base64|raw)$")): accept = request.headers.get("accept", "") if "text/html" in accept and format == "base64": return await get_web_page(subscription_id) - + user, tier = get_user_tier(subscription_id) if not user: raise HTTPException(404, "User not found") - + if not user.get("is_active"): raise HTTPException(403, "User is disabled") - + tier_config = settings.get("tiers", {}).get(tier, {}) servers_for_tier = get_servers_for_tier(tier) inbounds = deduplicate_inbounds(servers_for_tier, tier) - - if not inbounds: - raise HTTPException(404, "No servers available") - + all_links = get_cached_links(subscription_id) if all_links is None: all_links = [] seen_links = set() servers_processed = set() server_links = {} - + for srv, inbound in inbounds: srv_name = srv["name"] - + if srv_name not in servers_processed: servers_processed.add(srv_name) sub_path = srv["sub_path"].format(sub_id=subscription_id) url = f"{srv['subscription_url'].rstrip('/')}{sub_path}" - server_links[srv_name] = await fetch_vless_links(url) - + fetched = await fetch_vless_links(url) + if fetched: + server_links[srv_name] = fetched + links = server_links.get(srv_name, []) + if not links: + continue + srv_inbounds = [ib for s, ib in inbounds if s["name"] == srv_name] try: link_idx = srv_inbounds.index(inbound) except ValueError: continue - + if link_idx >= len(links): continue - + link = links[link_idx] clean_link = link.split('#')[0] if clean_link in seen_links: continue seen_links.add(clean_link) - + remark = f"{get_flag_emoji(srv.get('country', ''))} {srv_name.upper()} ({inbound['name']})" all_links.append(f"{clean_link}#{remark}") - + set_cached_links(subscription_id, all_links) - - if not all_links: - raise HTTPException(404, "No links found") - + if format == "json": return {"links": all_links, "count": len(all_links), "tier": tier} - + lines = [] motd_text = get_motd() announce_header = "" - + if motd_text: announce_header = f"base64:{base64.b64encode(motd_text.encode()).decode()}" elif settings.get("announcement"): announce_header = f"base64:{base64.b64encode(settings['announcement'].encode()).decode()}" - + host = settings.get("general", {}).get("host", "conn.zernmc.ru") title = settings.get("general", {}).get("title", "ZernProxy") support_url = settings.get("general", {}).get("support_url", "") update_interval = 12 - + web_url = f"https://{host}/sub/{subscription_id}" lines.append(f"#profile-web-page-url: {web_url}") lines.append(f"#profile-title: {title}") @@ -710,7 +957,7 @@ async def get_subscription(request: Request, subscription_id: str, format: str = lines.append("#hide-settings: 1") if support_url: lines.append(f"#support-url: {support_url}") - + expire_ts = user.get("tariff_end_at", 0) or 0 rem_days = get_remaining_days(user) if rem_days > 0: @@ -719,25 +966,25 @@ async def get_subscription(request: Request, subscription_id: str, format: str = lines.append("#sub-expire: 1") if support_url: lines.append(f"#sub-expire-button-link: {support_url}") - + traffic_limit = user.get("traffic_limit_gb") or tier_config.get("traffic_limit_gb", 0) traffic_limit_bytes = traffic_limit * 1073741824 if traffic_limit > 0 else 0 - + traffic = get_cached_traffic(subscription_id) if not traffic: traffic = get_traffic_stats(subscription_id) set_cached_traffic(subscription_id, traffic) upload = traffic.get("total_up", 0) download = traffic.get("total_down", 0) - + lines.append(f"#subscription-userinfo: upload={upload}; download={download}; total={traffic_limit_bytes}; expire={expire_ts}") - + if announce_header: lines.append(f"#announce: {announce_header}") - + lines.extend(all_links) content = "\n".join(lines) - + headers = { "Profile-Title": title, "Profile-Update-Interval": str(update_interval), @@ -754,8 +1001,7 @@ async def get_subscription(request: Request, subscription_id: str, format: str = headers["Sub-Expire"] = "1" if support_url: headers["Sub-Expire-Button-Link"] = support_url - - # WDTT-специфичные заголовки для клиента ZernProxy + is_wdtt_client = request.query_params.get("client") == "wdtt" or "WDTT" in request.headers.get("User-Agent", "") if is_wdtt_client or user.get("wdtt_password"): wdtt_servers = get_wdtt_servers() @@ -770,20 +1016,19 @@ async def get_subscription(request: Request, subscription_id: str, format: str = "country": s.get("country", ""), }) headers["X-WDTT-Servers"] = json.dumps(server_list) - + if user.get("wdtt_password"): headers["X-WDTT-Credentials"] = json.dumps({ "password": user["wdtt_password"], "expires_at": user.get("wdtt_expires_at", 0), "server_id": user.get("wdtt_server_id", 0), }) - - # ZernHash: случайные хеши для клиента + if is_wdtt_client or "ZernHash" in request.headers.get("User-Agent", ""): hashes = get_random_hashes(4) if hashes: headers["X-WDTT-Hashes"] = json.dumps(hashes) - + if format == "base64": return Response(content=base64.b64encode(content.encode()).decode(), media_type="text/plain; charset=utf-8", headers=headers) return Response(content=content, media_type="text/plain; charset=utf-8", headers=headers) @@ -792,30 +1037,30 @@ async def get_web_page(subscription_id: str): user, tier = get_user_tier(subscription_id) if not user: raise HTTPException(404, "User not found") - + tier_config = settings.get("tiers", {}).get(tier, {}) servers_for_tier = get_servers_for_tier(tier) inbounds = deduplicate_inbounds(servers_for_tier, tier) - + host = settings.get("general", {}).get("host", "conn.zernmc.ru") title = settings.get("general", {}).get("title", "ZernProxy") announcement = get_motd() or settings.get("announcement", "") da_config = settings.get("payments", {}).get("donationalerts", {}) - + sub_url = f"https://{host}/sub/{subscription_id}" qr_base64 = generate_qr_base64(sub_url, size=300) logo_base64 = get_logo_base64() - + logo_html = f'Logo' if logo_base64 else '
⚡
' - announcement_html = f'
{announcement}
' if announcement else '' - + announcement_html = f'
{h(announcement)}
' if announcement else '' + tier_color = "#4CAF50" if tier == "paid" else "#757575" tier_name = tier_config.get("name", "Free") - tier_badge = f'{tier_name}' - + tier_badge = f'{h(tier_name)}' + days_remaining = get_remaining_days(user) days_info = f"

⏳ Осталось дней: {days_remaining}

" if tier == "paid" and days_remaining > 0 else "" - + traffic = get_cached_traffic(subscription_id) if not traffic: traffic = get_traffic_stats(subscription_id) @@ -823,58 +1068,58 @@ async def get_web_page(subscription_id: str): traffic_limit_val = user.get('traffic_limit_gb') or tier_config.get('traffic_limit_gb', 0) traffic_limit_str = "∞" if traffic_limit_val == 0 else f"{traffic_limit_val} GB" traffic_info = f"

📊 Лимит: {traffic_limit_str}

⬆️ {format_bytes(traffic['total_up'])} | ⬇️ {format_bytes(traffic['total_down'])}

" - + traffic_details = "" if traffic.get("by_server"): for srv_name, data in traffic["by_server"].items(): if data["up"] or data["down"]: - traffic_details += f'
{srv_name}: ⬆ {format_bytes(data["up"])} ⬇ {format_bytes(data["down"])}
' - + traffic_details += f'
{h(srv_name)}: ⬆ {format_bytes(data["up"])} ⬇ {format_bytes(data["down"])}
' + info_html = f'
{days_info}{traffic_info}{traffic_details}
' if days_info or traffic_info or traffic_details else '' - - servers_html = "".join(f'{get_flag_emoji(srv.get("country", ""))} {srv["name"].upper()}' for srv in servers_for_tier) - + + servers_html = "".join(f'{get_flag_emoji(srv.get("country", ""))} {h(srv["name"].upper())}' for srv in servers_for_tier) + support_btn = "" if da_config.get("enabled"): da_url = da_config.get("url", "#") support_btn = f''' - + Поддержать проект ''' - + html = render_template("sub.html", - title=title, logo_html=logo_html, tier_badge=tier_badge, + title=h(title), logo_html=logo_html, tier_badge=tier_badge, announcement_html=announcement_html, info_html=info_html, servers_html=servers_html, qr_base64=qr_base64, - sub_url=sub_url, support_btn=support_btn) + sub_url=h(sub_url), support_btn=support_btn) return HTMLResponse(content=html) @app.post("/payment/webhook/donationalerts") async def webhook_donationalerts(request: Request): da_config = settings.get("payments", {}).get("donationalerts", {}) webhook_secret = da_config.get("webhook_secret", "") - + if webhook_secret: provided = request.headers.get("X-Webhook-Secret", "") if provided != webhook_secret: return JSONResponse({"error": "Invalid secret"}, status_code=403) - + try: data = await request.json() except: return JSONResponse({"error": "Invalid JSON"}, status_code=400) - + amount = data.get("amount", 0) username = data.get("username", "") message = data.get("message", "") donation_id = data.get("id", 0) - + logger.info(f"DA webhook: id={donation_id} amount={amount} username={username}") - + user = None message_parts = message.split() if message else [] - + for part in message_parts: conn = get_db() try: @@ -886,21 +1131,21 @@ async def webhook_donationalerts(request: Request): break finally: conn.close() - + if not user and username: conn = get_db() try: user = conn.execute("SELECT * FROM users WHERE username = ? COLLATE NOCASE", (username,)).fetchone() finally: conn.close() - + if not user: return JSONResponse({"status": "ignored", "reason": "user_not_found"}) - + tiers_config = settings.get("tiers", {}) tier = None days = 0 - + for tier_name, tier_data in tiers_config.items(): prices = tier_data.get("prices", {}) tier_days = tier_data.get("days", {}) @@ -911,15 +1156,15 @@ async def webhook_donationalerts(request: Request): break if tier: break - + if not tier: return JSONResponse({"status": "ignored", "reason": "amount_not_recognized"}) - + conn = get_db() try: now_ts = int(time.time()) conn.execute(""" - UPDATE users SET + UPDATE users SET tier = ?, tariff_days_bought = tariff_days_bought + ?, tariff_end_at = MAX(COALESCE(tariff_end_at, 0), ?) + ? * 86400, @@ -927,63 +1172,63 @@ async def webhook_donationalerts(request: Request): WHERE id = ? """, (tier, days, now_ts, days, amount, user["id"])) conn.commit() - + updated = conn.execute("SELECT * FROM users WHERE id = ?", (user["id"],)).fetchone() if updated: asyncio.create_task(sync_wdtt_password(dict(updated))) finally: conn.close() - + logger.info(f"VPN payment: {username} paid {amount} RUB, +{days} days") return JSONResponse({"status": "ok", "user": user["username"], "days": days}) @app.get("/admin/users") async def admin_users(request: Request): if not check_admin(request): - return HTMLResponse(content=render_template("admin_login.html", title="Вход", error_msg="")) + return HTMLResponse(content=render_template("admin_login.html", title=h(settings.get("general", {}).get("title", "ZernProxy")), error_msg="")) conn = get_db() try: users = conn.execute("SELECT * FROM users ORDER BY created_at DESC").fetchall() finally: conn.close() - + da_config = settings.get("payments", {}).get("donationalerts", {}) - + title = settings.get("general", {}).get("title", "ZernProxy") tier_options = "".join( - f'' + f'' for k, v in settings.get("tiers", {}).items()) users_rows = "" for u in users: traffic = f'{u["traffic_limit_gb"]} GB' if u['traffic_limit_gb'] > 0 else '∞' active = '✓' if u['is_active'] else '✗' - tier_display = u['tier'].upper() + tier_display = h(u['tier'].upper()) rem_days = get_remaining_days(u) users_rows += f''' {u['id']} - {u['username']} - {u['subscription_id']} - {tier_display} + {h(u['username'])} + {h(u['subscription_id'])} + {tier_display} {rem_days} {u['total_paid_rubles']}₽ {traffic} {active} {u['created_at'][:10]} - - + + ''' html = render_template("admin_users.html", - title=title, user_count=len(users), users_rows=users_rows, + title=h(title), user_count=len(users), users_rows=users_rows, tier_options=tier_options) return HTMLResponse(content=html) @app.get("/admin/dashboard") async def admin_dashboard(request: Request): if not check_admin(request): - return HTMLResponse(content=render_template("admin_login.html", title="Вход", error_msg="")) - + return HTMLResponse(content=render_template("admin_login.html", title=h(settings.get("general", {}).get("title", "ZernProxy")), error_msg="")) + conn = get_db() try: total_users = conn.execute("SELECT COUNT(*) as c FROM users").fetchone()["c"] @@ -994,9 +1239,11 @@ async def admin_dashboard(request: Request): total_donations = conn.execute("SELECT SUM(total_paid_rubles) as s FROM users WHERE total_paid_rubles > 0").fetchone()["s"] or 0 finally: conn.close() - + online_count = 0 for srv in servers: + if not is_server_alive(srv.get("name", "")): + continue for inbound in srv.get("inbounds", []): try: api = Api(host=inbound["api_host"], username=inbound["api_user"], password=inbound["api_pass"], use_tls_verify=False) @@ -1005,13 +1252,13 @@ async def admin_dashboard(request: Request): online_count += len(online) except: pass - + title = settings.get("general", {}).get("title", "ZernProxy") servers_rows = "".join( - f'
{get_flag_emoji(s.get("country",""))} {s["name"].upper()}{len(s.get("inbounds",[]))} inbound
' + f'
{get_flag_emoji(s.get("country",""))} {h(s["name"].upper())}{len(s.get("inbounds",[]))} inbound
' for s in servers) html = render_template("admin_dashboard.html", - title=title, total_users=total_users, free_users=free_users, + title=h(title), total_users=total_users, free_users=free_users, paid_users=paid_users, test_users=test_users, online_count=online_count, total_revenue=total_revenue, server_count=len(servers), @@ -1023,9 +1270,9 @@ async def admin_dashboard(request: Request): async def home_page(): title = settings.get("general", {}).get("title", "ZernProxy") da_url = settings.get("payments", {}).get("donationalerts", {}).get("url", "") - + statuses = await fetch_servers_status() - + srv_cards = "" for i, s in enumerate(statuses): chk = s.get("checks", {}) @@ -1035,7 +1282,7 @@ async def home_page(): srv_name = s.get("server_name", s["name"].upper()) delay = 0.45 + i * 0.15 online = s.get("online", False) - + srv_inbounds = [] for srv_cfg in servers: if srv_cfg["name"] == s["name"]: @@ -1047,14 +1294,14 @@ async def home_page(): else: has_free = True has_paid = True - + if has_free and has_paid: badge = 'Free+Premium' elif has_free: badge = 'Free' else: badge = 'Premium' - + if online: _, hc, hl = calc_health_score(chk) s1 = cpu if cpu is not None else 0 @@ -1065,21 +1312,21 @@ async def home_page(): hc = "#475569" hl = "Offline" stats = '
Offline
' - + inbound_cards = "" for ib in srv_inbounds: ib_name = ib.get("name", "unknown") ib_type = ib.get("network", ib.get("flow", "")) - inbound_cards += f'
{ib_name}{ib_type}
\n' - + inbound_cards += f'
{h(ib_name)}{h(ib_type)}
\n' + ru_note = "" if s.get("country", "").lower() == "ru" and online: ru_note = '
★ Нет рекламы в YouTube · Адаптирован к будущему ограничению 15 ГБ зарубежного трафика через Proxy
' - + srv_cards += f'''
- {get_flag_emoji(s.get("country",""))} {srv_name} {badge} + {get_flag_emoji(s.get("country",""))} {h(srv_name)} {badge} {hl}
@@ -1087,8 +1334,8 @@ async def home_page():
{inbound_cards}
{ru_note}
''' - - return HTMLResponse(content=render_template("home.html", title=title, da_url=da_url, srv_cards=srv_cards)) + + return HTMLResponse(content=render_template("home.html", title=h(title), da_url=h(da_url), srv_cards=srv_cards)) async def fetch_servers_status() -> list: @@ -1110,6 +1357,7 @@ async def fetch_servers_status() -> list: else: entry["checks"]["error"] = {"value": f"HTTP {resp.status_code}"} except Exception as e: + logger.warning(f"Status check failed for {srv.get('name','?')}: {e}") entry["checks"]["error"] = {"value": str(e)[:50]} results.append(entry) return results @@ -1122,25 +1370,24 @@ async def tg_webhook(): return JSONResponse({"status": "ok", "bot": False, "message": "Telegram bot not configured"}) def create_3xui_client(username: str, sub_id: str, inbound: dict, traffic_gb: int = 0) -> dict: - """Создаёт клиента на 3x-UI сервере""" api_host = inbound.get("api_host") api_user = inbound.get("api_user") api_pass = inbound.get("api_pass") inbound_id = inbound.get("id") - + if not all([api_host, api_user, api_pass, inbound_id]): return {"success": False, "error": "missing_credentials"} - + try: api = Api(host=api_host, username=api_user, password=api_pass, use_tls_verify=False) api.login() - + inbound_name = inbound.get('name', 'default') email = f"{username}_{inbound_name}@vless.local" total_bytes = traffic_gb * 1073741824 if traffic_gb > 0 else 0 - + logger.info(f"Creating client: email={email}, total_bytes={total_bytes}, inbound={inbound_name}") - + from py3xui.client import Client if inbound.get("protocol") == "trojan": client = Client( @@ -1163,20 +1410,20 @@ def create_3xui_client(username: str, sub_id: str, inbound: dict, traffic_gb: in limit_ip=0, subId=sub_id ) - + try: existing = api.client.get_by_email(email) if existing: logger.info(f"Deleting existing client: {existing.id}") - api.client.delete(existing.id, inbound_id) + api.client.delete(inbound_id, existing.id) except: pass - + logger.info(f"About to add client: {client}") api.client.add(inbound_id=inbound_id, clients=[client]) logger.info(f"Client created successfully on {inbound_name}") return {"success": True, "email": email} - + except Exception as e: logger.error(f"Error creating client: {e}") import traceback @@ -1184,19 +1431,18 @@ def create_3xui_client(username: str, sub_id: str, inbound: dict, traffic_gb: in return {"success": False, "error": str(e)[:150]} def delete_3xui_client(username: str, sub_id: str, inbound: dict) -> dict: - """Удаляет клиента с 3x-UI сервера""" api_host = inbound.get("api_host") api_user = inbound.get("api_user") api_pass = inbound.get("api_pass") inbound_id = inbound.get("id") - + if not all([api_host, api_user, api_pass, inbound_id]): return {"success": False, "error": "missing_credentials"} - + try: api = Api(host=api_host, username=api_user, password=api_pass, use_tls_verify=False) api.login() - + inbounds = api.inbound.get_list() for ib in inbounds: if ib.id == inbound_id and ib.client_stats: @@ -1205,9 +1451,9 @@ def delete_3xui_client(username: str, sub_id: str, inbound: dict) -> dict: api.client.delete(inbound_id, client.uuid) logger.info(f"Deleted client from {inbound.get('name')}") return {"success": True} - + return {"success": True, "error": "not_found"} - + except Exception as e: return {"success": False, "error": str(e)[:100]} @@ -1252,14 +1498,16 @@ def propagate_server_sync(server_name: str) -> dict: target_srv = next((s for s in servers if s["name"] == server_name), None) if not target_srv: return {"error": "Server not found"} + if not is_server_alive(server_name): + return {"error": "Target server is dead"} conn = get_db() try: users = conn.execute("SELECT username, subscription_id, traffic_limit_gb, is_active FROM users WHERE is_active = 1").fetchall() finally: conn.close() - other_servers = [s for s in servers if s.get("is_active", True) and s["name"] != server_name] + other_servers = [s for s in servers if s.get("is_active", True) and is_server_alive(s.get("name", "")) and s["name"] != server_name] threshold = max(1, len(other_servers) // 2 + 1) - + target_ib_ids = fetch_server_sub_ids(target_srv) other_ids = {} for s in other_servers: @@ -1268,24 +1516,24 @@ def propagate_server_sync(server_name: str) -> dict: for ids in ib_dict.values(): flat.update(ids) other_ids[s["name"]] = flat - + total = len(users) added = skipped = failed = 0 inbound_stats = {} for ib in target_srv.get("inbounds", []): inbound_stats[ib.get("name", f"id_{ib.get('id','?')}")] = 0 - + results = [] for u in users: sub_id = u["subscription_id"] username = u["username"] - + count = sum(1 for s in other_servers if sub_id in other_ids.get(s["name"], set())) if count < threshold: skipped += 1 results.append({"username": username, "skipped": True, "servers": count, "total": len(other_servers)}) continue - + user_added = False for inbound in target_srv.get("inbounds", []): ib_id = inbound.get("id") @@ -1301,11 +1549,11 @@ def propagate_server_sync(server_name: str) -> dict: else: failed += 1 results.append({"username": username, "inbound": ib_name, "error": r.get("error", "")}) - + if user_added: added += 1 results.append({"username": username, "added": True}) - + already = sum(1 for u in users if u["subscription_id"] in {sid for ids in target_ib_ids.values() for sid in ids}) return {"server": server_name, "total": total, "already_on_server": already, "added": added, "skipped": skipped, "failed": failed, @@ -1328,15 +1576,17 @@ async def create_user(request: Request, data: dict): username = data.get("username", "").strip() if not username: return JSONResponse({"error": "username required"}, status_code=400) - + sub_id = generate_sub_id() conn = get_db() - + traffic_gb = int(data.get("traffic_limit_gb", 0) or 0) logger.info(f"Creating user: username={username}, traffic_gb={traffic_gb}") - + results = [] for srv in servers: + if not is_server_alive(srv.get("name", "")): + continue srv_result = {"server": srv["name"], "inbounds": []} for inbound in srv.get("inbounds", []): result = create_3xui_client(username, sub_id, inbound, traffic_gb) @@ -1346,7 +1596,7 @@ async def create_user(request: Request, data: dict): "error": result.get("error", "") }) results.append(srv_result) - + try: conn.execute(""" INSERT INTO users (username, subscription_id, tier, tariff_days_bought, tariff_days_remaining, total_paid_rubles, traffic_limit_gb, is_active) @@ -1358,10 +1608,10 @@ async def create_user(request: Request, data: dict): return JSONResponse({"error": "username exists"}, status_code=400) finally: conn.close() - + success_count = sum(1 for r in results for ib in r["inbounds"] if ib["success"]) total_count = sum(len(r["inbounds"]) for r in results) - + return JSONResponse({ "status": "ok", "username": username, @@ -1375,17 +1625,16 @@ async def create_user(request: Request, data: dict): async def update_user(request: Request, data: dict): if not check_admin(request): return JSONResponse({"error": "Unauthorized"}, status_code=401) - logger.info(f"RAW data received: {data}") - + user_id = int(data.get("id", 0)) tier = str(data.get("tier", "free")) tariff_days_remaining = int(data.get("tariff_days_remaining", 0) or 0) tariff_end_at = int(time.time()) + tariff_days_remaining * 86400 if tariff_days_remaining > 0 else 0 traffic_limit_gb = int(data.get("traffic_limit_gb", 0) or 0) is_active = 1 if data.get("is_active") in [True, "true", "on", "1", 1] else 0 - + logger.info(f"Update user: id={user_id}, tier='{tier}', days={tariff_days_remaining}, traffic={traffic_limit_gb}, active={is_active}") - + conn = get_db() try: conn.execute(""" @@ -1393,18 +1642,17 @@ async def update_user(request: Request, data: dict): WHERE id = ? """, (tier, tariff_days_remaining, tariff_end_at, traffic_limit_gb, is_active, user_id)) conn.commit() - + updated = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone() - logger.info(f"After update DB: {dict(updated)}") finally: conn.close() - + if updated: updated = dict(updated) updated["tier"] = tier clear_cache(updated["subscription_id"]) asyncio.create_task(sync_wdtt_password(updated)) - + return JSONResponse({"status": "ok"}) @app.post("/admin/api/users/delete") @@ -1413,82 +1661,48 @@ async def delete_user(request: Request, data: dict): return JSONResponse({"error": "Unauthorized"}, status_code=401) user_id = data.get("id") username = data.get("username", "") - + conn = get_db() try: user = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone() if not user: return JSONResponse({"error": "user not found"}, status_code=404) - + user = dict(user) sub_id = user["subscription_id"] - + for srv in servers: + if not is_server_alive(srv.get("name", "")): + continue for inbound in srv.get("inbounds", []): - result = delete_3xui_client(username, sub_id, inbound) - + delete_3xui_client(username, sub_id, inbound) + asyncio.create_task(remove_wdtt_password(user)) - + conn.execute("DELETE FROM users WHERE id = ?", (user_id,)) conn.commit() finally: conn.close() - + return JSONResponse({"status": "ok", "message": "User deleted from all servers"}) -async def rotate_shortids(): - while True: - try: - rotation_hours = settings.get("shortid_rotation_hours", 11) - await asyncio.sleep(rotation_hours * 3600) - - for srv in servers: - for inbound in srv.get("inbounds", []): - api_host = inbound.get("api_host") - api_user = inbound.get("api_user") - api_pass = inbound.get("api_pass") - - if not all([api_host, api_user, api_pass]): - continue - - try: - api = Api(host=api_host, username=api_user, password=api_pass, use_tls_verify=False) - api.login() - - new_shortid = secrets.token_hex(4) - - inbounds = api.inbound.get_inbounds() - for ib in inbounds: - if ib.id == inbound.get("id"): - ib.shortid = new_shortid - api.inbound.update_inbound(ib) - logger.info(f"ShortID rotated for {srv['name']}/{inbound['name']}: {new_shortid}") - break - except Exception as e: - logger.error(f"ShortID rotation error {srv['name']}/{inbound['name']}: {e}") - except Exception as e: - logger.error(f"ShortID rotation error: {e}") - await asyncio.sleep(3600) - def update_3xui_expiry(sub_id: str, days: int): - """Обновляет expiry_time на всех 3x-UI серверах""" - import time expiry_timestamp = int(time.time() + (days * 86400)) - + for srv in servers: for inbound in srv.get("inbounds", []): api_host = inbound.get("api_host") api_user = inbound.get("api_user") api_pass = inbound.get("api_pass") inbound_id = inbound.get("id") - + if not all([api_host, api_user, api_pass, inbound_id]): continue - + try: api = Api(host=api_host, username=api_user, password=api_pass, use_tls_verify=False) api.login() - + inbounds = api.inbound.get_list() for ib in inbounds: if ib.id == inbound_id and ib.client_stats: @@ -1499,119 +1713,7 @@ def update_3xui_expiry(sub_id: str, days: int): logger.info(f"Updated expiry for {srv['name']}/{inbound['name']}: {days} days") break except Exception as e: - logger.error(f"Error updating expiry {srv['name']}/{inbound['name']}: {e}") - -async def poll_donationalerts(): - global last_donation_id - while True: - try: - da_config = settings.get("payments", {}).get("donationalerts", {}) - if not da_config.get("enabled"): - await asyncio.sleep(300) - continue - - api_token = da_config.get("api_token", "") - if not api_token: - logger.warning("DonationAlerts API token not configured") - await asyncio.sleep(300) - continue - - interval = da_config.get("check_interval_minutes", 5) - await asyncio.sleep(interval * 60) - - async with httpx.AsyncClient(timeout=30.0) as client: - resp = await client.get( - "https://www.donationalerts.com/api/v1/alerts/donations", - headers={"Authorization": f"Bearer {api_token}"} - ) - - if resp.status_code != 200: - logger.error(f"DA API error: {resp.status_code}") - continue - - data = resp.json() - donations = data.get("data", []) - - for donation in reversed(donations): - donation_id = donation.get("id") - if donation_id <= last_donation_id: - continue - - amount = donation.get("amount", 0) - username = donation.get("username", "") - message = donation.get("message", "") - - tiers_config = settings.get("tiers", {}) - - tier = None - days = 0 - - for tier_name, tier_data in tiers_config.items(): - prices = tier_data.get("prices", {}) - tier_days = tier_data.get("days", {}) - for price_key, price_val in prices.items(): - if amount == price_val: - tier = tier_name - days = tier_days.get(price_key, 30) - break - if tier: - break - - if not tier: - logger.info(f"DA: ignoring amount {amount} RUB (not in config)") - last_donation_id = donation_id - continue - - user = None - message_parts = message.split() if message else [] - - conn = get_db() - try: - for part in message_parts: - if part.isdigit(): - user = conn.execute("SELECT * FROM users WHERE id = ?", (int(part),)).fetchone() - if not user: - user = conn.execute("SELECT * FROM users WHERE username = ? COLLATE NOCASE", (part,)).fetchone() - if user: - break - - if not user and username: - user = conn.execute("SELECT * FROM users WHERE username = ? COLLATE NOCASE", (username,)).fetchone() - - if user: - now_ts = int(time.time()) - current_end = user.get("tariff_end_at", 0) - if current_end > now_ts: - new_end = current_end + days * 86400 - else: - new_end = now_ts + days * 86400 - - conn.execute(""" - UPDATE users SET - tier = ?, - tariff_days_bought = tariff_days_bought + ?, - tariff_end_at = ?, - total_paid_rubles = total_paid_rubles + ? - WHERE id = ? - """, (tier, days, new_end, amount, user["id"])) - conn.commit() - - sub_id = user["subscription_id"] - update_3xui_expiry(sub_id, days) - - updated = conn.execute("SELECT * FROM users WHERE id = ?", (user["id"],)).fetchone() - if updated: - asyncio.create_task(sync_wdtt_password(dict(updated))) - - logger.info(f"VPN payment: {username} paid {amount} RUB, tier={tier}, +{days} days") - finally: - conn.close() - - last_donation_id = donation_id - - except Exception as e: - logger.error(f"DonationAlerts polling error: {e}") - await asyncio.sleep(300) + logger.warning(f"Skipping unavailable server {srv['name']}/{inbound['name']} for expiry update: {e}") @app.get("/health") async def health(): @@ -1625,6 +1727,31 @@ async def reload_configs(request: Request): clear_cache() return {"status": "ok"} +async def rotate_shortids(): + rotation_hours = settings.get("shortid_rotation_hours", 11) + for srv in servers: + if not is_server_alive(srv.get("name", "")): + continue + for inbound in srv.get("inbounds", []): + api_host = inbound.get("api_host") + api_user = inbound.get("api_user") + api_pass = inbound.get("api_pass") + if not all([api_host, api_user, api_pass]): + continue + try: + api = Api(host=api_host, username=api_user, password=api_pass, use_tls_verify=False) + api.login() + new_shortid = secrets.token_hex(4) + inbounds = api.inbound.get_inbounds() + for ib in inbounds: + if ib.id == inbound.get("id"): + ib.shortid = new_shortid + api.inbound.update_inbound(ib) + logger.info(f"ShortID rotated for {srv['name']}/{inbound['name']}: {new_shortid}") + break + except Exception as e: + logger.warning(f"ShortID rotation skipped for {srv['name']}/{inbound['name']}: {e}") + @app.get("/admin/api/rotate-shortids") async def manual_rotate(request: Request): if not check_admin(request): @@ -1633,8 +1760,4 @@ async def manual_rotate(request: Request): return {"status": "ok"} if __name__ == "__main__": - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - loop.create_task(rotate_shortids()) - loop.create_task(poll_donationalerts()) - uvicorn.run(app, host="0.0.0.0", port=8000, log_level="info") + uvicorn.run(app, host="127.0.0.1", port=8000, log_level="info")