web_api_tokens v0.3.1 [base]

# modules/web_api_tokens.py
"""Долгоживущие API-токены веб-панели + одноразовые ticket для WebSocket.

Токен выпускается владельцем (или администратором с правом all — для любого
web-пользователя) и наследует права пользователя: каждая проверка прав идёт
через живой web_users.get_right(), отозванные права сразу перестают работать.

Скоуп: None = все права пользователя (живое наследование, включая будущие),
список = явный allowlist (подмножество прав на момент выпуска; отзыв права
администратором всё равно сразу ограничивает токен).

Долгий токен в query не передаётся никогда: для WebSocket клиент меняет его
на одноразовый ticket (TTL 60 сек, одно использование) через
POST /api/auth/ticket и подключается с ?ticket=...

Хранение: cache/web_api_tokens.json (только sha256 токенов, не сами токены).
Ticket'ы — только в памяти (эфемерные, рестарт их обнуляет — так и задумано).
"""
import hashlib
import hmac
import json
import logging
import os
import secrets
import threading
import time

logger = logging.getLogger(__name__)

_lock = threading.RLock()
_TOKENS_FILE = None  # ставит configure() из Module (data/api_tokens.json)
_configured = False


def configure(tokens_file: str) -> None:
    """Привязка к путям ядра. Ленивая загрузка при первом обращении."""
    global _TOKENS_FILE, _configured
    _TOKENS_FILE = tokens_file
    _configured = True
    _store()

_TOKEN_PREFIX = "bot_"
_TICKET_PREFIX = "tkt_"
_MAX_NAME = 64
MAX_TOKENS_PER_USER = 20
TICKET_TTL = 60
_TOUCH_INTERVAL = 3600  # не чаще раза в час обновляем last_used на диске


