access v0.2.2 [core]

"""access.engine — единый владелец ADMIN_IDS и операций над правами.

Единственное место, где живёт список TG-администраторов: все модули
(admin-движок, сканер сети, notify, веб-панель) обращаются сюда через
ctx.api, а не к своим копиям ADMIN_IDS. Файл прав принадлежит ядру
(core.rights): чтение — живое (без TTL-кешей), запись — диффом через
сервис (атомарно, с mtime-перечитыванием у остальных).
"""
import hashlib
import json
import logging
import os
import re
import secrets
import time

logger = logging.getLogger(__name__)

_rights = None  # RightsService (ставит module.setup)
_admin_ids = []  # [int] TG-администраторы
_env_path = None  # modules/access/.env (для add/remove через веб/TG)

_KEY_RE = re.compile(r"^(telegram|vk|web):\d+$")


def configure(rights, admin_ids, env_path=None) -> None:
    global _rights, _admin_ids, _env_path
    _rights = rights
    _admin_ids = [int(x) for x in (admin_ids or [])]
    _env_path = env_path


def _norm(uid) -> str:
    uid = str(uid)
    if ":" in uid:
        return uid
    return f"telegram:{uid}"


def admin_ids():
    """Список TG id администраторов (копия)."""
    return list(_admin_ids)


def is_admin(uid) -> bool:
    """Входит ли uid в ADMIN_IDS (семантика старого AdminMenu.is_admin)."""
    try:
        num = int(str(uid).split(":")[-1])
    except (ValueError, TypeError):
        return False
    return num in _admin_ids


def load_rights() -> dict:
    """Все записи прав {qid: entry} (живьём из ядра)."""
    try:
        return {qid: _rights.get_entry(qid) for qid in _rights.known_users()}
    except Exception:  # noqa: BLE001
        return {}


def guest_defaults() -> dict:
    """Гостевые дефолты новой записи (динамика из реестра ядра)."""
    return _defaults()


def get_entry(key) -> dict:
    try:
        return _rights.get_entry(key) or {}
    except Exception:  # noqa: BLE001
        return {}


def get_right(uid, key) -> bool:
    """Право пользователя (telegram-ns по умолчанию; живьём).

    С учётом привязки (3a, union): привязанный qid с согласием inherit
    видит и web-права своего uid. Режим off — как раньше.
    """
    try:
        return bool(effective_can(_norm(uid), key))
    except Exception:  # noqa: BLE001
        return False


def _defaults() -> dict:
    """Гостевые дефолты из реестра прав ядра (динамика вместо статики)."""
    try:
        if _rights is not None:
            return _rights.guest_defaults()
    except Exception:  # noqa: BLE001
        pass
    return {"all": False}


def toggle_right(key, right) -> bool:
    """Переключить право; вернуть новое значение (семантика toggle_admin_right)."""
    if not _KEY_RE.match(key or ""):
        raise ValueError("Некорректный ключ пользователя")
    entry = get_entry(key)
    if not entry:
        entry = _defaults()
    entry[right] = not entry.get(right, False)
    _rights.set_entry(key, entry, actor="system:legacy")
    return bool(entry[right])


def save_rights(rights: dict) -> None:
    """Полная запись словаря прав (для вендорного движка): дифф через сервис."""
    current = load_rights()
    for key in list(current):
        if key not in rights:
            try:
                _rights.delete_entry(key, actor="system:seed")
            except Exception:  # noqa: BLE001
                pass
    for key, entry in (rights or {}).items():
        try:
            _rights.set_entry(key, dict(entry), actor="system:seed")
        except Exception as e:  # noqa: BLE001
            logger.warning("access save %s: %s", key, e)


def notify_targets(right):
    """[(chat_id, no_sound)] telegram-получателей с правом right."""
    out = []
    for qid, entry in load_rights().items():
        try:
            kind, _, num = (qid or "").partition(":")
            if kind != "telegram" or not num.lstrip("-").isdigit():
                continue
            if not (entry or {}).get(right, False):
                continue
            out.append((int(num),
                        bool((entry or {}).get("receive_notifications_no_sound", False))))
        except Exception:  # noqa: BLE001
            continue
    return out


