From 02ff43095c8314727fd3e3bf4dc0db3197ea44ba Mon Sep 17 00:00:00 2001 From: savsis Date: Mon, 14 Sep 2026 05:58:27 +0500 Subject: [PATCH 01/11] site: catch up the comparison table with v1.2.0 (hold/pause, node webhooks, custom admin path, drag-n-drop) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The table hadn't been touched since it first went up — missed host sorting entirely (had the feature, never listed it) and everything shipped tonight. Same sourcing discipline as before: only claiming a Remnawave/Marzban row when it traces back to their own docs.rw comparison table, not guessing. Co-Authored-By: Claude Sonnet 5 --- landing/index.html | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/landing/index.html b/landing/index.html index 9fea350..c962461 100644 --- a/landing/index.html +++ b/landing/index.html @@ -233,6 +233,10 @@ Мультиадминство (раздельные логины)✓—в разработке 2FA на вход в админку✓ TOTPесть (passkeys/OAuth)— Rate-limit на вход/2FA-код✓не документированоне документировано + Свой путь входа в админку✓заявлено— + Сортировка нод мышкой✓Web UIтолько через конфиг Xray + Пауза подписки без потери оплаченных дней✓—есть + Исходящие вебхуки (пользователи + ноды)✓✓только пользователи ЛицензияMITAGPL-3.0AGPL-3.0 From 2b34b1139179b0751de81efe3582186e0689225a Mon Sep 17 00:00:00 2001 From: savsis Date: Mon, 14 Sep 2026 07:35:44 +0500 Subject: [PATCH 02/11] =?UTF-8?q?fix:=2012=20admin=20actions=20failed=20co?= =?UTF-8?q?mpletely=20silently=20on=20error=20=E2=80=94=20no=20message,=20?= =?UTF-8?q?no=20visible=20change,=20nothing?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Found by systematically walking every async function in admin.html and checking whether it wraps its api() call in try/catch — 24 didn't. Two of them (createManualNode, generateGuide) are the exact forms whose backend validation this session added over the last several commits: type a duplicate node code, a bad port, anything the new checks reject, and the button just... does nothing. No error, no success message, the click looks like it didn't register. The backend was correctly rejecting bad input with a clear message (and, since two commits ago, that message even displays cleanly instead of as raw JSON) — none of it reached the screen because the calling function never caught the exception to display it. Triaged the other 22 by actual risk instead of fixing all of them: - 12 mutating actions where a silent failure leaves the admin unsure whether their click did anything — grant/revoke/hold/resume a subscription, delete a device, set an HWID limit, create/toggle/ delete a node, create a gift code, start 2FA setup, plus the two above. Fixed all 12. - The remaining ~14 are view-population loads (loadNodes, loadGifts, loadDashboard, etc.) and logout. Deferred, deliberately: their most likely real failure mode is an expired session, which api()'s own 401 handling already resolves by redirecting to the login screen before the exception even reaches the caller — the confusing "did it work" ambiguity that motivates this fix doesn't really apply to a read-only load the way it does to a deliberate action. Two feedback shapes depending on what's nearby: functions with an existing dedicated result
(createManualNode, generateGuide, createGift) route the error there, matching how every other form in the panel already shows its errors. Functions with no natural home for inline text (grant/revoke/hold/resume, node toggle/delete, device delete, HWID limit, 2FA setup) use a plain alert() — these are infrequent, deliberate single-action clicks, not something a blocking dialog would be disruptive for. All of them still run their normal refresh after a failure, not just after success, so the view never goes stale relative to what the backend actually did. Verification: pure client-side JS, no backend involved, so tested directly under Node with a mocked api()/alert()/refresh — representative cases from both feedback shapes: holdSub and toggleNode (alert-based, confirmed the real backend message reaches the alert and the refresh still fires on both success and failure), createManualNode (result-div- based, confirmed the error text renders and loadNodes is correctly NOT called when creation genuinely failed), and startEnableTotp (confirmed the early return after a failed setup call avoids a second, more confusing crash from reading .secret off an undefined response). Re-ran the full function-by-function try/catch audit afterward to confirm exactly the intended 12 were fixed and list what's still deferred, rather than assuming the diff did what I meant. Co-Authored-By: Claude Sonnet 5 --- admin.html | 89 +++++++++++++++++++++++++++++++++++++++++++----------- 1 file changed, 72 insertions(+), 17 deletions(-) diff --git a/admin.html b/admin.html index ce75d1b..07e0588 100644 --- a/admin.html +++ b/admin.html @@ -1335,7 +1335,13 @@ async function loadTotpStatus() { } async function startEnableTotp() { - const res = await api("/admin/api/2fa/setup", { method: "POST" }); + let res; + try { + res = await api("/admin/api/2fa/setup", { method: "POST" }); + } catch (e) { + alert("Не получилось: " + e.message); + return; + } document.getElementById("totp-secret-display").textContent = res.secret; document.getElementById("totp-setup-box").dataset.secret = res.secret; document.getElementById("totp-setup-box").style.display = "block"; @@ -1525,7 +1531,11 @@ async function grantSubscription() { const node = DD.ucGrantNode.getValue(); const plan = DD.ucGrantPlan.getValue(); if (!node || !plan) return; - await api(`/admin/api/users/${devicesTgId}/grant`, { method: "POST", body: JSON.stringify({ node, plan }) }); + try { + await api(`/admin/api/users/${devicesTgId}/grant`, { method: "POST", body: JSON.stringify({ node, plan }) }); + } catch (e) { + alert("Не получилось: " + e.message); + } openUserCard(devicesTgId, document.getElementById("devices-title").textContent.replace("Карточка: ", "")); } function renderDevices(devices) { @@ -1541,28 +1551,48 @@ function renderDevices(devices) { `).join("") : '
Нет привязанных устройств
'; } async function deleteDevice(deviceId) { - await api(`/admin/api/users/${devicesTgId}/devices/${deviceId}`, { method: "DELETE" }); + try { + await api(`/admin/api/users/${devicesTgId}/devices/${deviceId}`, { method: "DELETE" }); + } catch (e) { + alert("Не получилось: " + e.message); + } const data = await api(`/admin/api/users/${devicesTgId}`); renderDevices(data.devices); } async function saveHwidLimit() { const val = document.getElementById("devices-limit").value.trim(); - await api(`/admin/api/users/${devicesTgId}/hwid-limit`, { method: "POST", body: JSON.stringify({ limit: val || null }) }); + try { + await api(`/admin/api/users/${devicesTgId}/hwid-limit`, { method: "POST", body: JSON.stringify({ limit: val || null }) }); + } catch (e) { + alert("Не получилось: " + e.message); + } } async function revokeSub(uuid) { if (!confirm("Отозвать подписку?")) return; - await api(`/admin/api/subscriptions/${uuid}/revoke`, { method: "POST" }); + try { + await api(`/admin/api/subscriptions/${uuid}/revoke`, { method: "POST" }); + } catch (e) { + alert("Не получилось: " + e.message); + } loadSubscriptions(); } async function holdSub(uuid) { - await api(`/admin/api/subscriptions/${uuid}/hold`, { method: "POST" }); + try { + await api(`/admin/api/subscriptions/${uuid}/hold`, { method: "POST" }); + } catch (e) { + alert("Не получилось: " + e.message); + } loadSubscriptions(); } async function resumeSub(uuid) { - await api(`/admin/api/subscriptions/${uuid}/resume`, { method: "POST" }); + try { + await api(`/admin/api/subscriptions/${uuid}/resume`, { method: "POST" }); + } catch (e) { + alert("Не получилось: " + e.message); + } loadSubscriptions(); } @@ -1590,9 +1620,14 @@ async function loadGifts() { async function createGift() { const node = DD.giftNode.getValue(); const plan = DD.giftPlan.getValue(); - const res = await api("/admin/api/gift-codes", { method: "POST", body: JSON.stringify({ node, plan }) }); - document.getElementById("gift-result").innerHTML = `
${res.link}
`; - loadGifts(); + const result = document.getElementById("gift-result"); + try { + const res = await api("/admin/api/gift-codes", { method: "POST", body: JSON.stringify({ node, plan }) }); + result.innerHTML = `
${res.link}
`; + loadGifts(); + } catch (e) { + result.innerHTML = '

Не получилось: ' + esc(e.message) + '

'; + } } function copyText(t) { navigator.clipboard.writeText(t); } @@ -1720,12 +1755,20 @@ async function saveEditNode() { } async function toggleNode(code, enabled) { - await api(`/admin/api/nodes/${code}`, { method: "PATCH", body: JSON.stringify({ enabled }) }); + try { + await api(`/admin/api/nodes/${code}`, { method: "PATCH", body: JSON.stringify({ enabled }) }); + } catch (e) { + alert("Не получилось: " + e.message); + } loadNodes(); } async function deleteNode(code) { if (!confirm("Удалить ноду?")) return; - await api(`/admin/api/nodes/${code}`, { method: "DELETE" }); + try { + await api(`/admin/api/nodes/${code}`, { method: "DELETE" }); + } catch (e) { + alert("Не получилось: " + e.message); + } loadNodes(); } @@ -1745,8 +1788,15 @@ async function generateGuide() { const include_hysteria2 = document.getElementById("ng-hy").checked; const hysteria_port = parseInt(document.getElementById("ng-hy-port").value || "443"); if (!label || !address) return; - const res = await api("/admin/api/nodes/provision-guide", { method: "POST", body: JSON.stringify({ label, address, port, sni, include_ws, include_hysteria2, hysteria_port }) }); - document.getElementById("guide-result").innerHTML = ` + const guideResult = document.getElementById("guide-result"); + let res; + try { + res = await api("/admin/api/nodes/provision-guide", { method: "POST", body: JSON.stringify({ label, address, port, sni, include_ws, include_hysteria2, hysteria_port }) }); + } catch (e) { + guideResult.innerHTML = '

Не получилось: ' + esc(e.message) + '

'; + return; + } + guideResult.innerHTML = `

Выполни на новом сервере:

${res.command}

Ожидаю установки…

@@ -1774,9 +1824,14 @@ async function createManualNode() { shared_uuid: document.getElementById("nm-uuid").value.trim() || null, kind: document.getElementById("nm-uuid").value.trim() ? "external" : "managed", }; - await api("/admin/api/nodes", { method: "POST", body: JSON.stringify(body) }); - document.getElementById("manual-result").innerHTML = '

Нода добавлена.

'; - loadNodes(); + const result = document.getElementById("manual-result"); + try { + await api("/admin/api/nodes", { method: "POST", body: JSON.stringify(body) }); + result.innerHTML = '

Нода добавлена.

'; + loadNodes(); + } catch (e) { + result.innerHTML = '

Не получилось: ' + esc(e.message) + '

'; + } } function applyBrandName(name) { From 695771c79aa69d5dd8f4d8339bc7aaf7ee659d55 Mon Sep 17 00:00:00 2001 From: savsis Date: Mon, 14 Sep 2026 09:09:36 +0500 Subject: [PATCH 03/11] fix: node-provisioning status poll could run forever, silently or stuck on "waiting" MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-on from the last commit's error-handling sweep — one more spot that calls api() without a try/catch, but a different shape of problem than the others: this one's a setInterval, not a one-shot action, so a thrown/rejected promise inside it doesn't stop anything — the interval just keeps firing every 4s regardless, forever, with each failure only visible as an unhandled rejection in devtools. And even on the success path there was no upper bound at all: if the node never actually comes online (the admin closes the terminal before finishing the install command, say), "Ожидаю установки…" just sits there indefinitely with no way to know if it's still trying or has effectively given up. Now: a consecutive-error counter that gives up after 5 straight failures with a visible message pointing at the manual "Проверить" button, and an overall 150-attempt cap (10 minutes at the existing 4s interval) that stops polling and says so if the node genuinely never reports active. A single transient failure doesn't trip either — the error counter resets on any successful check, so one blip in an otherwise-working poll doesn't cut it short. Verification: extracted the poll callback's logic (can't spin up a real setInterval usefully in a one-shot Node script) and drove it by calling it directly in sequence, which is what setInterval does under the hood anyway. 6 cases: quick success, a transient error that self-heals by the next tick, 5 consecutive failures giving up with the right message at exactly attempt 5, the 150-attempt timeout firing when status never goes active, and confirming no further attempts happen at all once either give-up path triggers — not just that the message stops updating. Co-Authored-By: Claude Sonnet 5 --- admin.html | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/admin.html b/admin.html index 07e0588..54f151e 100644 --- a/admin.html +++ b/admin.html @@ -1802,12 +1802,28 @@ async function generateGuide() {

Ожидаю установки…

`; if (pollTimer) clearInterval(pollTimer); + let pollAttempts = 0, pollErrors = 0; pollTimer = setInterval(async () => { - const st = await api(`/admin/api/nodes/${res.code}/status`); + pollAttempts++; + let st; + try { + st = await api(`/admin/api/nodes/${res.code}/status`); + pollErrors = 0; + } catch (e) { + pollErrors++; + if (pollErrors >= 5) { + clearInterval(pollTimer); + document.getElementById("guide-status").innerHTML = 'Не получилось проверить статус (' + esc(e.message) + ') — проверь вручную кнопкой «Проверить» у ноды'; + } + return; + } if (st.status === "active") { clearInterval(pollTimer); document.getElementById("guide-status").innerHTML = 'Установлено и подключено'; loadNodes(); + } else if (pollAttempts >= 150) { + clearInterval(pollTimer); + document.getElementById("guide-status").innerHTML = 'Не дождались за 10 минут — нода появится в списке сама, когда установка на сервере закончится'; } }, 4000); } From 6546d5bed110fdd7c9dac1331e6bfe24fc77db5d Mon Sep 17 00:00:00 2001 From: savsis Date: Sat, 26 Sep 2026 22:29:15 +0500 Subject: [PATCH 04/11] feat: two-sided referral program MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Each user gets a short ref_code (backfilled lazily for pre-existing accounts too) and a shareable t.me/?start=ref_ link, new "Пригласить друга" menu item shows it plus how many referrals actually converted and any bonus days waiting to be applied. Reward fires once, on the referred user's first subscription of any kind (free, gift, or paid) — not on signup, so an unconverted click never pays out. Both sides get REFERRAL_BONUS_DAYS (config.py/.env, default 3): the referrer's day count comes from settings.py's live-read pattern, same as prices/HWID, so it's tunable without a restart even before a panel UI exists for it. Bonus extends an active subscription directly if the recipient has one, otherwise accumulates in bonus_days_pending and gets folded into whichever subscription they create next (redeemed automatically inside create_subscription, one choke point regardless of which of bot.py's several call sites created it — free trial, gift code, paid, or admin grant). Guards: no self-referral, referrer must exist, first-touch attribution only (a second ?start=ref_ link never overwrites it), and only takes for genuinely new accounts (no existing subscriptions) — attaching a referrer to an already-active user was never the intent. Tested two ways, matching this repo's usual db.py-can-be-imported- standalone / bot.py-needs-a-workaround split: 16 checks against a real isolated sqlite db for the db.py logic (attribution, both reward paths, double-reward guard, pending-bonus fold-in), then 9 more through an actual `import bot` — aiogram/fastapi now have Python 3.14 wheels so this imported for real rather than needing AST-extraction, modulo one old blocker (xray_manager still imports the Unix-only fcntl for its file lock) worked around with a tiny fake fcntl module in sys.modules, same spirit as the fcntl shim already used elsewhere in this project's history. Real start_deeplink and cb_referral calls, get_me() mocked to avoid a live Telegram API call. Not done: admin-panel UI toggle for REFERRAL_ENABLED/REFERRAL_BONUS_DAYS (currently .env-only, like several other business tunables were before they got a settings-page treatment) and a docs-tab writeup — happy to add both if wanted, scoped this pass to the mechanic itself. Co-Authored-By: Claude Sonnet 5 --- bot.py | 41 +++++++++++++++++++++ config.py | 3 ++ db.py | 101 ++++++++++++++++++++++++++++++++++++++++++++++++++-- settings.py | 14 ++++++++ 4 files changed, 156 insertions(+), 3 deletions(-) diff --git a/bot.py b/bot.py index 0a6d057..a86cbe9 100644 --- a/bot.py +++ b/bot.py @@ -33,6 +33,7 @@ def main_menu_kb(tg_id: int) -> InlineKeyboardMarkup: rows = [ [InlineKeyboardButton(text="Получить VPN", callback_data="menu:get")], [InlineKeyboardButton(text="Моя подписка", callback_data="menu:mysub")], + [InlineKeyboardButton(text="Пригласить друга", callback_data="menu:referral")], [InlineKeyboardButton(text="О сервисе", callback_data="menu:about")], ] if is_admin(tg_id): @@ -40,6 +41,14 @@ def main_menu_kb(tg_id: int) -> InlineKeyboardMarkup: return InlineKeyboardMarkup(inline_keyboard=rows) +async def get_bot_username() -> str: + global _bot_username + if _bot_username is None: + me = await bot.get_me() + _bot_username = me.username + return _bot_username + + def nodes_kb(prefix: str) -> InlineKeyboardMarkup: rows = [] for n in db.list_nodes(enabled_only=True): @@ -90,6 +99,12 @@ async def send_main_menu(message: Message): async def start_deeplink(message: Message, command: CommandObject): user = db.get_or_create_user(message.from_user.id, message.from_user.username) payload = command.args or "" + if payload.startswith("ref_") or payload.startswith("ref-"): + ref_code = payload[4:] + referrer = db.get_user_by_ref_code(ref_code) + if referrer and settings.get_referral_settings()["enabled"]: + db.set_referred_by(message.from_user.id, referrer["tg_id"]) + return await send_main_menu(message) if payload.startswith("gift_") or payload.startswith("gift-"): code = payload[5:] gift, err = db.redeem_gift_code(code, message.from_user.id) @@ -140,6 +155,32 @@ async def cb_about(cb: CallbackQuery): await cb.answer() +@dp.callback_query(F.data == "menu:referral") +async def cb_referral(cb: CallbackQuery): + kb = InlineKeyboardMarkup(inline_keyboard=[[InlineKeyboardButton(text="Назад", callback_data="menu:main")]]) + ref_settings = settings.get_referral_settings() + if not ref_settings["enabled"]: + await cb.message.edit_text("Реферальная программа сейчас отключена.", reply_markup=kb) + return await cb.answer() + user = db.get_or_create_user(cb.from_user.id, cb.from_user.username) + stats = db.referral_stats(cb.from_user.id) + username = await get_bot_username() + link = f"https://t.me/{username}?start=ref_{user['ref_code']}" + days = ref_settings["bonus_days"] + text = ( + f"Пригласи друга\n\n" + f"За каждого друга, который активирует подписку по твоей ссылке, " + f"вы оба получаете +{days} дн. к подписке.\n\n" + f"{DIVIDER}\n" + f"Твоя ссылка:\n{link}\n\n" + f"Приглашено: {stats['referred_count']}\n" + ) + if stats["bonus_days_pending"]: + text += f"Накоплено бонусных дней (зачислятся при следующей подписке): {stats['bonus_days_pending']}\n" + await cb.message.edit_text(text, reply_markup=kb) + await cb.answer() + + @dp.callback_query(F.data == "menu:get") async def cb_get(cb: CallbackQuery): await cb.message.edit_text("Выбери сервер:", reply_markup=nodes_kb("node")) diff --git a/config.py b/config.py index 7f67c5c..f5a1cf1 100644 --- a/config.py +++ b/config.py @@ -120,3 +120,6 @@ PLATEGA_SECRET = env("PLATEGA_SECRET", "") HWID_LIMIT_ENABLED = env("HWID_LIMIT_ENABLED", "false").lower() == "true" HWID_FALLBACK_LIMIT = int(env("HWID_FALLBACK_LIMIT", "3")) + +REFERRAL_ENABLED = env("REFERRAL_ENABLED", "true").lower() == "true" +REFERRAL_BONUS_DAYS = int(env("REFERRAL_BONUS_DAYS", "3")) diff --git a/db.py b/db.py index 68dd516..62dec11 100644 --- a/db.py +++ b/db.py @@ -133,6 +133,10 @@ _NEW_NODE_COLUMNS = { _NEW_USER_COLUMNS = { "hwid_limit": "INTEGER", + "ref_code": "TEXT", + "referred_by": "INTEGER", + "referral_rewarded": "INTEGER NOT NULL DEFAULT 0", + "bonus_days_pending": "INTEGER NOT NULL DEFAULT 0", } _NEW_ADMIN_SESSION_COLUMNS = { @@ -159,6 +163,7 @@ def _migrate(): for name, decl in _NEW_USER_COLUMNS.items(): if name not in ucols: conn.execute(f"ALTER TABLE users ADD COLUMN {name} {decl}") + conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_users_ref_code ON users(ref_code)") scols = {r["name"] for r in conn.execute("PRAGMA table_info(admin_sessions)").fetchall()} for name, decl in _NEW_ADMIN_SESSION_COLUMNS.items(): if name not in scols: @@ -353,22 +358,97 @@ def delete_node(code: str): conn.execute("DELETE FROM nodes WHERE code=?", (code,)) +def _generate_ref_code(conn) -> str: + for _ in range(20): + code = secrets.token_hex(4) + if not conn.execute("SELECT 1 FROM users WHERE ref_code=?", (code,)).fetchone(): + return code + raise RuntimeError("could not generate a unique ref_code") + + def get_or_create_user(tg_id: int, username: str | None): with get_conn() as conn: row = conn.execute("SELECT * FROM users WHERE tg_id=?", (tg_id,)).fetchone() if row: if username and row["username"] != username: conn.execute("UPDATE users SET username=? WHERE tg_id=?", (username, tg_id)) + if not row["ref_code"]: + conn.execute( + "UPDATE users SET ref_code=? WHERE tg_id=?", (_generate_ref_code(conn), tg_id) + ) + row = conn.execute("SELECT * FROM users WHERE tg_id=?", (tg_id,)).fetchone() return dict(row) token = secrets.token_hex(16) + ref_code = _generate_ref_code(conn) conn.execute( - "INSERT INTO users (tg_id, token, username, created_at) VALUES (?,?,?,?)", - (tg_id, token, username, now_iso()), + "INSERT INTO users (tg_id, token, username, ref_code, created_at) VALUES (?,?,?,?,?)", + (tg_id, token, username, ref_code, now_iso()), ) row = conn.execute("SELECT * FROM users WHERE tg_id=?", (tg_id,)).fetchone() return dict(row) +def get_user_by_ref_code(ref_code: str): + with get_conn() as conn: + row = conn.execute("SELECT * FROM users WHERE ref_code=?", (ref_code,)).fetchone() + return dict(row) if row else None + + +def set_referred_by(tg_id: int, referrer_tg_id: int) -> bool: + """First-touch attribution: only takes effect for a brand-new account + (no subscriptions yet) that isn't already attributed, and never to self.""" + if tg_id == referrer_tg_id: + return False + with get_conn() as conn: + row = conn.execute("SELECT referred_by FROM users WHERE tg_id=?", (tg_id,)).fetchone() + if not row or row["referred_by"] is not None: + return False + has_sub = conn.execute("SELECT 1 FROM subscriptions WHERE tg_id=?", (tg_id,)).fetchone() + if has_sub: + return False + referrer = conn.execute("SELECT 1 FROM users WHERE tg_id=?", (referrer_tg_id,)).fetchone() + if not referrer: + return False + conn.execute("UPDATE users SET referred_by=? WHERE tg_id=?", (referrer_tg_id, tg_id)) + return True + + +def _apply_bonus_days(conn, tg_id: int, days: int): + if days <= 0: + return + row = conn.execute( + "SELECT uuid, expires_at FROM subscriptions WHERE tg_id=? AND active=1 AND held_at IS NULL " + "ORDER BY expires_at DESC LIMIT 1", + (tg_id,), + ).fetchone() + if row: + new_expires = datetime.datetime.fromisoformat(row["expires_at"]) + datetime.timedelta(days=days) + conn.execute("UPDATE subscriptions SET expires_at=? WHERE uuid=?", (new_expires.isoformat(), row["uuid"])) + else: + conn.execute( + "UPDATE users SET bonus_days_pending = COALESCE(bonus_days_pending, 0) + ? WHERE tg_id=?", + (days, tg_id), + ) + + +def credit_bonus_days(tg_id: int, days: int): + with get_conn() as conn: + _apply_bonus_days(conn, tg_id, days) + + +def referral_stats(tg_id: int) -> dict: + with get_conn() as conn: + user = conn.execute("SELECT ref_code, bonus_days_pending FROM users WHERE tg_id=?", (tg_id,)).fetchone() + count = conn.execute( + "SELECT COUNT(*) c FROM users WHERE referred_by=? AND referral_rewarded=1", (tg_id,) + ).fetchone()["c"] + return { + "ref_code": user["ref_code"] if user else None, + "bonus_days_pending": user["bonus_days_pending"] if user else 0, + "referred_count": count, + } + + def get_user_by_token(token: str): with get_conn() as conn: row = conn.execute("SELECT * FROM users WHERE token=?", (token,)).fetchone() @@ -377,17 +457,32 @@ def get_user_by_token(token: str): def create_subscription(tg_id: int, node: str, plan_days: int, plan_code: str, source: str = "bot", client_uuid: str | None = None): import uuid as uuidlib + import config cid = client_uuid or str(uuidlib.uuid4()) created = datetime.datetime.utcnow() expires = created + datetime.timedelta(days=plan_days) with get_conn() as conn: + is_first = conn.execute("SELECT 1 FROM subscriptions WHERE tg_id=?", (tg_id,)).fetchone() is None + urow = conn.execute( + "SELECT referred_by, referral_rewarded, bonus_days_pending FROM users WHERE tg_id=?", (tg_id,) + ).fetchone() + pending = urow["bonus_days_pending"] if urow else 0 + if pending: + expires += datetime.timedelta(days=pending) + conn.execute("UPDATE users SET bonus_days_pending=0 WHERE tg_id=?", (tg_id,)) conn.execute( "INSERT INTO subscriptions (uuid, tg_id, node, plan, created_at, expires_at, active, source) " "VALUES (?,?,?,?,?,?,1,?)", (cid, tg_id, node, plan_code, created.isoformat(), expires.isoformat(), source), ) - return {"uuid": cid, "tg_id": tg_id, "node": node, "plan": plan_code, "expires_at": expires.isoformat()} + if is_first and urow and urow["referred_by"] and not urow["referral_rewarded"] and config.REFERRAL_ENABLED: + conn.execute("UPDATE users SET referral_rewarded=1 WHERE tg_id=?", (tg_id,)) + bonus = config.REFERRAL_BONUS_DAYS + _apply_bonus_days(conn, tg_id, bonus) + _apply_bonus_days(conn, urow["referred_by"], bonus) + expires_final = conn.execute("SELECT expires_at FROM subscriptions WHERE uuid=?", (cid,)).fetchone()["expires_at"] + return {"uuid": cid, "tg_id": tg_id, "node": node, "plan": plan_code, "expires_at": expires_final} def list_active_subscriptions(tg_id: int | None = None, node: str | None = None): diff --git a/settings.py b/settings.py index 3679ad5..a7394d5 100644 --- a/settings.py +++ b/settings.py @@ -87,3 +87,17 @@ def get_hwid_settings() -> dict: "enabled": _bool(raw.get("HWID_LIMIT_ENABLED"), config.HWID_LIMIT_ENABLED), "fallback_limit": limit if limit > 0 else config.HWID_FALLBACK_LIMIT, } + + +def get_referral_settings() -> dict: + raw = legal.read_env_vars(["REFERRAL_ENABLED", "REFERRAL_BONUS_DAYS"]) + days = _positive_int(raw.get("REFERRAL_BONUS_DAYS"), config.REFERRAL_BONUS_DAYS) + return { + "enabled": _bool(raw.get("REFERRAL_ENABLED"), config.REFERRAL_ENABLED), + "bonus_days": days if days > 0 else config.REFERRAL_BONUS_DAYS, + } + + +def set_referral_settings(enabled: bool, bonus_days: int): + legal.update_env_var("REFERRAL_ENABLED", "true" if enabled else "false") + legal.update_env_var("REFERRAL_BONUS_DAYS", str(int(bonus_days))) From da6200ae4a1e3087610a0067b7b1bff15b7343d3 Mon Sep 17 00:00:00 2001 From: savsis Date: Sun, 4 Oct 2026 03:51:16 +0500 Subject: [PATCH 05/11] feat: server chains (client -> A -> B -> internet), audit log, users list, csv export Chains are real Xray hops: a chain- inbound + outbound + routing rule on the entry node and a relay- client on the exit node, reconciled by sync_node with config test, port check, rollback on a failed restart and a per-node lock. Admin API gets chains CRUD/probe/check, node latency, audit log (middleware, no request bodies), users list with subscription counters and a csv export that neutralises formulas. --- api.py | 291 +++++++++++++++++++++++++++++++++++++++++++++++- backup.py | 3 + chains.py | 233 ++++++++++++++++++++++++++++++++++++++ db.py | 134 ++++++++++++++++++++++ links.py | 20 ++++ nodeprov.py | 141 +++++++++++++++++------ xray_manager.py | 177 +++++++++++++++++++++++------ 7 files changed, 926 insertions(+), 73 deletions(-) create mode 100644 chains.py diff --git a/api.py b/api.py index 7a39d45..0402def 100644 --- a/api.py +++ b/api.py @@ -1,17 +1,22 @@ import asyncio +import csv import datetime +import io import json import os import re import secrets import subprocess import urllib.request +import uuid as uuidlib +from concurrent.futures import ThreadPoolExecutor from fastapi import FastAPI, HTTPException, Request, Response, File, UploadFile from fastapi.middleware.cors import CORSMiddleware from fastapi import Body from fastapi.responses import HTMLResponse, PlainTextResponse, FileResponse import backup +import chains import db import legal import links @@ -52,6 +57,59 @@ def require_admin(request: Request): raise HTTPException(401, "unauthorized") +AUDIT_RULES = [ + ("POST", r"^/admin/api/nodes$", "node.add"), + ("POST", r"^/admin/api/nodes/provision-guide$", "node.provision"), + ("POST", r"^/admin/api/nodes/reorder$", "node.reorder"), + ("PATCH", r"^/admin/api/nodes/[^/]+$", "node.edit"), + ("DELETE", r"^/admin/api/nodes/[^/]+$", "node.delete"), + ("POST", r"^/admin/api/chains$", "chain.create"), + ("PATCH", r"^/admin/api/chains/[^/]+$", "chain.edit"), + ("DELETE", r"^/admin/api/chains/[^/]+$", "chain.delete"), + ("POST", r"^/admin/api/subscriptions/[^/]+/revoke$", "sub.revoke"), + ("POST", r"^/admin/api/subscriptions/[^/]+/hold$", "sub.hold"), + ("POST", r"^/admin/api/subscriptions/[^/]+/resume$", "sub.resume"), + ("POST", r"^/admin/api/subscriptions/[^/]+/reset-traffic$", "sub.reset_traffic"), + ("POST", r"^/admin/api/users/-?\d+/grant$", "user.grant"), + ("POST", r"^/admin/api/users/-?\d+/hwid-limit$", "user.hwid_limit"), + ("DELETE", r"^/admin/api/users/-?\d+/devices/\d+$", "user.device_delete"), + ("POST", r"^/admin/api/gift-codes$", "gift.create"), + ("POST", r"^/admin/api/admins$", "admin.add"), + ("DELETE", r"^/admin/api/admins/\d+$", "admin.delete"), + ("POST", r"^/admin/api/2fa/enable$", "2fa.enable"), + ("POST", r"^/admin/api/2fa/disable$", "2fa.disable"), + ("GET", r"^/admin/api/backup$", "backup.download"), + ("POST", r"^/admin/api/backup/restore$", "backup.restore"), + ("POST", r"^/admin/api/settings/bot$", "settings.bot"), + ("POST", r"^/admin/api/branding$", "settings.brand"), + ("POST", r"^/admin/api/webhook-settings$", "settings.webhook"), + ("POST", r"^/admin/api/hwid-settings$", "settings.hwid"), + ("POST", r"^/admin/api/payments/(yookassa-settings|platega-settings|plan-settings|legal-settings)$", "settings.payments"), +] +AUDIT_COMPILED = [(method, re.compile(pattern), action) for method, pattern, action in AUDIT_RULES] + + +def _audit_action(method: str, path: str): + for rule_method, pattern, action in AUDIT_COMPILED: + if rule_method == method and pattern.match(path): + return action + return None + + +@app.middleware("http") +async def audit_middleware(request: Request, call_next): + path = request.url.path + action = _audit_action(request.method, path) if path.startswith("/admin/api/") else None + admin_name = None + if action: + admin = await asyncio.to_thread(db.get_session_admin, request.cookies.get(ADMIN_COOKIE)) + admin_name = admin["username"] if admin else None + response = await call_next(request) + if action and response.status_code < 400: + await asyncio.to_thread(db.add_audit, admin_name, action, path, _client_ip(request)) + return response + + def _days_left(sub: dict) -> int: exp = datetime.datetime.fromisoformat(sub["expires_at"]) reference = datetime.datetime.fromisoformat(sub["held_at"]) if sub.get("held_at") else datetime.datetime.utcnow() @@ -614,11 +672,13 @@ def admin_login(request: Request, response: Response, body: dict = Body(...)): admin = db.verify_admin_login(username, password) if not admin: db.record_login_attempt(ip, "password") + db.add_audit(username[:64] or None, "login.failed", "", ip) raise HTTPException(401, "wrong username or password") db.clear_login_attempts(ip, "password") if admin.get("totp_secret"): pending_token = db.create_pending_totp(admin["id"]) return {"ok": True, "needs_totp": True, "pending_token": pending_token} + db.add_audit(admin["username"], "login.ok", "", ip) token = db.create_admin_session(admin["id"]) response.set_cookie(ADMIN_COOKIE, token, httponly=True, secure=True, samesite="strict", max_age=7 * 24 * 3600) return {"ok": True} @@ -637,9 +697,11 @@ def admin_login_totp(request: Request, response: Response, body: dict = Body(... admin = db.get_admin_by_id(pending["admin_id"]) if not admin or not admin.get("totp_secret") or not totp.verify(admin["totp_secret"], code): db.record_login_attempt(ip, "totp") + db.add_audit(admin["username"] if admin else None, "login.totp_failed", "", ip) raise HTTPException(401, "wrong code") db.clear_login_attempts(ip, "totp") db.delete_pending_totp(pending_token) + db.add_audit(admin["username"], "login.ok", "2fa", ip) token = db.create_admin_session(admin["id"]) response.set_cookie(ADMIN_COOKIE, token, httponly=True, secure=True, samesite="strict", max_age=7 * 24 * 3600) return {"ok": True} @@ -942,12 +1004,22 @@ def admin_reset_traffic(uuid: str, request: Request): return {"ok": ok} +@app.get("/admin/api/users") +def admin_list_users(request: Request, q: str = "", limit: int = 200): + require_admin(request) + return db.list_users(q=q, limit=limit) + + @app.get("/admin/api/users/{tg_id}") def admin_user_card(tg_id: int, request: Request): require_admin(request) user = db.get_user(tg_id) if not user: - raise HTTPException(404, "not found") + return { + "tg_id": tg_id, "username": None, "created_at": None, "token": None, "exists": False, + "subscriptions": [], "devices": [], "hwid_limit": None, + "hwid_fallback_limit": settings.get_hwid_settings()["fallback_limit"], + } subs = db.list_subscriptions_for_user(tg_id) nodes_by_code = {n["code"]: n for n in db.list_nodes()} plans_by_code = settings.get_plans_by_code() @@ -966,6 +1038,7 @@ def admin_user_card(tg_id: int, request: Request): "username": user["username"], "created_at": user["created_at"], "token": user["token"], + "exists": True, "subscriptions": out_subs, "devices": db.list_devices(tg_id), "hwid_limit": user.get("hwid_limit"), @@ -1231,6 +1304,222 @@ def admin_node_status(code: str, request: Request): return {"status": node["status"], "enabled": bool(node["enabled"])} +def _panel_latency(node: dict) -> dict: + samples = chains.tcp_connect_ms(node["address"], node["port"], samples=2, timeout=2.0) + return {"code": node["code"], "ms": chains.median_ms(samples)} + + +@app.get("/admin/api/nodes/latency") +def admin_nodes_latency(request: Request): + require_admin(request) + targets = [n for n in db.list_nodes() if n["address"] and n["status"] == "active" and n["enabled"]] + if not targets: + return [] + with ThreadPoolExecutor(max_workers=min(8, len(targets))) as pool: + return list(pool.map(_panel_latency, targets)) + + +def _chain_view(chain: dict, nodes_by_code: dict) -> dict: + entry = nodes_by_code.get(chain["entry_node"]) + exit_node = nodes_by_code.get(chain["exit_node"]) + return { + "code": chain["code"], "label": chain["label"], + "entry_node": chain["entry_node"], "exit_node": chain["exit_node"], + "entry_label": entry["label"] if entry else chain["entry_node"], + "exit_label": exit_node["label"] if exit_node else chain["exit_node"], + "entry_address": entry["address"] if entry else None, + "port": chain["port"], "enabled": bool(chain["enabled"]), + "created_at": chain["created_at"], + } + + +def _csv_safe(value) -> str: + text = str(value) + if text and text[0] in "=+-@\t\r": + return "'" + text + return text + + +def _sync_chain_nodes(chain: dict) -> dict: + results = {} + for code in dict.fromkeys([chain["entry_node"], chain["exit_node"]]): + node = db.get_node(code) + if not node: + continue + try: + res = xray_manager.sync_node(node) + results[code] = {"ok": not res["problems"], "problems": res["problems"]} + except Exception as e: + results[code] = {"ok": False, "problems": [str(e)]} + return results + + +def _chain_failures(results: dict) -> list: + failures = [] + for code, res in results.items(): + for problem in res["problems"]: + failures.append(f"{code}: {problem}") + return failures + + +@app.get("/admin/api/chains") +def admin_list_chains(request: Request): + require_admin(request) + nodes_by_code = {n["code"]: n for n in db.list_nodes()} + return [_chain_view(c, nodes_by_code) for c in db.list_chains()] + + +@app.post("/admin/api/chains") +def admin_create_chain(request: Request, body: dict = Body(...)): + require_admin(request) + entry_code = str(body.get("entry", "")).strip() + exit_code = str(body.get("exit", "")).strip() + entry = db.get_node(entry_code) + exit_node = db.get_node(exit_code) + if not entry or not exit_node: + raise HTTPException(400, "выбери входной и выходной серверы из списка нод") + if entry_code == exit_code: + raise HTTPException(400, "вход и выход цепочки должны быть разными серверами") + if entry["kind"] not in chains.CHAIN_KINDS_ENTRY: + raise HTTPException(400, "входной сервер должен быть под управлением панели (локальный или управляемый)") + if exit_node["kind"] not in chains.CHAIN_KINDS_EXIT: + raise HTTPException(400, "этот тип ноды нельзя использовать как выход цепочки") + for node in (entry, exit_node): + if not node["enabled"] or node["status"] != "active": + raise HTTPException(400, f"нода «{node['label']}» выключена или ещё не установлена") + relay_uuid = None + if exit_node["kind"] == "external": + if not exit_node.get("shared_uuid"): + raise HTTPException(400, "у внешней ноды не задан shared UUID — через неё цепочку не построить") + else: + relay_uuid = str(uuidlib.uuid4()) + label = str(body.get("label", "")).strip()[:80] or f"{entry['label']} → {exit_node['label']}" + try: + chain = db.create_chain(label, entry_code, exit_code, relay_uuid) + except ValueError as e: + raise HTTPException(400, str(e)) + results = _sync_chain_nodes(chain) + failures = _chain_failures(results) + if failures: + db.delete_chain(chain["code"]) + _sync_chain_nodes(chain) + raise HTTPException(502, "цепочка не применилась, всё откатили назад: " + "; ".join(failures)) + webhooks.send("chain.created", { + "code": chain["code"], "label": chain["label"], "entry": entry_code, "exit": exit_code, + }) + nodes_by_code = {n["code"]: n for n in db.list_nodes()} + return _chain_view(chain, nodes_by_code) + + +@app.patch("/admin/api/chains/{code}") +def admin_update_chain(code: str, request: Request, body: dict = Body(...)): + require_admin(request) + chain = db.get_chain(code) + if not chain: + raise HTTPException(404, "not found") + fields = {} + if "label" in body: + label = str(body["label"]).strip()[:80] + if not label: + raise HTTPException(400, "название не может быть пустым") + fields["label"] = label + toggled = "enabled" in body and bool(body["enabled"]) != bool(chain["enabled"]) + if "enabled" in body: + fields["enabled"] = 1 if body["enabled"] else 0 + updated = db.update_chain(code, **fields) + if toggled: + failures = _chain_failures(_sync_chain_nodes(updated)) + if failures: + db.update_chain(code, enabled=chain["enabled"]) + _sync_chain_nodes(chain) + raise HTTPException(502, "не получилось применить: " + "; ".join(failures)) + webhooks.send("chain.enabled" if updated["enabled"] else "chain.disabled", { + "code": code, "label": updated["label"], + }) + nodes_by_code = {n["code"]: n for n in db.list_nodes()} + return _chain_view(db.get_chain(code), nodes_by_code) + + +@app.delete("/admin/api/chains/{code}") +def admin_delete_chain(code: str, request: Request): + require_admin(request) + chain = db.get_chain(code) + if not chain: + raise HTTPException(404, "not found") + db.delete_chain(code) + results = _sync_chain_nodes(chain) + webhooks.send("chain.deleted", {"code": code, "label": chain["label"]}) + return {"ok": True, "warnings": _chain_failures(results)} + + +def _hop_probe(entry: dict, exit_node: dict) -> dict: + try: + samples = xray_manager.probe_from_node(entry, exit_node["address"], exit_node["port"]) + except Exception as e: + return {"rtt_ms": None, "samples": [], "level": "unknown", "error": str(e)} + rtt = chains.median_ms(samples) + return {"rtt_ms": rtt, "samples": samples, "level": chains.latency_level(rtt)} + + +@app.get("/admin/api/chains/probe") +def admin_probe_chain(entry: str, exit: str, request: Request): + require_admin(request) + entry_node = db.get_node(entry) + exit_node = db.get_node(exit) + if not entry_node or not exit_node or entry == exit: + raise HTTPException(400, "нужны две разные ноды") + return _hop_probe(entry_node, exit_node) + + +@app.post("/admin/api/chains/{code}/check") +def admin_check_chain(code: str, request: Request): + require_admin(request) + chain = db.get_chain(code) + if not chain: + raise HTTPException(404, "not found") + entry = db.get_node(chain["entry_node"]) + exit_node = db.get_node(chain["exit_node"]) + if not entry or not exit_node: + raise HTTPException(400, "одна из нод цепочки удалена") + entry_alive = nodeprov.check_node_alive(entry["address"], chain["port"]) + hop = _hop_probe(entry, exit_node) + return {"entry_alive": entry_alive, **hop} + + +@app.get("/admin/api/audit") +def admin_audit(request: Request, limit: int = 100): + require_admin(request) + return db.list_audit(limit=max(1, min(limit, 500))) + + +@app.get("/admin/api/subscriptions/export.csv") +def admin_export_subscriptions(request: Request): + require_admin(request) + nodes_by_code = {n["code"]: n for n in db.list_nodes()} + plans_by_code = settings.get_plans_by_code() + buf = io.StringIO() + writer = csv.writer(buf) + writer.writerow(["tg_id", "username", "node", "plan", "created_at", "expires_at", "days_left", "status", "uuid"]) + for s in db.list_all_subscriptions(limit=100000): + node = nodes_by_code.get(s["node"]) + plan = plans_by_code.get(s["plan"]) + if s.get("held_at"): + status = "held" + elif not s["active"] or s["expires_at"] <= db.now_iso(): + status = "expired" + else: + status = "active" + writer.writerow([_csv_safe(v) for v in [ + s["tg_id"], s.get("username") or "", node["label"] if node else s["node"], + plan["label"] if plan else s["plan"], s["created_at"], s["expires_at"], + _days_left(s), status, s["uuid"], + ]]) + return Response( + content="" + buf.getvalue(), media_type="text/csv; charset=utf-8", + headers={"Content-Disposition": 'attachment; filename="subscriptions.csv"'}, + ) + + ADMIN_HTML_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "admin.html") diff --git a/backup.py b/backup.py index a40ad53..59049d6 100644 --- a/backup.py +++ b/backup.py @@ -123,5 +123,8 @@ def restore_backup(data: bytes) -> dict: os.chmod(tmp_db_path, 0o600) os.replace(tmp_db_path, DB_PATH) + import db + db.init_db() + _prune_old_safety_copies() return {"restored_env": restored_env, "safety_copy": safety_copy} diff --git a/chains.py b/chains.py new file mode 100644 index 0000000..d43c7fa --- /dev/null +++ b/chains.py @@ -0,0 +1,233 @@ +import copy +import json +import re +import socket +import time + +BASE_TAGS = ("vless-tcp-reality", "vless-grpc-reality", "vless-xhttp-reality", "vless-ws-tls") +TCP_TAG = "vless-tcp-reality" +VISION = "xtls-rprx-vision" +CHAIN_PREFIX = "chain-" +RELAY_EMAIL_PREFIX = "relay-" +MAX_SERVERS = 2 +PORT_MIN = 10443 +PORT_MAX = 10999 +CODE_RE = re.compile(r"^[a-z0-9]{1,16}$") +HOST_RE = re.compile(r"^[A-Za-z0-9.-]{1,253}$") +CHAIN_KINDS_ENTRY = ("local", "managed") +CHAIN_KINDS_EXIT = ("local", "managed", "external") + + +class ChainConfigError(Exception): + pass + + +def inbound_tag(code): + return CHAIN_PREFIX + code + + +def outbound_tag(code): + return CHAIN_PREFIX + code + "-out" + + +def relay_email(code): + return RELAY_EMAIL_PREFIX + code + + +def is_chain_inbound_tag(tag): + return bool(tag) and tag.startswith(CHAIN_PREFIX) and not tag.endswith("-out") + + +def is_user_tag(tag): + return tag in BASE_TAGS or is_chain_inbound_tag(tag) + + +def flow_for_tag(tag): + if tag == TCP_TAG or is_chain_inbound_tag(tag): + return VISION + return None + + +def sync_clients(clients, wanted, flow): + kept = [] + seen = set() + for c in clients: + cid = c.get("id") + if cid in wanted and cid not in seen: + kept.append(c) + seen.add(cid) + for cid in wanted: + if cid in seen: + continue + entry = {"id": cid, "email": wanted[cid]} + if flow: + entry["flow"] = flow + kept.append(entry) + return kept + + +def find_inbound(cfg, tag): + for ib in cfg.get("inbounds", []): + if ib.get("tag") == tag: + return ib + return None + + +def build_chain_inbound(template, chain, wanted, old_clients): + reality = (template.get("streamSettings") or {}).get("realitySettings") + if not reality: + raise ChainConfigError("у входной ноды нет TCP+Reality inbound — цепочку строить не из чего") + ib = copy.deepcopy(template) + ib["tag"] = inbound_tag(chain["code"]) + ib["port"] = chain["port"] + ib["streamSettings"]["realitySettings"]["shortIds"] = [chain["short_id"]] + ib["settings"]["clients"] = sync_clients(old_clients, wanted, VISION) + return ib + + +def build_chain_outbound(chain, exit_node, relay_uuid): + return { + "tag": outbound_tag(chain["code"]), + "protocol": "vless", + "settings": { + "vnext": [{ + "address": exit_node["address"], + "port": int(exit_node["port"]), + "users": [{"id": relay_uuid, "encryption": "none", "flow": VISION}], + }], + }, + "streamSettings": { + "network": "tcp", + "security": "reality", + "realitySettings": { + "serverName": exit_node["sni"], + "fingerprint": "chrome", + "publicKey": exit_node["public_key"], + "shortId": exit_node["short_id"], + "spiderX": "", + }, + }, + } + + +def build_chain_rule(chain): + return { + "type": "field", + "inboundTag": [inbound_tag(chain["code"])], + "outboundTag": outbound_tag(chain["code"]), + } + + +def relay_for_chain(chain, exit_node): + if exit_node["kind"] == "external": + return exit_node.get("shared_uuid") + return chain.get("relay_uuid") + + +def split_busy_chains(cfg, entry_chains, busy_ports): + new_ports = set(new_ports_needed(cfg, entry_chains)) + usable = [] + problems = [] + for chain in entry_chains: + if chain["port"] in new_ports and chain["port"] in busy_ports: + problems.append(f"{chain['code']}: порт {chain['port']} уже занят другим процессом, цепочка не применена") + continue + usable.append(chain) + return usable, problems + + +def sync_config(cfg, wanted, relay_wanted, entry_chains, exit_nodes, apply_chains=True): + before = json.dumps(cfg, sort_keys=True) + problems = [] + + template = find_inbound(cfg, TCP_TAG) + + for ib in cfg["inbounds"]: + tag = ib.get("tag") + if not is_user_tag(tag): + continue + want = dict(wanted) + if tag == TCP_TAG: + want.update(relay_wanted) + ib["settings"]["clients"] = sync_clients(ib["settings"]["clients"], want, flow_for_tag(tag)) + + if not apply_chains: + changed = json.dumps(cfg, sort_keys=True) != before + return changed, problems + + old_chain_inbounds = {} + for ib in cfg["inbounds"]: + if is_chain_inbound_tag(ib.get("tag")): + old_chain_inbounds[ib["tag"]] = ib + + kept_inbounds = [ib for ib in cfg["inbounds"] if not is_chain_inbound_tag(ib.get("tag"))] + kept_outbounds = [ob for ob in cfg.get("outbounds", []) if not (ob.get("tag") or "").startswith(CHAIN_PREFIX)] + routing = cfg.setdefault("routing", {}) + kept_rules = [r for r in routing.get("rules", []) if not (r.get("outboundTag") or "").startswith(CHAIN_PREFIX)] + + for chain in entry_chains: + exit_node = exit_nodes.get(chain["exit_node"]) + if template is None: + problems.append(f"{chain['code']}: нет TCP+Reality inbound на входной ноде") + continue + if not exit_node: + problems.append(f"{chain['code']}: выходная нода не найдена") + continue + relay_uuid = relay_for_chain(chain, exit_node) + if not relay_uuid: + problems.append(f"{chain['code']}: у выходной ноды нет ключа для цепочки") + continue + old = old_chain_inbounds.get(inbound_tag(chain["code"])) + old_clients = old["settings"]["clients"] if old else [] + kept_inbounds.append(build_chain_inbound(template, chain, wanted, old_clients)) + kept_outbounds.append(build_chain_outbound(chain, exit_node, relay_uuid)) + kept_rules.append(build_chain_rule(chain)) + + cfg["inbounds"] = kept_inbounds + cfg["outbounds"] = kept_outbounds + routing["rules"] = kept_rules + + changed = json.dumps(cfg, sort_keys=True) != before + return changed, problems + + +def new_ports_needed(cfg, entry_chains): + existing = set() + for ib in cfg.get("inbounds", []): + if is_chain_inbound_tag(ib.get("tag")): + existing.add(ib["tag"]) + ports = [] + for chain in entry_chains: + if inbound_tag(chain["code"]) not in existing: + ports.append(chain["port"]) + return ports + + +def tcp_connect_ms(host, port, samples=3, timeout=3.0): + results = [] + for _ in range(samples): + start = time.perf_counter() + try: + with socket.create_connection((host, int(port)), timeout=timeout): + pass + results.append(round((time.perf_counter() - start) * 1000)) + except OSError: + results.append(-1) + return results + + +def median_ms(samples): + good = sorted(s for s in samples if s >= 0) + if not good: + return None + return good[len(good) // 2] + + +def latency_level(rtt_ms): + if rtt_ms is None: + return "unknown" + if rtt_ms < 40: + return "low" + if rtt_ms < 120: + return "medium" + return "high" diff --git a/db.py b/db.py index 62dec11..c2573eb 100644 --- a/db.py +++ b/db.py @@ -6,6 +6,7 @@ import secrets import datetime import contextlib +import chains as chainsmod from config import DB_PATH SCHEMA = """ @@ -113,6 +114,31 @@ CREATE TABLE IF NOT EXISTS nodes ( created_at TEXT NOT NULL ); +CREATE TABLE IF NOT EXISTS chains ( + code TEXT PRIMARY KEY, + label TEXT NOT NULL, + entry_node TEXT NOT NULL, + exit_node TEXT NOT NULL, + port INTEGER NOT NULL, + short_id TEXT NOT NULL, + relay_uuid TEXT, + enabled INTEGER NOT NULL DEFAULT 1, + sort_order INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS audit_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ts TEXT NOT NULL, + admin TEXT, + action TEXT NOT NULL, + detail TEXT, + ip TEXT +); + +CREATE UNIQUE INDEX IF NOT EXISTS idx_chains_pair ON chains (entry_node, exit_node); +CREATE UNIQUE INDEX IF NOT EXISTS idx_chains_entry_port ON chains (entry_node, port); +CREATE INDEX IF NOT EXISTS idx_audit_ts ON audit_log (ts); CREATE INDEX IF NOT EXISTS idx_subs_active_expires ON subscriptions (active, expires_at); CREATE INDEX IF NOT EXISTS idx_subs_tg_id ON subscriptions (tg_id); CREATE INDEX IF NOT EXISTS idx_subs_node ON subscriptions (node); @@ -355,9 +381,87 @@ def delete_node(code: str): ).fetchone()["c"] if active: raise ValueError(f"node has {active} active subscriptions, revoke them first") + in_chains = conn.execute( + "SELECT COUNT(*) c FROM chains WHERE entry_node=? OR exit_node=?", (code, code) + ).fetchone()["c"] + if in_chains: + raise ValueError(f"node is used in {in_chains} chain(s), delete them first") conn.execute("DELETE FROM nodes WHERE code=?", (code,)) +def list_chains(enabled_only: bool = False): + q = "SELECT * FROM chains" + if enabled_only: + q += " WHERE enabled=1" + q += " ORDER BY sort_order ASC, created_at ASC" + with get_conn() as conn: + rows = conn.execute(q).fetchall() + return [dict(r) for r in rows] + + +def get_chain(code: str): + with get_conn() as conn: + row = conn.execute("SELECT * FROM chains WHERE code=?", (code,)).fetchone() + return dict(row) if row else None + + +def create_chain(label: str, entry_node: str, exit_node: str, relay_uuid: str | None): + with get_conn() as conn: + dup = conn.execute( + "SELECT 1 FROM chains WHERE entry_node=? AND exit_node=?", (entry_node, exit_node) + ).fetchone() + if dup: + raise ValueError("такая цепочка уже есть") + used = {r["port"] for r in conn.execute( + "SELECT port FROM chains WHERE entry_node=?", (entry_node,) + ).fetchall()} + port = None + for candidate in range(chainsmod.PORT_MIN, chainsmod.PORT_MAX + 1): + if candidate not in used: + port = candidate + break + if port is None: + raise ValueError("закончились свободные порты под цепочки на этой ноде") + row = conn.execute("SELECT MAX(sort_order) m FROM chains").fetchone() + next_order = (row["m"] or 0) + 1 + code = "c" + secrets.token_hex(3) + conn.execute( + "INSERT INTO chains (code, label, entry_node, exit_node, port, short_id, relay_uuid, enabled, sort_order, created_at) " + "VALUES (?,?,?,?,?,?,?,1,?,?)", + (code, label, entry_node, exit_node, port, secrets.token_hex(8), relay_uuid, next_order, now_iso()), + ) + return get_chain(code) + + +def update_chain(code: str, **fields): + if not fields: + return get_chain(code) + cols = ", ".join(f"{k}=?" for k in fields) + with get_conn() as conn: + conn.execute(f"UPDATE chains SET {cols} WHERE code=?", (*fields.values(), code)) + return get_chain(code) + + +def delete_chain(code: str): + with get_conn() as conn: + conn.execute("DELETE FROM chains WHERE code=?", (code,)) + + +def add_audit(admin: str | None, action: str, detail: str = "", ip: str | None = None): + with get_conn() as conn: + conn.execute( + "INSERT INTO audit_log (ts, admin, action, detail, ip) VALUES (?,?,?,?,?)", + (now_iso(), admin, action, detail[:500], ip), + ) + conn.execute("DELETE FROM audit_log WHERE id <= (SELECT MAX(id) FROM audit_log) - 5000") + + +def list_audit(limit: int = 100): + with get_conn() as conn: + rows = conn.execute("SELECT * FROM audit_log ORDER BY id DESC LIMIT ?", (limit,)).fetchall() + return [dict(r) for r in rows] + + def _generate_ref_code(conn) -> str: for _ in range(20): code = secrets.token_hex(4) @@ -740,6 +844,32 @@ def get_subscription(client_uuid: str): return dict(row) if row else None +def list_users(q: str = "", limit: int = 200): + limit = max(1, min(int(limit), 1000)) + q = (q or "").strip() + where = "" + params = [now_iso(), now_iso()] + if q: + if q.lstrip("-").isdigit(): + where = "WHERE u.tg_id = ? OR instr(lower(COALESCE(u.username, '')), ?) > 0" + params += [int(q), q.lower()] + else: + where = "WHERE instr(lower(COALESCE(u.username, '')), ?) > 0" + params.append(q.lower().lstrip("@")) + params.append(limit) + query = ( + "SELECT u.tg_id, u.username, u.created_at, " + "(SELECT COUNT(*) FROM subscriptions s WHERE s.tg_id=u.tg_id) AS subs_total, " + "(SELECT COUNT(*) FROM subscriptions s WHERE s.tg_id=u.tg_id AND s.active=1 AND s.held_at IS NULL AND s.expires_at > ?) AS subs_active, " + "(SELECT MAX(s.expires_at) FROM subscriptions s WHERE s.tg_id=u.tg_id AND s.active=1 AND s.held_at IS NULL AND s.expires_at > ?) AS active_until, " + "(SELECT COUNT(*) FROM devices d WHERE d.tg_id=u.tg_id) AS devices " + "FROM users u " + where + " ORDER BY u.created_at DESC LIMIT ?" + ) + with get_conn() as conn: + rows = conn.execute(query, params).fetchall() + return [dict(r) for r in rows] + + def get_user(tg_id: int): with get_conn() as conn: row = conn.execute("SELECT * FROM users WHERE tg_id=?", (tg_id,)).fetchone() @@ -771,12 +901,16 @@ def stats(): total_subs = conn.execute("SELECT COUNT(*) c FROM subscriptions").fetchone()["c"] gifts_created = conn.execute("SELECT COUNT(*) c FROM gift_codes").fetchone()["c"] gifts_used = conn.execute("SELECT COUNT(*) c FROM gift_codes WHERE used_by IS NOT NULL").fetchone()["c"] + nodes_n = conn.execute("SELECT COUNT(*) c FROM nodes WHERE enabled=1").fetchone()["c"] + chains_n = conn.execute("SELECT COUNT(*) c FROM chains WHERE enabled=1").fetchone()["c"] return { "users": users_n, "active_subscriptions": active_n, "total_subscriptions": total_subs, "gifts_created": gifts_created, "gifts_used": gifts_used, + "nodes": nodes_n, + "chains": chains_n, } diff --git a/links.py b/links.py index a1d652d..a5bbcf6 100644 --- a/links.py +++ b/links.py @@ -89,6 +89,17 @@ def vless_uris_for_node(client_uuid: str, node: dict, base_name: str) -> list[st )] +def chain_remark(entry_node: dict, exit_node: dict) -> str: + return f"{display_name(entry_node['label'])} → {display_name(exit_node['label'])}" + + +def chain_uri(client_uuid: str, entry_node: dict, chain: dict, remark: str) -> str: + return _tcp_reality_uri( + client_uuid, entry_node["address"], chain["port"], entry_node["public_key"], + chain["short_id"], entry_node["sni"], "xtls-rprx-vision", remark, + ) + + def build_subscription_text(subs: list[dict]) -> str: import db @@ -99,6 +110,10 @@ def build_subscription_text(subs: list[dict]) -> str: best_by_node[s["node"]] = s nodes_by_code = {n["code"]: n for n in db.list_nodes()} + chains_by_entry = {} + for chain in db.list_chains(enabled_only=True): + chains_by_entry.setdefault(chain["entry_node"], []).append(chain) + lines = [] for node_code, s in best_by_node.items(): node = nodes_by_code.get(node_code) @@ -109,5 +124,10 @@ def build_subscription_text(subs: list[dict]) -> str: hy = hysteria_uri_for_node(node, base_name) if hy: lines.append(hy) + for chain in chains_by_entry.get(node_code, []): + exit_node = nodes_by_code.get(chain["exit_node"]) + if not node["enabled"] or not exit_node or not exit_node["enabled"]: + continue + lines.append(chain_uri(s["uuid"], node, chain, chain_remark(node, exit_node))) raw = "\n".join(lines) return base64.b64encode(raw.encode()).decode() diff --git a/nodeprov.py b/nodeprov.py index 4c3f877..6eda908 100644 --- a/nodeprov.py +++ b/nodeprov.py @@ -1,3 +1,6 @@ +import contextlib +import fcntl +import hashlib import json import secrets import socket @@ -5,12 +8,12 @@ import subprocess import paramiko +import chains from config import PANEL_DOMAIN MGMT_KEY_PATH = "/root/.ssh/mbs_nodes_ed25519" MGMT_KNOWN_HOSTS_PATH = "/root/.ssh/mbs_nodes_known_hosts" -LOCAL_TAGS = {"vless-tcp-reality", "vless-grpc-reality", "vless-xhttp-reality", "vless-ws-tls"} -TAG_FLOW = {"vless-tcp-reality": "xtls-rprx-vision"} +REMOTE_CONFIG_PATH = "/usr/local/etc/xray/config.json" ONE_COMMAND_TEMPLATE = "bash <(curl -Ls https://{panel}/install/{token}.sh)" @@ -254,26 +257,45 @@ class RemoteConfigError(Exception): pass -def _remote_edit_clients(node: dict, mutate_fn): +def _busy_ports(client) -> set: + _, stdout, _ = client.exec_command("ss -ltnH 2>/dev/null | awk '{print $4}'", timeout=10) + busy = set() + for line in stdout.read().decode(errors="replace").splitlines(): + tail = line.rsplit(":", 1)[-1] + if tail.isdigit(): + busy.add(int(tail)) + return busy + + +@contextlib.contextmanager +def _node_lock(address: str): + key = hashlib.sha1(address.encode()).hexdigest()[:12] + with open(f"/tmp/mbs-node-{key}.lock", "w") as lock_file: + fcntl.flock(lock_file, fcntl.LOCK_EX) + try: + yield + finally: + fcntl.flock(lock_file, fcntl.LOCK_UN) + + +def _remote_edit_config(node: dict, mutate_fn): + with _node_lock(node["address"]): + return _remote_edit_config_locked(node, mutate_fn) + + +def _remote_edit_config_locked(node: dict, mutate_fn): client = _mgmt_connect(node["address"]) try: sftp = client.open_sftp() - with sftp.open("/usr/local/etc/xray/config.json") as f: + with sftp.open(REMOTE_CONFIG_PATH) as f: cfg = json.loads(f.read().decode()) - changed = False - for ib in cfg["inbounds"]: - if ib.get("tag") not in LOCAL_TAGS: - continue - clients = ib["settings"]["clients"] - new_clients = mutate_fn(clients, ib["tag"]) - if new_clients is not None: - ib["settings"]["clients"] = new_clients - changed = True - if not changed: + result = mutate_fn(cfg, client) + if not result["changed"]: sftp.close() - return + return result data = json.dumps(cfg, indent=2).encode() - tmp_path = "/usr/local/etc/xray/config.json.validate.tmp" + tmp_path = REMOTE_CONFIG_PATH + ".validate.tmp" + prev_path = REMOTE_CONFIG_PATH + ".mbs-prev" with sftp.open(tmp_path, "wb") as f: f.write(data) _, stdout, stderr = client.exec_command(f"/usr/local/bin/xray run -test -format=json -config {tmp_path}", timeout=15) @@ -283,23 +305,43 @@ def _remote_edit_clients(node: dict, mutate_fn): client.exec_command(f"rm -f {tmp_path}") sftp.close() raise RemoteConfigError(f"config test failed on {node['address']}: {test_out}") - client.exec_command(f"mv {tmp_path} /usr/local/etc/xray/config.json")[1].channel.recv_exit_status() + client.exec_command(f"cp -p {REMOTE_CONFIG_PATH} {prev_path}")[1].channel.recv_exit_status() + client.exec_command(f"mv {tmp_path} {REMOTE_CONFIG_PATH}")[1].channel.recv_exit_status() sftp.close() - _, stdout, stderr = client.exec_command("systemctl restart xray", timeout=20) + _, stdout, stderr = client.exec_command("systemctl restart xray && sleep 1 && systemctl is-active xray", timeout=30) restart_exit = stdout.channel.recv_exit_status() if restart_exit != 0: err = stderr.read().decode(errors="replace").strip() - raise RemoteConfigError(f"xray restart failed on {node['address']}: {err}") + client.exec_command(f"cp -p {prev_path} {REMOTE_CONFIG_PATH} && systemctl restart xray")[1].channel.recv_exit_status() + raise RemoteConfigError(f"xray не поднялся на {node['address']}, конфиг откатили назад: {err}") + for port in result.get("new_ports") or []: + client.exec_command(f"command -v ufw >/dev/null 2>&1 && ufw allow {int(port)}/tcp || true")[1].channel.recv_exit_status() + return result finally: client.close() +def _remote_edit_clients(node: dict, mutate_fn): + def mutate(cfg, client): + changed = False + for ib in cfg["inbounds"]: + tag = ib.get("tag") + if not chains.is_user_tag(tag): + continue + new_clients = mutate_fn(ib["settings"]["clients"], tag) + if new_clients is not None: + ib["settings"]["clients"] = new_clients + changed = True + return {"changed": changed} + _remote_edit_config(node, mutate) + + def remote_add_client(node: dict, client_uuid: str, email: str): def mutate(clients, tag): if any(c["id"] == client_uuid for c in clients): return None entry = {"id": client_uuid, "email": email} - flow = TAG_FLOW.get(tag) + flow = chains.flow_for_tag(tag) if flow: entry["flow"] = flow clients.append(entry) @@ -314,24 +356,49 @@ def remote_remove_client(node: dict, client_uuid: str): _remote_edit_clients(node, mutate) -def remote_sync(node: dict, active_subs: list[dict]): - active_by_id = {s["uuid"]: s for s in active_subs} +def remote_reconcile(node: dict, wanted: dict, relay_wanted: dict, entry_chains: list, exit_nodes: dict, apply_chains: bool = True): + def mutate(cfg, client): + usable = entry_chains + skipped = [] + if apply_chains: + usable, skipped = chains.split_busy_chains(cfg, entry_chains, _busy_ports(client)) + new_ports = chains.new_ports_needed(cfg, usable) if apply_chains else [] + changed, problems = chains.sync_config(cfg, wanted, relay_wanted, usable, exit_nodes, apply_chains=apply_chains) + return {"changed": changed, "new_ports": new_ports, "problems": skipped + problems} + return _remote_edit_config(node, mutate) - def mutate(clients, tag): - current_ids = {c["id"] for c in clients} - if current_ids == set(active_by_id.keys()): - return None - new_clients = [c for c in clients if c["id"] in active_by_id] - existing_ids = {c["id"] for c in new_clients} - flow = TAG_FLOW.get(tag) - for cid in active_by_id: - if cid not in existing_ids: - entry = {"id": cid, "email": cid} - if flow: - entry["flow"] = flow - new_clients.append(entry) - return new_clients - _remote_edit_clients(node, mutate) + +PROBE_SCRIPT = """for i in 1 2 3; do + s=$(date +%s%N) + if timeout 3 bash -c 'exec 3<>/dev/tcp/{host}/{port}' 2>/dev/null; then + e=$(date +%s%N) + echo $(( (e - s) / 1000000 )) + else + echo -1 + fi +done +""" + + +def remote_probe(node: dict, host: str, port: int) -> list: + if not chains.HOST_RE.match(host or ""): + raise ValueError("bad host") + port = int(port) + client = _mgmt_connect(node["address"]) + try: + stdin, stdout, _ = client.exec_command("bash -s", timeout=30) + stdin.write(PROBE_SCRIPT.format(host=host, port=port)) + stdin.channel.shutdown_write() + out = stdout.read().decode(errors="replace") + finally: + client.close() + samples = [] + for line in out.split(): + try: + samples.append(int(line)) + except ValueError: + continue + return samples def remote_query_stats(node: dict) -> dict: diff --git a/xray_manager.py b/xray_manager.py index 1c66010..e13ddc5 100644 --- a/xray_manager.py +++ b/xray_manager.py @@ -1,14 +1,15 @@ import json import os +import socket import subprocess import fcntl import contextlib +import time -from config import XRAY_CONFIG_PATH, DE1_TRANSPORTS +import chains +from config import XRAY_CONFIG_PATH _LOCK_PATH = XRAY_CONFIG_PATH + ".lock" -_LOCAL_TAGS = {t["tag"] for t in DE1_TRANSPORTS} -_TAG_FLOW = {t["tag"]: t.get("flow") for t in DE1_TRANSPORTS} @contextlib.contextmanager @@ -101,7 +102,7 @@ def _reload_xray(): def _local_inbounds(cfg): - return [ib for ib in cfg["inbounds"] if ib.get("tag") in _LOCAL_TAGS] + return [ib for ib in cfg["inbounds"] if chains.is_user_tag(ib.get("tag"))] def add_client(client_uuid: str, email: str): @@ -113,7 +114,7 @@ def add_client(client_uuid: str, email: str): if any(c["id"] == client_uuid for c in clients): continue entry = {"id": client_uuid, "email": email} - flow = _TAG_FLOW.get(ib["tag"]) + flow = chains.flow_for_tag(ib["tag"]) if flow: entry["flow"] = flow clients.append(entry) @@ -138,38 +139,138 @@ def remove_client(client_uuid: str): _reload_xray() +def _node_usable(node): + return bool(node["enabled"]) and node["status"] == "active" + + +def desired_state(node): + import db as dbmod + + active = dbmod.list_active_subscriptions(node=node["code"]) + wanted = {s["uuid"]: s["uuid"] for s in active} + nodes_by_code = {n["code"]: n for n in dbmod.list_nodes()} + entry_chains = [] + exit_nodes = {} + relay_wanted = {} + for chain in dbmod.list_chains(enabled_only=True): + entry = nodes_by_code.get(chain["entry_node"]) + exit_node = nodes_by_code.get(chain["exit_node"]) + if not entry or not exit_node: + continue + if not _node_usable(entry) or not _node_usable(exit_node): + continue + if chain["entry_node"] == node["code"]: + entry_chains.append(chain) + exit_nodes[chain["exit_node"]] = exit_node + if chain["exit_node"] == node["code"] and node["kind"] in ("local", "managed") and chain.get("relay_uuid"): + relay_wanted[chain["relay_uuid"]] = chains.relay_email(chain["code"]) + return wanted, relay_wanted, entry_chains, exit_nodes + + +def _read_config_text(): + with open(XRAY_CONFIG_PATH, "r", encoding="utf-8") as f: + return f.read() + + +def _restore_config_text(text): + tmp = XRAY_CONFIG_PATH + ".restore.tmp" + with open(tmp, "w", encoding="utf-8") as f: + f.write(text) + os.replace(tmp, XRAY_CONFIG_PATH) + subprocess.run(["systemctl", "restart", "xray"], timeout=20) + + +def _reload_and_verify(): + subprocess.run(["systemctl", "restart", "xray"], check=True, timeout=20) + time.sleep(1) + state = subprocess.run(["systemctl", "is-active", "xray"], capture_output=True, text=True).stdout.strip() + if state != "active": + raise ConfigValidationError("xray не поднялся после применения конфига, вернули старый") + + +def _port_busy(port): + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + try: + sock.bind(("0.0.0.0", int(port))) + return False + except OSError: + return True + finally: + sock.close() + + +def _open_firewall(port): + subprocess.run( + ["sh", "-c", f"command -v ufw >/dev/null 2>&1 && ufw allow {int(port)}/tcp || true"], + timeout=20, + ) + + +def _reconcile_local(wanted, relay_wanted, entry_chains, exit_nodes, apply_chains=True): + with _locked(): + before_text = _read_config_text() + cfg = json.loads(before_text) + usable = entry_chains + skipped = [] + new_ports = [] + if apply_chains: + busy = {c["port"] for c in entry_chains if _port_busy(c["port"])} + usable, skipped = chains.split_busy_chains(cfg, entry_chains, busy) + new_ports = chains.new_ports_needed(cfg, usable) + changed, problems = chains.sync_config(cfg, wanted, relay_wanted, usable, exit_nodes, apply_chains=apply_chains) + if changed: + _save(cfg) + try: + _reload_and_verify() + except Exception: + _restore_config_text(before_text) + raise + for port in new_ports: + _open_firewall(port) + return {"changed": changed, "new_ports": new_ports, "problems": skipped + problems} + + +def sync_node(node): + wanted, relay_wanted, entry_chains, exit_nodes = desired_state(node) + if node["kind"] == "managed": + import nodeprov + reconcile = nodeprov.remote_reconcile + args = (node, wanted, relay_wanted, entry_chains, exit_nodes) + elif node["kind"] == "local": + reconcile = _reconcile_local + args = (wanted, relay_wanted, entry_chains, exit_nodes) + else: + return {"changed": False, "new_ports": [], "problems": []} + try: + return reconcile(*args) + except Exception as first_error: + if not entry_chains: + raise + result = reconcile(*args, apply_chains=False) + result["problems"].append(f"цепочки не применились, клиенты синхронизированы: {first_error}") + return result + + def sync_from_db(): import db as dbmod expired = dbmod.deactivate_expired() - active = dbmod.list_active_subscriptions(node="de1") - active_by_id = {s["uuid"]: s for s in active} + node = dbmod.get_node("de1") + result = sync_node(node) + wanted = desired_state(node)[0] + return { + "removed_expired": len(expired), "active_now": len(wanted), + "reloaded": result["changed"], "problems": result["problems"], + } - with _locked(): - cfg = _load() - changed = False - for ib in _local_inbounds(cfg): - clients = ib["settings"]["clients"] - current_ids = {c["id"] for c in clients} - if current_ids == set(active_by_id.keys()): - continue - new_clients = [c for c in clients if c["id"] in active_by_id] - existing_ids = {c["id"] for c in new_clients} - flow = _TAG_FLOW.get(ib["tag"]) - for cid, sub in active_by_id.items(): - if cid not in existing_ids: - entry = {"id": cid, "email": cid} - if flow: - entry["flow"] = flow - new_clients.append(entry) - ib["settings"]["clients"] = new_clients - changed = True - if changed: - _save(cfg) - _reload_xray() - - return {"removed_expired": len(expired), "active_now": len(active_by_id), "reloaded": changed} +def probe_from_node(node: dict, host: str, port: int): + if node["kind"] == "local": + return chains.tcp_connect_ms(host, port) + if node["kind"] == "managed": + import nodeprov + return nodeprov.remote_probe(node, host, port) + raise ValueError("нода не под управлением панели, замерить с неё нельзя") def add_client_to_node(node: dict, client_uuid: str, email: str): @@ -266,7 +367,6 @@ def local_node_status() -> dict: def sync_all(): import db as dbmod - import nodeprov expired = dbmod.deactivate_expired() results = {} @@ -276,8 +376,15 @@ def sync_all(): elif node["kind"] == "managed": active = dbmod.list_active_subscriptions(node=node["code"]) try: - nodeprov.remote_sync(node, active) - results[node["code"]] = {"active_now": len(active), "ok": True} + res = sync_node(node) + results[node["code"]] = { + "active_now": len(active), "ok": True, + "changed": res["changed"], "problems": res["problems"], + } except Exception as e: results[node["code"]] = {"active_now": len(active), "ok": False, "error": str(e)} - return {"removed_expired": len(expired), "nodes": results} + reloaded = any(r.get("changed") or r.get("reloaded") for r in results.values()) + return { + "removed_expired": len(expired), "active_now": len(dbmod.list_active_subscriptions()), + "reloaded": reloaded, "nodes": results, + } From 2c5ec8b06ce50a0879a844d11b091053c29febdb Mon Sep 17 00:00:00 2001 From: savsis Date: Sun, 4 Oct 2026 03:51:16 +0500 Subject: [PATCH 06/11] feat: admin panel redesign, chain builder, users page, audit log, command palette New visual system with accent presets, toasts and custom confirms. Chain builder: hold the left button on Client and drag the wire through servers to Internet (max two servers, latency warning with measured RTT). Users page lists everyone incl. people without subscriptions, with search and grant by telegram id. Live node ping, Ctrl+K palette. --- admin.html | 1846 ++++++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 1664 insertions(+), 182 deletions(-) diff --git a/admin.html b/admin.html index 54f151e..87b26e7 100644 --- a/admin.html +++ b/admin.html @@ -4,272 +4,580 @@ MBS Panel + - + @@ -301,40 +609,80 @@
MBS Panel
+
Обзор
Дашборд
+
Сеть
+
Ноды
+
Цепочки
+
Трафик
+
Продажи
+
Юзеры
Подписки
Гифт-коды
-
Ноды
-
Трафик
Платежи
+
Система
+
Журнал
Документация
Настройки
-
MBS Panel v1.2.0 · обновления
+
MBS Panel v1.6.0 · обновления
+
+
MBS Panel/Дашборд
+
+
Поиск и командыCtrl K
+
+
онлайн
+
+
Дашборд
-
Общая статистика по сервису
+
Что происходит с сервисом прямо сейчас
+
+
+

Сеть

+
+
+
+

Последние действия

+
+
+
-

Последние подписки

+

Последние подписки

ПользовательСерверТарифИстекаетСтатус
+
+
Юзеры
+
Все, кто уже есть в системе, в том числе те, у кого ни разу не было подписки (в «Подписках» их не видно). Открой карточку и выдай подписку на любую ноду и срок.
+
+
+
+
+
+
+
+ +
ПользовательПодпискиАктивна доУстройствЗарегистрирован
+
+
Подписки
Все выданные подписки
+
@@ -363,7 +711,7 @@
ПользовательСерверТарифВыданаИстекаетСтатус
- +
ЛокацияАдресТипСтатусLiveЛокацияАдресТипСтатусПингLive
@@ -421,6 +769,62 @@
+
+
Цепочки
+
Собери маршрут клиент → сервер → сервер → интернет. Зажми левую кнопку мыши на «Клиенте» и протяни провод через серверы из пула к «Интернету». Один сервер — это обычное подключение, два — цепочка. Больше двух нельзя: каждый прыжок добавляет задержку.
+
+
+
+ +
+ +
+
Клиент
+
зажми ЛКМ и тяни
+
+
+
Пул серверов
+
+
+
+ +
+
Интернет
+
+
+
+
зажми ЛКМ на клиенте, веди провод через серверы, отпусти на интернете · клик по серверу тоже добавляет его · Esc — сбросить
+
+

Маршрут

+
+
+
+
+ + + +
+
+
+
+
+

Мои цепочки

+
+
+
+ +
+
Журнал
+
Кто и что менял в панели — последние 300 действий. Пароли и ключи сюда не попадают, только факт действия.
+
+
+
+
+
+ +
ВремяАдминДействиеДеталиIP
+
+
+
+
↑↓ выбратьEnter открытьEsc закрыть
+