web_api_tokens v0.2.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) ---
# links-файл: {"telegram:ID": {"uid": N, "linked_at": ts}}.
# link-коды — одноразовые, только в памяти (TTL 10 мин).

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


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)."""
    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)."""
    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."""
    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."""
    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:
    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 (при удалении пользователя)."""
    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)

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

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

/link <код> — привязать TG к web-аккаунту (код из веб-кабинета).
/token [имя] — выпустить токен (полное наследование прав, показать один раз).
/tokens — список своих токенов. /token_del <id> — отозвать. /unlink — отвязать.
"""

TG_COMMANDS = {
    "link": {"desc": "привязать Telegram: /link <код из кабинета>",
             "right": "tokens_tg", "menu": False},
    "unlink": {"desc": "отвязать Telegram от web-аккаунта",
               "right": "tokens_tg", "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 <код>.")


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 _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 handle_command(ctx, config, command: str, text: str, inv=None):
    inv = inv or {}
    me = inv.get("user_qid", "?")

    if command == "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
        return ("✅ Telegram привязан к web-аккаунту (uid %s). "
                "Теперь доступны /token, /tokens. Для ИИ в TG: выпустите "
                "/token, затем в личке /apitoken <токен>." % uid)

    if command == "unlink":
        ok = ctx.api.call("web_api_tokens.unlink", me)
        return "✅ Привязка снята." if ok else "Привязки не было."

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

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":
        qid = ctx.api.call("web_api_tokens.linked_telegram", uid)
        return {"ok": True, "telegram": qid}

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

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