fileman v0.2.0 [extra]

"""fileman.engine — файловый менеджер дерева бота (только веб).

Полный доступ к дереву проекта (корень — каталог рядом с boot.py):
чтение/запись/создание/удаление любых файлов, кроме секретов.
Скрыты и запрещены: .env* (секреты — только через modsafe.env),
*.session* (сессии TG), .git, __pycache__, *.pyc, .lock/*.tmp.
Лимиты: чтение/запись текста — 1 МБ (utf-8), загрузка бинаря — 10 МБ.
Запись — tmp+rename (атомарно). Каждая мутация — в audit (через module).
"""
import logging
import os

logger = logging.getLogger(__name__)

_MAX_READ_BYTES = 1024 * 1024
_MAX_UPLOAD_BYTES = 10 * 1024 * 1024
_MAX_LIST = 500

#: Имена/префиксы, скрытые всегда (секреты, мусор, служебное).
_DENY_NAMES = {".env", ".git", "__pycache__", "node_modules", ".lock"}
_DENY_SUFFIX = (".session", ".session-journal", ".pyc", ".pyo", ".tmp")


def _root():
    mod_dir = os.path.dirname(os.path.abspath(__file__))
    return os.path.dirname(os.path.dirname(mod_dir))


def _safe_rel(root, rel):
    """Нормализованный путь внутри корня или None."""
    parts = [p for p in (rel or "").replace("\\", "/").split("/")
             if p not in ("", ".")]
    if not parts or any(p == ".." for p in parts):
        return None
    low_first = parts[0].lower()
    if low_first.startswith("."):
        return None
    target = os.path.join(root, *parts)
    if not os.path.abspath(target).startswith(os.path.abspath(root) + os.sep):
        return None
    return "/".join(parts)


def _hidden(name):
    low = name.lower()
    if name in _DENY_NAMES:
        return True
    return low.endswith(_DENY_SUFFIX)


def _entry(root, rel):
    full = os.path.join(root, *rel.split("/"))
    try:
        st = os.stat(full)
    except OSError:
        return None
    import time as _time
    return {"path": rel,
            "dir": os.path.isdir(full),
            "size": st.st_size if not os.path.isdir(full) else 0,
            "mtime": _time.strftime("%Y-%m-%d %H:%M:%S",
                                    _time.localtime(st.st_mtime))}


def list_dir(rel=""):
    """Содержимое каталога [{path, dir, size, mtime}]."""
    root = _root()
    norm = _safe_rel(root, rel or "")
    if norm is None and (rel or "") not in ("", "/"):
        raise ValueError("плохой путь: %s" % rel)
    norm = norm or ""
    full = os.path.join(root, *norm.split("/")) if norm else root
    if not os.path.isdir(full):
        raise ValueError("не каталог: %s" % (rel or "/"))
    out = []
    try:
        names = sorted(os.listdir(full))
    except OSError as e:
        raise ValueError("не читается: %s" % e)
    for name in names:
        if _hidden(name):
            continue
        e = _entry(root, ((norm + "/" + name) if norm else name).strip("/"))
        if e is not None:
            out.append(e)
        if len(out) >= _MAX_LIST:
            break
    out.sort(key=lambda e: (not e["dir"], e["path"].lower()))
    return {"path": norm or "/", "entries": out}


def _writable_rel(rel):
    """Куда разрешена запись: всё дерево, кроме скрытого (.env/.git/...)."""
    return True


def _uploadable_rel(rel):
    """Куда разрешена загрузка бинаря: всё дерево, кроме скрытого."""
    return True


def read_file(rel):
    """Текст файла (utf-8, лимит 1 МБ)."""
    root = _root()
    norm = _safe_rel(root, rel or "")
    if norm is None:
        raise ValueError("плохой путь: %s" % rel)
    if any(_hidden(p) for p in norm.split("/")):
        raise ValueError("файл скрыт: %s" % rel)
    full = os.path.join(root, *norm.split("/"))
    if not os.path.isfile(full) or os.path.islink(full):
        raise ValueError("не файл: %s" % rel)
    if os.path.getsize(full) > _MAX_READ_BYTES:
        raise ValueError("файл больше 1 МБ: %s" % rel)
    try:
        with open(full, "r", encoding="utf-8") as f:
            return f.read()
    except (OSError, ValueError, UnicodeDecodeError):
        raise ValueError("не текстовый файл: %s" % rel)


