web_users v0.4.3 [core]

"""Пакетный config-шим для engine (вендор web_users.py пишет config.X)."""

_vals: dict = {}


def install(values: dict, files: dict) -> None:
    _vals.clear()
    _vals.update(values)
    _vals.update(files)
    try:
        _vals["WEB_SESSION_TTL_HOURS"] = int(values.get("WEB_SESSION_TTL_HOURS", "168") or 168)
    except (ValueError, TypeError):
        _vals["WEB_SESSION_TTL_HOURS"] = 168
    _vals["WEB_OPEN_REGISTRATION"] = (values.get("WEB_OPEN_REGISTRATION", "1") == "1")


def __getattr__(name: str):
    try:
        return _vals[name]
    except KeyError:
        raise AttributeError("web_users.config: нет ключа %s" % name) from None

# modules/web_users.py
"""Учётные записи и сессии веб-панели.

Хранение: web_users.json (логин + PBKDF2-хэш пароля). Права — в общем
admin_rights.json под ключами "web:<id>" (тот же механизм, что у Telegram:
DEFAULT_RIGHTS => новый пользователь по умолчанию гость, доп. права выдаёт
администратор через обычный редактор прав).

Сессии — в файле cache/web_sessions.json (переживают перезапуск бота).
"""
import hashlib
import hmac
import json
import logging
import os
import secrets
import threading
import time

from . import config

logger = logging.getLogger(__name__)

_lock = threading.RLock()
_SESSIONS_FILE = None  # ставит configure() из Module (data/sessions.json)
_rights = None  # RightsService, ставит configure()


def configure(sessions_file: str, rights) -> None:
    """Привязка к путям/сервисам ядра. Вызывает Module.setup, ленивая загрузка."""
    global _SESSIONS_FILE, _rights
    _SESSIONS_FILE = sessions_file
    _rights = rights
    _store()  # прогрев + чистка протухших


def _store():
    """Ленивый доступ к сессиям (загрузка при первом обращении)."""
    global _sessions
    if _sessions is None:
        _sessions = _load_sessions_file()
        now = time.time()
        fresh = {k: v for k, v in _sessions.items() if v.get("expires", 0) > now}
        if len(fresh) != len(_sessions):
            _sessions = fresh
            _save_sessions_file(_sessions)
    return _sessions


def _load_sessions_file():
    try:
        with open(_SESSIONS_FILE, "r", encoding="utf-8") as f:
            data = json.load(f)
            if isinstance(data, dict):
                return data
    except FileNotFoundError:
        pass
    except Exception as e:
        logger.warning("Не удалось прочитать сессии %s: %s", _SESSIONS_FILE, e)
    return {}


def _save_sessions_file(sessions):
    try:
        os.makedirs(os.path.dirname(_SESSIONS_FILE) or ".", exist_ok=True)
        tmp = _SESSIONS_FILE + ".tmp"
        with open(tmp, "w", encoding="utf-8") as f:
            json.dump(sessions, f, ensure_ascii=False)
        os.replace(tmp, _SESSIONS_FILE)
    except Exception as e:
        logger.warning("Не удалось сохранить сессии %s: %s", _SESSIONS_FILE, e)


_sessions = None  # лениво через _store(); token -> {"uid", "username", "expires"}

_PBKDF2_ITERS = 200_000
_MIN_LOGIN = 3
_MIN_PASSWORD = 6
_LOGIN_WINDOW = 300
_LOGIN_MAX_FAILS = 10
_fails: dict[str, list] = {}

_RESERVED_USERNAMES = frozenset({
    "admin", "administrator", "admins", "root", "system", "sysadmin",
    "superadmin", "support", "operator", "webmaster", "postmaster", "owner",
})


def _hash_password(password: str, salt: str) -> str:
    dk = hashlib.pbkdf2_hmac("sha256", password.encode(), bytes.fromhex(salt), _PBKDF2_ITERS)
    return dk.hex()


