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, + }