def write_file(rel, content):
    """Создать/перезаписать текстовый файл (любой, кроме скрытых)."""
    root = _root()
    norm = _safe_rel(root, rel or "")
    if norm is None:
        raise ValueError("плохой путь: %s" % rel)
    if any(_hidden(p) for p in norm.split("/")):
        raise ValueError("файл скрыт: %s" % rel)
    data = content or ""
    if len(data.encode("utf-8")) > _MAX_READ_BYTES:
        raise ValueError("текст больше 1 МБ")
    try:
        data.encode("utf-8")
        if "\x00" in data:
            raise ValueError("не текстовый файл: %s" % rel)
    except (ValueError, UnicodeEncodeError):
        raise ValueError("не текстовый файл: %s" % rel)
    full = os.path.join(root, *norm.split("/"))
    if os.path.isdir(full):
        raise ValueError("это каталог: %s" % rel)
    os.makedirs(os.path.dirname(full), exist_ok=True)
    tmp = full + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        f.write(data)
    os.replace(tmp, full)
    logger.info("fileman: записан %s (%d Б)", norm, len(data.encode("utf-8")))
    return {"ok": True, "path": norm}


def upload_file(rel, body, filename=""):
    """Загрузить бинарный файл (любой, кроме скрытых; лимит 10 МБ).

    rel — целевой путь (может включать имя файла); filename — fallback
    имени, если rel указывает на каталог или пуст внутри дерева.
    Симлинки не создаются (открываем только обычный файл).
    """
    root = _root()
    data = body or b""
    if len(data) > _MAX_UPLOAD_BYTES:
        raise ValueError("файл больше 10 МБ")
    norm = _safe_rel(root, rel or "")
    full = None
    if norm is not None:
        full = os.path.join(root, *norm.split("/"))
        if os.path.isdir(full):
            base = (filename or "").replace("\\", "/").split("/")[-1].strip()
            if not base or base in (".", ".."):
                raise ValueError("нужно имя файла")
            norm = norm + "/" + base
            full = os.path.join(full, base)
    if norm is None or full is None:
        raise ValueError("плохой путь: %s" % rel)
    if any(_hidden(p) for p in norm.split("/")):
        raise ValueError("файл скрыт: %s" % rel)
    if os.path.isdir(full):
        raise ValueError("это каталог: %s" % rel)
    if os.path.islink(full):
        raise ValueError("симлинк: %s" % rel)
    os.makedirs(os.path.dirname(full), exist_ok=True)
    tmp = full + ".tmp"
    with open(tmp, "wb") as f:
        f.write(data)
    os.replace(tmp, full)
    logger.info("fileman: загружен %s (%d Б)", norm, len(data))
    return {"ok": True, "path": norm, "size": len(data)}


def delete_file(rel):
    """Удалить файл (любой, кроме скрытых и корня)."""
    root = _root()
    norm = _safe_rel(root, rel or "")
    if norm is None:
        raise ValueError("плохой путь: %s" % rel)
    parts = norm.split("/")
    if any(_hidden(p) for p in parts):
        raise ValueError("файл скрыт: %s" % rel)
    full = os.path.join(root, *parts)
    if not os.path.isfile(full) or os.path.islink(full):
        raise ValueError("не файл: %s" % rel)
    os.unlink(full)
    logger.info("fileman: удалён %s", norm)
    return {"ok": True, "path": norm}

"""fileman.Module — файловый менеджер дерева бота (только веб).

Своё право fileman_manage. Полный доступ к дереву (кроме секретов
.env*/.session* и служебного мусора). TG-команд нет, ctx.api соседям
не раздаётся (только веб-роуты ниже) — для защиты.
"""
import logging
import os

from core.base_module import BaseModule

logger = logging.getLogger(__name__)


