modsafe v0.4.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_]*$,
манифест с обязательными полями, отказ при существующем каталоге.
"""
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 _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)
    if os.path.exists(os.path.join(modules_dir, name)):
        shutil.rmtree(tmp)
        raise ValueError("модуль %s уже установлен" % name)
    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)",
                name, manifest.get("version"), summary)
    return {"ok": True, "name": name, "version": manifest.get("version") or "",
            "description": manifest.get("description") or "",
            "summary": summary, "findings": findings[:200],
            "staged": True}


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 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("недопустимое имя")
    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)
    # запись 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"}


#: Файлы, никогда не попадающие в архив без секретов.
_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
            if not with_secrets and (
                    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):
    """Упаковать модуль в zip. Возвращает (filename, bytes).

    with_secrets=False (дефолт): без .env/*.session — безопасно делиться.
    """
    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))
    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):
    """Удалить модуль: каталог + запись modstate + стейджинг.

    Только выключенные (сначала modsafe.enable=false + рестарт).
    modsafe и транспорты удалять нельзя.
    """
    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)
    logger.info("modsafe: удалён %s", name)
    return {"ok": True, "name": name}


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

"""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", "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", "market", self.api_market)
        ctx.api.register("modsafe", "delete", self.api_delete)
        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)

    # --- 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

        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 api_list(self):
        """Все известные модули/транспорты со статусом (право modules_manage)."""
        out = []
        for name in sorted(self._audit()):
            a = self._audit()[name]
            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")),
                "summary": a.get("summary") or {},
                "hashes": a.get("hashes") or {},
            })
        return {"modules": out,
                "note": "Вкл/выкл применяется после рестарта бота."}

    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). Себя выключить нельзя."""
        if (name or "") == "modsafe" and not enabled:
            raise ValueError("modsafe нельзя выключить (иначе нечем включить)")
        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 '?')
        return {"ok": True, "name": name, "enabled": bool(enabled),
                "apply": "restart"}

    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"}
            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 api_market(self):
        """Маркет модулей — ЗАГЛУШКА до запуска страницы."""
        url = ""
        try:
            url = (self.config.get("MODSAFE_MARKET_URL", "") or "").strip()
        except Exception:  # noqa: BLE001
            pass
        return {"ok": True, "stub": True, "url": url,
                "modules": [],
                "note": "Маркет ещё не запущен. Загрузка — архивом выше."}

    def api_delete(self, name, actor=''):
        """Удалить модуль (только выключенный; право 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)
            self._audit_rec('modules.delete', name, actor)
            return out
        except ValueError as e:
            raise ValueError(str(e))

    def api_pack(self, name, with_secrets=False):
        """Упаковать модуль в zip (b64; право modules_manage)."""
        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))
        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 _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]}

"""modsafe.ui_web — API кабинета + карточка «Модули и безопасность»."""
ROUTES = [
    ("GET", "/api/modsafe/modules", "modules", {"right": "modules_manage",
     "desc": "Модули: статус, аудит, хеши"}),
    ("POST", "/api/modsafe/enable", "enable", {"right": "modules_manage",
     "desc": "Вкл/выкл модуля {name, enabled} (применится после рестарта)"}),
    ("POST", "/api/modsafe/rescan", "rescan", {"right": "modules_manage",
     "desc": "Обновить эталон хешей {name?}"}),
    ("POST", "/api/modsafe/upload", "upload", {"right": "modules_manage",
     "desc": "Загрузить архив модуля (multipart file=*.zip) в стейджинг"}),
    ("GET", "/api/modsafe/staged", "staged", {"right": "modules_manage",
     "desc": "Стейджинг загруженных модулей"}),
    ("POST", "/api/modsafe/install", "install", {"right": "modules_manage",
     "desc": "Установить модуль из стейджинга {name} (выключенным)"}),
    ("GET", "/api/modsafe/market", "market", {"right": "modules_manage",
     "desc": "Маркет модулей (заглушка до запуска страницы)"}),
    ("POST", "/api/modsafe/delete", "delete", {"right": "modules_manage",
     "desc": "Удалить модуль {name} (только выключенный)"}),
    ("GET", "/api/modsafe/pack", "pack", {"right": "modules_manage",
     "desc": "Скачать модуль zip (?name, ?secrets=1)"}),
    ("GET", "/api/modsafe/env", "env_keys", {"right": "modules_manage",
     "desc": "Ключи секретов модуля (?name, без значений)"}),
    ("POST", "/api/modsafe/env", "env_set", {"right": "modules_manage",
     "desc": "Сохранить секреты модуля {name, values}"}),
    ("GET", "/api/modsafe/inspect", "inspect", {"right": "modules_manage",
     "desc": "Паспорт модуля (?name): манифест, API, права, вклады, README"}),
    ("GET", "/api/modsafe/orphans", "orphans", {"right": "modules_manage",
     "desc": "pip-сироты: пакеты, которых не требует никто (только показ)"}),
    ("POST", "/api/modsafe/orphans", "uninstall_orphans", {"right": "modules_manage",
     "desc": "Удалить пакеты-сироты {packages: [...]} (с перепроверкой)"}),
]

SERVICE = {"key": "modsafe", "icon": "🧩", "title": "Модули и безопасность",
           "right": "modules_manage", "phase": 1,
           "desc": "Вкл/выкл модулей, аудит кода, сверка хешей"}

PANELS = []

def _actor(req):
    sess = req.get('session') or {}
    return 'web:%s' % sess.get('uid') if sess.get('uid') else '?'



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

    if method == "modules":
        query = req.get("query") or {}
        if query.get("findings"):
            try:
                return ctx.api.call("modsafe.audit_one", query.get("findings"))
            except ValueError as e:
                return 400, {"ok": False, "error": str(e)}
        return ctx.api.call("modsafe.list")
    if method == "enable":
        body = req.get("body") or {}
        try:
            return ctx.api.call("modsafe.set_enabled",
                                body.get("name"), body.get("enabled"),
                                _actor(req))
        except ValueError as e:
            return 400, {"ok": False, "error": str(e)}
    if method == "rescan":
        body = req.get("body") or {}
        try:
            return ctx.api.call("modsafe.rescan", body.get("name"))
        except ValueError as e:
            return 400, {"ok": False, "error": str(e)}
    if method == "upload":
        files = (req.get("files") or {}).get("file") or []
        if not files:
            return 400, {"ok": False, "error": "Файл не передан (multipart file=*.zip)"}
        fi = files[0]
        try:
            return ctx.api.call("modsafe.stage", fi.get("filename") or "upload.zip",
                                fi.get("content_type") or "", fi.get("body") or b"")
        except ValueError as e:
            return 400, {"ok": False, "error": str(e)}
    if method == "staged":
        return ctx.api.call("modsafe.staged")
    if method == "install":
        body = req.get("body") or {}
        try:
            return ctx.api.call("modsafe.install", body.get("name"),
                                _actor(req))
        except ValueError as e:
            return 400, {"ok": False, "error": str(e)}
    if method == "market":
        return ctx.api.call("modsafe.market")
    if method == "delete":
        body = req.get("body") or {}
        try:
            return ctx.api.call("modsafe.delete", body.get("name"),
                                _actor(req))
        except ValueError as e:
            return 400, {"ok": False, "error": str(e)}
    if method == "pack":
        query = req.get("query") or {}
        try:
            return ctx.api.call("modsafe.pack", query.get("name"),
                                str(query.get("secrets") or "") == "1")
        except ValueError as e:
            return 400, {"ok": False, "error": str(e)}
    if method == "env_keys":
        query = req.get("query") or {}
        try:
            return ctx.api.call("modsafe.env_keys", query.get("name"))
        except ValueError as e:
            return 400, {"ok": False, "error": str(e)}
    if method == "env_set":
        body = req.get("body") or {}
        if not isinstance(body.get("values"), dict):
            return 400, {"ok": False, "error": "values должен быть объектом"}
        try:
            return ctx.api.call("modsafe.env_set",
                                body.get("name"), body.get("values"),
                                _actor(req))
        except ValueError as e:
            return 400, {"ok": False, "error": str(e)}
    if method == "inspect":
        query = req.get("query") or {}
        try:
            return ctx.api.call("modsafe.inspect", query.get("name") or "")
        except ValueError as e:
            return 400, {"ok": False, "error": str(e)}
    if method == "orphans":
        if req.get("http_method") == "GET":
            return ctx.api.call("modsafe.orphans")
        body = req.get("body") or {}
        try:
            return ctx.api.call("modsafe.uninstall_orphans",
                                body.get("packages") or [])
        except ValueError as e:
            return 400, {"ok": False, "error": str(e)}
    raise UserError("неизвестный метод: %s" % method)