feat: server chains (client -> A -> B -> internet), audit log, users list, csv export

Chains are real Xray hops: a chain-<code> inbound + outbound + routing rule on the
entry node and a relay-<code> 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

@@ -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")
 

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}

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"

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

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()

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"}
+REMOTE_CONFIG_PATH = "/usr/local/etc/xray/config.json"
-TAG_FLOW = {"vless-tcp-reality": "xtls-rprx-vision"}
 
 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
+        result = mutate_fn(cfg, client)
-        for ib in cfg["inbounds"]:
+        if not result["changed"]:
-            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:
             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]):
+def remote_reconcile(node: dict, wanted: dict, relay_wanted: dict, entry_chains: list, exit_nodes: dict, apply_chains: bool = True):
-    active_by_id = {s["uuid"]: s for s in active_subs}
+    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}
+PROBE_SCRIPT = """for i in 1 2 3; do
-        if current_ids == set(active_by_id.keys()):
+  s=$(date +%s%N)
-            return None
+  if timeout 3 bash -c 'exec 3<>/dev/tcp/{host}/{port}' 2>/dev/null; then
-        new_clients = [c for c in clients if c["id"] in active_by_id]
+    e=$(date +%s%N)
-        existing_ids = {c["id"] for c in new_clients}
+    echo $(( (e - s) / 1000000 ))
-        flow = TAG_FLOW.get(tag)
+  else
-        for cid in active_by_id:
+    echo -1
-            if cid not in existing_ids:
+  fi
-                entry = {"id": cid, "email": cid}
+done
-                if flow:
+"""
-                    entry["flow"] = flow
+
-                new_clients.append(entry)
+
-        return new_clients
+def remote_probe(node: dict, host: str, port: int) -> list:
-    _remote_edit_clients(node, mutate)
+    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:

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")
+    node = dbmod.get_node("de1")
-    active_by_id = {s["uuid"]: s for s in active}
+    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:
+def probe_from_node(node: dict, host: str, port: int):
-            _save(cfg)
+    if node["kind"] == "local":
-            _reload_xray()
+        return chains.tcp_connect_ms(host, port)
-
+    if node["kind"] == "managed":
-    return {"removed_expired": len(expired), "active_now": len(active_by_id), "reloaded": changed}
+        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)
+                res = sync_node(node)
-                results[node["code"]] = {"active_now": len(active), "ok": True}
+                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,
+    }