class Module(BaseModule):
    def setup(self, ctx) -> None:
        self.config = ctx.module_config("fileman")
        self.ctx = ctx
        api = ctx.api.register
        api("fileman", "list", self.api_list)
        api("fileman", "read", self.api_read)
        api("fileman", "write", self.api_write)
        api("fileman", "delete", self.api_delete)
        api("fileman", "upload", self.api_upload)

    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_list(self, path=""):
        """Содержимое каталога (право fileman_manage)."""
        from . import engine as _eng
        try:
            return {"ok": True, **_eng.list_dir(path or "")}
        except ValueError as e:
            raise ValueError(str(e))

    def api_read(self, path):
        """Текст файла (право fileman_manage)."""
        from . import engine as _eng
        try:
            return {"ok": True, "path": path,
                    "text": _eng.read_file(path or "")}
        except ValueError as e:
            raise ValueError(str(e))

    def api_write(self, path, text, actor=""):
        """Создать/перезаписать текстовый файл (право fileman_manage)."""
        from . import engine as _eng
        try:
            out = _eng.write_file(path or "", text or "")
        except ValueError as e:
            self._audit_rec("fileman.write", path or "", actor or "?",
                            str(e)[:200], False)
            raise ValueError(str(e))
        self._audit_rec("fileman.write", out["path"], actor or "?")
        return out

    def api_upload(self, path, body, filename="", actor=""):
        """Загрузить бинарный файл (право fileman_manage, до 10 МБ)."""
        from . import engine as _eng
        try:
            out = _eng.upload_file(path or "", body or b"", filename or "")
        except ValueError as e:
            self._audit_rec("fileman.upload", path or "", actor or "?",
                            str(e)[:200], False)
            raise ValueError(str(e))
        self._audit_rec("fileman.upload", out["path"], actor or "?")
        return out

    def api_delete(self, path, actor=""):
        """Удалить файл (право fileman_manage)."""
        from . import engine as _eng
        try:
            out = _eng.delete_file(path or "")
        except ValueError as e:
            self._audit_rec("fileman.delete", path or "", actor or "?",
                            str(e)[:200], False)
            raise ValueError(str(e))
        self._audit_rec("fileman.delete", out["path"], actor or "?")
        return out

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

"""fileman.ui_web — кабинет «Файлы»."""

ROUTES = [
    ("GET", "/api/fileman/list", "list", {"right": "fileman_manage",
     "desc": "Каталог (?path)"}),
    ("GET", "/api/fileman/read", "read", {"right": "fileman_manage",
     "desc": "Текст файла (?path)"}),
    ("POST", "/api/fileman/write", "write", {"right": "fileman_manage",
     "desc": "Создать/перезаписать текстовый файл {path, text}"}),
    ("POST", "/api/fileman/upload", "upload", {"right": "fileman_manage",
     "desc": "Загрузить файл (multipart file, ?path, до 10 МБ)"}),
    ("POST", "/api/fileman/delete", "delete", {"right": "fileman_manage",
     "desc": "Удалить файл {path}"}),
]

SERVICE = [
    {"key": "fileman_manage", "icon": "📁", "title": "Файлы", "right": "fileman_manage",
     "phase": 1, "desc": "Дерево проекта: просмотр, правка data/web, удаление в data"},
]

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 == "list":
        query = req.get("query") or {}
        try:
            return ctx.api.call("fileman.list", query.get("path") or "")
        except ValueError as e:
            return 400, {"ok": False, "error": str(e)}
    if method == "read":
        query = req.get("query") or {}
        try:
            return ctx.api.call("fileman.read", query.get("path") or "")
        except ValueError as e:
            return 400, {"ok": False, "error": str(e)}
    if method == "write":
        body = req.get("body") or {}
        try:
            return ctx.api.call("fileman.write", body.get("path") or "",
                                body.get("text") or "", _actor(req))
        except ValueError as e:
            return 400, {"ok": False, "error": str(e)}
    if method == "delete":
        body = req.get("body") or {}
        try:
            return ctx.api.call("fileman.delete", body.get("path") or "",
                                _actor(req))
        except ValueError as e:
            return 400, {"ok": False, "error": str(e)}
    if method == "upload":
        query = req.get("query") or {}
        files = (req.get("files") or {}).get("file") or []
        if not files:
            return 400, {"ok": False, "error": "Файл не передан (multipart file=*)"}
        fi = files[0]
        try:
            return ctx.api.call("fileman.upload",
                                query.get("path") or "", fi.get("body") or b"",
                                fi.get("filename") or "", _actor(req))
        except ValueError as e:
            return 400, {"ok": False, "error": str(e)}
    raise UserError("неизвестный метод: %s" % method)