def seed() -> None:
    """Синхронизировать права со списком ADMIN_IDS (бывший check_and_update_*).

    - каждому админу из списка — запись с недостающими ключами из
      гостевых дефолтов ядра (динамика);
    - telegram:* вне списка — удалить (web:/vk: не трогаем);
    - pve_manage_hosts дотягивает pve-модуль (свои хосты знает только он).
    """
    rights = load_rights()
    modified = False
    for admin in _admin_ids:
        key = f"telegram:{admin}"
        entry = rights.get(key)
        if entry is None:
            rights[key] = _defaults()
            modified = True
        else:
            for k, default_value in _defaults().items():
                if k not in entry:
                    entry[k] = default_value
                    modified = True
    valid = {f"telegram:{a}" for a in _admin_ids}
    try:
        linked = set(_links_store()) if _LINKS_FILE else set()
    except Exception:  # noqa: BLE001
        linked = set()
    for qid in list(rights):
        if qid.startswith("telegram:") and qid not in valid \
                and qid not in linked:
            del rights[qid]
            modified = True
    if modified:
        save_rights(rights)
        logger.info("access: права синхронизированы с ADMIN_IDS.")


def _write_env(ids) -> None:
    """Перезаписать строку ADMIN_IDS в .env модуля (сохраняя остальное)."""
    import os as _os
    if not _env_path:
        raise RuntimeError("access .env недоступен")
    with open(_env_path, "r", encoding="utf-8") as f:
        lines = f.read().splitlines()
    new_lines, done = [], False
    for ln in lines:
        if re.match(r"^\s*ADMIN_IDS\s*=", ln):
            new_lines.append("ADMIN_IDS=" + ",".join(str(x) for x in ids))
            done = True
        else:
            new_lines.append(ln)
    if not done:
        new_lines.append("ADMIN_IDS=" + ",".join(str(x) for x in ids))
    tmp = _env_path + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        f.write("\n".join(new_lines) + "\n")
    os.replace(tmp, _env_path)


def add_admin_id(uid) -> tuple:
    """Добавить TG-админа: .env + seed. Вернуть (changed, msg)."""
    global _admin_ids
    try:
        num = int(str(uid).strip())
    except (ValueError, TypeError):
        raise ValueError("ID должен быть числом")
    if num in _admin_ids:
        return False, f"Администратор {num} уже в списке"
    _admin_ids.append(num)
    _write_env(_admin_ids)
    seed()
    return True, (f"Администратор {num} добавлен (вступит после перезапуска; "
                   "права выданы сразу)")


def remove_admin_id(uid) -> tuple:
    """Убрать TG-админа: .env + права. Вернуть (cfg_changed, rights_changed)."""
    global _admin_ids
    try:
        num = int(str(uid).strip())
    except (ValueError, TypeError):
        raise ValueError("ID должен быть числом")
    cfg_changed = False
    if num in _admin_ids:
        _admin_ids = [x for x in _admin_ids if x != num]
        _write_env(_admin_ids)
        cfg_changed = True
    rights_changed = False
    try:
        if _rights.delete_entry(f"telegram:{num}", actor="system:seed"):
            rights_changed = True
    except Exception:  # noqa: BLE001
        pass
    if not cfg_changed and not rights_changed:
        raise LookupError(f"Администратор {num} не найден")
    return cfg_changed, rights_changed


# --- привязка мессенджер <-> web-аккаунт -----------------------------------
# Владелец links здесь (ядро идентичности), а не в web_api_tokens: токены —
# лишь потребитель привязки. Формат links-файла:
#   {"telegram:ID": {"uid": N, "linked_at": ts, "inherit": bool}}
# Кардинальность: один TG-qid -> один uid; один uid -> много qid.
# Коды одноразовые, персистентные (codes-файл, TTL), два вектора:
#   web->tg: кабинет mint_link_code -> /link <код> в боте;
#   tg->web: /bind в боте (mint_tg_code) -> ввод кода в кабинете.
# "inherit" — согласие владельца на наследование web-прав в TG (3a, union);
# без него привязка работает в старом режиме (только токены).