def _load_file():
    try:
        with open(_TOKENS_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", _TOKENS_FILE, e)
    return {}


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


_tokens = None  # лениво через _store(); tid -> entry
_t_tokens_mtime = 0  # mtime для межпроцессной свежести (файл правят и снаружи)
_tickets = {}  # ticket -> {"uid","username","scope","expires"}


def _store():
    """Ленивый доступ к токенам (загрузка при первом обращении).

    Перечитывает файл при смене mtime — токены создают/отзывают и из
    других процессов (one-shot, второй инстанс), кеш в памяти иначе врёт.
    """
    global _tokens, _t_tokens_mtime
    try:
        mt = os.path.getmtime(_TOKENS_FILE) if _TOKENS_FILE else -1
    except OSError:
        mt = -1
    if _tokens is None or mt != _t_tokens_mtime:
        _tokens = _load_file()
        _t_tokens_mtime = mt
    return _tokens


def _phash(token: str) -> str:
    return hashlib.sha256(token.encode()).hexdigest()


def _public(entry, tid):
    return {
        "id": tid,
        "uid": entry.get("uid"),
        "username": entry.get("username"),
        "name": entry.get("name"),
        "prefix": entry.get("prefix"),
        "scope": entry.get("scope"),  # None = все права пользователя
        "created": entry.get("created"),
        "last_used": entry.get("last_used"),
        "created_by": entry.get("created_by"),
    }


def available_scope(effective_rights: dict) -> list:
    """Права, которые реально есть у пользователя (кандидаты в скоуп).

    Принимает готовый effective-словарь (считает ui через web_users API),
    чтобы engine не зависел от соседнего модуля.
    """
    try:
        return sorted(k for k, v in (effective_rights or {}).items() if v)
    except Exception:  # noqa: BLE001
        return []


def normalize_scope(requested, available, catalog=()):
    """Проверяет запрошенный скоуп. Возвращает (scope|None, ошибка|None).

    requested=None (поле отсутствует) = полный доступ. Полный набор галочек
    тоже сворачивается в None, чтобы токен наследовал и будущие права.
    catalog — известные ключи прав (из web_users API).
    """
    if requested is None:
        return None, None
    if not isinstance(requested, list):
        return None, "scope должен быть списком прав или отсутствовать"
    avail = set(available or [])
    req = [r for r in requested if isinstance(r, str)]
    if len(req) != len(requested):
        return None, "scope: только строковые ключи прав"
    unknown = [r for r in req if r not in set(catalog or [])]
    if unknown:
        return None, f"scope: неизвестные права: {', '.join(unknown[:8])}"
    denied = [r for r in req if r not in avail]
    if denied:
        return None, f"scope: нет у пользователя: {', '.join(denied[:8])}"
    if not req:
        return None, "scope: выберите хотя бы одно право"
    if set(req) == avail:
        return None, None
    return sorted(set(req)), None


def create_token(uid: int, username: str, name: str, scope, created_by: int):
    """Создаёт токен. Возвращает (token|None, public|None, ошибка|None).

    Сам токен возвращается только здесь (показать один раз), в файле —
    только его sha256.
    """
    name = (name or "api").strip()[:_MAX_NAME] or "api"
    with _lock:
        tokens = _store()
        owned = [t for t, e in tokens.items() if e.get("uid") == uid]
        if len(owned) >= MAX_TOKENS_PER_USER:
            return None, None, f"Лимит токенов пользователя ({MAX_TOKENS_PER_USER})"
        token = _TOKEN_PREFIX + secrets.token_urlsafe(32)
        tid = secrets.token_hex(8)
        now = time.time()
        tokens[tid] = {
            "uid": uid,
            "username": username,
            "name": name,
            "prefix": token[:12],
            "phash": _phash(token),
            "scope": scope,  # None или отсортированный список
            "created": now,
            "last_used": now,
            "created_by": created_by,
        }
        _save_file(tokens)
        logger.info("Выпущен API-токен %s... для web:%s (scope=%s)",
                    token[:12], uid, "full" if scope is None else len(scope))
        return token, _public(tokens[tid], tid), None


def lookup_bearer(token: str):
    """Проверяет Bearer-токен. Возвращает сессию для web.py или None."""
    if not token or not token.startswith(_TOKEN_PREFIX):
        return None
    digest = _phash(token)
    now = time.time()
    with _lock:
        tokens = _store()
        for tid, e in tokens.items():
            if hmac.compare_digest(e.get("phash", ""), digest):
                if now - e.get("last_used", 0) > _TOUCH_INTERVAL:
                    e["last_used"] = now
                    _save_file(tokens)
                return {
                    "uid": e["uid"],
                    "username": e.get("username"),
                    "via": "api",
                    "api_scope": e.get("scope"),
                    "token_id": tid,
                    "token_name": e.get("name"),
                    "token_prefix": e.get("prefix"),
                }
    return None


def list_tokens(uid: int):
    with _lock:
        tokens = _store()
        out = [_public(e, t) for t, e in tokens.items() if e.get("uid") == uid]
    out.sort(key=lambda x: x.get("created") or 0, reverse=True)
    return out


def revoke_token(requester_uid: int, tid: str, is_admin: bool):
    with _lock:
        tokens = _store()
        e = tokens.get(tid)
        if not e:
            return None, "Токен не найден"
        if e.get("uid") != requester_uid and not is_admin:
            return None, "Токен не найден"
        del tokens[tid]
        _save_file(tokens)
        logger.info("Отозван API-токен %s... (web:%s)", e.get("prefix"), e.get("uid"))
        return True, None


def revoke_all_for_uid(uid: int) -> int:
    with _lock:
        tokens = _store()
        doomed = [t for t, e in tokens.items() if e.get("uid") == uid]
        for t in doomed:
            del tokens[t]
        if doomed:
            _save_file(tokens)
            logger.info("Отозваны все API-токены web:%s (%d шт.)", uid, len(doomed))
        return len(doomed)


def _purge_tickets(now=None):
    now = now or time.time()
    expired = [t for t, e in _tickets.items() if e.get("expires", 0) <= now]
    for t in expired:
        del _tickets[t]


def mint_ticket(uid: int, username: str, scope):
    """Одноразовый ticket для WebSocket. Возвращает (ticket, ttl)."""
    with _lock:
        _purge_tickets()
        ticket = _TICKET_PREFIX + secrets.token_urlsafe(24)
        _tickets[ticket] = {
            "uid": uid,
            "username": username,
            "scope": scope,
            "expires": time.time() + TICKET_TTL,
        }
        return ticket, TICKET_TTL


def consume_ticket(ticket: str):
    """Одноразово меняет ticket на сессию. Повтор/просрочка → None."""
    if not ticket or not ticket.startswith(_TICKET_PREFIX):
        return None
    with _lock:
        _purge_tickets()
        e = _tickets.pop(ticket, None)
    if not e:
        return None
    return {
        "uid": e["uid"],
        "username": e.get("username"),
        "via": "api",
        "api_scope": e.get("scope"),
        "token_name": "ticket",
    }


# --- привязка Telegram -> web-аккаунт (для /token в TG и bot_api) ---
# Владелец — access (ядро идентичности): links-файл identity_links.json,
# коды персистентные, два вектора (кабинет->бот и бот->кабинет), union-права,
# волшебный вход. Здесь — тонкие прокси для совместимости: UI ходит сюда,
# engine делегирует в access, если тот настроен (всегда в проде), иначе —
# legacy-стор ниже (только для изолированных unit-тестов).
# links-файл legacy: {"telegram:ID": {"uid": N, "linked_at": ts}}.
# link-коды legacy — одноразовые, только в памяти (TTL 10 мин).

_LINKS_FILE = None
_links = None
_t_links_mtime = 0
_link_codes = {}  # code -> {"uid", "expires"}
LINK_CODE_TTL = 600


def _owner():
    """Движок access, если его хранилище настроено (прод)."""
    try:
        from botmod_access import engine as _acc
        if _acc.links_configured():
            return _acc
    except Exception:  # noqa: BLE001
        pass
    return None


def configure_links(links_file: str) -> None:
    global _LINKS_FILE
    _LINKS_FILE = links_file
    _links_store()


def _links_store():
    global _links, _t_links_mtime
    try:
        mt = os.path.getmtime(_LINKS_FILE) if _LINKS_FILE else -1
    except OSError:
        mt = -1
    if _links is None or mt != _t_links_mtime:
        _links = _links_load()
        _t_links_mtime = mt
    return _links


def _links_load():
    data = {}
    try:
        if _LINKS_FILE and os.path.exists(_LINKS_FILE):
            with open(_LINKS_FILE, "r", encoding="utf-8") as f:
                loaded = json.load(f)
            if isinstance(loaded, dict):
                data = loaded
    except Exception as e:  # noqa: BLE001
        logger.warning("Не удалось прочитать привязки %s: %s",
                       _LINKS_FILE, e)
    return data


def _links_save():
    if not _LINKS_FILE:
        return
    try:
        os.makedirs(os.path.dirname(_LINKS_FILE) or ".", exist_ok=True)
        tmp = _LINKS_FILE + ".tmp"
        with open(tmp, "w", encoding="utf-8") as f:
            json.dump(_links_store(), f, ensure_ascii=False)
        os.replace(tmp, _LINKS_FILE)
    except Exception as e:  # noqa: BLE001
        logger.warning("Не удалось сохранить привязки %s: %s",
                       _LINKS_FILE, e)


def mint_link_code(uid: int):
    """Одноразовый код привязки TG (из веб-кабинета). Возвращает (code, ttl)."""
    owner = _owner()
    if owner is not None:
        return owner.mint_link_code(uid)
    with _lock:
        now = time.time()
        expired = [c for c, e in _link_codes.items()
                   if e.get("expires", 0) <= now]
        for c in expired:
            del _link_codes[c]
        raw = secrets.token_urlsafe(6).replace("-", "").replace("_", "")
        code = (raw + "XXXXXXXX")[:8].upper()
        code = "%s-%s" % (code[:4], code[4:])
        _link_codes[code] = {"uid": int(uid), "expires": now + LINK_CODE_TTL}
        return code, LINK_CODE_TTL


def consume_link_code(code: str, telegram_qid: str):
    """Гасит код из /link. Возвращает (uid|None, ошибка|None)."""
    owner = _owner()
    if owner is not None:
        return owner.consume_link_code(code, telegram_qid)
    code = (code or "").strip().upper()
    qid = (telegram_qid or "").strip()
    if not qid.startswith("telegram:"):
        return None, "только для Telegram"
    with _lock:
        e = _link_codes.pop(code, None)
    if not e:
        return None, "Неверный или просроченный код"
    if e.get("expires", 0) <= time.time():
        return None, "Код просрочен — выпустите новый"
    links = _links_store()
    links[qid] = {"uid": int(e["uid"]), "linked_at": time.time()}
    with _lock:
        _links_save()
    logger.info("Привязан %s -> web:%s", qid, e["uid"])
    return int(e["uid"]), None


def linked_uid(telegram_qid: str):
    """web-uid, привязанный к TG-qid, или None."""
    owner = _owner()
    if owner is not None:
        return owner.linked_uid(telegram_qid)
    try:
        return _links_store().get(telegram_qid, {}).get("uid")
    except Exception:  # noqa: BLE001
        return None


def linked_telegram(uid: int):
    """TG-qid, привязанный к web-uid, или None."""
    owner = _owner()
    if owner is not None:
        return owner.linked_telegram(uid)
    try:
        for qid, e in _links_store().items():
            if isinstance(e, dict) and e.get("uid") == int(uid):
                return qid
    except Exception:  # noqa: BLE001
        pass
    return None


def unlink_telegram(telegram_qid: str) -> bool:
    owner = _owner()
    if owner is not None:
        return owner.unlink_telegram(telegram_qid)
    with _lock:
        if telegram_qid in _links_store():
            del _links_store()[telegram_qid]
            _links_save()
            return True
    return False


def unlink_uid(uid: int) -> int:
    """Снять все привязки web-uid (при удалении пользователя)."""
    owner = _owner()
    if owner is not None:
        return owner.unlink_uid(uid)
    with _lock:
        doomed = [q for q, e in _links_store().items()
                  if isinstance(e, dict) and e.get("uid") == int(uid)]
        for q in doomed:
            del _links_store()[q]
        if doomed:
            _links_save()
        return len(doomed)


# --- новые identity-методы: только через владельца (access) ---
# В изолированных тестах (без access) — безопасные заглушки.

def linked_telegrams(uid: int):
    """Все qid web-uid: [{qid, linked_at, inherit}]."""
    owner = _owner()
    if owner is not None:
        return owner.linked_telegrams(uid)
    first = linked_telegram(uid)
    return ([{"qid": first, "linked_at": 0, "inherit": False}]
            if first else [])


def canonical_owner(qid: str) -> str:
    """Канонический владелец пользовательских данных для qid."""
    owner = _owner()
    if owner is not None:
        return owner.canonical_owner(qid)
    qid = (qid or "").strip()
    uid = linked_uid(qid)
    return ("web:%s" % uid) if uid is not None else qid


def link_tg_code(qid: str):
    """Код tg->web (из /bind). Возвращает (code, ttl) или (None, err)."""
    owner = _owner()
    if owner is not None:
        return owner.mint_tg_code(qid)
    return None, "модуль access недоступен"


def link_tg_consume(code: str, uid: int):
    """Погасить tg-код в кабинете. Возвращает (qid|None, err|None)."""
    owner = _owner()
    if owner is not None:
        return owner.consume_tg_code(code, uid)
    return None, "модуль access недоступен"


def link_set_inherit(qid: str, flag: bool):
    """Согласие на наследование web-прав в TG. Возвращает bool."""
    owner = _owner()
    if owner is not None:
        return owner.set_inherit(qid, flag)
    return False


def link_list():
    """Все привязки: [{qid, uid, linked_at, inherit}]."""
    owner = _owner()
    if owner is not None:
        return owner.link_list()
    return []


def link_admin(uid: int, qid: str, actor: str = ""):
    """Админская связка пары. Возвращает (ok, err)."""
    owner = _owner()
    if owner is not None:
        return owner.link_admin(uid, qid, actor)
    return False, "модуль access недоступен"


def effective_can(qid: str, right: str) -> bool:
    """Право TG-qid с учётом привязки (union)."""
    owner = _owner()
    if owner is not None:
        return owner.effective_can(qid, right)
    return False


def magic_issue(uid: int, qid: str):
    """Ticket волшебного входа. Возвращает (ticket|None, ttl|None, err|None)."""
    owner = _owner()
    if owner is not None:
        return owner.magic_issue(uid, qid)
    return None, None, "модуль access недоступен"


def magic_consume(ticket: str):
    """Погасить ticket. Возвращает (uid, qid)|None."""
    owner = _owner()
    if owner is not None:
        return owner.magic_consume(ticket)
    return None
    with _lock:
        doomed = [q for q, e in _links_store().items()
                  if isinstance(e, dict) and e.get("uid") == int(uid)]
        for q in doomed:
            del _links_store()[q]
        if doomed:
            _links_save()
        return len(doomed)

"""web_api_tokens.Module — Bearer-токены (sha256 в data/) + ticket'ы (память).

Скоуп валидируется через web_users API (каталог прав + effective-права).
"""
import logging
import os

from core.base_module import BaseModule

from . import engine as tokens

logger = logging.getLogger(__name__)


class Module(BaseModule):
    def setup(self, ctx) -> None:
        mod_dir = os.path.dirname(os.path.abspath(__file__))
        self.config = ctx.module_config("web_api_tokens")
        self.ctx = ctx
        tokens.configure(
            (self.config.get("API_TOKENS_FILE", "") or "").strip()
            or os.path.join(ctx.data_dir, "api_tokens.json")
        )
        tokens.configure_links(
            (self.config.get("TG_LINKS_FILE", "") or "").strip()
            or os.path.join(ctx.data_dir, "tg_links.json")
        )
        ctx.api.register("web_api_tokens", "lookup", tokens.lookup_bearer)
        ctx.api.register("web_api_tokens", "create", tokens.create_token)
        ctx.api.register("web_api_tokens", "list", tokens.list_tokens)
        ctx.api.register("web_api_tokens", "revoke", tokens.revoke_token)
        ctx.api.register("web_api_tokens", "revoke_all", tokens.revoke_all_for_uid)
        ctx.api.register("web_api_tokens", "mint_ticket", tokens.mint_ticket)
        ctx.api.register("web_api_tokens", "consume_ticket", tokens.consume_ticket)
        ctx.api.register("web_api_tokens", "normalize_scope", tokens.normalize_scope)
        ctx.api.register("web_api_tokens", "available_scope", tokens.available_scope)
        ctx.api.register("web_api_tokens", "link_code", tokens.mint_link_code)
        ctx.api.register("web_api_tokens", "link_consume", tokens.consume_link_code)
        ctx.api.register("web_api_tokens", "linked_uid", tokens.linked_uid)
        ctx.api.register("web_api_tokens", "linked_telegram", tokens.linked_telegram)
        ctx.api.register("web_api_tokens", "unlink", tokens.unlink_telegram)
        ctx.api.register("web_api_tokens", "unlink_uid", tokens.unlink_uid)
        ctx.api.register(
            "web_api_tokens", "linked_telegrams", tokens.linked_telegrams)
        ctx.api.register(
            "web_api_tokens", "canonical_owner", tokens.canonical_owner)
        ctx.api.register(
            "web_api_tokens", "link_tg_code", tokens.link_tg_code)
        ctx.api.register(
            "web_api_tokens", "link_tg_consume", tokens.link_tg_consume)
        ctx.api.register(
            "web_api_tokens", "link_set_inherit",
            tokens.link_set_inherit)
        ctx.api.register("web_api_tokens", "link_list", tokens.link_list)
        ctx.api.register("web_api_tokens", "link_admin", tokens.link_admin)
        ctx.api.register(
            "web_api_tokens", "effective_can", tokens.effective_can)
        ctx.api.register(
            "web_api_tokens", "magic_issue", tokens.magic_issue)
        ctx.api.register(
            "web_api_tokens", "magic_consume", tokens.magic_consume)

    def health(self) -> dict:
        return {"ok": True, "module": self.name}

"""web_api_tokens.ui_tg — привязка Telegram и токены из бота (нативный режим).

/link <код> — привязать TG к web-аккаунту (код из веб-кабинета; без права —
сам код и есть авторизация; только личка).
/unlink — отвязать (без права).
/bind — выпустить код для ввода в кабинете (обратный вектор; без права,
только личка).
/weblogin — ссылка волшебного входа в кабинет (только привязанные, только
личка, флаг MAGIC_LOGIN в access).
/token [имя] — выпустить токен (полное наследование прав, показать один раз).
/tokens — список своих токенов. /token_del <id> — отозвать. /unlink — отвязать.

Владелец привязок — access (ядро идентичности); здесь только команды.
"""

TG_COMMANDS = {
    "link": {"desc": "привязать Telegram: /link <код из кабинета>",
             "menu": False},
    "unlink": {"desc": "отвязать Telegram от web-аккаунта",
               "menu": False},
    "bind": {"desc": "код привязки для ввода в кабинете: /bind",
             "menu": False},
    "binds": {"desc": "все привязки TG<->web (админ)",
              "right": "manage_rights", "menu": False},
    "weblogin": {"desc": "ссылка входа в веб-кабинет без пароля",
                 "menu": False},
    "token": {"desc": "выпустить API-токен: /token [имя]",
              "right": "tokens_tg"},
    "tokens": {"desc": "мои API-токены", "right": "tokens_tg"},
    "token_del": {"desc": "отозвать токен: /token_del <id>",
                  "right": "tokens_tg", "menu": False},
}

_NO_LINK = ("🔗 Сначала привяжите Telegram: веб-кабинет → API-доступ → "
            "«Привязать Telegram», затем /link <код>. Либо в боте /bind "
            "и введите код в кабинете.")


def _uid_of(ctx, inv):
    try:
        return ctx.api.call("web_api_tokens.linked_uid",
                            (inv or {}).get("user_qid", "?"))
    except Exception:  # noqa: BLE001
        return None


def _my_uid(inv):
    try:
        return int(str((inv or {}).get("user_qid", "?")).split(":")[-1])
    except (ValueError, TypeError):
        return None


def _is_private(inv) -> bool:
    """Личка ли это: в приватном чате chat_id равен id пользователя."""
    try:
        return int((inv or {}).get("chat_id") or 0) == (_my_uid(inv) or -1)
    except (ValueError, TypeError):
        return False


def _web_username(ctx, uid):
    try:
        for u in ctx.api.call("web_users.list") or []:
            if isinstance(u, dict) and u.get("id") == uid:
                return u.get("username") or ("uid-%s" % uid)
    except Exception:  # noqa: BLE001
        pass
    return "uid-%s" % uid


def _audit(ctx, actor, action, target="", detail=""):
    try:
        ctx.api.call("audit.record", actor, action, target, detail)
    except Exception:  # noqa: BLE001
        pass


def _can(ctx, qid, right) -> bool:
    try:
        return bool(ctx.rights.can(qid, right))
    except Exception:  # noqa: BLE001
        pass
    try:
        return bool(ctx.api.call("web_api_tokens.effective_can", qid, right))
    except Exception:  # noqa: BLE001
        return False


def handle_command(ctx, config, command: str, text: str, inv=None):
    inv = inv or {}
    me = inv.get("user_qid", "?")

    if command == "link":
        if not _is_private(inv):
            return "🔒 /link — только в личке с ботом (код светить в группы нельзя)."
        code = (text or "").strip().split()[0] if (text or "").strip() else ""
        if not code:
            return "Использование: /link <код из веб-кабинета>"
        uid, err = ctx.api.call("web_api_tokens.link_consume", code, me)
        if err:
            return "⚠️ %s" % err
        _audit(ctx, me, "identity.link", "web:%s" % uid, "via=bot")
        return ("✅ Telegram привязан к web-аккаунту (uid %s). "
                "Теперь доступны /token, /tokens, /weblogin. Для ИИ в TG: выпустите "
                "/token, затем в личке /apitoken <токен>." % uid)

    if command == "unlink":
        parts = (text or "").strip().split()
        if parts:
            # Админская отвязка чужого qid: /unlink telegram:123.
            if not _can(ctx, me, "manage_rights"):
                return "🚫 Доступ запрещён!"
            target = parts[0]
            ok = ctx.api.call("web_api_tokens.unlink", target)
            if ok:
                _audit(ctx, me, "identity.unlink", target, "via=admin")
                return "✅ Привязка %s снята." % target
            return "Привязки %s не было." % target
        ok = ctx.api.call("web_api_tokens.unlink", me)
        if ok:
            _audit(ctx, me, "identity.unlink", me, "via=bot")
            return "✅ Привязка снята."
        return "Привязки не было."

    if command == "bind":
        if not _is_private(inv):
            return "🔒 /bind — только в личке с ботом (код светить в группы нельзя)."
        parts = (text or "").strip().split()
        if len(parts) == 2:
            # Админская связка: /bind <uid> <telegram:ID|ID>.
            if not _can(ctx, me, "manage_rights"):
                return "🚫 Доступ запрещён!"
            try:
                target_uid = int(parts[0])
            except (ValueError, TypeError):
                return "Использование: /bind <uid> <telegram-id>"
            target_qid = parts[1] if ":" in parts[1] \
                else "telegram:%s" % parts[1]
            ok, err = ctx.api.call(
                "web_api_tokens.link_admin", target_uid, target_qid, me)
            if not ok:
                return "⚠️ %s" % (err or "ошибка")
            _audit(ctx, me, "identity.link_admin", target_qid,
                     "web:%s" % target_uid)
            return ("✅ Связано: %s → web:%s." % (target_qid, target_uid))
        res = ctx.api.call("web_api_tokens.link_tg_code", me)
        code, second = (res + (None, None))[:2]
        if not code:
            return "⚠️ %s" % (second or "не удалось выпустить код")
        ttl = second if isinstance(second, int) else 600
        return ("🔗 Код для кабинета (одноразовый, %s мин.): `%s`\n"
                "Веб-кабинет → API-доступ → «Код из бота» → введите код."
                % (ttl // 60, code))

    if command == "weblogin":
        if not _is_private(inv):
            return "🔒 /weblogin — только в личке с ботом."
        uid = _uid_of(ctx, inv)
        if uid is None:
            return _NO_LINK
        ticket, ttl, err = ctx.api.call(
            "web_api_tokens.magic_issue", uid, me)
        if err:
            return "⚠️ %s" % err
        base = ""
        try:
            base = ((config or {}).get("MAGIC_BASE_URL", "") or "").strip()
        except Exception:  # noqa: BLE001
            base = ""
        if base:
            url = base.rstrip("/") + "/api/auth/magic?ticket=" + ticket
            tail = ("Откройте ссылку (одноразовая, %s мин., никому не пересылайте), "
                    "затем — главную страницу панели:\n%s"
                    % (ttl // 60, url))
        else:
            tail = ("Откройте в браузере (одноразовый пропуск, %s мин., никому "
                    "не пересылайте):\nhttp(s)://<адрес-вашей-панели>"
                    "/api/auth/magic?ticket=`%s`\n"
                    "После открытия ссылки перейдите на главную страницу панели. "
                    "(Администратор может настроить MAGIC_BASE_URL — тогда здесь "
                    "будет готовая ссылка.)"
                    % (ttl // 60, ticket))
        _audit(ctx, me, "identity.magic_issue", "web:%s" % uid, "via=bot")
        return "🔑 " + tail

    if command == "binds":
        try:
            items = ctx.api.call("web_api_tokens.link_list") or []
        except Exception:  # noqa: BLE001
            items = []
        if not items:
            return "Привязок нет."
        lines = []
        for it in items:
            uname = _web_username(ctx, it.get("uid"))
            inh = " +наследование" if it.get("inherit") else ""
            lines.append("%s → web:%s (%s)%s" % (
                it.get("qid"), it.get("uid"), uname, inh))
        return "🔗 Привязки (%d):\n" % len(lines) + "\n".join(lines)

    uid = _uid_of(ctx, inv)
    if uid is None and command in ("token", "tokens", "token_del"):
        return _NO_LINK

    if command == "token":
        name = ((text or "").strip()[:40] or "tg")
        token, public, err = ctx.api.call(
            "web_api_tokens.create", uid, _web_username(ctx, uid),
            name, None, uid)
        if err:
            return "⚠️ %s" % err
        return ("🔑 Токен `%s` (показываю один раз — сохраните, "
                "потом удалите это сообщение):\n`%s`\n\nДля ИИ в TG: "
                "в личке с ботом скажите /apitoken и вставьте этот токен." % (
                    (public or {}).get("name"), token))

    if command == "tokens":
        items = ctx.api.call("web_api_tokens.list", uid) or []
        if not items:
            return "Токенов нет. /token [имя] — выпустить."
        return "🔑 Мои токены:\n" + "\n".join(
            "`%s` %s (%s)" % (t.get("id"), t.get("name") or "",
                              "все права" if t.get("scope") is None
                              else "%d прав" % len(t.get("scope") or []))
            for t in items)

    if command == "token_del":
        tid = (text or "").strip().split()[0] if (text or "").strip() else ""
        if not tid:
            return "Использование: /token_del <id> (id — из /tokens)"
        ok, err = ctx.api.call("web_api_tokens.revoke", uid, tid, False)
        return "✅ Отозван." if ok else "⚠️ %s" % (err or "ошибка")
    return None

"""web_api_tokens.ui_web — выпуск/список/отзыв токенов, ticket для WS.

Выпуск себе — любая сессия; за другого (for_user) — только с правом all.
"""

ROUTES = [
    ("POST", "/api/auth/token", "token", {"right": None}),
    ("GET", "/api/auth/tokens", "tokens", {"right": None}),
    ("POST", "/api/auth/token/revoke", "revoke", {"right": None}),
    ("POST", "/api/auth/ticket", "ticket", {"right": None}),
    ("POST", "/api/auth/link_code", "link_code", {"right": None}),
    ("GET", "/api/auth/link", "link", {"right": None}),
    ("POST", "/api/auth/unlink", "unlink", {"right": None}),
    ("POST", "/api/auth/link_tg_consume", "link_tg_consume",
     {"right": None}),
    ("POST", "/api/auth/link_inherit", "link_inherit", {"right": None}),
    ("GET", "/api/auth/magic", "magic", {"right": None}),
]

PANELS = []


def _who(req):
    sess = req.get("session")
    if not sess:
        return None
    return sess["uid"], sess.get("username", "")


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

    who = _who(req)
    if who is None:
        return 401, {"ok": False, "error": "auth"}
    uid, username = who
    qid = "web:%s" % uid

    if method == "tokens":
        return {"ok": True, "tokens": ctx.api.call("web_api_tokens.list", uid)}

    if method == "token":
        body = req.get("body") or {}
        target_uid = body.get("for_user", uid)
        try:
            target_uid = int(target_uid)
        except (ValueError, TypeError):
            raise UserError("for_user: числовой id")
        if target_uid != uid and not ctx.rights.can(qid, "all"):
            return 403, {"ok": False, "error": "forbidden"}
        catalog = ctx.api.call("web_users.right_catalog")
        eff = ctx.api.call("web_users.rights_of", target_uid)
        available = ctx.api.call("web_api_tokens.available_scope", eff)
        scope, err = ctx.api.call(
            "web_api_tokens.normalize_scope", body.get("scope"), available, catalog)
        if err:
            return 400, {"ok": False, "error": err}
        tname = username
        if target_uid != uid:
            tname = "uid-%s" % target_uid
        token, public, err = ctx.api.call(
            "web_api_tokens.create", target_uid, tname,
            body.get("name"), scope, uid)
        if err:
            return 400, {"ok": False, "error": err}
        public = dict(public, token=token)
        return {"ok": True, "token": public}

    if method == "revoke":
        body = req.get("body") or {}
        ok, err = ctx.api.call(
            "web_api_tokens.revoke", uid, body.get("id"),
            ctx.rights.can(qid, "all"))
        if err:
            return 404 if err == "Токен не найден" else 400, {"ok": False, "error": err}
        return {"ok": True}

    if method == "ticket":
        sess = req.get("session") or {}
        ticket, ttl = ctx.api.call(
            "web_api_tokens.mint_ticket", uid, username, sess.get("api_scope"))
        return {"ok": True, "ticket": ticket, "expires_in": ttl}

    if method == "link_code":
        code, ttl = ctx.api.call("web_api_tokens.link_code", uid)
        return {"ok": True, "code": code, "expires_in": ttl}

    if method == "link":
        items = ctx.api.call("web_api_tokens.linked_telegrams", uid)
        first = items[0]["qid"] if items else None
        return {"ok": True, "telegram": first, "telegrams": items}

    if method == "unlink":
        body = req.get("body") or {}
        qid = (body.get("qid") or "").strip()
        if qid:
            # Отвязка конкретного qid — только своего.
            mine = {t["qid"] for t in
                    ctx.api.call("web_api_tokens.linked_telegrams", uid)}
            if qid not in mine:
                return 404, {"ok": False, "error": "Привязка не найдена"}
            ok = ctx.api.call("web_api_tokens.unlink", qid)
            action, target = "identity.unlink", qid
        else:
            n = ctx.api.call("web_api_tokens.unlink_uid", uid)
            ok, action, target = bool(n), "identity.unlink", "web:%s" % uid
        if ok:
            try:
                ctx.api.call("audit.record", "web:%s" % uid, action,
                             target, "via=web")
            except Exception:  # noqa: BLE001
                pass
        return {"ok": True}

    if method == "link_tg_consume":
        body = req.get("body") or {}
        code = (body.get("code") or "").strip()
        if not code:
            return 400, {"ok": False, "error": "Нужен код из бота (/bind)"}
        qid, err = ctx.api.call(
            "web_api_tokens.link_tg_consume", code, uid)
        if err:
            return 400, {"ok": False, "error": err}
        try:
            ctx.api.call("audit.record", "web:%s" % uid, "identity.link",
                         qid, "via=web,tgcode")
        except Exception:  # noqa: BLE001
            pass
        return {"ok": True, "telegram": qid}

    if method == "link_inherit":
        body = req.get("body") or {}
        qid = (body.get("qid") or "").strip()
        flag = bool(body.get("inherit", True))
        mine = {t["qid"] for t in
                ctx.api.call("web_api_tokens.linked_telegrams", uid)}
        if not qid or qid not in mine:
            return 404, {"ok": False, "error": "Привязка не найдена"}
        ok = ctx.api.call("web_api_tokens.link_set_inherit", qid, flag)
        if not ok:
            return 400, {"ok": False, "error": "Не удалось сохранить"}
        try:
            ctx.api.call("audit.record", "web:%s" % uid,
                         "identity.inherit", qid, "inherit=%s" % flag)
        except Exception:  # noqa: BLE001
            pass
        return {"ok": True, "qid": qid, "inherit": flag}

    if method == "magic":
        query = req.get("query") or {}
        body = req.get("body") or {}
        ticket = (query.get("ticket") or body.get("ticket") or "").strip()
        if not ticket:
            return 400, {"ok": False, "error": "Нужен ticket из /weblogin"}
        pair = ctx.api.call("web_api_tokens.magic_consume", ticket)
        if not pair:
            return 401, {"ok": False,
                         "error": "Неверная или просроченная ссылка — "
                                  "выпустите новую (/weblogin)"}
        m_uid, m_qid = pair
        username = "uid-%s" % m_uid
        try:
            for u in ctx.api.call("web_users.list") or []:
                if isinstance(u, dict) and u.get("id") == m_uid:
                    username = u.get("username") or username
                    break
        except Exception:  # noqa: BLE001
            pass
        try:
            token = ctx.api.call(
                "web_users.create_session",
                {"id": m_uid, "username": username})
        except Exception:  # noqa: BLE001
            return 500, {"ok": False, "error": "Не удалось создать сессию"}
        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", m_qid, "auth.magic_login",
                         "web:%s" % m_uid, "")
        except Exception:  # noqa: BLE001
            pass
        # Уведомление в привязанную личку — угон TG будет виден владельцу.
        try:
            _send = ctx.api.get("transport_telegram.send") \
                if ctx.api.has("transport_telegram.send") else None
        except Exception:  # noqa: BLE001
            _send = None
        if callable(_send):
            try:
                import re as _re
                _m = _re.fullmatch(r"telegram:(-?\d+)", m_qid or "")
                if _m:
                    _send(int(_m.group(1)),
                          "🔑 В веб-кабинет вошли по волшебной ссылке "
                          "(uid %s). Если это были не вы — смените пароль "
                          "и выполните /unlink." % m_uid)
            except Exception:  # noqa: BLE001
                pass
        return 200, {"ok": True,
                     "user": {"id": m_uid, "username": username},
                     "via": "magic"}, {
            "set_cookies": [("websess", token, {"max_age": ttl})],
        }

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