def load_users():
    try:
        with open(config.WEB_USERS_FILE, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception:
        return {"users": []}


def save_users(data):
    with open(config.WEB_USERS_FILE, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, ensure_ascii=False)
    try:
        os.chmod(config.WEB_USERS_FILE, 0o600)  # внутри — хэши паролей
    except OSError:
        pass


def validate_username(username: str):
    if not username or len(username) < _MIN_LOGIN or len(username) > 32:
        return f"Логин: от {_MIN_LOGIN} до 32 символов"
    allowed = set("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_.-")
    if not set(username) <= allowed:
        return "Логин: только латиница, цифры и символы _ . -"
    if username.lower() in _RESERVED_USERNAMES:
        return "Логин: имя зарезервировано"
    return None


def validate_password(password: str):
    if not password or len(password) < _MIN_PASSWORD:
        return f"Пароль: минимум {_MIN_PASSWORD} символов"
    return None


def create_user(username: str, password: str):
    """Создаёт пользователя и гостевые права. Возвращает (user|None, ошибка|None)."""
    err = validate_username(username) or validate_password(password)
    if err:
        return None, err
    with _lock:
        data = load_users()
        for u in data["users"]:
            if u["username"].lower() == username.lower():
                return None, "Логин уже занят"
        new_id = max([u["id"] for u in data["users"]], default=0) + 1
        salt = secrets.token_hex(16)
        user = {
            "id": new_id,
            "username": username,
            "salt": salt,
            "pwhash": _hash_password(password, salt),
            "created": time.time(),
        }
        data["users"].append(user)
        try:
            save_users(data)
        except Exception as e:
            logger.error("Не удалось сохранить web_users.json: %s", e)
            return None, "Ошибка сохранения"
        ensure_guest_rights(new_id)
        logger.info("Зарегистрирован web-пользователь %s (web:%s)", username, new_id)
        return user, None


def verify_login(username: str, password: str):
    with _lock:
        for u in load_users()["users"]:
            if u["username"].lower() == (username or "").lower():
                calc = _hash_password(password or "", u["salt"])
                if hmac.compare_digest(calc, u["pwhash"]):
                    return u
        return None


def set_password(uid: int, old_password: str, new_password: str):
    """Смена своего пароля: проверка старого обязательна."""
    err = validate_password(new_password)
    if err:
        return None, err
    with _lock:
        data = load_users()
        for u in data.get("users", []):
            if u.get("id") == uid:
                calc = _hash_password(old_password or "", u["salt"])
                if not hmac.compare_digest(calc, u["pwhash"]):
                    return None, "Старый пароль неверный"
                salt = secrets.token_hex(16)
                u["salt"] = salt
                u["pwhash"] = _hash_password(new_password, salt)
                try:
                    save_users(data)
                except Exception as e:  # noqa: BLE001
                    logger.error("Не удалось сохранить web_users.json: %s", e)
                    return None, "Ошибка сохранения"
                logger.info("Смена пароля web:%s", uid)
                return u, None
        return None, "Пользователь не найден"


def admin_set_password(uid: int, new_password: str):
    """Сброс пароля администратором (старый не нужен)."""
    err = validate_password(new_password)
    if err:
        return None, err
    with _lock:
        data = load_users()
        for u in data.get("users", []):
            if u.get("id") == uid:
                salt = secrets.token_hex(16)
                u["salt"] = salt
                u["pwhash"] = _hash_password(new_password, salt)
                try:
                    save_users(data)
                except Exception as e:  # noqa: BLE001
                    logger.error("Не удалось сохранить web_users.json: %s", e)
                    return None, "Ошибка сохранения"
                logger.info("Сброс пароля web:%s администратором", uid)
                return u, None
        return None, "Пользователь не найден"


# --- сессии -------------------------------------------------------------

def create_session(user) -> str:
    token = secrets.token_urlsafe(32)
    with _lock:
        sessions = _store()
        sessions[token] = {
            "uid": user["id"],
            "username": user["username"],
            "expires": time.time() + config.WEB_SESSION_TTL_HOURS * 3600,
        }
        _cleanup_sessions()
        _save_sessions_file(sessions)
    return token


def session_ttl():
    """TTL сессии в секундах (для скользящего продления куки)."""
    try:
        return max(3600, int(config.WEB_SESSION_TTL_HOURS) * 3600)
    except Exception:  # noqa: BLE001
        return 168 * 3600


def get_session(token: str):
    if not token:
        return None
    with _lock:
        sessions = _store()
        s = sessions.get(token)
        if not s:
            return None
        now = time.time()
        if s["expires"] < now:
            del sessions[token]
            _save_sessions_file(sessions)
            return None
        # Скользящее продление: в последней четверти срока — продлить
        # (киоск-планшет не вылетает при активности).
        ttl = session_ttl()
        if s["expires"] - now < ttl / 4:
            s["expires"] = now + ttl
            _save_sessions_file(sessions)
        return dict(s, token=token)


def drop_session(token: str):
    with _lock:
        sessions = _store()
        if token in sessions:
            sessions.pop(token, None)
            _save_sessions_file(sessions)


def _cleanup_sessions():
    now = time.time()
    sessions = _store()
    expired = [t for t, s in sessions.items() if s.get("expires", 0) < now]
    for t in expired:
        del sessions[t]
    if expired:
        _save_sessions_file(sessions)


# --- права ---------------------------------------------------------------

def get_right(uid, key) -> bool:
    """Право web-пользователя через RightsService ядра (семантика effective)."""
    if _rights is None:
        return False
    eff = _rights.effective(f"web:{uid}")
    return bool(eff.get(key, False))


def get_all_rights(uid) -> dict:
    """Полный словарь прав пользователя (для отображения в кабинете)."""
    if _rights is None:
        return {"all": False}
    return _rights.effective(f"web:{uid}")


def ensure_guest_rights(uid):
    """Гостевой набор прав для нового web-пользователя (идемпотентно)."""
    if _rights is None:
        return
    key = f"web:{uid}"
    if _rights.get_entry(key):
        return
    _rights.set_entry(key, _rights.guest_defaults())
    logger.info("Выданы гостевые права web:%s", uid)


def is_ratelimited(ip: str) -> bool:
    with _lock:
        now = time.time()
        fails = [t for t in _fails.get(ip, []) if now - t < _LOGIN_WINDOW]
        _fails[ip] = fails
        return len(fails) >= _LOGIN_MAX_FAILS


def note_fail(ip: str) -> None:
    with _lock:
        _fails.setdefault(ip, []).append(time.time())


def find_uid_by_username(username: str):
    for u in load_users()["users"]:
        if u["username"].lower() == (username or "").lower():
            return u["id"]
    return None


def delete_user(uid: int):
    """Удаляет web-пользователя: запись в web_users.json, права в admin_rights.json
    и все его сессии. Возвращает (ok|None, ошибка|None)."""
    with _lock:
        data = load_users()
        before = len(data["users"])
        data["users"] = [u for u in data["users"] if u["id"] != uid]
        if len(data["users"]) == before:
            return None, "Пользователь не найден"
        try:
            save_users(data)
        except Exception as e:
            logger.error("Не удалось сохранить web_users.json: %s", e)
            return None, "Ошибка сохранения"
        rights = None
        if _rights is not None:
            rights = _rights.get_entry(f"web:{uid}")
        if rights:
            _rights.delete_entry(f"web:{uid}")
        sessions = _store()
        dropped = [t for t, s in sessions.items() if s.get("uid") == uid]
        for t in dropped:
            del sessions[t]
        if dropped:
            _save_sessions_file(sessions)
        logger.info("Удалён web-пользователь web:%s", uid)
        return True, None

"""web_users.Module — учётные записи (PBKDF2) и сессии веб-панели.

Хранение — data/: users.json (0600), sessions.json. Права — в общем
admin_rights.json через ctx.rights (ключи web:<id>).
"""
import logging
import os

from core.base_module import BaseModule

from . import config as pkg_config
from . import engine as users

logger = logging.getLogger(__name__)


def _right_meta(ctx):
    """Права с атрибуцией: {name: {module, desc, origin}}.

    origin: manifest (объявлено модулем) | defaults (ядро) | file
    (встречается только в admin_rights.json — наследие/ручная правка).
    """
    meta = {}
    try:
        declared = ctx.rights.list_rights()
    except Exception:  # noqa: BLE001
        declared = []
    for r in declared:
        if isinstance(r, dict) and r.get("name"):
            meta[r["name"]] = {
                "module": r.get("module") or "?",
                "desc": r.get("description") or "",
                "origin": "manifest",
            }
    try:
        file_keys = set()
        for qid in ctx.rights.known_users():
            entry = ctx.rights.get_entry(qid) or {}
            file_keys.update(k for k in entry if isinstance(k, str))
    except Exception:  # noqa: BLE001
        file_keys = set()
    for name in sorted(file_keys):
        if name not in meta and name != "pve_manage_hosts":
            meta[name] = {"module": "", "desc": "",
                          "origin": "file"}
    return meta


class Module(BaseModule):
    def setup(self, ctx) -> None:
        mod_dir = os.path.dirname(os.path.abspath(__file__))
        self.config = ctx.module_config("web_users")
        self.ctx = ctx

        def _f(key, default):
            return (self.config.get(key, "") or "").strip() or default

        def _fp(key, default):
            from core.paths import abs_data
            return abs_data(ctx.data_dir,
                            (self.config.get(key, "") or "").strip(),
                            default)

        files = {
            "WEB_USERS_FILE": _fp("WEB_USERS_FILE", os.path.join(ctx.data_dir, "users.json")),
            "WEB_SESSIONS_FILE": _fp(
                "WEB_SESSIONS_FILE", os.path.join(ctx.data_dir, "sessions.json")),
        }
        pkg_config.install(dict(self.config.values), files)
        users.configure(files["WEB_SESSIONS_FILE"], ctx.rights)

        ctx.api.register("web_users", "create", users.create_user)
        ctx.api.register("web_users", "verify", users.verify_login)
        ctx.api.register("web_users", "set_password", users.set_password)
        ctx.api.register(
            "web_users", "admin_set_password", users.admin_set_password)
        ctx.api.register("web_users", "session", users.get_session)
        ctx.api.register("web_users", "session_ttl", users.session_ttl)
        ctx.api.register(
            "web_users", "create_session", users.create_session)
        ctx.api.register("web_users", "drop_session", users.drop_session)
        ctx.api.register("web_users", "delete", users.delete_user)
        ctx.api.register(
            "web_users", "list",
            lambda: [dict(u) for u in users.load_users().get("users", [])])
        ctx.api.register(
            "web_users", "rights_of",
            lambda uid: ctx.rights.effective("web:%s" % uid),
        )
        ctx.api.register(
            "web_users", "right_catalog",
            # "all" — служебный ключ (не право модуля), но выдаётся
            # через тот же тоггл — держим в каталоге явно.
            lambda: sorted({"all"} | {
                r["name"] for r in ctx.rights.list_rights()}),
        )
        ctx.api.register("web_users", "right_meta",
                         lambda: _right_meta(ctx))
        ctx.api.register(
            "web_users", "all_ignore", lambda: sorted(ctx.rights.all_ignored()))
        ctx.api.register(
            "web_users", "open_registration",
            lambda: (self.config.get("WEB_OPEN_REGISTRATION", "1") == "1"),
        )

    def health(self) -> dict:
        try:
            n = len(users.load_users().get("users", []))
        except Exception:  # noqa: BLE001
            n = -1
        return {"ok": n >= 0, "module": self.name, "users": n}

"""web_users.ui_web — /api/auth/*, /api/me, /api/admin/users*, /api/admin/rights.

Порт Register/Login/Logout/Me + AdminUsers/AdminUserDelete/AdminRightsToggle
из старого messengers/web.py. /api/me — форма 1:1 со старым ботом
(authenticated/user/rights/pve_hosts/is_admin/via), иначе SPA не встанет.
POST принимают только application/json (транспорт уже отбил 415 иначе).
login/logout управляют cookie websess через extra-ответ (см. контракт v1.1).
"""

ROUTES = [
    ("POST", "/api/auth/register", "register", {"right": None}),
    ("POST", "/api/auth/login", "login", {"right": None}),
    ("POST", "/api/auth/logout", "logout", {"right": None}),
    ("POST", "/api/auth/password", "password", {"right": None}),
    ("GET", "/api/me", "me", {"right": None}),
    ("GET", "/api/admin/users", "admin_users", {"right": "manage_rights"}),
    ("POST", "/api/admin/users/delete", "admin_user_delete",
     {"right": "manage_rights"}),
    ("POST", "/api/admin/users/password", "admin_user_password",
     {"right": "manage_rights"}),
    ("POST", "/api/admin/rights", "admin_right_toggle",
     {"right": "manage_rights"}),
]

SERVICE = {"key": "manage_rights", "icon": "🔑", "title": "Редактор прав",
           "right": "manage_rights", "phase": 3,
           "desc": "Права пользователей Telegram и Web"}

PANELS = []


def handle_api(ctx, config, method, req):
    from core.errors import UserError

    if method == "register":
        if not ctx.api.call("web_users.open_registration"):
            raise UserError("регистрация закрыта")
        body = req.get("body") or {}
        user, err = ctx.api.call(
            "web_users.create", (body.get("username") or "").strip(), body.get("password") or "")
        if err:
            return 400, {"ok": False, "error": err}
        try:
            ctx.api.call("audit.record", "?", "auth.register",
                         "web:%s" % user["id"], user["username"])
        except Exception:  # noqa: BLE001
            pass
        return {"ok": True, "user": {"id": user["id"], "username": user["username"]}}

    if method == "login":
        body = req.get("body") or {}
        from . import engine as users

        ip = req.get("remote_ip") or "?"
        if users.is_ratelimited(ip):
            return 429, {"ok": False, "error": "Слишком много попыток, попробуйте позже"}
        user = ctx.api.call(
            "web_users.verify", (body.get("username") or "").strip(), body.get("password") or "")
        if user is None:
            users.note_fail(ip)
            try:
                ctx.api.call("audit.record", "?", "auth.login_fail",
                             (body.get("username") or "").strip()[:60],
                             "ip=%s" % ip, False)
            except Exception:  # noqa: BLE001
                pass
            return 401, {"ok": False, "error": "Неверный логин или пароль"}
        token = users.create_session(user)
        try:
            ttl = int((config.get("WEB_SESSION_TTL_HOURS", "168") if config else "168") or 168)
        except (ValueError, TypeError, AttributeError):
            ttl = 168
        ttl *= 3600
        try:
            ctx.api.call("audit.record", "web:%s" % user["id"], "auth.login",
                         user["username"], "ip=%s" % ip)
        except Exception:  # noqa: BLE001
            pass
        return 200, {"ok": True, "user": {"id": user["id"], "username": user["username"]}}, {
            "set_cookies": [("websess", token, {"max_age": ttl})],
        }

    if method == "logout":
        sess = req.get("session") or {}
        if sess and sess.get("via") != "api" and sess.get("token"):
            ctx.api.call("web_users.drop_session", sess["token"])
        return 200, {"ok": True}, {"clear_cookies": ["websess"]}

    if method == "password":
        sess = req.get("session") or {}
        if not sess or not sess.get("uid"):
            return 401, {"ok": False, "error": "Нужно войти"}
        body = req.get("body") or {}
        try:
            user, err = ctx.api.call(
                "web_users.set_password", sess["uid"],
                body.get("old") or "", body.get("new") or "")
        except Exception as e:  # noqa: BLE001
            return 400, {"ok": False, "error": str(e)}
        if err:
            return 400, {"ok": False, "error": err}
        try:
            ctx.api.call("audit.record", "web:%s" % user["id"],
                         "auth.password", user["username"], "")
        except Exception:  # noqa: BLE001
            pass
        return {"ok": True}

    if method == "admin_user_password":
        import re as _re
        sess = req.get("session") or {}
        body = req.get("body") or {}
        key = str(body.get("key") or "")
        m = _re.fullmatch(r"web:(\d+)", key)
        if not m:
            return 400, {"ok": False, "error": "Некорректный key"}
        uid = int(m.group(1))
        if uid == sess.get("uid"):
            return 400, {"ok": False, "error": "Свой пароль — через форму ниже"}
        try:
            user, err = ctx.api.call(
                "web_users.admin_set_password", uid,
                body.get("new") or "")
        except Exception as e:  # noqa: BLE001
            return 400, {"ok": False, "error": str(e)}
        if err:
            return 400, {"ok": False, "error": err}
        try:
            actor = "web:%s" % sess.get("uid") if sess.get("uid") else "?"
            ctx.api.call("audit.record", actor, "users.password", key, "")
        except Exception:  # noqa: BLE001
            pass
        return {"ok": True, "key": key}

    if method == "me":
        sess = req.get("session")
        if not sess:
            return {"authenticated": False}
        uid = sess["uid"]
        try:
            entry = ctx.rights.get_entry(f"web:{uid}")
        except Exception:  # noqa: BLE001
            entry = {}
        try:
            is_admin = bool(ctx.api.call("transport_web.check", sess, "all"))
        except Exception:  # noqa: BLE001
            is_admin = bool((entry or {}).get("all"))
        out = {
            "authenticated": True,
            "user": {"id": uid, "username": sess.get("username"), "ns": "web"},
            "rights": ctx.api.call("web_users.rights_of", uid),
            "pve_hosts": dict((entry or {}).get("pve_manage_hosts") or {}),
            "is_admin": is_admin,
            "via": sess.get("via") or "web",
        }
        if sess.get("via") == "api":
            out["token"] = {
                "id": sess.get("token_id"),
                "name": sess.get("token_name"),
                "prefix": sess.get("token_prefix"),
            }
            out["scope"] = sess.get("api_scope")  # None = все права пользователя
        try:
            out["telegrams"] = ctx.api.call(
                "web_api_tokens.linked_telegrams", uid)
        except Exception:  # noqa: BLE001
            out["telegrams"] = []
        return out

    if method == "admin_users":
        # Редактор прав: пользователи всех неймспейсов + каталог прав.
        import re as _re
        web_names = {}
        try:
            for u in ctx.api.call("web_users.list"):
                web_names[f"web:{u['id']}"] = u.get("username") or ""
        except Exception:  # noqa: BLE001
            pass
        links_by_uid = {}
        try:
            for lk in ctx.api.call("web_api_tokens.link_list") or []:
                links_by_uid.setdefault(lk.get("uid"), []).append(lk.get("qid"))
        except Exception:  # noqa: BLE001
            pass
        users = []

        def _display(ns, uid):
            if ns == "web":
                return web_names.get(f"web:{uid}") or ""
            if ns in ("telegram", "vk"):
                try:
                    return ctx.api.call("tg_names.resolve", str(uid)) or ""
                except Exception:  # noqa: BLE001
                    return ""
            return ""

        def _add(key):
            try:
                entry = ctx.rights.get_entry(key) or {}
            except Exception:  # noqa: BLE001
                entry = {}
            m = _re.fullmatch(r"(web|telegram|vk):([^:]+)", key or "")
            if not m:
                return
            ns, uid = m.group(1), m.group(2)
            try:
                catalog = ctx.api.call("web_users.right_catalog")
            except Exception:  # noqa: BLE001
                catalog = []
            users.append({
                "key": key,
                "ns": ns,
                "id": uid,
                "name": _display(ns, uid),
                "is_all": bool(entry.get("all", False)),
                "active": sum(1 for k in catalog
                              if k != "all" and bool(entry.get(k, False))),
                "rights": {k: bool(entry.get(k, False)) for k in catalog},
                "pve_hosts": dict(entry.get("pve_manage_hosts") or {}),
                "linked": (links_by_uid.get(int(uid), [])
                           if ns == "web" and str(uid).isdigit() else []),
            })

        try:
            known = ctx.rights.known_users()
        except Exception:  # noqa: BLE001
            known = []
        for key in known:
            if key.startswith(("web:", "vk:", "telegram:")) and \
                    all(u["key"] != key for u in users):
                _add(key)
        users.sort(key=lambda u: ({"telegram": 0, "web": 1, "vk": 2}.get(u["ns"], 3),
                                  int(u["id"]) if u["id"].isdigit() else 0))
        # is_admin: TG-админы из единого списка access; остальным — False.
        # can_manage_admins — по праву manage_admins.
        try:
            admin_ids = {str(x) for x in ctx.api.call("access.admin_ids")}
        except Exception:  # noqa: BLE001
            admin_ids = set()
        try:
            can_manage_admins = bool(
                ctx.api.call("transport_web.check", req.get("session"), "manage_admins"))
        except Exception:  # noqa: BLE001
            can_manage_admins = False
        for u in users:
            u["is_admin"] = u["ns"] == "telegram" and u["id"] in admin_ids
        try:
            catalog = ctx.api.call("web_users.right_catalog")
        except Exception:  # noqa: BLE001
            catalog = []
        try:
            ignore_all = ctx.api.call("web_users.all_ignore")
        except Exception:  # noqa: BLE001
            ignore_all = []
        try:
            vms = ctx.api.call("pve.list_vms")
            pve_vms = [{"vmid": str(v.get("vmid")), "name": v.get("name") or ""}
                       for v in vms or [] if v.get("vmid") is not None]
        except Exception:  # noqa: BLE001
            pve_vms = []
        try:
            rights_meta = ctx.api.call("web_users.right_meta")
        except Exception:  # noqa: BLE001
            rights_meta = {}
        return {
            "catalog": catalog,
            "ignore_all": ignore_all,
            "rights_meta": rights_meta,
            "users": users,
            "pve_vms": pve_vms,
            "admin_ids": sorted(admin_ids),
            "can_manage_admins": can_manage_admins,
        }

    if method == "admin_user_delete":
        import re as _re
        sess = req.get("session") or {}
        body = req.get("body") or {}
        key = str(body.get("key") or "")
        m = _re.fullmatch(r"web:(\d+)", key)
        if not m:
            return 400, {"ok": False, "error": "Некорректный key"}
        uid = int(m.group(1))
        if uid == sess.get("uid"):
            return 400, {"ok": False, "error": "Нельзя удалить самого себя"}
        try:
            ok, err = ctx.api.call("web_users.delete", uid)
        except Exception as e:  # noqa: BLE001
            return 400, {"ok": False, "error": str(e)}
        if not ok:
            return 400, {"ok": False, "error": err}
        try:
            ctx.api.call("web_api_tokens.revoke_all", uid)
        except Exception:  # noqa: BLE001
            pass
        try:
            ctx.api.call("web_api_tokens.unlink_uid", uid)
        except Exception:  # noqa: BLE001
            pass
        try:
            actor = "web:%s" % sess.get("uid") if sess.get("uid") else "?"
            ctx.api.call("audit.record", actor, "users.delete", key, "")
        except Exception:  # noqa: BLE001
            pass
        return {"ok": True, "key": key}

    if method == "admin_right_toggle":
        # Переключение одного права (семантика AdminMenu.toggle_admin_right).
        import re as _re
        body = req.get("body") or {}
        sess = req.get("session") or {}
        actor = "web:%s" % sess.get("uid") if sess.get("uid") else "?"
        key = str(body.get("key") or "")
        right = str(body.get("right") or "")
        if not _re.match(r"^(telegram|vk|web):\d+$", key):
            return 400, {"ok": False, "error": "Некорректный ключ пользователя"}
        try:
            catalog = ctx.api.call("web_users.right_catalog")
        except Exception:  # noqa: BLE001
            catalog = []
        if right not in catalog:
            return 400, {"ok": False, "error": "Неизвестное право"}
        try:
            entry = ctx.rights.get_entry(key) or {}
            if not entry:
                entry = ctx.rights.guest_defaults()
            entry[right] = not entry.get(right, False)
            ctx.rights.set_entry(key, entry, actor=actor)
        except Exception as e:  # noqa: BLE001
            return 500, {"ok": False, "error": str(e)}
        # notify пользователю о смене его права (как в проде).
        try:
            ctx.api.call("notify.rights_changed", key, right,
                         bool(entry[right]))
        except Exception:  # noqa: BLE001
            pass
        return {"ok": True, "key": key, "right": right,
                "value": bool(entry[right])}

    raise UserError("неизвестный метод: %s" % method)