fix: server health ping, rate limiting, security hardening, DA backoff

- Server health ping every 30s — dead servers excluded from all API calls, auto-recovered when back online
- Fixed py3xui client.delete() argument order (inbound_id first, client_uuid second)
- Rate limiting on /admin/login (max 10 attempts per 60s per IP)
- XSS prevention: HTML-escape all user-controlled fields in templates
- CSP headers on all HTML pages (CSP, X-Frame-Options, X-Content-Type-Options)
- CORS restricted to configured host instead of wildcard
- Secure + SameSite=Lax cookie flags for admin_token
- Bind to 127.0.0.1 instead of 0.0.0.0
- Async locks for global state (last_donation_id)
- Graceful shutdown: all background tasks cancelled on exit
- DA polling exponential backoff: 1s→max 1h, immediate 1h on 401
This commit is contained in:
SashegDev
2026-06-16 21:23:03 +00:00
parent d1f018dba1
commit 255caeb069
2 changed files with 513 additions and 377 deletions
+16 -3
View File
@@ -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 серверов
+360 -237
View File
@@ -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,11 +136,9 @@ 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:
@@ -119,7 +170,6 @@ 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
@@ -137,7 +187,6 @@ async def sync_wdtt_password(user: dict):
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:
@@ -149,12 +198,10 @@ async def sync_wdtt_password(user: dict):
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()
@@ -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:
@@ -418,6 +462,8 @@ def get_traffic_stats(sub_id: str) -> dict:
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", []):
@@ -441,7 +487,7 @@ 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}
@@ -454,30 +500,35 @@ 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")
@@ -506,7 +557,7 @@ 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
@@ -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
@@ -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,16 +853,26 @@ 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='<div class="login-err">Слишком много попыток. Подождите.</div>'),
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='<div class="login-err">Неверный пароль</div>'), status_code=401)
return HTMLResponse(
content=render_template("admin_login.html", title=h(settings.get("general", {}).get("title", "ZernProxy")),
error_msg='<div class="login-err">Неверный пароль</div>'),
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)$")):
@@ -643,9 +891,6 @@ async def get_subscription(request: Request, subscription_id: str, format: str =
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 = []
@@ -660,9 +905,14 @@ async def get_subscription(request: Request, subscription_id: str, format: str =
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)
@@ -683,9 +933,6 @@ async def get_subscription(request: Request, subscription_id: str, format: str =
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}
@@ -755,7 +1002,6 @@ async def get_subscription(request: Request, subscription_id: str, format: str =
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()
@@ -778,7 +1024,6 @@ async def get_subscription(request: Request, subscription_id: str, format: str =
"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:
@@ -807,11 +1052,11 @@ async def get_web_page(subscription_id: str):
logo_base64 = get_logo_base64()
logo_html = f'<img src="data:image/png;base64,{logo_base64}" alt="Logo" class="logo-img">' if logo_base64 else '<div class="logo-emoji">⚡</div>'
announcement_html = f'<div class="announcement">{announcement}</div>' if announcement else ''
announcement_html = f'<div class="announcement">{h(announcement)}</div>' if announcement else ''
tier_color = "#4CAF50" if tier == "paid" else "#757575"
tier_name = tier_config.get("name", "Free")
tier_badge = f'<span class="tier-badge" style="background: {tier_color}">{tier_name}</span>'
tier_badge = f'<span class="tier-badge" style="background: {tier_color}">{h(tier_name)}</span>'
days_remaining = get_remaining_days(user)
days_info = f"<p>⏳ Осталось дней: {days_remaining}</p>" if tier == "paid" and days_remaining > 0 else ""
@@ -828,26 +1073,26 @@ async def get_web_page(subscription_id: str):
if traffic.get("by_server"):
for srv_name, data in traffic["by_server"].items():
if data["up"] or data["down"]:
traffic_details += f'<div class="traffic-server"><span class="server-name">{srv_name}</span>: <span class="traffic-values">⬆ {format_bytes(data["up"])}{format_bytes(data["down"])}</span></div>'
traffic_details += f'<div class="traffic-server"><span class="server-name">{h(srv_name)}</span>: <span class="traffic-values">⬆ {format_bytes(data["up"])}{format_bytes(data["down"])}</span></div>'
info_html = f'<div class="info-block">{days_info}{traffic_info}{traffic_details}</div>' if days_info or traffic_info or traffic_details else ''
servers_html = "".join(f'<span class="srv-tag">{get_flag_emoji(srv.get("country", ""))} {srv["name"].upper()}</span>' for srv in servers_for_tier)
servers_html = "".join(f'<span class="srv-tag">{get_flag_emoji(srv.get("country", ""))} {h(srv["name"].upper())}</span>' for srv in servers_for_tier)
support_btn = ""
if da_config.get("enabled"):
da_url = da_config.get("url", "#")
support_btn = f'''
<a href="{da_url}" class="btn btn-support" target="_blank">
<a href="{h(da_url)}" class="btn btn-support" target="_blank">
Поддержать проект
</a>
'''
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")
@@ -940,7 +1185,7 @@ async def webhook_donationalerts(request: Request):
@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()
@@ -951,38 +1196,38 @@ async def admin_users(request: Request):
title = settings.get("general", {}).get("title", "ZernProxy")
tier_options = "".join(
f'<option value="{k}">{v.get("name", k.capitalize())}</option>'
f'<option value="{h(k)}">{h(v.get("name", k.capitalize()))}</option>'
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 '&#8734;'
active = '&#10003;' if u['is_active'] else '&#10007;'
tier_display = u['tier'].upper()
tier_display = h(u['tier'].upper())
rem_days = get_remaining_days(u)
users_rows += f'''<tr>
<td>{u['id']}</td>
<td style="font-weight:600;color:var(--text)">{u['username']}</td>
<td><code>{u['subscription_id']}</code></td>
<td><span class="badge badge-{u['tier']}">{tier_display}</span></td>
<td style="font-weight:600;color:var(--text)">{h(u['username'])}</td>
<td><code>{h(u['subscription_id'])}</code></td>
<td><span class="badge badge-{h(u['tier'])}">{tier_display}</span></td>
<td>{rem_days}</td>
<td>{u['total_paid_rubles']}&#8381;</td>
<td>{traffic}</td>
<td>{active}</td>
<td>{u['created_at'][:10]}</td>
<td>
<button class="btn-sm" onclick="editUser({u['id']}, '{u['tier']}', {rem_days}, {u['traffic_limit_gb']}, {1 if u['is_active'] else 0})">&#9997;&#65039;</button>
<button class="btn-sm danger" onclick="deleteUser({u['id']}, '{u['username']}')">&#128465;&#65039;</button>
<button class="btn-sm" onclick="editUser({u['id']}, '{h(u['tier'])}', {rem_days}, {u['traffic_limit_gb']}, {1 if u['is_active'] else 0})">&#9997;&#65039;</button>
<button class="btn-sm danger" onclick="deleteUser({u['id']}, '{h(u['username'])}')">&#128465;&#65039;</button>
</td>
</tr>'''
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:
@@ -997,6 +1242,8 @@ async def admin_dashboard(request: Request):
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)
@@ -1008,10 +1255,10 @@ async def admin_dashboard(request: Request):
title = settings.get("general", {}).get("title", "ZernProxy")
servers_rows = "".join(
f'<div class="srv-row"><span class="srv-name">{get_flag_emoji(s.get("country",""))} {s["name"].upper()}</span><span class="srv-ibs">{len(s.get("inbounds",[]))} inbound</span><button class="btn-sm" onclick="syncServer(\'{s["name"]}\')">&#128260;</button></div>'
f'<div class="srv-row"><span class="srv-name">{get_flag_emoji(s.get("country",""))} {h(s["name"].upper())}</span><span class="srv-ibs">{len(s.get("inbounds",[]))} inbound</span><button class="btn-sm" onclick="syncServer(\'{h(s["name"])}\')">&#128260;</button></div>'
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),
@@ -1070,7 +1317,7 @@ async def home_page():
for ib in srv_inbounds:
ib_name = ib.get("name", "unknown")
ib_type = ib.get("network", ib.get("flow", ""))
inbound_cards += f'<div class="ib-card"><span class="ib-dot"></span><span class="ib-name">{ib_name}</span><span class="ib-type">{ib_type}</span></div>\n'
inbound_cards += f'<div class="ib-card"><span class="ib-dot"></span><span class="ib-name">{h(ib_name)}</span><span class="ib-type">{h(ib_type)}</span></div>\n'
ru_note = ""
if s.get("country", "").lower() == "ru" and online:
@@ -1079,7 +1326,7 @@ async def home_page():
srv_cards += f'''
<div class="s-card" style="--d: {delay}s; --hl: {hc}">
<div class="s-head">
<span class="flag">{get_flag_emoji(s.get("country",""))}</span> {srv_name} {badge}
<span class="flag">{get_flag_emoji(s.get("country",""))}</span> {h(srv_name)} {badge}
<span class="h-dot" style="background:{hc}"></span>
<span class="h-lbl" style="color:{hc}">{hl}</span>
</div>
@@ -1088,7 +1335,7 @@ async def home_page():
{ru_note}
</div>'''
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,7 +1370,6 @@ 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")
@@ -1168,7 +1415,7 @@ def create_3xui_client(username: str, sub_id: str, inbound: dict, traffic_gb: in
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
@@ -1184,7 +1431,6 @@ 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")
@@ -1252,12 +1498,14 @@ 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)
@@ -1337,6 +1585,8 @@ async def create_user(request: Request, data: dict):
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)
@@ -1375,7 +1625,6 @@ 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"))
@@ -1395,7 +1644,6 @@ async def update_user(request: Request, data: dict):
conn.commit()
updated = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
logger.info(f"After update DB: {dict(updated)}")
finally:
conn.close()
@@ -1424,8 +1672,10 @@ async def delete_user(request: Request, data: dict):
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))
@@ -1436,43 +1686,7 @@ async def delete_user(request: Request, data: dict):
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:
@@ -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")