modsafe v0.12.1 [base]

"""modsafe.installer — загрузка модулей архивом: стейджинг, проверки, установка.

Пайплайн (всё под правом modules_manage):
  upload (POST multipart) -> data/modstaging/<name>/ + аудит сканером;
  review (🔍 в таблице) — человек смотрит находки;
  install -> modules/<name>/, запись modstate ВЫКЛЮЧЕННЫМ, эталон хешей.
Включение — вручную (modsafe.enable) + рестарт. Маркет (список со страницы)
— заглушка до запуска страницы (modsafe.market).

Защита: только .zip/.tar.gz, лимиты размера/числа файлов, запрет ..,
абсолютных путей, симлинков и бинарных расширений, имя ^[a-z][a-z0-9_]*$,
манифест с обязательными полями; архив уже установленного модуля —
кандидат на обновление (флаг update, откат через update_from_staging).
"""
import io
import logging
import os
import re
import shutil
import tarfile
import zipfile

logger = logging.getLogger(__name__)

_NAME_RX = re.compile(r"^[a-z][a-z0-9_]*$")
_BLOCK_EXT = {".so", ".dll", ".exe", ".bin", ".apk"}


def _limits(config):
    def _int(key, default):
        try:
            return int((config.get(key, default) if config else default) or default)
        except (ValueError, TypeError, AttributeError):
            return default
    max_mb = _int("MODSAFE_MAX_UPLOAD_MB", 5)
    return max_mb * 1024 * 1024, 200


def _read_manifest(d):
    import json
    for fn in ("manifest.yaml", "manifest.yml", "manifest.json"):
        p = os.path.join(d, fn)
        if os.path.isfile(p):
            with open(p, "r", encoding="utf-8") as f:
                if fn.endswith(".json"):
                    data = json.load(f)
                else:
                    import yaml
                    data = yaml.safe_load(f)
            if not isinstance(data, dict):
                raise ValueError("манифест %s должен быть объектом" % fn)
            return data
    raise ValueError("в архиве нет manifest.yaml/json")


def _check_manifest(data, dirname):
    from core.base_module import MANIFEST_REQUIRED
    for field in MANIFEST_REQUIRED:
        if not data.get(field):
            raise ValueError("манифест: нет обязательного поля %s" % field)
    if data.get("name") != dirname:
        raise ValueError("манифест name=%s не совпадает с именем %s"
                         % (data.get("name"), dirname))
    if not _NAME_RX.match(data.get("name") or ""):
        raise ValueError("недопустимое имя модуля (латиница, цифры, _)")


def _manifest_version(d):
    """Версия из манифеста каталога модуля (пустая, если прочитать нельзя)."""
    try:
        return str(_read_manifest(d).get("version") or "")
    except Exception:  # noqa: BLE001
        return ""


def _cmp_versions(a, b):
    """Сверка версий модулей: -1 (a<b), 0 (равны), 1 (a>b).

    Та же логика, что у market_client._release_key (числовые чанки по
    разделителям .-+_): архивное обновление и маркет сортируют одинаково.
    Равные строки всегда дают 0, независимо от парсинга. Пустая версия —
    ниже любой непустой.
    """
    a = a or ""
    b = b or ""
    if a == b:
        return 0
    if not a:
        return -1
    if not b:
        return 1
    from . import market_client as _mc
    ka = _mc._release_key({"version": a or ""})
    kb = _mc._release_key({"version": b or ""})
    return (ka > kb) - (ka < kb)


def _safe_members_zip(zf):
    out = []
    for info in zf.infolist():
        if info.is_dir():
            continue
        name = info.filename
        out.append((name, info.file_size, False))
    return out


def _safe_members_tar(tf):
    out = []
    for m in tf.getmembers():
        if m.isdir():
            continue
        out.append((m.name, m.size, m.issym() or m.islnk()))
    return out


def _validate_members(members, max_bytes):
    total = 0
    if len(members) > 200 or not members:
        raise ValueError("в архиве %d файлов (пусто или больше 200)" % len(members))
    for name, size, is_link in members:
        if is_link:
            raise ValueError("симлинки запрещены: %s" % name)
        if not name or name.startswith("/") or name.startswith("\\"):
            raise ValueError("абсолютный путь: %s" % name)
        parts = [p for p in name.replace("\\", "/").split("/") if p not in ("", ".")]
        if not parts or any(p == ".." for p in name.replace("\\", "/").split("/")):
            raise ValueError("выход за каталог: %s" % name)
        if os.path.splitext(parts[-1])[1].lower() in _BLOCK_EXT:
            raise ValueError("бинарные файлы запрещены: %s" % name)
        total += size or 0
        if total > max_bytes:
            raise ValueError("архив больше лимита")
    return total