_LINKS_FILE = None
_CODES_FILE = None
_links = None
_t_links_mtime = 0
_codes = None
_t_codes_mtime = 0
_UNIFIED_MODE = "off"  # off | union (3a)
_MAGIC_ENABLED = False  # волшебный вход /weblogin (3b)

_LINK_QID_RE = re.compile(r"^(telegram|vk):\d+$")

LINK_CODE_TTL = 600
_LINK_FAIL_WINDOW = 300
_LINK_MAX_FAILS = 10
_link_fails: dict = {}  # key("tg:<qid>"|"web:<uid>") -> [ts]

MAGIC_TTL = 300
_MAGIC_MIN_INTERVAL = 60  # не чаще 1 выпуска в минуту на uid
_MAGIC_MAX_ACTIVE = 3  # активных ticket'ов на uid
_magic: dict = {}  # phash -> {"uid", "qid", "expires"}
_magic_last_issue: dict = {}  # uid -> ts


def configure_links(links_file: str, codes_file: str = "") -> None:
    """Файлы привязок и кодов. Вызывает Module.setup."""
    global _LINKS_FILE, _CODES_FILE
    _LINKS_FILE = links_file or None
    _CODES_FILE = codes_file or None
    _links_store()
    _codes_store()


def configure_identity(mode: str = "off", magic_enabled: bool = False) -> None:
    """Режим наследования прав (off|union) и флаг волшебного входа."""
    global _UNIFIED_MODE, _MAGIC_ENABLED
    _UNIFIED_MODE = "union" if str(mode or "").strip().lower() == "union" else "off"
    _MAGIC_ENABLED = bool(magic_enabled)


def links_configured() -> bool:
    """Настроено ли хранилище (для делегирования из web_api_tokens)."""
    return bool(_LINKS_FILE)


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 migrate_from_legacy(path: str) -> int:
    """Разовый переезд из tg_links.json web_api_tokens (inherit=False).

    Копирует только если целевое хранилище пусто, а наследие непусто.
    Возвращает число перенесённых записей.
    """
    if not _LINKS_FILE or not path:
        return 0
    try:
        if _links_store():
            return 0
        if not os.path.exists(path):
            return 0
        with open(path, "r", encoding="utf-8") as f:
            legacy = json.load(f)
        if not isinstance(legacy, dict) or not legacy:
            return 0
        n = 0
        for qid, e in legacy.items():
            if not _LINK_QID_RE.match(str(qid or "")):
                continue
            try:
                uid = int((e or {}).get("uid"))
            except (ValueError, TypeError):
                continue
            _links_store()[str(qid)] = {
                "uid": uid,
                "linked_at": float((e or {}).get("linked_at") or 0) or time.time(),
                "inherit": False,
            }
            n += 1
        if n:
            _links_save()
            logger.info("identity: мигрировано %d привязок из %s", n, path)
        return n
    except Exception as e:  # noqa: BLE001
        logger.warning("identity migrate %s: %s", path, e)
        return 0


def _codes_store():
    global _codes, _t_codes_mtime
    try:
        mt = os.path.getmtime(_CODES_FILE) if _CODES_FILE else -1
    except OSError:
        mt = -1
    if _codes is None or mt != _t_codes_mtime:
        _codes = _codes_load()
        _t_codes_mtime = mt
    return _codes