def stage_upload(data_dir, modules_dir, config, filename, body):
    """Принять архив, разложить в стейджинг, проверить, просканировать."""
    from core import audit as _audit
    max_bytes, _ = _limits(config)
    filename = (filename or "").strip()
    body = body or b""
    if not body:
        raise ValueError("пустой файл")
    if len(body) > max_bytes:
        raise ValueError("файл больше лимита %d МБ" % (max_bytes // 1048576))
    low = filename.lower()
    kind = "zip" if low.endswith(".zip") else (
        "tar" if low.endswith((".tar.gz", ".tgz", ".tar")) else "")
    if not kind:
        raise ValueError("только .zip / .tar.gz")
    staging = os.path.join(data_dir, "modstaging")
    os.makedirs(staging, exist_ok=True)
    tmp = os.path.join(staging, "_incoming")
    if os.path.isdir(tmp):
        shutil.rmtree(tmp)
    os.makedirs(tmp)
    try:
        if kind == "zip":
            with zipfile.ZipFile(io.BytesIO(body)) as zf:
                members = _safe_members_zip(zf)
                _validate_members(members, max_bytes)
                for name, _, _ in members:
                    dest = os.path.join(tmp, *name.replace("\\", "/").split("/"))
                    if dest.startswith(tmp + os.sep):
                        os.makedirs(os.path.dirname(dest), exist_ok=True)
                        with zf.open(name) as src, open(dest, "wb") as dst:
                            shutil.copyfileobj(src, dst, 65536)
        else:
            with tarfile.open(fileobj=io.BytesIO(body), mode="r:*") as tf:
                members = _safe_members_tar(tf)
                _validate_members(members, max_bytes)
                for m in tf.getmembers():
                    if m.isdir():
                        continue
                    dest = os.path.join(tmp, *m.name.replace("\\", "/").split("/"))
                    if not dest.startswith(tmp + os.sep):
                        raise ValueError("выход за каталог: %s" % m.name)
                    os.makedirs(os.path.dirname(dest), exist_ok=True)
                    src = tf.extractfile(m)
                    if src is None:
                        raise ValueError("не читается: %s" % m.name)
                    with src, open(dest, "wb") as dst:
                        shutil.copyfileobj(src, dst, 65536)
    except (zipfile.BadZipFile, tarfile.TarError) as e:
        raise ValueError("битый архив: %s" % e)
    # unwrap: единственный корневой каталог -> его содержимое
    top = [e for e in sorted(os.listdir(tmp))
           if os.path.isdir(os.path.join(tmp, e))]
    files_top = [e for e in sorted(os.listdir(tmp))
                 if os.path.isfile(os.path.join(tmp, e))]
    root = tmp
    if len(top) == 1 and not files_top:
        root = os.path.join(tmp, top[0])
    manifest = _read_manifest(root)
    name = manifest.get("name")
    _check_manifest(manifest, name)
    update = False
    local_version = ""
    downgrade = False
    installed = os.path.join(modules_dir, name)
    if os.path.exists(installed):
        if not os.path.isdir(installed):
            shutil.rmtree(tmp)
            raise ValueError("по пути модуля %s не каталог" % name)
        # Уже установлен — это кандидат на ОБНОВЛЕНИЕ архивом: стейджим
        # так же; решение принимает человек (install/update_staged).
        local_version = _manifest_version(installed)
        update = True
        downgrade = _cmp_versions(manifest.get("version") or "",
                                  local_version) < 0
    dest = os.path.join(staging, name)
    if os.path.isdir(dest):
        shutil.rmtree(dest)
    if root == tmp:
        os.rename(tmp, dest)
    else:
        shutil.move(root, dest)
        shutil.rmtree(tmp)
    findings = _audit.scan_module(dest, manifest.get("audit_allow") or [])
    summary = _audit.summarize(findings)
    logger.info("modsafe: застейджен %s v%s (%s)%s",
                name, manifest.get("version"), summary,
                " (обновление поверх %s)" % local_version if update else "")
    return {"ok": True, "name": name, "version": manifest.get("version") or "",
            "description": manifest.get("description") or "",
            "summary": summary, "findings": findings[:200],
            "staged": True, "update": update,
            "local_version": local_version, "downgrade": downgrade}


def list_staged(data_dir):
    """Стейджинг: [{name, version, description}] с валидным манифестом."""
    staging = os.path.join(data_dir, "modstaging")
    out = []
    if not os.path.isdir(staging):
        return out
    for entry in sorted(os.listdir(staging)):
        if entry.startswith("_"):
            continue
        d = os.path.join(staging, entry)
        if not os.path.isdir(d):
            continue
        try:
            m = _read_manifest(d)
            out.append({"name": entry, "version": m.get("version") or "",
                        "description": m.get("description") or ""})
        except Exception:  # noqa: BLE001
            out.append({"name": entry, "version": "", "description": "битый стейджинг"})
    return out


def _adopt_staged_env(staged_dir, name, data_dir):
    """Забрать <mod>/_env/<name>.env из стейджинга в data/env/<name>.env.

    Обратная сторона pack with_secrets: секреты едут отдельно от кода
    и кладутся в data/env (0600), а не в modules/. Без _env — ничего.
    """
    src = os.path.join(staged_dir, "_env", name + ".env")
    if not os.path.isfile(src):
        return False
    env_dir = os.path.join(data_dir, "env")
    os.makedirs(env_dir, exist_ok=True)
    dest = os.path.join(env_dir, name + ".env")
    if os.path.isfile(dest):
        logger.info("modsafe: env %s уже есть — архивный не затираем", dest)
    else:
        shutil.copy2(src, dest)
        try:
            os.chmod(dest, 0o600)
        except OSError:
            pass
        logger.info("modsafe: env принят: %s", dest)
    shutil.rmtree(os.path.join(staged_dir, "_env"), ignore_errors=True)
    return True


def install_staged(data_dir, modules_dir, name):
    """Переместить стейджинг в modules/, выключенным, с эталоном хешей."""
    from core import audit as _audit
    from core import modstate as _ms
    if not _NAME_RX.match(name or ""):
        raise ValueError("недопустимое имя")
    if _ms.corrupt_info(data_dir):
        raise ValueError(_ms.CORRUPT_ERR)
    src = os.path.join(data_dir, "modstaging", name)
    dest = os.path.join(modules_dir, name)
    if not os.path.isdir(src):
        raise ValueError("в стейджинге нет %s" % name)
    if os.path.exists(dest):
        raise ValueError("модуль %s уже установлен" % name)
    manifest = _read_manifest(src)
    _check_manifest(manifest, name)
    shutil.move(src, dest)
    _adopt_staged_env(dest, name, data_dir)
    # запись modstate: известен, ВЫКЛЮЧЕН; эталон = текущие файлы
    state = _ms.load(data_dir)
    mods = state.setdefault("modules", {})
    rec = mods.get(name)
    if not isinstance(rec, dict):
        rec = {"enabled": False, "known": True, "hashes": {},
               "hashes_at": "", "audit": {}, "audited_at": "",
               "missing": False}
        mods[name] = rec
    rec["enabled"] = False
    rec["known"] = True
    rec["missing"] = False
    _ms.save(data_dir, state)
    _ms.update_hashes(data_dir, name, _audit.hash_module(dest))
    logger.info("modsafe: установлен %s (выключен)", name)
    return {"ok": True, "name": name, "enabled": False, "apply": "restart"}


def update_from_staging(data_dir, modules_dir, name, force=False):
    """Обновить установленный модуль из стейджинга: бекап + атомарный swap.

    Как маркет-обновление (api_update_module), но источник — загруженный
    архив (его проверки сделал stage_upload). Порядок: сверка версий
    (равные без force — отказ с подсказкой; даунгрейд — разрешён, флаг
    downgrade), бекап кода в data/backups/modules/, swap .__new__/__old__
    (единый путь для всех модулей; обязателен для modsafe — исполняемый
    код не трогается в процессе), env из <mod>/_env без перезаписи,
    modstate known/missing, эталон хешей. enabled и рестарт — на совести
    вызывающего (modsafe.api_update_staged).
    """
    import time as _time
    from core import audit as _audit
    from core import modstate as _ms
    if not _NAME_RX.match(name or ""):
        raise ValueError("недопустимое имя")
    if _ms.corrupt_info(data_dir):
        raise ValueError(_ms.CORRUPT_ERR)
    src = os.path.join(data_dir, "modstaging", name)
    mod_dir = os.path.join(modules_dir, name)
    if not os.path.isdir(src):
        raise ValueError("в стейджинге нет %s" % name)
    if not os.path.isdir(mod_dir):
        raise ValueError("модуль %s не установлен "
                         "(установка — modsafe.install)" % name)
    manifest = _read_manifest(src)
    _check_manifest(manifest, name)
    new_version = manifest.get("version") or ""
    old_version = _manifest_version(mod_dir)
    order_cmp = _cmp_versions(new_version, old_version)
    if order_cmp == 0 and not force:
        raise ValueError("версия %s уже установлена "
                         "(перезалить — force)" % new_version)
    downgrade = order_cmp < 0
    stamp = _time.strftime("%Y%m%d_%H%M%S")
    backup = os.path.join(data_dir, "backups", "modules",
                          "%s-%s-%s" % (name, old_version or "noversion",
                                        stamp))
    os.makedirs(os.path.dirname(backup), exist_ok=True)
    shutil.copytree(mod_dir, backup, symlinks=False,
                    ignore=shutil.ignore_patterns("__pycache__"))
    new_dir = mod_dir + ".__new__"
    shutil.rmtree(new_dir, ignore_errors=True)
    try:
        shutil.copytree(src, new_dir, symlinks=False)
        _adopt_staged_env(new_dir, name, data_dir)
    except Exception as e:  # noqa: BLE001
        shutil.rmtree(new_dir, ignore_errors=True)
        raise ValueError("запись не удалась: %s" % str(e)[:200])
    dead_dir = mod_dir + ".__old__"
    shutil.rmtree(dead_dir, ignore_errors=True)
    try:
        os.rename(mod_dir, dead_dir)
        os.rename(new_dir, mod_dir)
    except Exception as e:  # noqa: BLE001
        try:
            if os.path.isdir(dead_dir) and not os.path.isdir(mod_dir):
                os.rename(dead_dir, mod_dir)
        except Exception:  # noqa: BLE001
            pass
        shutil.rmtree(new_dir, ignore_errors=True)
        shutil.rmtree(dead_dir, ignore_errors=True)
        raise ValueError("swap не удался, откачено: %s" % str(e)[:200])
    shutil.rmtree(dead_dir, ignore_errors=True)
    shutil.rmtree(src, ignore_errors=True)
    # modstate: известен, не отсутствует; enabled НЕ трогаем (сохраняет
    # вызывающий); эталон хешей = новые файлы.
    state = _ms.load(data_dir)
    mods = state.setdefault("modules", {})
    rec = mods.get(name)
    if not isinstance(rec, dict):
        rec = {"enabled": False, "known": True, "hashes": {},
               "hashes_at": "", "audit": {}, "audited_at": "",
               "missing": False}
        mods[name] = rec
    rec["known"] = True
    rec["missing"] = False
    _ms.save(data_dir, state)
    _ms.update_hashes(data_dir, name, _audit.hash_module(mod_dir))
    logger.info("modsafe: обновлён из архива %s %s->%s (бекап %s%s)",
                name, old_version, new_version, backup,
                ", даунгрейд" if downgrade else "")
    return {"ok": True, "name": name, "version": new_version,
            "old_version": old_version, "updated": True, "backup": backup,
            "downgrade": downgrade, "apply": "restart"}


def write_files(dest_dir, files):
    """Записать {relpath: bytes} в каталог (валидация путей, без секретов).

    Общая точка для маркет-установки и обновления: пути уже проверены
    в market_client, здесь — повторная защита от .. и абсолютных.
    """
    for rel in sorted(files):
        parts = [p for p in (rel or "").replace("\\", "/").split("/")
                 if p not in ("", ".")]
        if (not parts or any(p == ".." for p in parts)
                or parts[0].startswith(".") or os.path.isabs(rel or "")):
            raise ValueError("плохой путь: %s" % rel)
        target = os.path.join(dest_dir, *parts)
        if not os.path.abspath(target).startswith(
                os.path.abspath(dest_dir) + os.sep):
            raise ValueError("выход за каталог: %s" % rel)
        os.makedirs(os.path.dirname(target), exist_ok=True)
        with open(target, "wb") as f:
            f.write(files[rel])
    return len(files)


def stage_files(data_dir, name, files):
    """Положить файлы модуля в стейджинг напрямую (маркет, без архива).

    Проверки те же, что у stage_upload: манифест, имя, отсутствие
    установленного каталога. Возвращает путь стейджинга.
    """
    from core import audit as _audit
    if not _NAME_RX.match(name or ""):
        raise ValueError("недопустимое имя")
    staging = os.path.join(data_dir, "modstaging")
    os.makedirs(staging, exist_ok=True)
    dest = os.path.join(staging, name)
    if os.path.isdir(dest):
        shutil.rmtree(dest)
    os.makedirs(dest)
    try:
        write_files(dest, files)
    except ValueError:
        shutil.rmtree(dest, ignore_errors=True)
        raise
    manifest = _read_manifest(dest)
    _check_manifest(manifest, name)
    findings = _audit.scan_module(dest, manifest.get("audit_allow") or [])
    summary = _audit.summarize(findings)
    logger.info("modsafe: застейджен с сайта %s v%s (%s)",
                name, manifest.get("version"), summary)
    return dest


#: Файлы, никогда не попадающие в архив без секретов.
_SECRET_NAMES = {".env"}
_SECRET_EXT = {".session"}
#: Мусор, не попадающий в архив никогда.
_SKIP_DIRS = {"__pycache__", ".git", "node_modules"}
_SKIP_EXT = {".pyc", ".pyo"}


def _module_files(mod_dir, with_secrets):
    out = []
    for root, dirs, files in os.walk(mod_dir):
        dirs[:] = sorted(d for d in dirs if d not in _SKIP_DIRS)
        for fn in sorted(files):
            if os.path.splitext(fn)[1] in _SKIP_EXT:
                continue
            # .env внутри модуля — legacy (миграция в data/env/): в архив
            # не кладём никогда; секреты едут только из data/env (см. pack).
            if fn in _SECRET_NAMES or os.path.splitext(fn)[1] in _SECRET_EXT:
                continue
            p = os.path.join(root, fn)
            if os.path.isfile(p) and not os.path.islink(p):
                out.append(p)
    return out


def pack_module(modules_dir, name, with_secrets=False, env_dir=None):
    """Упаковать модуль в zip. Возвращает (filename, bytes).

    with_secrets=False (дефолт): без секретов — безопасно делиться.
    with_secrets=True: плюс data/env/<name>.env как <name>/_env/<name>.env
    (при установке кладётся обратно в data/env/<name>.env, 0600).
    Legacy .env внутри модуля не пакуется никогда (мигрирован в data/).
    Раньше экспорт с env не работал: env переехали в data/, а pack их
    не видел — теперь читает env_dir напрямую.
    """
    if not _NAME_RX.match(name or ""):
        raise ValueError("недопустимое имя")
    if name == "modsafe":
        raise ValueError("modsafe не упаковывается")
    mod_dir = os.path.join(modules_dir, name)
    if not os.path.isdir(mod_dir):
        raise ValueError("модуль %s не найден" % name)
    manifest = _read_manifest(mod_dir)
    _check_manifest(manifest, name)
    files = _module_files(mod_dir, with_secrets)
    if not files:
        raise ValueError("в модуле нет файлов для упаковки")
    buf = io.BytesIO()
    with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
        for p in files:
            rel = os.path.relpath(p, mod_dir).replace(os.sep, "/")
            zf.write(p, os.path.join(name, rel))
        if with_secrets and env_dir:
            ep = os.path.join(env_dir, name + ".env")
            if os.path.isfile(ep):
                zf.write(ep, os.path.join(name, "_env", name + ".env"))
    data = buf.getvalue()
    logger.info("modsafe: упакован %s (%d файлов, %d байт, secrets=%s)",
                name, len(files), len(data), bool(with_secrets))
    return "%s%s.zip" % (name, "-secrets" if with_secrets else ""), data


def delete_module(data_dir, modules_dir, name, must_be_disabled=True,
                  purge_cache=False, purge_env=False):
    """Удалить модуль: каталог + запись modstate + стейджинг.

    Только выключенные (сначала modsafe.enable=false + рестарт).
    modsafe и транспорты удалять нельзя.
    purge_cache: снести data/cache/<name>/ (по умолчанию кеш остаётся —
    пригодится при переустановке). purge_env: снести data/env/<name>.env
    (по умолчанию секреты остаются — переживают переустановку).
    """
    from core import modstate as _ms
    if not _NAME_RX.match(name or ""):
        raise ValueError("недопустимое имя")
    if name == "modsafe":
        raise ValueError("modsafe удалять нельзя")
    mod_dir = os.path.join(modules_dir, name)
    if not os.path.isdir(mod_dir):
        raise ValueError("модуль %s не найден" % name)
    state = _ms.load(data_dir)
    rec = (state.get("modules") or {}).get(name) or {}
    if must_be_disabled and rec.get("enabled"):
        raise ValueError("модуль %s включён: сначала выключите и перезапустите" % name)
    shutil.rmtree(mod_dir)
    if name in (state.get("modules") or {}):
        del state["modules"][name]
        _ms.save(data_dir, state)
    staging = os.path.join(data_dir, "modstaging", name)
    if os.path.isdir(staging):
        shutil.rmtree(staging)
    removed = {"cache": False, "env": False}
    if purge_cache:
        cache_dir = os.path.join(data_dir, "cache", name)
        if os.path.isdir(cache_dir) and not os.path.islink(cache_dir):
            shutil.rmtree(cache_dir)
            removed["cache"] = True
    if purge_env:
        env_file = os.path.join(data_dir, "env", name + ".env")
        if os.path.isfile(env_file):
            os.remove(env_file)
            removed["env"] = True
    logger.info("modsafe: удалён %s (cache=%s env=%s)", name,
                removed["cache"], removed["env"])
    return {"ok": True, "name": name, "purged": removed}


def _env_path(modules_dir, name, env_dir=None):
    """Путь .env модуля: data/env/<name>.env; legacy modules/<name>/.env
    учитывается только для флага has_env (миграция — через install.sh)."""
    main = os.path.join(env_dir, name + ".env") if env_dir else None
    legacy = os.path.join(modules_dir, name, ".env")
    has_legacy = os.path.isfile(legacy)
    if main and os.path.isfile(main):
        return main, has_legacy
    if has_legacy:
        return legacy, False
    # нет нигде — писать в основной (если есть куда), иначе legacy
    return (main or legacy), False


def env_keys(modules_dir, name, env_dir=None):
    """Ключи секретов модуля: схема + ключи из .env (БЕЗ значений).

    Динамические секреты (auth_ref точек outbound и т.п.) нельзя перечислить
    в схеме — их имена берутся из самого .env (dynamic: true).
    """
    import json
    if not _NAME_RX.match(name or ""):
        raise ValueError("недопустимое имя")
    mod_dir = os.path.join(modules_dir, name)
    if not os.path.isdir(mod_dir):
        raise ValueError("модуль %s не найден" % name)
    schema = {}
    sp = os.path.join(mod_dir, "config.schema.json")
    if os.path.isfile(sp):
        try:
            schema = json.load(open(sp, encoding="utf-8"))
        except (OSError, ValueError):
            pass
    required = list(schema.get("required") or [])
    defaults = dict(schema.get("optional") or {})
    have = set()
    ep, legacy = _env_path(modules_dir, name, env_dir)
    if os.path.isfile(ep):
        try:
            with open(ep, encoding="utf-8") as f:
                for line in f:
                    line = line.strip()
                    if line and not line.startswith("#") and "=" in line:
                        have.add(line.split("=", 1)[0].strip())
        except OSError:
            pass
    keys = [{"key": k, "required": k in required, "schema": True,
             "set": k in have, "default": str(defaults.get(k, ""))}
            for k in required + [k for k in defaults if k not in required]]
    seen = {k["key"] for k in keys}
    for k in sorted(have - seen):
        keys.append({"key": k, "required": False, "schema": False,
                     "set": True, "default": ""})
    out = {"name": name, "keys": keys,
           "has_env": os.path.isfile(ep)}
    if legacy:
        out["legacy_env"] = True
    return out


def env_set(modules_dir, name, values, env_dir=None):
    """Сохранить секреты модуля (.env 0600). Пустые значения — оставить как есть.

    Ключ допустим, если он в схеме (required+optional) ЛИБО похож на имя
    переменной окружения ([A-Z][A-Z0-9_], до 64): динамические секреты
    (auth_ref и т.п.) в схеме не перечислить. Остальное — отказ.
    """
    import json
    import re as _re
    if not _NAME_RX.match(name or ""):
        raise ValueError("недопустимое имя")
    mod_dir = os.path.join(modules_dir, name)
    if not os.path.isdir(mod_dir):
        raise ValueError("модуль %s не найден" % name)
    schema = {}
    sp = os.path.join(mod_dir, "config.schema.json")
    if os.path.isfile(sp):
        try:
            schema = json.load(open(sp, encoding="utf-8"))
        except (OSError, ValueError):
            pass
    allowed = set(schema.get("required") or []) | set((schema.get("optional") or {}))
    if not allowed:
        raise ValueError("у модуля %s нет схемы секретов" % name)
    values = values or {}
    unknown = sorted(k for k in values
                     if k not in allowed
                     and not _re.fullmatch(r"[A-Z][A-Z0-9_]{0,63}", k or ""))
    if unknown:
        raise ValueError("недопустимые ключи: %s" % ", ".join(unknown))
    ep, _legacy = _env_path(modules_dir, name, env_dir)
    current = {}
    order = []
    if os.path.isfile(ep):
        try:
            with open(ep, encoding="utf-8") as f:
                for line in f:
                    s = line.strip()
                    if s and not s.startswith("#") and "=" in s:
                        k, v = s.split("=", 1)
                        k = k.strip()
                        if k not in current:
                            order.append(k)
                        current[k] = v.strip()
        except OSError:
            pass
    changed = []
    for k, v in values.items():
        v = (v or "").strip()
        if not v:
            continue  # пустое = не менять
        if current.get(k) != v:
            current[k] = v
            changed.append(k)
        if k not in order:
            order.append(k)
    if changed:
        parent = os.path.dirname(ep)
        if parent:
            os.makedirs(parent, exist_ok=True)
        tmp = ep + ".tmp"
        with open(tmp, "w", encoding="utf-8") as f:
            for k in order:
                if k in current:
                    f.write("%s=%s\n" % (k, current[k]))
        os.chmod(tmp, 0o600)
        os.replace(tmp, ep)
    logger.info("modsafe: секреты %s: изменено %d", name, len(changed))
    return {"ok": True, "name": name, "changed": changed}

#: Базовые пакеты venv — никогда не сироты.
_BASE_PKGS = frozenset({"pip", "setuptools", "wheel"})


def _norm_pkg(name):
    import re as _re
    return _re.sub(r"[-_.]+", "-", (name or "").lower()).strip("-")


def _req_names(path):
    """Имена пакетов из requirements.txt (без версий/маршрутов)."""
    import re as _re
    out = set()
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as f:
            for line in f:
                line = line.strip()
                if not line or line.startswith("#"):
                    continue
                line = line.split(";", 1)[0].strip()
                m = _re.match(r"^([A-Za-z0-9_.-]+)", line)
                if m:
                    out.add(_norm_pkg(m.group(1)))
    except OSError:
        pass
    return out


def pip_orphans(modules_dir, transports_dir, root_dir):
    """Пакеты в venv, которых не требует никто (только показ, без удаления).

    required = union requirements.txt корня + транспортов + УСТАНОВЛЕННЫХ
    модулей (удалённого модуля уже нет в дереве — его пакеты и всплывут).
    Замыкание по зависимостям зависимостей — через Requires-Dist.
    Возвращает {orphans: [{name, version}], required_n, installed_n}.
    """
    try:
        from importlib import metadata as _md
    except ImportError:
        return {"orphans": [], "required_n": 0, "installed_n": 0,
                "error": "нет importlib.metadata"}
    roots = set()
    rp = os.path.join(root_dir, "requirements.txt")
    if os.path.isfile(rp):
        roots |= _req_names(rp)
    for base in (modules_dir, transports_dir):
        try:
            entries = sorted(os.listdir(base))
        except OSError:
            continue
        for entry in entries:
            fp = os.path.join(base, entry, "requirements.txt")
            if os.path.isfile(fp):
                roots |= _req_names(fp)
    installed = {}
    for dist in _md.distributions():
        try:
            n = _norm_pkg(dist.metadata["Name"] or "")
        except Exception:  # noqa: BLE001
            continue
        if n:
            installed.setdefault(n, str(dist.version or ""))
    # замыкание: требуемое + всё, что тянут их Requires-Dist
    keep = set(roots) | set(_BASE_PKGS)
    queue = sorted(keep)
    while queue:
        cur = queue.pop()
        try:
            reqs = _md.requires(cur) or []
        except Exception:  # noqa: BLE001
            continue
        for r in reqs:
            m = _norm_pkg(r.split(";", 1)[0].strip().split()[0]
                          .split("[")[0].split("(")[0].rstrip(","))
            import re as _re
            mm = _re.match(r"^([A-Za-z0-9_.-]+)", m)
            name = _norm_pkg(mm.group(1)) if mm else ""
            if name and name not in keep:
                keep.add(name)
                queue.append(name)
    orphans = [{"name": n, "version": installed[n]}
               for n in sorted(installed) if n not in keep]
    return {"orphans": orphans, "required_n": len(keep),
            "installed_n": len(installed)}


def pip_uninstall_orphans(modules_dir, transports_dir, root_dir, packages):
    """Удалить пакеты-сироты тем же интерпретатором (pip uninstall -y).

    Безопасность: каждый пакет перепроверяется на сиротство ПЕРЕД удалением
    (свежее замыкание, клиентскому списку не доверяем); базовые пакеты
    (pip/setuptools/wheel) и пустые имена — отказ; лимит 20 за вызов.
    Возвращает {removed: [...], skipped: [{name, reason}], output}.
    """
    import re as _re
    import subprocess as _sp
    import sys as _sys
    wanted = []
    for p in packages or []:
        n = _norm_pkg(p or "")
        if n and n not in wanted:
            wanted.append(n)
    if not wanted:
        raise ValueError("пустой список пакетов")
    if len(wanted) > 20:
        raise ValueError("больше 20 за раз нельзя")
    fresh = pip_orphans(modules_dir, transports_dir, root_dir)
    still = {o["name"] for o in fresh.get("orphans") or []}
    removed, skipped, out_lines = [], [], []
    for n in wanted:
        if n in _BASE_PKGS:
            skipped.append({"name": n, "reason": "базовый пакет venv"})
            continue
        if n not in still:
            skipped.append({"name": n, "reason": "уже требуется (не сирота)"})
            continue
        if not _re.fullmatch(r"[a-z0-9][a-z0-9_.-]*", n):
            skipped.append({"name": n, "reason": "плохое имя"})
            continue
        try:
            r = _sp.run([_sys.executable, "-m", "pip", "uninstall",
                         "-y", n],
                        capture_output=True, text=True, timeout=120)
            tail = ((r.stdout or "") + "\n" + (r.stderr or "")).strip().splitlines()
            out_lines += ["== %s (rc=%d)" % (n, r.returncode)]
            out_lines += [l[:200] for l in tail[-4:]]
            if r.returncode == 0:
                removed.append(n)
            else:
                skipped.append({"name": n,
                                "reason": "pip rc=%d" % r.returncode})
        except Exception as e:  # noqa: BLE001
            skipped.append({"name": n, "reason": "%s: %s" % (
                type(e).__name__, e)})
    logger.info("modsafe: pip-сироты удалены: %s, пропущены: %s",
                removed, [s["name"] for s in skipped])
    return {"ok": True, "removed": removed, "skipped": skipped,
            "output": "\n".join(out_lines)[:4000]}


#: Экспорт кеша: только .json неймспейса data/cache/<name>/.
#: .lock и *.tmp (флаки/незавершённая запись) не едут никогда.
_CACHE_EXT = {".json"}
_CACHE_SKIP = {".lock"}
_CACHE_MAX_BYTES = 50 * 1024 * 1024
_CACHE_MAX_FILES = 500


def _cache_dir(data_dir, name):
    if not _NAME_RX.match(name or ""):
        raise ValueError("недопустимое имя")
    return os.path.join(data_dir, "cache", name)


def cache_list(data_dir, name):
    """Файлы кеша модуля [{path, size}]. Нет каталога — пусто."""
    d = _cache_dir(data_dir, name)
    out = []
    if not os.path.isdir(d):
        return out
    for root, dirs, files in os.walk(d):
        dirs[:] = sorted(x for x in dirs if not x.startswith("."))
        for fn in sorted(files):
            if fn in _CACHE_SKIP or fn.endswith(".tmp"):
                continue
            if os.path.splitext(fn)[1].lower() not in _CACHE_EXT:
                continue
            p = os.path.join(root, fn)
            if os.path.isfile(p) and not os.path.islink(p):
                out.append({"path": os.path.relpath(p, d).replace(os.sep, "/"),
                            "size": os.path.getsize(p)})
    return out


def export_cache(data_dir, name):
    """Упаковать кеш модуля в zip. Возвращает (filename, bytes)."""
    files = cache_list(data_dir, name)
    if not files:
        raise ValueError("у модуля %s нет кеша для экспорта" % name)
    total = sum(f["size"] for f in files)
    if len(files) > _CACHE_MAX_FILES or total > _CACHE_MAX_BYTES:
        raise ValueError("кеш слишком большой (%d файлов, %d байт)"
                         % (len(files), total))
    d = _cache_dir(data_dir, name)
    buf = io.BytesIO()
    with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
        for f in files:
            zf.write(os.path.join(d, *f["path"].split("/")),
                     os.path.join(name, "cache", f["path"]))
    data = buf.getvalue()
    logger.info("modsafe: кеш %s упакован (%d файлов, %d байт)",
                name, len(files), len(data))
    return "%s-cache.zip" % name, data


def import_cache(data_dir, name, body):
    """Восстановить кеш модуля из zip (с бекапом текущего).

    Проверки как у stage_upload: zip, без .. и абсолютных, без симлинков,
    только .json, лимиты. Текущий кеш — в data/backups/cache/<name>-<ts>/.
    """
    import time as _time
    if not _NAME_RX.match(name or ""):
        raise ValueError("недопустимое имя")
    body = body or b""
    if not body:
        raise ValueError("пустой файл")
    if len(body) > _CACHE_MAX_BYTES:
        raise ValueError("архив больше лимита")
    try:
        zf = zipfile.ZipFile(io.BytesIO(body))
        members = _safe_members_zip(zf)
    except zipfile.BadZipFile as e:
        raise ValueError("битый архив: %s" % e)
    if not members or len(members) > _CACHE_MAX_FILES:
        raise ValueError("в архиве %d файлов (пусто или больше %d)"
                         % (len(members), _CACHE_MAX_FILES))
    total = 0
    clean = []
    for fname, size, _link in members:
        parts = [p for p in fname.replace("\\", "/").split("/")
                 if p not in ("", ".")]
        if (not parts or any(p == ".." for p in parts)
                or parts[0].startswith(".") or fname.startswith("/")):
            raise ValueError("плохой путь: %s" % fname)
        # корень вида <name>/cache/... (как пишет export_cache)
        if len(parts) >= 3 and parts[0] == name and parts[1] == "cache":
            rel = "/".join(parts[2:])
        else:
            rel = "/".join(parts)
        if os.path.splitext(rel)[1].lower() not in _CACHE_EXT:
            raise ValueError("только .json: %s" % fname)
        total += size or 0
        if total > _CACHE_MAX_BYTES:
            raise ValueError("архив больше лимита")
        clean.append((fname, rel))
    d = _cache_dir(data_dir, name)
    backup = None
    if os.path.isdir(d):
        stamp = _time.strftime("%Y%m%d_%H%M%S")
        backup = os.path.join(data_dir, "backups", "cache",
                              "%s-%s" % (name, stamp))
        os.makedirs(os.path.dirname(backup), exist_ok=True)
        shutil.copytree(d, backup, symlinks=False,
                        ignore=shutil.ignore_patterns(".lock"))
        shutil.rmtree(d)
    os.makedirs(d, exist_ok=True)
    try:
        with zipfile.ZipFile(io.BytesIO(body)) as zf:
            for fname, rel in clean:
                dest = os.path.join(d, *rel.split("/"))
                if not os.path.abspath(dest).startswith(
                        os.path.abspath(d) + os.sep):
                    raise ValueError("выход за каталог: %s" % fname)
                os.makedirs(os.path.dirname(dest), exist_ok=True)
                with zf.open(fname) as src, open(dest, "wb") as dst:
                    shutil.copyfileobj(src, dst, 65536)
    except Exception:
        shutil.rmtree(d, ignore_errors=True)
        if backup and os.path.isdir(backup):
            shutil.copytree(backup, d)
        raise
    logger.info("modsafe: кеш %s импортирован (%d файлов, бекап %s)",
                name, len(clean), backup or "—")
    return {"ok": True, "name": name, "files": len(clean),
            "backup": backup or ""}

"""modsafe.market_client — чтение маркета сайта (stdlib, без зависимостей).

Источники (см. site-report-2026-09-16, §1):
  живой маркет  GET {base}/api/market/modules[?slug=bot-modular]
    -> {"ok": true, "modules": [{name, version, description, category,
        url, sha, integrity}]}
  витрина       GET {base}/projects/bot-modular/market.json
    -> элементы с files[] ({path, size, content|truncated|binary})
  релизы        GET {base}/releases/index.json
    -> {"bot-modular": [{version, tag, name, body, prerelease, files[]}]}

Правила:
  - url вида /market/<name>-<version>.zip — legacy-шаблон без файлов,
    код берём из витрины (files[].content), не из url;
  - сверяем integrity (sha256-каноника имя+версия+коммит+пути/размеры+
    первые 4КБ), а не sha (sha может совпадать у разных модулей);
  - файлы с truncated/binary или отсутствующим content — отказ с честной
    ошибкой (предлагаем git), а не молчаливую неполную установку.
"""
import hashlib
import json
import logging
import urllib.parse
import urllib.request

logger = logging.getLogger(__name__)

_TIMEOUT = 25
# market.json везёт files[].content всех модулей инлайном (~11МБ
# на 20 модулей в 2026-10 и растёт) — капы файлов/модуля отдельно
# в extract_module_files (200КБ/файл, 1МБ/модуль).
_MAX_JSON_BYTES = 32 * 1024 * 1024
_MAX_FILE_BYTES = 100 * 1024 * 1024

_DENY_PREFIXES = (".", ".git/", "__pycache__/", "data/")
_DENY_NAMES = {".env"}


def _get_json(url):
    req = urllib.request.Request(url, headers={
        "Accept": "application/json",
        "User-Agent": "bot_modular-modsafe/1.0",
    })
    try:
        with urllib.request.urlopen(req, timeout=_TIMEOUT) as r:
            if r.status != 200:
                raise ValueError("HTTP %s: %s" % (r.status, url))
            raw = r.read(_MAX_JSON_BYTES + 1)
    except ValueError:
        raise
    except Exception as e:  # noqa: BLE001
        raise ValueError("маркет недоступен (%s: %s)"
                         % (type(e).__name__, str(e)[:120]))
    if len(raw) > _MAX_JSON_BYTES:
        raise ValueError("ответ маркета больше лимита")
    try:
        data = json.loads(raw.decode("utf-8"))
    except ValueError:
        raise ValueError("маркет вернул не JSON")
    if not isinstance(data, dict):
        raise ValueError("маркет: корень не объект")
    return data


def _get_bytes(url, max_bytes=_MAX_FILE_BYTES):
    req = urllib.request.Request(url, headers={
        "User-Agent": "bot_modular-modsafe/1.0",
    })
    try:
        with urllib.request.urlopen(req, timeout=60) as r:
            if r.status != 200:
                raise ValueError("HTTP %s: %s" % (r.status, url))
            raw = r.read(max_bytes + 1)
    except ValueError:
        raise
    except Exception as e:  # noqa: BLE001
        raise ValueError("скачивание не удалось (%s: %s)"
                         % (type(e).__name__, str(e)[:120]))
    if len(raw) > max_bytes:
        raise ValueError("файл больше лимита %d МБ" % (max_bytes // 1048576))
    return raw


def fetch_market(base_url):
    """Живой маркет §1.1. Возвращает [{name, version, ...}]."""
    base = (base_url or "").rstrip("/")
    if not base:
        raise ValueError("не задан MODSAFE_MARKET_URL")
    data = _get_json(base + "/api/market/modules?slug=bot-modular")
    if data.get("ok") is False:
        raise ValueError("маркет: %s" % str(data.get("error") or "ok=false"))
    mods = data.get("modules") or []
    if not isinstance(mods, list):
        raise ValueError("маркет: modules не список")
    out = []
    for m in mods:
        if not isinstance(m, dict) or not m.get("name"):
            continue
        out.append({
            "name": str(m.get("name") or ""),
            "version": str(m.get("version") or ""),
            "description": str(m.get("description") or ""),
            "category": str(m.get("category") or "extra"),
            "url": str(m.get("url") or ""),
            "sha": str(m.get("sha") or ""),
            "integrity": str(m.get("integrity") or ""),
        })
    return out


def fetch_vitrina(base_url, query=""):
    """Витрина §1.2. Возвращает {name: элемент с files[]}.

    query — точечный запрос одного модуля (?q=name): сайт фильтрует
    на своей стороне, в ответе только он (сотни КБ вместо всех 11МБ).
    Для установки/обновления всегда точечно; полный список — только
    для отображения requires в листинге (кеш 10 мин).
    """
    base = (base_url or "").rstrip("/")
    if not base:
        raise ValueError("не задан адрес сайта")
    url = base + "/projects/bot-modular/market.json"
    if (query or "").strip():
        url += "?q=" + urllib.parse.quote(str(query).strip(), safe="")
    data = _get_json(url)
    items = data if isinstance(data, list) else data.get("modules") \
        if isinstance(data, dict) else None
    if items is None:
        items = data.get("items") if isinstance(data, dict) else None
    if not isinstance(items, list):
        raise ValueError("витрина: нет списка модулей")
    out = {}
    for m in items:
        if isinstance(m, dict) and m.get("name"):
            out[str(m["name"])] = m
    return out


def _release_key(rel):
    """Ключ сортировки релизов: версия (числами) -> тег. Новейший последний."""
    import re as _re
    ver = str((rel or {}).get("version") or "")
    parts = []
    for chunk in _re.split(r"[.\-+_]", ver):
        parts.append((0, int(chunk)) if chunk.isdigit() else (1, chunk))
    return (parts, str((rel or {}).get("tag") or ""))


def fetch_releases(base_url):
    """Релизы §1.3. Возвращает [записи bot-modular] (черновики отсеяны).

    Сайт не гарантирует порядок (старые могут идти первыми), поэтому
    сортируем по версии: новейший — последний. Выбирайте rels[-1].
    """
    base = (base_url or "").rstrip("/")
    if not base:
        raise ValueError("не задан адрес сайта")
    data = _get_json(base + "/releases/index.json")
    items = (data.get("bot-modular") or []) if isinstance(data, dict) else []
    rels = [r for r in items if isinstance(r, dict)
            and not r.get("draft", r.get("prerelease", False)) is True
            and r.get("tag")]
    rels.sort(key=_release_key)
    return rels


def _safe_relpath(path):
    """Проверка пути файла модуля. Возвращает норму или None."""
    p = (path or "").replace("\\", "/").strip()
    if not p or p.startswith("/"):
        return None
    parts = [x for x in p.split("/") if x not in ("", ".")]
    if not parts or any(x == ".." for x in p.split("/")):
        return None
    low = "/".join(parts).lower()
    if low.split("/")[0].startswith(".") or low.startswith("."):
        return None
    for prefix in _DENY_PREFIXES:
        if low == prefix.rstrip("/") or low.startswith(prefix):
            return None
    if parts[-1] in _DENY_NAMES:
        return None
    return "/".join(parts)


def vitrina_files(vitrina, name, vitrina_error=""):
    """files[] одного модуля витрины -> {relpath: bytes}.

    Если витрина не скачалась (сеть/лимит) — честная причина вместо
    вводящего «нет files[]»: та ошибка означала бы «модуля нет на сайте»,
    а по факту бот витрину вообще не видел.
    """
    entry = (vitrina or {}).get(name) or {}
    if not entry.get("files") and vitrina_error:
        raise ValueError("витрина недоступна (%s) — установка только "
                         "через git" % str(vitrina_error)[:150])
    return extract_module_files(entry)


def extract_module_files(entry):
    """files[] витрины -> {relpath: bytes}. Отказ при truncated/binary."""
    files = entry.get("files") or []
    if not isinstance(files, list) or not files:
        raise ValueError("витрина: у модуля нет files[] "
                         "(установка только через git)")
    out = {}
    for f in files:
        if not isinstance(f, dict):
            continue
        rel = _safe_relpath(f.get("path") or "")
        if rel is None:
            raise ValueError("витрина: плохой путь %r" % (f.get("path") or ""))
        if f.get("truncated") or f.get("binary"):
            raise ValueError(
                "витрина: файл %s урезан (truncated/binary) — "
                "установка только через git" % rel)
        content = f.get("content")
        if content is None:
            raise ValueError("витрина: нет content у %s — только через git"
                             % rel)
        data = content.encode("utf-8") if isinstance(content, str) else bytes(content)
        if len(data) > 200 * 1024:
            raise ValueError("витрина: файл %s больше лимита" % rel)
        if isinstance(f.get("size"), int) and f["size"] != len(data):
            raise ValueError(
                "витрина: файл %s урезан на отдаче "
                "(заявлено %d, пришло %d) — integrity не сойдётся, "
                "нужна пересборка сайта" % (rel, f["size"], len(data)))
        out[rel] = data
    if not out:
        raise ValueError("витрина: пустой набор файлов")
    total = sum(len(v) for v in out.values())
    if total > 1024 * 1024:
        raise ValueError("витрина: модуль больше 1 МБ")
    return out


def check_integrity(entry, files):
    """Сверка integrity §1.1. Возвращает (ok, detail).

    Каноника — буквально из tools/gen_registry.py сайта (module_integrity):
    name\\0version\\0commit\\0 + на файл: path\\0size\\0 + первые 4096
    СИМВОЛОВ content (.encode, не байты!). size — заявленный витриной,
    порядок — как в files[] (не сортируем). Несовпадение — отказ
    в установке (fail-closed), не warning.
    """
    want = (entry or {}).get("integrity") or ""
    if not want.startswith("sha256:"):
        return False, "нет integrity в маркете"
    h = hashlib.sha256()
    h.update(("%s\x00%s\x00%s\x00" % (
        entry.get("name", ""), entry.get("version", ""),
        entry.get("commit") or entry.get("sha", ""))).encode())
    for rel, blob in files.items():
        h.update(("%s\x00%d\x00" % (rel, len(blob))).encode())
        try:
            text = blob.decode("utf-8")
        except ValueError:
            text = blob.decode("utf-8", errors="replace")
        h.update(text[:4096].encode(errors="replace"))
    got = "sha256:" + h.hexdigest()
    if got != want:
        return False, "integrity не сошлось (маркет %s.., факт %s..)" % (
            want[7:15], got[7:15])
    return True, "ok"


def download_release_file(base_url, rel_path):
    """Ассет релиза §1.3: GET {base}/releases/<rel_path>."""
    base = (base_url or "").rstrip("/")
    if not base:
        raise ValueError("не задан адрес сайта")
    rel = (rel_path or "").lstrip("/")
    if not rel or ".." in rel.split("/"):
        raise ValueError("плохой путь ассета")
    return _get_bytes(base + "/releases/" + rel)

"""modsafe.Module — охрана модулей: список, аудит, вкл/выкл, хеши.

Читает ctx.module_audit (считает loader при старте: сканер + сверка хешей
ДО импорта кода) и core.modstate (вкл/выкл, эталон хешей). Изменения
вкл/выкл и эталона применяются после рестарта бота (loader-time).
"""
import logging
import os

from core.base_module import BaseModule

logger = logging.getLogger(__name__)


class Module(BaseModule):
    def setup(self, ctx) -> None:
        self.ctx = ctx
        self.config = ctx.module_config("modsafe")
        from core import modstate as _ms
        self._ms = _ms
        ctx.api.register("modsafe", "list", self.api_list)
        ctx.api.register("modsafe", "log_errors", self.api_log_errors)
        ctx.api.register("modsafe", "audit_one", self.api_audit_one)
        ctx.api.register("modsafe", "set_enabled", self.api_set_enabled)
        ctx.api.register("modsafe", "rescan", self.api_rescan)
        ctx.api.register("modsafe", "stage", self.api_stage)
        ctx.api.register("modsafe", "staged", self.api_staged)
        ctx.api.register("modsafe", "install", self.api_install)
        ctx.api.register("modsafe", "update_staged", self.api_update_staged)
        ctx.api.register("modsafe", "market", self.api_market)
        ctx.api.register("modsafe", "market_install", self.api_market_install)
        ctx.api.register("modsafe", "update_check", self.api_update_check)
        ctx.api.register("modsafe", "update_module", self.api_update_module)
        ctx.api.register("modsafe", "skeleton_check", self.api_skeleton_check)
        ctx.api.register("modsafe", "update_skeleton",
                         self.api_update_skeleton)
        ctx.api.register("modsafe", "delete", self.api_delete)
        ctx.api.register("modsafe", "delete_preview", self.api_delete_preview)
        ctx.api.register("modsafe", "pack", self.api_pack)
        ctx.api.register("modsafe", "env_keys", self.api_env_keys)
        ctx.api.register("modsafe", "env_set", self.api_env_set)
        ctx.api.register("modsafe", "inspect", self.api_inspect)
        ctx.api.register("modsafe", "orphans", self.api_orphans)
        ctx.api.register("modsafe", "uninstall_orphans",
                         self.api_uninstall_orphans)
        ctx.api.register("modsafe", "cache_list", self.api_cache_list)
        ctx.api.register("modsafe", "cache_export", self.api_cache_export)
        ctx.api.register("modsafe", "cache_import", self.api_cache_import)
        ctx.api.register("modsafe", "pending", self.api_pending)
        ctx.api.register("modsafe", "restart", self.api_restart)
        ctx.api.register("modsafe", "state_confirm", self.state_confirm)
        # Сброс маркера «ждёт рестарта»: переживший рестарт код видит
        # чистый маркер — значит, всё применено.
        try:
            self._pending_clear()
        except Exception:  # noqa: BLE001
            pass

    # --- маркер «ждёт рестарта» ---

    def _pending_path(self):
        from core.paths import abs_data
        return abs_data(getattr(self.ctx, "data_dir", "."),
                        "cache/modsafe/pending_restart.json",
                        "cache/modsafe/pending_restart.json")

    def _pending_load(self):
        import json as _json
        try:
            with open(self._pending_path(), "r",
                      encoding="utf-8") as f:
                d = _json.load(f)
            return d if isinstance(d, dict) else {}
        except Exception:  # noqa: BLE001
            return {}

    def _pending_save(self, d):
        import json as _json
        import time as _time
        p = self._pending_path()
        try:
            os.makedirs(os.path.dirname(p), exist_ok=True)
            d = dict(d or {})
            d["at"] = _time.strftime("%Y-%m-%d %H:%M:%S")
            with open(p, "w", encoding="utf-8") as f:
                _json.dump(d, f, ensure_ascii=False, indent=2)
        except Exception:  # noqa: BLE001
            pass

    def _pending_add(self, kind, name, detail=""):
        d = self._pending_load()
        items = d.get("items") or []
        items = [i for i in items
                 if not (i.get("kind") == kind and i.get("name") == name)]
        items.append({"kind": kind, "name": name, "detail": detail or ""})
        d["items"] = items
        self._pending_save(d)

    def _pending_clear(self):
        try:
            os.remove(self._pending_path())
        except OSError:  # noqa: BLE001
            pass

    def api_pending(self):
        """Что обновлено, но ещё не применено (ждёт рестарта)."""
        d = self._pending_load()
        return {"ok": True, "items": d.get("items") or [],
                "at": d.get("at") or ""}

    def api_restart(self, actor=''):
        """Перезапуск бота из кабинета (право modules_manage)."""
        self._audit_rec('service.restart', 'bot', actor or '?')
        logger.info("modsafe: рестарт из кабинета (%s)", actor or "?")
        try:
            from core import restart as _restart
            _restart.restart_self()
        except Exception as e:  # noqa: BLE001
            raise ValueError("рестарт не удался: %s" % str(e)[:200])
        return {"ok": True}

    # --- версия скелета ---

    def _skeleton_version_path(self):
        from core.paths import abs_data
        return abs_data(getattr(self.ctx, "data_dir", "."),
                        "cache/modsafe/skeleton_version.json",
                        "cache/modsafe/skeleton_version.json")

    def _skeleton_local(self):
        import json as _json
        try:
            with open(self._skeleton_version_path(), "r",
                      encoding="utf-8") as f:
                d = _json.load(f)
            if isinstance(d, dict) and d.get("tag"):
                return d
        except Exception:  # noqa: BLE001
            pass
        return {"tag": "", "version": ""}

    # --- helpers ---

    def _audit(self):
        return getattr(self.ctx, "module_audit", {}) or {}

    def _mod_dir(self, name):
        base = os.path.join(getattr(self.ctx, "modules_dir", ""), name or "")
        if os.path.isfile(os.path.join(base, "manifest.yaml")) or \
                os.path.isfile(os.path.join(base, "manifest.json")):
            return base
        return ""

    def _any_dir(self, name):
        """Каталог модуля или транспорта (для inspect)."""
        d = self._mod_dir(name)
        if d:
            return d, False
        tbase = os.path.join(
            getattr(self.ctx, "transports_dir", ""), name or "")
        if os.path.isfile(os.path.join(tbase, "manifest.yaml")):
            return tbase, True
        return "", False

    def api_inspect(self, name):
        """Полный паспорт модуля/транспорта: манифест, API, права, вклады
        (роуты, сервисы, панели, nav, TG-кнопки, WS), README, файлы, health,
        аудит. Право modules_manage. Всё best-effort: отсутствующее — пусто.
        """
        import importlib as _il
        name = (name or "").strip()
        if not name or any(c not in
                           "abcdefghijklmnopqrstuvwxyz0123456789-_"
                           for c in name.lower()):
            raise ValueError("плохое имя: %s" % name)
        out = {"name": name}
        mod_dir, is_transport = self._any_dir(name)
        if not mod_dir:
            raise ValueError("модуль %s неизвестен" % name)
        out["transport"] = is_transport

        manifest = {}
        for fn in ("manifest.yaml", "manifest.json"):
            fp = os.path.join(mod_dir, fn)
            if os.path.isfile(fp):
                try:
                    if fn.endswith(".yaml"):
                        import yaml as _yaml
                        manifest = _yaml.safe_load(
                            open(fp, encoding="utf-8")) or {}
                    else:
                        import json as _json
                        manifest = _json.load(
                            open(fp, encoding="utf-8"))
                except Exception:  # noqa: BLE001
                    manifest = {}
                break
        out["manifest"] = manifest
        try:
            audit_all = self._audit()
            out["required_by"] = sorted(
                n for n, rec in audit_all.items()
                if name in ((rec or {}).get("requires") or []))
        except Exception:  # noqa: BLE001
            out["required_by"] = []

        loaded = None
        try:
            for m in self.ctx.loaded_modules or []:
                if m.name == name:
                    loaded = m
                    break
        except Exception:  # noqa: BLE001
            loaded = None
        out["loaded"] = loaded is not None
        if loaded is not None:
            try:
                out["health"] = loaded.health()
            except Exception as e:  # noqa: BLE001
                out["health"] = {"ok": False,
                                 "error": "%s: %s" % (type(e).__name__, e)}

        try:
            described = (self.ctx.api.describe() or {}).get(name) or []
        except Exception:  # noqa: BLE001
            described = []
        out["api_live"] = described

        try:
            rights = [r for r in self.ctx.rights.list_rights()
                      if (r.get("module") or "") == name]
        except Exception:  # noqa: BLE001
            rights = []
        out["rights"] = rights
        try:
            out["entry_defaults"] = self.ctx.rights.entry_defaults(name)
        except Exception:  # noqa: BLE001
            out["entry_defaults"] = {}

        ui = {}
        try:
            mod = _il.import_module("botmod_%s.ui_web" % name)
            for attr in ("ROUTES", "SERVICE", "PANELS", "NAV"):
                raw = getattr(mod, attr, None)
                if raw is None:
                    continue
                items = raw if isinstance(raw, list) else [raw]
                ui[attr.lower()] = [_safe(o) for o in items if o]
        except Exception:  # noqa: BLE001
            pass
        out["ui_web"] = ui
        try:
            tg = _il.import_module("botmod_%s.ui_tg" % name)
            menu = getattr(tg, "TG_MENU", None)
            items = menu if isinstance(menu, list) else [menu]
            out["tg_menu"] = [
                {"id": i.get("id"), "title": i.get("title"),
                 "right": i.get("right")}
                for i in items if isinstance(i, dict)]
        except Exception:  # noqa: BLE001
            out["tg_menu"] = []
        try:
            ws = _il.import_module("botmod_%s.ui_ws" % name)
            out["ws"] = {"path": getattr(ws, "WS_PATH", "") or "",
                         "right": getattr(ws, "RIGHT", "") or ""}
        except Exception:  # noqa: BLE001
            out["ws"] = {}

        readme = ""
        for fn in ("README.md", "README"):
            fp = os.path.join(mod_dir, fn)
            if os.path.isfile(fp):
                try:
                    with open(fp, "r", encoding="utf-8",
                              errors="replace") as f:
                        readme = f.read(30000)
                except Exception:  # noqa: BLE001
                    readme = ""
                break
        out["readme"] = readme

        a = self._audit().get(name) or {}
        out["enabled"] = True if name == "modsafe" else bool(a.get("enabled"))
        out["audit_summary"] = a.get("summary") or {}
        out["hashes"] = a.get("hashes") or {}
        out["files"] = []
        try:
            from core import modstate as _ms
            rec = (_ms.load(
                getattr(self.ctx, "data_dir", ".")) or {}).get(
                    "modules", {}).get(name) or {}
            out["files"] = sorted((rec.get("hashes") or {}).keys())
        except Exception:  # noqa: BLE001
            pass
        return out



    # --- api ---

    def _corrupt_info(self) -> dict:
        """Метаданные повреждённого modules_state.json ({} — чисто).

        Маркер выставляет loader при старте; если повреждение появилось уже
        после старта (маркера ещё нет), load() фиксирует его сам: копия
        .corrupt-<ts> + маркер. Иначе кабинет показал бы «чисто».
        """
        data_dir = getattr(self.ctx, "data_dir", ".")
        try:
            info = self._ms.corrupt_info(data_dir) or {}
            if not info and os.path.exists(
                    os.path.join(data_dir, self._ms.STATE_FILE)):
                self._ms.load(data_dir)
                info = self._ms.corrupt_info(data_dir) or {}
        except Exception:  # noqa: BLE001
            return {}
        return info

    def state_confirm(self, actor=""):
        """Подтвердить повреждённое состояние (право modules_manage).

        Снимает маркер повреждённости и пересевает эталоны хешей текущими
        файлами (после того как админ их увидел). Копии битого файла
        (modules_state.json.corrupt-<ts>) остаются на диске.
        """
        data_dir = getattr(self.ctx, "data_dir", ".")
        info = self._corrupt_info()
        if not info:
            raise ValueError("повреждённого состояния нет — подтверждать нечего")
        logger.error("modsafe: подтверждение повреждённого состояния (%s): %s "
                     "(sha256 %s, размер %s)", actor or "?", info.get("path")
                     or "", info.get("sha256") or "", info.get("size") or "")
        if not self._ms.clear_corrupt(data_dir, actor or "?"):
            raise ValueError("маркер повреждённости не снят — см. журнал")
        # Текущее состояние (known-записи loader'а) фиксируем на диске,
        # иначе пустой слепок соблазнит ensure() включить всё массово.
        try:
            self._ms.save(data_dir, getattr(self.ctx, "modstate", None)
                          or {"version": self._ms.VERSION, "modules": {}})
        except OSError as e:  # noqa: BLE001
            raise ValueError("не удалось записать подтверждённое состояние: %s"
                             % str(e)[:200])
        res = self.api_rescan()
        try:
            ctx_c = getattr(self.ctx, "modstate_corrupt", None)
            if ctx_c is not None:
                self.ctx.modstate_corrupt = None
        except Exception:  # noqa: BLE001
            pass
        self._audit_rec("modules.state_confirm", "modules_state.json",
                        actor or "?", str(info.get("path") or ""))
        self._pending_add("state", "modules_state.json", "confirmed")
        return {"ok": True, "confirmed": True, "corrupt": None,
                "rescanned": res.get("rescanned") or [],
                "apply": "restart",
                "note": "Эталон хешей засеван текущими файлами; флаги "
                        "вкл/выкл подтверждённых модулей — включать вручную."}

    def api_list(self):
        """Все известные модули/транспорты со статусом (право modules_manage)."""
        audit = self._audit()
        corrupt = self._corrupt_info()
        required_by: dict[str, list] = {}
        for nm, rec in audit.items():
            for dep in (rec or {}).get("requires") or []:
                required_by.setdefault(dep, []).append(nm)
        out = []
        for name in sorted(audit):
            a = audit[name]
            reqs = list(a.get("requires") or [])
            states = []
            for dep in reqs:
                dep_rec = audit.get(dep)
                if dep_rec is None:
                    st = "missing"
                elif dep_rec.get("transport") or dep_rec.get("enabled"):
                    st = "ok"
                else:
                    st = "off"
                states.append({"name": dep, "state": st})
            h = a.get("hashes") or {}
            out.append({
                "name": name,
                "version": a.get("version") or "",
                "description": a.get("description") or "",
                "category": a.get("category") or "extra",
                # modsafe грузится всегда (инфраструктура) — показываем факт
                "enabled": True if name == "modsafe" else bool(a.get("enabled")),
                "transport": bool(a.get("transport")),
                "requires": states,
                "required_by": sorted(required_by.get(name) or []),
                "summary": a.get("summary") or {},
                "hashes": h,
                # эталон не подтверждён (состояние было повреждено, seeding
                # пропущен) — детект подмены включится после state_confirm
                "baseline_unverified": bool(corrupt)
                and not h.get("baseline_seeded", True),
            })
        return {"modules": out,
                "skipped": self._skipped_problems(),
                "corrupt": corrupt or {},
                "note": ("Состояние модулей повреждено: не-core модули "
                         "выключены, эталон хешей не подтверждён — "
                         "подтвердите в кабинете (modsafe.state_confirm)"
                         if corrupt else
                         "Вкл/выкл применяется после рестарта бота.")}

    def _skipped_problems(self):
        """Не загрузившиеся модули с причинами (без намеренно выключенных).

        ctx.skipped_modules ставит loader: там и ошибки (нет зависимостей,
        цикл, падение setup), и штатно выключенные
        («отключён (modsafe...)») — последние здесь отсекаем, они и так
        видны в таблице как «выкл».
        """
        try:
            skipped = getattr(self.ctx, "skipped_modules", None) or {}
        except Exception:  # noqa: BLE001
            return []
        out = []
        for name in sorted(skipped):
            reason = str(skipped.get(name) or "")
            if not reason or reason.startswith("отключён (modsafe"):
                continue
            out.append({"name": name, "reason": reason[:300]})
        return out

    def api_log_errors(self):
        """Ошибки/варнинги в журналах с момента запуска (право modules_manage).

        Источник — data/logs/bot.log (+ ротации), фильтр по времени записи
        >= booted_at. Группировка по модулю (botmod_<name> -> <name>).
        Подсказка для кабинета: детали — в syslog (Логи бота).
        """
        from core import logparse as _lp
        data_dir = getattr(self.ctx, "data_dir", ".")
        booted = 0.0
        try:
            booted = float(getattr(self.ctx, "booted_at", 0) or 0)
        except (TypeError, ValueError):
            booted = 0.0
        recs = _lp.read_records(data_dir)
        per: dict[str, dict] = {}
        total_w = total_e = 0
        for r in recs:
            if booted and r.get("ts", 0) < booted:
                continue
            lvl = r.get("level") or ""
            if lvl not in ("WARNING", "ERROR", "CRITICAL"):
                continue
            name = r.get("module") or "?"
            slot = per.setdefault(name, {"warning": 0, "error": 0})
            if lvl == "WARNING":
                slot["warning"] += 1
                total_w += 1
            else:
                slot["error"] += 1
                total_e += 1
        boot_s = ""
        if booted:
            try:
                import datetime as _dt
                boot_s = _dt.datetime.fromtimestamp(booted).strftime(
                    "%Y-%m-%d %H:%M:%S")
            except Exception:  # noqa: BLE001
                boot_s = ""
        files = [p.split("/")[-1] for p in _lp.log_files(data_dir)]
        return {"ok": True, "booted_at": booted, "booted_at_str": boot_s,
                "since_boot": bool(booted),
                "per_module": per, "total_warning": total_w,
                "total_error": total_e, "files": files,
                "note": "Детали — в syslog: Логи бота → ⚠ Варнинги и ошибки."}

    def api_audit_one(self, name):
        """Находки одного модуля (право modules_manage)."""
        a = self._audit().get(name or "")
        if a is None:
            raise ValueError("Модуль %s неизвестен" % name)
        return {"name": name, "summary": a.get("summary") or {},
                "findings": a.get("findings") or []}

    def _audit_rec(self, action, target, actor="", detail="", ok=True):
        try:
            self.ctx.api.call('audit.record', actor or '?', action,
                              target, detail, ok)
        except Exception:  # noqa: BLE001
            pass

    def api_set_enabled(self, name, enabled, actor=''):
        """Вкл/выкл модуля (право modules_manage). Себя выключить нельзя.

        Возвращает warnings: что сломается после рестарта (при включении —
        выключенные/отсутствующие зависимости; при выключении — включённые
        модули, зависящие от этого). Кабинет показывает их confirm до
        применения; API-вызывающим — после (флаг уже переключён).
        """
        if (name or "") == "modsafe" and not enabled:
            raise ValueError("modsafe нельзя выключить (иначе нечем включить)")
        warnings = self._toggle_warnings(name, bool(enabled))
        ok, err = self._ms.set_enabled(
            getattr(self.ctx, "data_dir", "."), name, bool(enabled))
        if not ok:
            raise ValueError(err)
        # освежаем кэш аудита для UI
        try:
            self._audit()[name]["enabled"] = bool(enabled)
        except Exception:  # noqa: BLE001
            pass
        logger.info("modsafe: модуль %s -> %s",
                    name, "enabled" if enabled else "disabled")
        self._audit_rec('modules.enable' if enabled else 'modules.disable',
                        name, actor or '?')
        self._pending_add("toggle", name,
                          "enabled" if enabled else "disabled")
        return {"ok": True, "name": name, "enabled": bool(enabled),
                "apply": "restart", "warnings": warnings}

    def _toggle_warnings(self, name, enabled) -> list:
        """Предупреждения о зависимостях до переключения (best-effort)."""
        try:
            audit = self._audit() or {}
        except Exception:  # noqa: BLE001
            return []
        out = []
        rec = audit.get(name or "") or {}
        if enabled:
            for dep in (rec.get("requires") or []):
                dep_rec = audit.get(dep)
                if dep_rec is None:
                    out.append("зависимости нет среди известных: %s — "
                               "%s не загрузится после рестарта"
                               % (dep, name))
                elif not (dep_rec.get("transport")
                          or dep_rec.get("enabled")):
                    out.append("зависимость выключена: %s — включите её "
                               "тоже, иначе %s не загрузится после рестарта"
                               % (dep, name))
            try:
                reason = (getattr(self.ctx, "skipped_modules", None)
                          or {}).get(name or "")
            except Exception:  # noqa: BLE001
                reason = ""
            if reason and not str(reason).startswith("отключён (modsafe"):
                out.append("при старте модуль был пропущен: %s" % reason)
        else:
            for nm, r in audit.items():
                if nm == name:
                    continue
                if name in ((r or {}).get("requires") or []) and \
                        bool((r or {}).get("enabled")) and \
                        not bool((r or {}).get("transport")):
                    out.append("от %s зависит включённый %s — после "
                               "рестарта он не загрузится (нет зависимостей)"
                               % (name, nm))
        return out[:10]

    def api_rescan(self, name=None):
        """Пересчитать эталон хешей (и свежий скан) модуля или всех.

        Право modules_manage. Явное действие — фиксирует текущие файлы
        как доверенные (после проверки находок).
        """
        from core import audit as _audit
        data_dir = getattr(self.ctx, "data_dir", ".")
        names = [name] if name else sorted(self._audit())
        done = []
        for n in names:
            d = self._mod_dir(n)
            if not d and not self._audit().get(n, {}).get("transport"):
                continue
            if not d:
                # транспорт: ищем каталог в transports_dir
                tbase = os.path.join(
                    getattr(self.ctx, "transports_dir", ""), n or "")
                if os.path.isdir(tbase):
                    d = tbase
                else:
                    continue
            hashes = _audit.hash_module(d)
            self._ms.update_hashes(data_dir, n, hashes)
            try:
                self._audit()[n]["hashes"] = {
                    "ok": True, "changed": [], "new": [], "missing": [],
                    "baseline_at": "now", "baseline_seeded": True,
                    "unverified": False}
                self._audit()[n]["baseline_unverified"] = False
            except Exception:  # noqa: BLE001
                pass
            done.append(n)
        logger.info("modsafe: эталон хешей обновлён: %s", ", ".join(done))
        return {"ok": True, "rescanned": done}

    def api_stage(self, filename, content_type, body):
        """Принять архив модуля в стейджинг + аудит (право modules_manage)."""
        from . import installer as _inst
        try:
            return _inst.stage_upload(
                getattr(self.ctx, "data_dir", "."),
                getattr(self.ctx, "modules_dir", "modules"),
                self.config, filename, body)
        except ValueError as e:
            raise ValueError(str(e))

    def api_staged(self):
        """Список стейджинга (право modules_manage)."""
        from . import installer as _inst
        return {"staged": _inst.list_staged(getattr(self.ctx, "data_dir", "."))}

    def api_install(self, name, actor=''):
        """Установить из стейджинга (выключенным; право modules_manage)."""
        from . import installer as _inst
        try:
            out = _inst.install_staged(
                getattr(self.ctx, "data_dir", "."),
                getattr(self.ctx, "modules_dir", "modules"), name)
            self._audit_rec('modules.install', name, actor)
            return out
        except ValueError as e:
            raise ValueError(str(e))

    def _market_base(self):
        try:
            base = (self.config.get("MODSAFE_MARKET_URL", "") or "").strip()
        except Exception:  # noqa: BLE001
            base = ""
        return base.rstrip("/")

    def _market_cached(self, max_age=600, force=False):
        """Живой маркет + витрина с кешем (TTL, в памяти процесса)."""
        import time as _time
        from . import market_client as _mc
        base = self._market_base()
        if not base:
            raise ValueError("не задан MODSAFE_MARKET_URL "
                             "(data/env/modsafe.env)")
        now = _time.time()
        cache = getattr(self, "_market_cache", None) or {}
        if not force and cache.get("base") == base and now - cache.get("at", 0) < max_age:
            return cache["market"], cache["vitrina"], base
        market = _mc.fetch_market(base)
        try:
            vitrina = _mc.fetch_vitrina(base)
        except Exception as e:  # noqa: BLE001
            logger.warning("modsafe: витрина недоступна: %s", e)
            vitrina = {}
        self._market_cache = {"base": base, "at": now,
                               "market": market, "vitrina": vitrina}
        return market, vitrina, base

    def _local_version(self, name):
        a = self._audit().get(name or "") or {}
        return a.get("version") or ""

    @staticmethod
    def _vitrina_requires(entry):
        """requires модуля маркета из manifest.yaml витрины (best-effort).

        Витрина уже загружена (_market_cached) — лишних запросов нет.
        Нет манифеста/битый — пусто, не ошибка.
        """
        try:
            files = (entry or {}).get("files") or []
            for f in files:
                if not isinstance(f, dict):
                    continue
                path = str(f.get("path") or "")
                if not path.endswith(("manifest.yaml", "manifest.yml",
                                      "manifest.json")):
                    continue
                content = f.get("content")
                if not isinstance(content, str) or not content.strip():
                    continue
                if len(content) > 20000:
                    continue
                if path.endswith(".json"):
                    import json as _json
                    man = _json.loads(content)
                else:
                    import yaml as _yaml
                    man = _yaml.safe_load(content)
                if isinstance(man, dict):
                    reqs = man.get("requires") or []
                    return [str(r) for r in reqs if r]
                return []
        except Exception:  # noqa: BLE001
            pass
        return []

    def _module_files(self, vitrina, name):
        """files[] одного модуля -> (files, ventry для commit).

        Берём из кеша витрины; нет — точечный ?q=name (сотни КБ),
        а не вся витрина 11МБ. Честная ошибка, если витрина недоступна.
        """
        from . import market_client as _mc
        ventry = (vitrina or {}).get(name) or {}
        if ventry.get("files"):
            return _mc.extract_module_files(ventry), ventry
        try:
            single = _mc.fetch_vitrina(self._market_base(), query=name)
        except Exception as e:  # noqa: BLE001
            raise ValueError("витрина недоступна (%s) — установка "
                             "только через git" % str(e)[:150])
        return _mc.vitrina_files(single, name, ""), single.get(name) or {}

    def api_market(self, refresh=False):
        """Маркет сайта: живые записи + локальные версии (право modules_manage)."""
        try:
            market, vitrina, base = self._market_cached(force=bool(refresh))
        except ValueError as e:
            return {"ok": False, "stub": False, "url": self._market_base(),
                    "modules": [],
                    "error": str(e)}
        vitrina = vitrina or {}
        out = []
        for m in market:
            local = self._local_version(m["name"])
            out.append({**m, "installed": local,
                        "update": bool(local) and local != m["version"],
                        "requires": self._vitrina_requires(
                            vitrina.get(m["name"]) or {})})
        return {"ok": True, "stub": False, "url": base, "modules": out,
                "note": "update=true — доступна новая версия."}

    def api_market_install(self, name, actor=''):
        """Установить модуль С САЙТА (право modules_manage).

        Маркет -> витрина files[] (точечно ?q=) -> сверка integrity ->
        стейджинг -> install (выключенным) -> pip-зависимости. Env НЕ тянется
        (секреты — через modsafe.env_set / кабинет).
        """
        from . import market_client as _mc
        from . import installer as _inst
        name = (name or "").strip()
        if not _inst._NAME_RX.match(name):
            raise ValueError("недопустимое имя")
        market, vitrina, _base = self._market_cached()
        entry = next((m for m in market if m["name"] == name), None)
        if entry is None:
            raise ValueError("модуля %s нет в маркете" % name)
        if self._mod_dir(name):
            raise ValueError("модуль %s уже установлен "
                             "(обновление — modsafe.update_module)" % name)
        files, ventry = self._module_files(vitrina, name)
        ok, detail = _mc.check_integrity(
            {**entry, "commit": ventry.get("commit")}, files)
        if not ok:
            raise ValueError(detail)
        data_dir = getattr(self.ctx, "data_dir", ".")
        dest = _inst.stage_files(data_dir, name, files)
        try:
            out = _inst.install_staged(
                data_dir, getattr(self.ctx, "modules_dir", "modules"),
                name)
        except Exception:
            raise
        self._pip_install_module(name)
        try:
            self._audit()[name] = self._audit().get(name) or {}
        except Exception:  # noqa: BLE001
            pass
        self._pending_add("install", name,
                          entry.get("version") or "")
        self._audit_rec('modules.market_install', "%s v%s" % (
            name, entry.get("version") or "?"), actor or '?')
        logger.info("modsafe: маркет-установка %s v%s из %s",
                    name, entry.get("version"), dest)
        return {**out, "version": entry.get("version") or "",
                "source": "market"}

    def _pip_install_module(self, name):
        """pip install -r requirements.txt модуля тем же интерпретатором."""
        import subprocess as _sp
        import sys as _sys
        req = os.path.join(getattr(self.ctx, "modules_dir", "modules"),
                           name or "", "requirements.txt")
        if not os.path.isfile(req):
            return {"ok": True, "skipped": True}
        try:
            r = _sp.run([_sys.executable, "-m", "pip", "install", "-q",
                         "-r", req],
                        capture_output=True, text=True, timeout=600)
        except Exception as e:  # noqa: BLE001
            logger.warning("modsafe: pip %s: %s", name, e)
            return {"ok": False, "error": "%s: %s" % (
                type(e).__name__, str(e)[:200])}
        if r.returncode != 0:
            tail = ((r.stdout or "") + "\n" + (r.stderr or "")).strip()
            tail = "\n".join(tail.splitlines()[-5:])
            logger.warning("modsafe: pip %s rc=%d: %s", name,
                           r.returncode, tail[:500])
            return {"ok": False, "error": "pip rc=%d: %s" % (
                r.returncode, tail[:300])}
        return {"ok": True}

    def api_update_check(self, refresh=False):
        """Сверка локальных версий с маркетом (право modules_manage).

        Возвращает [{name, installed, market, update}]; nightly-CI сайта
        может дёргать это для бейджа «X → Y».
        """
        try:
            market, _vitrina, base = self._market_cached(force=bool(refresh))
        except ValueError as e:
            return {"ok": False, "url": self._market_base(), "updates": [],
                    "error": str(e)}
        wanted = {m["name"] for m in market}
        out = []
        for name in sorted(self._audit()):
            if self._mod_dir(name) and name in wanted:
                local = self._local_version(name)
                remote = next(m["version"] for m in market
                              if m["name"] == name)
                if local and local != remote:
                    out.append({"name": name, "installed": local,
                                "market": remote, "update": True})
        return {"ok": True, "url": base, "updates": out}

    def _backup_dir(self, kind, name, version):
        """Каталог бекапа data/backups/<kind>/<name>-<ver>-<ts>/."""
        import time as _time
        stamp = _time.strftime("%Y%m%d_%H%M%S")
        dest = os.path.join(
            getattr(self.ctx, "data_dir", "."),
            "backups", kind,
            "%s-%s-%s" % (name, version or "noversion", stamp))
        os.makedirs(os.path.dirname(dest), exist_ok=True)
        return dest

    def api_update_module(self, name, actor='', force=False):
        """Обновить установленный модуль с сайта (право modules_manage).

        Бекап кода в data/backups/modules/, замена файлами с сайта
        (integrity fail-closed), сохранение флага enabled, rescan хешей,
        pip-зависимости. Env НЕ трогается. Применяется сразу, но новый
        код подхватывается после рестарта (loader-time импорт).
        Откат: остановить бота, вернуть каталог из бекапа, рестарт.
        modsafe обновляет и сам себя: замена идёт в новый каталог,
        затем атомарный swap — исполняемый код не трогается в процессе.
        """
        from . import market_client as _mc
        from . import installer as _inst
        import shutil as _sh
        name = (name or "").strip()
        if not _inst._NAME_RX.match(name):
            raise ValueError("недопустимое имя")
        mod_dir = self._mod_dir(name)
        if not mod_dir:
            raise ValueError("модуль %s не установлен "
                             "(установка — modsafe.market_install)" % name)
        market, vitrina, _base = self._market_cached()
        entry = next((m for m in market if m["name"] == name), None)
        if entry is None:
            raise ValueError("модуля %s нет в маркете" % name)
        local = self._local_version(name)
        if local and local == entry.get("version") and not force:
            return {"ok": True, "name": name, "version": local,
                    "updated": False, "note": "уже актуально"}
        files, ventry = self._module_files(vitrina, name)
        ok, detail = _mc.check_integrity(
            {**entry, "commit": ventry.get("commit")}, files)
        if not ok:
            raise ValueError(detail)
        backup = self._backup_dir("modules", name, local)
        _sh.copytree(mod_dir, backup, symlinks=False,
                     ignore=_sh.ignore_patterns("__pycache__"))
        was_enabled = bool((self._audit().get(name) or {}).get("enabled"))
        if name == "modsafe":
            # Самообновление: пишем в новый каталог, затем атомарно
            # меняем местами (исполняемый код не трогаем в процессе).
            new_dir = mod_dir + ".__new__"
            _sh.rmtree(new_dir, ignore_errors=True)
            try:
                _inst.write_files(new_dir, files)
            except Exception as e:  # noqa: BLE001
                _sh.rmtree(new_dir, ignore_errors=True)
                raise ValueError("запись не удалась: %s" % str(e)[:200])
            dead_dir = mod_dir + ".__old__"
            _sh.rmtree(dead_dir, ignore_errors=True)
            try:
                os.rename(mod_dir, dead_dir)
                os.rename(new_dir, mod_dir)
            except Exception as e:  # noqa: BLE001
                try:
                    if os.path.isdir(dead_dir) and not os.path.isdir(mod_dir):
                        os.rename(dead_dir, mod_dir)
                except Exception:  # noqa: BLE001
                    pass
                _sh.rmtree(new_dir, ignore_errors=True)
                raise ValueError("swap не удался, откачено: %s"
                                 % str(e)[:200])
            _sh.rmtree(dead_dir, ignore_errors=True)
        else:
            _sh.rmtree(mod_dir)
            try:
                _inst.write_files(mod_dir, files)
            except Exception as e:  # noqa: BLE001
                _sh.rmtree(mod_dir, ignore_errors=True)
                _sh.copytree(backup, mod_dir)
                raise ValueError("запись не удалась, откачено из бекапа: %s"
                                 % str(e)[:200])
        self.api_rescan(name)
        try:
            from core import modstate as _ms
            _ms.set_enabled(getattr(self.ctx, "data_dir", "."),
                            name, was_enabled)
        except Exception:  # noqa: BLE001
            pass
        self._pip_install_module(name)
        self._audit_rec('modules.update', "%s %s->%s" % (
            name, local or "?", entry.get("version") or "?"), actor or '?')
        logger.info("modsafe: обновлён %s %s->%s (бекап %s)",
                    name, local, entry.get("version"), backup)
        self._pending_add("module", name, "%s -> %s" % (
            local or "?", entry.get("version") or "?"))
        return {"ok": True, "name": name, "version": entry.get("version"),
                "updated": True, "backup": backup,
                "enabled": was_enabled, "apply": "restart"}

    def api_update_staged(self, name, actor='', force=False):
        """Обновить установленный модуль из стейджинга (право modules_manage).

        Как update_module (маркет), но источник — загруженный архив
        (POST /upload уже проверил архив и прошёл аудит). Порядок:
        installer.update_from_staging (сверка версий, бекап, swap),
        rescan хешей, сохранение флага enabled, pip-зависимости,
        аудит-запись и маркер pending_restart. Равная версия без force —
        отказ; даунгрейд разрешён (флаг downgrade — UI предупредит).
        Integrity не требуется: архив собственный, не из витрины.
        """
        from . import installer as _inst
        name = (name or "").strip()
        if not _inst._NAME_RX.match(name):
            raise ValueError("недопустимое имя")
        data_dir = getattr(self.ctx, "data_dir", ".")
        try:
            out = _inst.update_from_staging(
                data_dir, getattr(self.ctx, "modules_dir", "modules"),
                name, force)
        except ValueError as e:
            raise ValueError(str(e))
        self.api_rescan(name)
        was_enabled = bool((self._audit().get(name) or {}).get("enabled"))
        try:
            from core import modstate as _ms
            _ms.set_enabled(data_dir, name, was_enabled)
        except Exception:  # noqa: BLE001
            pass
        self._pip_install_module(name)
        self._audit_rec('modules.update', "%s %s->%s (архив)" % (
            name, out.get("old_version") or "?", out.get("version") or "?"),
            actor or '?')
        self._pending_add("module", name, "%s -> %s (архив)" % (
            out.get("old_version") or "?", out.get("version") or "?"))
        return {**out, "enabled": was_enabled}

    def api_skeleton_check(self):
        """Свежая версия скелета на сайте (право modules_manage)."""
        from . import market_client as _mc
        try:
            base = self._market_base()
            rels = _mc.fetch_releases(base)
        except ValueError as e:
            return {"ok": False, "url": self._market_base(),
                    "error": str(e)}
        if not rels:
            return {"ok": True, "releases": []}
        latest = rels[-1]  # fetch_releases: новейший последний
        local = self._skeleton_local()
        update = bool(latest.get("tag")) and latest.get("tag") != local.get("tag")
        return {"ok": True, "url": base, "local": local, "update": update,
                "latest": {
            "version": latest.get("version") or "",
            "tag": latest.get("tag") or "",
            "name": latest.get("name") or "",
            "files": latest.get("files") or []},
            "releases": [{"version": r.get("version") or "",
                          "tag": r.get("tag") or ""} for r in rels]}

    def api_update_skeleton(self, actor='', force=False):
        """Обновить скелет (ядро) с сайта (право modules_manage).

        Трогает ТОЛЬКО {boot.py, core/, transports/, requirements.txt,
        scripts/}. Никогда: modules/, data/, .env. Старое — в
        data/backups/skeleton/. pip install -r requirements.txt.
        force: перезалить ту же версию (починить локальные правки ядра).
        """
        from . import market_client as _mc
        import shutil as _sh
        import tarfile as _tf
        import io as _io
        base = self._market_base()
        rels = _mc.fetch_releases(base)
        if not rels:
            raise ValueError("на сайте нет релизов скелета")
        latest = rels[-1]  # fetch_releases: новейший последний
        files = latest.get("files") or []
        if not files:
            raise ValueError("в релизе %s нет файлов" % latest.get("tag"))
        arc = next((f for f in files
                    if str(f.get("file") or "").endswith(
                        (".tar.gz", ".tgz"))), files[0])
        blob = _mc.download_release_file(base, arc.get("path") or "")
        if len(blob) != int(arc.get("size") or len(blob)):
            raise ValueError("размер архива не сошёлся "
                             "(ожидалось %s, пришло %s)"
                             % (arc.get("size"), len(blob)))
        root = os.path.dirname(os.path.abspath(
            getattr(self.ctx, "modules_dir", "modules")))
        tmp = os.path.join(getattr(self.ctx, "data_dir", "."),
                           "modstaging", "_skeleton")
        if os.path.isdir(tmp):
            _sh.rmtree(tmp)
        os.makedirs(tmp)
        try:
            with _tf.open(fileobj=_io.BytesIO(blob), mode="r:*") as tf:
                members = [m for m in tf.getmembers()
                           if m.isfile() and not m.issym()
                           and not m.islnk()]
                for m in members:
                    parts = [p for p in m.name.replace("\\", "/").split("/")
                             if p not in ("", ".")]
                    if (not parts or any(p == ".." for p in parts)
                            or parts[0].startswith(".")):
                        raise ValueError("плохой путь в архиве: %s" % m.name)
                tf.extractall(tmp, members=members)
        except (ValueError, _tf.TarError) as e:
            _sh.rmtree(tmp, ignore_errors=True)
            raise ValueError("битый архив скелета: %s" % str(e)[:200])
        top = [e for e in sorted(os.listdir(tmp))
               if os.path.isdir(os.path.join(tmp, e))]
        only_files = [e for e in sorted(os.listdir(tmp))
                      if os.path.isfile(os.path.join(tmp, e))]
        src = os.path.join(tmp, top[0]) if len(top) == 1 and not only_files \
            else tmp
        allow_files = {"boot.py", "requirements.txt"}
        allow_dirs = {"core", "transports", "scripts"}
        backup = self._backup_dir("skeleton", "skeleton",
                                  latest.get("version"))
        replaced = []
        try:
            for entry in sorted(os.listdir(src)):
                if entry in allow_files or entry in allow_dirs:
                    s = os.path.join(src, entry)
                    d = os.path.join(root, entry)
                    b = os.path.join(backup, entry)
                    if os.path.isdir(s):
                        if os.path.isdir(d):
                            _sh.copytree(d, b, symlinks=False,
                                         ignore=_sh.ignore_patterns(
                                             "__pycache__"))
                            _sh.rmtree(d)
                        _sh.copytree(s, d, symlinks=False,
                                     ignore=_sh.ignore_patterns("__pycache__"))
                    else:
                        if os.path.isfile(d):
                            os.makedirs(backup, exist_ok=True)
                            _sh.copy2(d, b)
                        _sh.copy2(s, d)
                    replaced.append(entry)
            if not replaced:
                raise ValueError("в архиве нет файлов скелета "
                                 "(boot.py/core/transports)")
        except ValueError:
            raise
        except Exception as e:  # noqa: BLE001
            raise ValueError("замена не удалась: %s" % str(e)[:200])
        finally:
            _sh.rmtree(tmp, ignore_errors=True)
        self._pip_install_module_files(
            os.path.join(root, "requirements.txt"))
        self.api_rescan()
        self._audit_rec('skeleton.update', latest.get("tag") or "?",
                        actor or '?')
        logger.info("modsafe: скелет обновлён до %s (бекап %s)",
                    latest.get("tag"), backup)
        import json as _json
        try:
            sp = self._skeleton_version_path()
            os.makedirs(os.path.dirname(sp), exist_ok=True)
            with open(sp, "w", encoding="utf-8") as f:
                _json.dump({"tag": latest.get("tag") or "",
                            "version": latest.get("version") or ""},
                           f, ensure_ascii=False, indent=2)
        except Exception:  # noqa: BLE001
            pass
        self._pending_add("skeleton", "skeleton",
                          latest.get("tag") or "")
        return {"ok": True, "version": latest.get("version") or "",
                "tag": latest.get("tag") or "", "replaced": replaced,
                "backup": backup, "apply": "restart"}

    def _pip_install_module_files(self, req_path):
        """pip install -r произвольного requirements (скелет)."""
        import subprocess as _sp
        import sys as _sys
        if not os.path.isfile(req_path):
            return {"ok": True, "skipped": True}
        try:
            r = _sp.run([_sys.executable, "-m", "pip", "install", "-q",
                         "-r", req_path],
                        capture_output=True, text=True, timeout=900)
        except Exception as e:  # noqa: BLE001
            return {"ok": False, "error": "%s: %s" % (
                type(e).__name__, str(e)[:200])}
        if r.returncode != 0:
            tail = ((r.stdout or "") + "\n" + (r.stderr or "")).strip()
            return {"ok": False, "error": "pip rc=%d: %s" % (
                r.returncode, "\n".join(tail.splitlines()[-5:])[:500])}
        return {"ok": True}

    def api_delete(self, name, actor='', purge_cache=False,
                   purge_env=False):
        """Удалить модуль (только выключенный; право modules_manage)."""
        from core import modstate as _ms
        if (name or "") in _ms.CORE_MODULES:
            raise ValueError("модуль %s системный (core), удалить нельзя" % name)
        from . import installer as _inst
        try:
            out = _inst.delete_module(
                getattr(self.ctx, "data_dir", "."),
                getattr(self.ctx, "modules_dir", "modules"), name,
                purge_cache=bool(purge_cache), purge_env=bool(purge_env))
            # Стартовый кеш аудита (ctx.module_audit) иначе продолжит
            # показывать удалённый модуль до рестарта: повторное удаление
            # падало бы «не найден», а маркет предлагал бы «обновить».
            try:
                audit = self._audit()
                if name in audit:
                    del audit[name]
            except Exception:  # noqa: BLE001
                pass
            self._audit_rec('modules.delete', name, actor,
                            "purge_cache=%s purge_env=%s" % (
                                bool(purge_cache), bool(purge_env)))
            return out
        except ValueError as e:
            raise ValueError(str(e))

    def api_delete_preview(self, name):
        """Что будет удалено: файлы кода/кеша/env с размерами (право modules_manage)."""
        from core import modstate as _ms
        if (name or "") in _ms.CORE_MODULES:
            raise ValueError("модуль %s системный (core), удалить нельзя" % name)
        mod_dir = self._mod_dir(name)
        if not mod_dir:
            raise ValueError("модуль %s не найден" % name)
        data_dir = getattr(self.ctx, "data_dir", ".")

        def _tree(root):
            out, total = [], 0
            for base, _dirs, files in os.walk(root):
                for f in sorted(files):
                    p = os.path.join(base, f)
                    try:
                        size = os.path.getsize(p)
                    except OSError:
                        continue
                    total += size
                    out.append({"path": os.path.relpath(p, root),
                                "size": size})
            return {"files": out, "total": total, "count": len(out)}

        code = _tree(mod_dir)
        cache_dir = os.path.join(data_dir, "cache", name)
        cache = _tree(cache_dir) if os.path.isdir(cache_dir) else \
            {"files": [], "total": 0, "count": 0}
        env_file = os.path.join(data_dir, "env", name + ".env")
        try:
            env_size = os.path.getsize(env_file)
            env = {"files": [{"path": name + ".env", "size": env_size}],
                   "total": env_size, "count": 1}
        except OSError:
            env = {"files": [], "total": 0, "count": 0}
        return {"ok": True, "name": name, "code": code, "cache": cache,
                "env": env}

    def api_pack(self, name, with_secrets=False):
        """Упаковать модуль в zip (b64; право modules_manage).

        with_secrets=true: плюс data/env/<name>.env (в архив как
        _env/<name>.env, при установке вернётся в data/env).
        """
        import base64 as _b64
        from . import installer as _inst
        try:
            filename, data = _inst.pack_module(
                getattr(self.ctx, "modules_dir", "modules"),
                name, bool(with_secrets), self._env_dir())
        except ValueError as e:
            raise ValueError(str(e))
        return {"ok": True, "filename": filename, "size": len(data),
                "with_secrets": bool(with_secrets),
                "b64": _b64.b64encode(data).decode("ascii")}

    def _env_dir(self):
        try:
            return os.path.join(self.ctx.data_dir, "env")
        except Exception:  # noqa: BLE001
            return None

    def api_env_keys(self, name):
        """Ключи секретов модуля без значений (право modules_manage)."""
        from . import installer as _inst
        try:
            return _inst.env_keys(
                getattr(self.ctx, "modules_dir", "modules"), name,
                self._env_dir())
        except ValueError as e:
            raise ValueError(str(e))

    def api_env_set(self, name, values, actor=''):
        """Сохранить секреты модуля (право modules_manage)."""
        from . import installer as _inst
        try:
            out = _inst.env_set(
                getattr(self.ctx, "modules_dir", "modules"),
                name, values or {}, self._env_dir())
            self._audit_rec('secrets.set', name, actor,
                            'keys=%s' % ','.join(out.get('changed') or []))
            return out
        except ValueError as e:
            raise ValueError(str(e))

    def api_orphans(self):
        """pip-пакеты, которых не требует никто (только показ; modules_manage)."""
        from . import installer as _inst
        import os as _os
        mods = getattr(self.ctx, "modules_dir", "modules")
        root = _os.path.dirname(_os.path.abspath(mods))
        return _inst.pip_orphans(mods, _os.path.join(root, "transports"),
                                 root)

    def api_uninstall_orphans(self, packages):
        """Удалить пакеты-сироты (перепроверка + modules_manage)."""
        from . import installer as _inst
        import os as _os
        mods = getattr(self.ctx, "modules_dir", "modules")
        root = _os.path.dirname(_os.path.abspath(mods))
        try:
            return _inst.pip_uninstall_orphans(
                mods, _os.path.join(root, "transports"), root,
                packages or [])
        except ValueError as e:
            raise ValueError(str(e))

    def health(self) -> dict:
        n = len(self._audit())
        crit = sum((a.get("summary") or {}).get("critical", 0)
                   for a in self._audit().values())
        return {"ok": True, "module": self.name,
                "known": n, "critical": crit}

    def api_cache_list(self, name):
        """Файлы кеша модуля data/cache/<name>/ (право modules_manage)."""
        from . import installer as _inst
        if not self._mod_dir(name):
            raise ValueError("модуль %s неизвестен" % name)
        return {"name": name,
                "files": _inst.cache_list(
                    getattr(self.ctx, "data_dir", "."), name)}

    def api_cache_export(self, name):
        """Скачать кеш модуля zip (b64; право modules_manage)."""
        import base64 as _b64
        from . import installer as _inst
        if not self._mod_dir(name):
            raise ValueError("модуль %s неизвестен" % name)
        try:
            filename, data = _inst.export_cache(
                getattr(self.ctx, "data_dir", "."), name)
        except ValueError as e:
            raise ValueError(str(e))
        return {"ok": True, "filename": filename, "size": len(data),
                "b64": _b64.b64encode(data).decode("ascii")}

    def api_cache_import(self, name, body_b64, actor=''):
        """Восстановить кеш модуля из zip (с бекапом; право modules_manage)."""
        import base64 as _b64
        from . import installer as _inst
        if not self._mod_dir(name):
            raise ValueError("модуль %s неизвестен" % name)
        try:
            body = _b64.b64decode(body_b64 or "", validate=True)
        except Exception:  # noqa: BLE001
            raise ValueError("битый base64")
        try:
            out = _inst.import_cache(
                getattr(self.ctx, "data_dir", "."), name, body)
        except ValueError as e:
            raise ValueError(str(e))
        self._audit_rec('modules.cache_import', name, actor or '?',
                        'files=%s' % out.get("files"))
        return out

def _safe(o):
    """JSON-безопасная копия вклада (tuples ROUTES -> lists)."""
    import json as _json
    try:
        return _json.loads(_json.dumps(o, ensure_ascii=False, default=str))
    except Exception:  # noqa: BLE001
        return {"repr": str(o)[:500]}