def _codes_load():
    data = {}
    try:
        if _CODES_FILE and os.path.exists(_CODES_FILE):
            with open(_CODES_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", _CODES_FILE, e)
    return data


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


def _mint_code() -> str:
    raw = secrets.token_urlsafe(6).replace("-", "").replace("_", "")
    code = (raw + "XXXXXXXX")[:8].upper()
    return "%s-%s" % (code[:4], code[4:])


def _prune_codes(now=None):
    now = now or time.time()
    codes = _codes_store()
    dead = [c for c, e in codes.items()
            if not isinstance(e, dict) or e.get("expires", 0) <= now]
    for c in dead:
        del codes[c]
    if dead:
        _codes_save()


def mint_link_code(uid: int):
    """Одноразовый код web->tg (из кабинета). Возвращает (code, ttl)."""
    code = _mint_code()
    codes = _codes_store()
    codes[code] = {"uid": int(uid), "qid": None,
                   "expires": time.time() + LINK_CODE_TTL}
    _codes_save()
    return code, LINK_CODE_TTL


def mint_tg_code(qid: str):
    """Одноразовый код tg->web (из /bind). Возвращает (code, ttl)."""
    qid = (qid or "").strip()
    if not _LINK_QID_RE.match(qid):
        return None, "только Telegram/VK"
    code = _mint_code()
    codes = _codes_store()
    codes[code] = {"uid": None, "qid": qid,
                   "expires": time.time() + LINK_CODE_TTL}
    _codes_save()
    return code, LINK_CODE_TTL


def _fail_key_consume(qid: str) -> str:
    return "tg:%s" % qid


def _note_fail(key: str) -> None:
    now = time.time()
    lst = [t for t in _link_fails.get(key, []) if now - t < _LINK_FAIL_WINDOW]
    lst.append(now)
    _link_fails[key] = lst


def _is_ratelimited(key: str) -> bool:
    now = time.time()
    lst = [t for t in _link_fails.get(key, []) if now - t < _LINK_FAIL_WINDOW]
    _link_fails[key] = lst
    return len(lst) >= _LINK_MAX_FAILS


def consume_link_code(code: str, qid: str):
    """Гасит web-код из /link. Возвращает (uid|None, ошибка|None).

    Один qid — один uid: повторная привязка чужого/занятого только
    через явный /unlink.
    """
    code = (code or "").strip().upper()
    qid = (qid or "").strip()
    if not _LINK_QID_RE.match(qid):
        return None, "только для Telegram/VK"
    if _is_ratelimited(_fail_key_consume(qid)):
        return None, "Слишком много попыток, попробуйте позже"
    _prune_codes()
    codes = _codes_store()
    e = codes.pop(code, None)
    if not e or e.get("qid") is not None:
        _note_fail(_fail_key_consume(qid))
        _codes_save()
        return None, "Неверный или просроченный код"
    if e.get("expires", 0) <= time.time():
        _note_fail(_fail_key_consume(qid))
        _codes_save()
        return None, "Код просрочен — выпустите новый"
    links = _links_store()
    cur = links.get(qid)
    if isinstance(cur, dict) and int(cur.get("uid") or 0) != int(e["uid"]):
        codes[code] = e  # код не гасим — чужая привязка мешает
        _codes_save()
        return None, ("Этот Telegram уже привязан к uid %s — "
                      "сначала /unlink" % cur.get("uid"))
    links[qid] = {"uid": int(e["uid"]), "linked_at": time.time(),
                  "inherit": False}
    _codes_save()
    _links_save()
    logger.info("Привязан %s -> web:%s", qid, e["uid"])
    return int(e["uid"]), None


def consume_tg_code(code: str, uid: int):
    """Гасит tg-код в кабинете. Возвращает (qid|None, ошибка|None)."""
    code = (code or "").strip().upper()
    try:
        uid = int(uid)
    except (ValueError, TypeError):
        return None, "Некорректный пользователь"
    if _is_ratelimited("web:%s" % uid):
        return None, "Слишком много попыток, попробуйте позже"
    _prune_codes()
    codes = _codes_store()
    e = codes.pop(code, None)
    if not e or e.get("uid") is not None:
        _note_fail("web:%s" % uid)
        _codes_save()
        return None, "Неверный или просроченный код"
    if e.get("expires", 0) <= time.time():
        _note_fail("web:%s" % uid)
        _codes_save()
        return None, "Код просрочен — выпустите новый в боте (/bind)"
    qid = e.get("qid") or ""
    links = _links_store()
    cur = links.get(qid)
    if isinstance(cur, dict) and int(cur.get("uid") or 0) != uid:
        codes[code] = e
        _codes_save()
        return None, ("Этот Telegram уже привязан к uid %s — "
                      "сначала отвяжите его" % cur.get("uid"))
    links[qid] = {"uid": uid, "linked_at": time.time(), "inherit": False}
    _codes_save()
    _links_save()
    logger.info("Привязан %s -> web:%s (код из бота)", qid, uid)
    return qid, None


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


def linked_telegrams(uid: int):
    """Все qid, привязанные к web-uid: [{qid, linked_at, inherit}]."""
    out = []
    try:
        for qid, e in _links_store().items():
            if isinstance(e, dict) and int(e.get("uid") or -1) == int(uid):
                out.append({"qid": qid,
                            "linked_at": e.get("linked_at") or 0,
                            "inherit": bool(e.get("inherit", False))})
    except Exception:  # noqa: BLE001
        pass
    out.sort(key=lambda x: x.get("linked_at") or 0)
    return out


def linked_telegram(uid: int):
    """Первый qid web-uid (legacy, 1:1). Множественные — linked_telegrams."""
    items = linked_telegrams(uid)
    return items[0]["qid"] if items else None


def canonical_owner(qid: str) -> str:
    """Канонический владелец пользовательских данных для qid.

    Привязан -> "web:<uid>" (общий раздел связки), иначе — сам qid
    (раздельные данные). Единственное место правила: модули с
    пользовательскими разделами (web/TG-интерфейсы) считают владельца
    только здесь, а не ветвлением у себя.
    """
    qid = (qid or "").strip()
    uid = linked_uid(qid)
    if uid is None:
        return qid
    return "web:%s" % uid


def link_list():
    """Все привязки: [{qid, uid, linked_at, inherit}] (для админки)."""
    out = []
    try:
        for qid, e in _links_store().items():
            if isinstance(e, dict) and e.get("uid") is not None:
                try:
                    uid = int(e.get("uid"))
                except (ValueError, TypeError):
                    continue
                out.append({"qid": qid, "uid": uid,
                            "linked_at": e.get("linked_at") or 0,
                            "inherit": bool(e.get("inherit", False))})
    except Exception:  # noqa: BLE001
        pass
    out.sort(key=lambda x: (x["uid"], x["qid"]))
    return out


def link_admin(uid: int, qid: str, actor: str = ""):
    """Админская связка пары (минуя коды). Возвращает (ok, ошибка|None).

    Перезапись чужой привязки запрещена: сначала unlink.
    """
    try:
        uid = int(uid)
    except (ValueError, TypeError):
        return False, "Некорректный uid"
    qid = (qid or "").strip()
    if not _LINK_QID_RE.match(qid):
        return False, "Некорректный qid (telegram|vk:<число>)"
    links = _links_store()
    cur = links.get(qid)
    if isinstance(cur, dict):
        try:
            cur_uid = int(cur.get("uid"))
        except (ValueError, TypeError):
            cur_uid = None
        if cur_uid != uid:
            return False, ("Уже привязан к uid %s — сначала отвяжите"
                           % cur.get("uid"))
        return True, None
    links[qid] = {"uid": uid, "linked_at": time.time(), "inherit": False}
    _links_save()
    logger.info("Админская привязка %s -> web:%s (actor=%s)", qid, uid,
                actor or "?")
    return True, None


def set_inherit(qid: str, flag: bool):
    """Согласие на наследование web-прав в TG (3a). Возвращает bool."""
    qid = (qid or "").strip()
    links = _links_store()
    e = links.get(qid)
    if not isinstance(e, dict):
        return False
    e["inherit"] = bool(flag)
    _links_save()
    logger.info("Наследование %s: %s", qid, bool(flag))
    return True


def unlink_telegram(qid: str) -> bool:
    links = _links_store()
    if (qid or "").strip() in links:
        del links[(qid or "").strip()]
        _links_save()
        logger.info("Отвязан %s", qid)
        return True
    return False


def unlink_uid(uid: int) -> int:
    """Снять все привязки web-uid (при удалении пользователя)."""
    doomed = []
    try:
        for q, e in _links_store().items():
            if isinstance(e, dict):
                try:
                    match = int(e.get("uid")) == int(uid)
                except (ValueError, TypeError):
                    match = False
                if match:
                    doomed.append(q)
    except Exception:  # noqa: BLE001
        pass
    for q in doomed:
        try:
            del _links_store()[q]
        except KeyError:
            pass
    if doomed:
        _links_save()
        logger.info("Сняты привязки web:%s (%d шт.)", uid, len(doomed))
    return len(doomed)


def effective_can(qid: str, right: str) -> bool:
    """Проверка права TG-qid с учётом привязки (3a, режим union).

    off — только своя запись. union — своя запись OR web-запись
    привязанного uid при согласии inherit. all не наследуется никогда:
    web all=true наследование не открывает (только явная TG-запись).
    """
    base = False
    try:
        if _rights is not None:
            base = bool(_rights.can(qid, right))
    except Exception:  # noqa: BLE001
        base = False
    if base or _UNIFIED_MODE != "union" or not right:
        return base
    if str(right) == "all":
        return base
    uid = linked_uid(qid)
    if uid is None:
        return base
    try:
        e = _links_store().get(qid) or {}
        if not e.get("inherit", False):
            return base
        if _rights is None:
            return base
        web_entry = _rights.get_entry("web:%s" % uid) or {}
        if web_entry.get("all"):
            return base  # all через наследование не передаётся
        return bool(_rights.can("web:%s" % uid, right))
    except Exception:  # noqa: BLE001
        return base


# --- волшебный вход (3b) ----------------------------------------------------
# Ticket из привязанной лички ТГ -> одноразовая web-сессия. Храним только
# sha256 ticket'а; TTL 5 мин; 1 выпуск в минуту на uid; макс. 3 активных.

def _magic_phash(ticket: str) -> str:
    return hashlib.sha256(ticket.encode()).hexdigest()


def _magic_purge(now=None):
    now = now or time.time()
    dead = [h for h, e in _magic.items() if e.get("expires", 0) <= now]
    for h in dead:
        del _magic[h]


def magic_issue(uid: int, qid: str):
    """Выпустить ticket волшебного входа. Возвращает (ticket|None, ttl|None, ошибка|None)."""
    if not _MAGIC_ENABLED:
        return None, None, "Волшебный вход отключён администратором"
    try:
        uid = int(uid)
    except (ValueError, TypeError):
        return None, None, "Некорректный пользователь"
    qid = (qid or "").strip()
    if linked_uid(qid) != uid:
        return None, None, "Telegram не привязан к этому аккаунту"
    now = time.time()
    _magic_purge(now)
    last = _magic_last_issue.get(uid, 0)
    if now - last < _MAGIC_MIN_INTERVAL:
        wait = int(_MAGIC_MIN_INTERVAL - (now - last)) + 1
        return None, None, "Подождите %s сек. перед новым выпуском" % wait
    active = [h for h, e in _magic.items() if e.get("uid") == uid]
    if len(active) >= _MAGIC_MAX_ACTIVE:
        return None, None, "Слишком много активных ссылок — дождитесь сгорания"
    ticket = "tkt_%s" % secrets.token_urlsafe(24)
    _magic[_magic_phash(ticket)] = {"uid": uid, "qid": qid,
                                    "expires": now + MAGIC_TTL}
    _magic_last_issue[uid] = now
    logger.info("magic ticket выпущен web:%s (%s)", uid, qid)
    return ticket, MAGIC_TTL, None


def magic_consume(ticket: str):
    """Погасить ticket (одноразово, fail closed). Возвращает (uid, qid)|None."""
    if not ticket or not str(ticket).startswith("tkt_"):
        return None
    _magic_purge()
    e = _magic.pop(_magic_phash(str(ticket)), None)
    if not e:
        return None
    try:
        return int(e["uid"]), str(e.get("qid") or "")
    except (ValueError, TypeError):
        return None

"""access.Module — единый владелец ADMIN_IDS и операций над правами.

Файл прав принадлежит ядру; этот модуль — API-слой: admin_ids/is_admin,
get_right/toggle_right, notify_targets, seed(). Сидинг — в start():
реестр прав полон только после всех setup'ов.
"""
import logging
import os

from core.base_module import BaseModule

from . import engine as access

logger = logging.getLogger(__name__)

_DEFAULT_ADMIN_IDS = ""  # пусто: админы только из access.env (дефолтов нет)


class Module(BaseModule):
    def setup(self, ctx) -> None:
        mod_dir = os.path.dirname(os.path.abspath(__file__))
        self.config = ctx.module_config("access", mod_dir)
        self.ctx = ctx
        raw = (self.config.get("ADMIN_IDS", "") or "").strip() or _DEFAULT_ADMIN_IDS
        ids = [int(x) for x in raw.split(",") if x.strip().isdigit()]
        try:
            env_path = os.path.join(ctx.data_dir, "env", "access.env")
        except Exception:  # noqa: BLE001
            env_path = os.path.join(mod_dir, ".env")
        access.configure(ctx.rights, ids, env_path)
        from core.paths import abs_data as _abs
        links_file = (self.config.get("IDENTITY_LINKS_FILE", "") or "").strip() \
            or _abs(ctx.data_dir, "identity_links.json", "identity_links.json")
        codes_file = (self.config.get("IDENTITY_CODES_FILE", "") or "").strip() \
            or _abs(ctx.data_dir, "identity_codes.json", "identity_codes.json")
        access.configure_links(links_file, codes_file)
        access.configure_identity(
            self.config.get("IDENTITY_UNIFIED_RIGHTS", "off") or "off",
            str(self.config.get("MAGIC_LOGIN", "0") or "0").strip() == "1")
        # Разовый переезд из tg_links.json web_api_tokens (пусто + наследие).
        try:
            legacy = _abs(ctx.data_dir, "tg_links.json", "tg_links.json")
            access.migrate_from_legacy(legacy)
        except Exception:  # noqa: BLE001
            pass
        ctx.api.register("access", "admin_ids", access.admin_ids)
        ctx.api.register("access", "is_admin", access.is_admin)
        ctx.api.register("access", "get_right", access.get_right)
        ctx.api.register("access", "toggle_right", access.toggle_right)
        ctx.api.register("access", "notify_targets", access.notify_targets)
        ctx.api.register("access", "seed", access.seed)
        ctx.api.register("access", "add_admin", access.add_admin_id)
        ctx.api.register("access", "remove_admin", access.remove_admin_id)
        ctx.api.register("access", "link_code", access.mint_link_code)
        ctx.api.register("access", "link_consume", access.consume_link_code)
        ctx.api.register("access", "linked_uid", access.linked_uid)
        ctx.api.register("access", "linked_telegram", access.linked_telegram)
        ctx.api.register(
            "access", "linked_telegrams", access.linked_telegrams)
        ctx.api.register(
            "access", "canonical_owner", access.canonical_owner)
        ctx.api.register("access", "link_tg_code", access.mint_tg_code)
        ctx.api.register(
            "access", "link_tg_consume", access.consume_tg_code)
        ctx.api.register(
            "access", "link_set_inherit", access.set_inherit)
        ctx.api.register("access", "link_list", access.link_list)
        ctx.api.register("access", "link_admin", access.link_admin)
        ctx.api.register("access", "unlink", access.unlink_telegram)
        ctx.api.register("access", "unlink_uid", access.unlink_uid)
        ctx.api.register(
            "access", "effective_can", access.effective_can)
        ctx.api.register("access", "magic_issue", access.magic_issue)
        ctx.api.register("access", "magic_consume", access.magic_consume)

    def start(self) -> None:
        # Реестр прав собирается по ходу setup'ов — сидим здесь,
        # когда все манифесты уже зарегистрированы.
        try:
            access.seed()
        except Exception as e:  # noqa: BLE001
            logger.warning("access seed: %s", e)

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