SenSu/services/web_panel/routes/files.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
File Manager Backend API

Endpoints:
  GET  {prefix}/api/files/list?path=&show_hidden=0       — list directory
  POST {prefix}/api/files/mkdir   {path, name}            — create directory
  POST {prefix}/api/files/touch   {path, name}            — create empty file
  POST {prefix}/api/files/delete  {path}                  — delete file/dir
  POST {prefix}/api/files/rename  {path, new_name}        — rename
  POST {prefix}/api/files/upload  (multipart)              — upload file(s)
  GET  {prefix}/api/files/download?path=                   — download file
  GET  {prefix}/api/files/read?path=                       — read text file
  POST {prefix}/api/files/write   {path, content}          — write text file
  GET  {prefix}/api/files/info?path=                       — file/dir stat

Root directory is restricted to configurable ROOTS (default: [project_root, data/]).
Path-traversal is blocked.
"""

import os
import shutil
import stat
import time
import json
import logging
import mimetypes
from pathlib import Path
from aiohttp import web
from ..utils.auth import panel_auth

logger = logging.getLogger(__name__)

# ── Security: allowed root directories ──
_PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent.parent
_ALLOWED_ROOTS = []

if os.name == 'nt':
    # Windows: enumerate all available drive letters
    import string
    for letter in string.ascii_uppercase:
        drive = Path(letter + ":\\")
        if drive.exists():
            _ALLOWED_ROOTS.append(drive)
else:
    # Linux / macOS / Android — 默认仅限项目目录 + data/，生产安全
    _ALLOWED_ROOTS = []
_ALLOWED_ROOTS.append(_PROJECT_ROOT)
# 从环境变量读取额外允许路径 (逗号分隔)
_extra_roots = os.environ.get("SENSU_FILE_ROOTS", "")
for r in _extra_roots.split(","):
    r = r.strip()
    if r:
        _ALLOWED_ROOTS.append(Path(r))
# Deduplicate and keep only existing
_seen = set()
_filtered = []
for p in _ALLOWED_ROOTS:
    rp = p.resolve() if p.exists() else None
    if rp and str(rp) not in _seen:
        _seen.add(str(rp))
        _filtered.append(rp)
_ALLOWED_ROOTS = _filtered or [Path("/") if os.name != 'nt' else Path("C:\\")]

MAX_READ_SIZE = 1 * 1024 * 1024    # 1 MB for text read
MAX_UPLOAD_SIZE = 50 * 1024 * 1024  # 50 MB per upload
TEXT_EXTENSIONS = {
    '.txt','.py','.js','.ts','.html','.css','.json','.yaml','.yml',
    '.md','.ini','.cfg','.conf','.log','.sh','.bat','.env','.xml',
    '.toml','.csv','.sql','.Makefile','.gitignore','.dockerfile',
    '.c','.h','.cpp','.hpp','.java','.kt','.rs','.go','.rb','.php',
    '.swift','.r','.lua','.pl','.scala','.dart',
}

# ── Helpers ──

def _normalize(path_str: str) -> Path:
    """Normalize path (resolve .. and .) WITHOUT following symlinks.
    Returns a Path that may be a symlink itself."""
    if not path_str:
        return _ALLOWED_ROOTS[0]

    # Use os.path.normpath which resolves .. and . without following symlinks
    normalized = os.path.normpath(path_str)
    return Path(normalized)

def _resolve(path_str: str) -> Path:
    """Resolve and validate path against allowed roots.
    Returns the LOGICAL path (may still be a symlink, not resolved)."""
    candidate = _normalize(path_str)
    # Security: ensure it's within an allowed root
    resolved = candidate.resolve()
    for root in _ALLOWED_ROOTS:
        try:
            resolved.relative_to(root)
            return candidate  # Return the logical path, not resolved
        except ValueError:
            continue
    raise ValueError(f"Path not within allowed roots: {path_str}")

def _ensure_exists(p: Path, must_exist: bool = True):
    if not p.exists():
        if must_exist:
            raise FileNotFoundError(str(p))
    return p

def _stat(p: Path) -> dict:
    """Stat a path: follow symlinks normally; fall back to lstat for broken ones."""
    try:
        s = p.stat()  # Follow symlinks — resolves dir/file correctly
        return {
            "name": p.name,
            "path": str(p),
            "size": s.st_size,
            "is_dir": p.is_dir(),
            "is_file": p.is_file(),
            "is_symlink": p.is_symlink(),
            "mtime": int(s.st_mtime),
            "mtime_str": time.strftime("%Y-%m-%d %H:%M", time.localtime(s.st_mtime)),
            "mode": stat.filemode(s.st_mode),
            "ext": p.suffix.lower() if p.is_file() else "",
            "readable": os.access(p, os.R_OK),
            "writable": os.access(p, os.W_OK),
        }
    except FileNotFoundError:
        # Broken symlink or missing target
        try:
            s = p.lstat()
            return {
                "name": p.name,
                "path": str(p),
                "size": s.st_size,
                "is_dir": False,
                "is_file": False,
                "is_symlink": True,
                "mtime": int(s.st_mtime),
                "mtime_str": time.strftime("%Y-%m-%d %H:%M", time.localtime(s.st_mtime)),
                "mode": stat.filemode(s.st_mode),
                "ext": "",
                "readable": False,
                "writable": False,
                "broken": True,
            }
        except Exception:
            pass  # Give up entirely
    except (OSError, PermissionError):
        pass
    # Ultimate fallback
    return {
        "name": p.name, "path": str(p), "size": 0,
        "is_dir": False, "is_file": False, "is_symlink": p.is_symlink(),
        "mtime": 0, "mtime_str": "—", "mode": "?---------", "ext": "",
        "readable": False, "writable": False, "broken": True,
    }

# ── Route setup ──

def setup_file_routes(app, service_manager, prefix=''):
    """Register all file-manager routes on the aiohttp app."""
    logger.info(f"📁 文件管理路由已注册 ({prefix}/api/files)")

    # List directory
    async def list_dir(req):
        try:
            path = req.query.get("path", "")
            show_hidden = req.query.get("show_hidden", "0") == "1"
            p = _resolve(path)
            _ensure_exists(p)
            if not p.is_dir():
                return web.json_response({"error": "Not a directory"}, status=400)

            items = []
            try:
                for entry in sorted(p.iterdir(), key=lambda x: (not x.is_dir(), x.name.lower())):
                    if not show_hidden and entry.name.startswith('.'):
                        continue
                    try:
                        items.append(_stat(entry))
                    except Exception:
                        pass  # Skip entries that can't be stat'd
            except PermissionError:
                return web.json_response({"error": "Permission denied"}, status=403)

            # Breadcrumbs: split path by OS separator, every segment clickable
            def make_breadcrumbs(path_obj):
                raw = str(path_obj)
                if os.name == 'nt' and len(raw) >= 2 and raw[1] == ':':
                    # Windows: C:\Users\... → crumbs: [C:\, Users, ...]
                    crumbs = [{"label": raw[:2] + "\\", "path": raw[:2] + "\\"}]
                    tail = raw[3:]
                else:
                    crumbs = [{"label": "/", "path": "/"}]
                    tail = raw.lstrip("/")
                parts = [x for x in tail.replace("\\", "/").split("/") if x]
                acc = crumbs[0]["path"].rstrip("\\/")
                for part in parts:
                    acc += ("\\" if os.name == 'nt' and acc.endswith(":") else "") + "/" + part
                    crumbs.append({"label": part, "path": acc.replace("\\", "/")})
                return crumbs

            crumbs = make_breadcrumbs(p)
            # Logical parent: None at filesystem root (Unix: /, Windows: C:\)
            p_str = str(p)
            is_root = p_str == "/" or (os.name == 'nt' and len(p_str) == 3 and p_str[1] == ':')
            logical_parent = str(Path(p_str).parent) if not is_root else None
            resolved = p.resolve()
            resolved_crumbs = None
            if str(resolved) != str(p):
                resolved_crumbs = make_breadcrumbs(resolved)

            return web.json_response({
                "current": str(p),
                "resolved": str(resolved) if str(resolved) != str(p) else None,
                "breadcrumbs": crumbs,
                "resolved_breadcrumbs": resolved_crumbs,
                "items": items,
                "parent": logical_parent,
                "allowed_roots": [str(r) for r in _ALLOWED_ROOTS],
            })
        except (FileNotFoundError, ValueError) as e:
            return web.json_response({"error": str(e)}, status=404)
        except Exception as e:
            logger.error(f"list_dir: {e}")
            return web.json_response({"error": str(e)}, status=500)

    # Create directory
    async def mkdir(req):
        try:
            data = await req.json()
            p = _resolve(data.get("path", ""))
            name = data.get("name", "").strip()
            if not name or "/" in name or "\\" in name:
                return web.json_response({"error": "Invalid name"}, status=400)
            target = (p / name)
            target.mkdir(parents=False, exist_ok=False)
            logger.info(f"📁 创建目录: {target}")
            return web.json_response({"ok": True, "path": str(target)})
        except FileExistsError:
            return web.json_response({"error": "Already exists"}, status=409)
        except Exception as e:
            return web.json_response({"error": str(e)}, status=500)

    # Create empty file
    async def touch(req):
        try:
            data = await req.json()
            p = _resolve(data.get("path", ""))
            name = data.get("name", "").strip()
            if not name or "/" in name or "\\" in name:
                return web.json_response({"error": "Invalid name"}, status=400)
            target = (p / name)
            target.touch(exist_ok=False)
            logger.info(f"📄 创建文件: {target}")
            return web.json_response({"ok": True, "path": str(target)})
        except FileExistsError:
            return web.json_response({"error": "Already exists"}, status=409)
        except Exception as e:
            return web.json_response({"error": str(e)}, status=500)

    # Delete file or empty directory (recursive for non-empty dirs)
    async def delete(req):
        try:
            data = await req.json()
            p = _resolve(data.get("path", ""))
            _ensure_exists(p)
            # Safety: refuse to delete project root
            if p == _PROJECT_ROOT:
                return web.json_response({"error": "Cannot delete project root"}, status=403)
            if p.is_dir():
                shutil.rmtree(p)
            else:
                p.unlink()
            logger.info(f"🗑 删除: {p}")
            # 审计日志
            db = request.app.get("sensu_db")
            if db:
                db.log_audit(request.get("user", {}).get("username", "?"), "file_delete", str(p))
            return web.json_response({"ok": True})
        except Exception as e:
            return web.json_response({"error": str(e)}, status=500)

    # Rename
    async def rename(req):
        try:
            data = await req.json()
            p = _resolve(data.get("path", ""))
            new_name = data.get("new_name", "").strip()
            if not new_name or "/" in new_name or "\\" in new_name:
                return web.json_response({"error": "Invalid name"}, status=400)
            target = p.parent / new_name
            p.rename(target)
            logger.info(f"✏ 重命名: {p} → {target}")
            return web.json_response({"ok": True, "path": str(target)})
        except Exception as e:
            return web.json_response({"error": str(e)}, status=500)

    # Upload
    async def upload(req):
        try:
            reader = await req.multipart()
            target_dir_str = req.query.get("path", "")
            target_dir = _resolve(target_dir_str)
            _ensure_exists(target_dir)
            if not target_dir.is_dir():
                return web.json_response({"error": "Target not a directory"}, status=400)

            uploaded = []
            while True:
                part = await reader.next()
                if part is None:
                    break
                if part.name == "file":
                    fname = part.filename
                    if not fname:
                        continue
                    # Sanitize filename
                    fname = Path(fname).name
                    dest = target_dir / fname
                    size = 0
                    with open(dest, 'wb') as f:
                        while True:
                            chunk = await part.read_chunk(65536)
                            if not chunk:
                                break
                            size += len(chunk)
                            if size > MAX_UPLOAD_SIZE:
                                f.close()
                                dest.unlink()
                                return web.json_response({"error": f"File too large: {fname}"}, status=413)
                            f.write(chunk)
                    uploaded.append(_stat(dest))
            logger.info(f"📤 上传 {len(uploaded)} 文件到 {target_dir}")
            return web.json_response({"ok": True, "files": uploaded})
        except Exception as e:
            return web.json_response({"error": str(e)}, status=500)

    # Download
    async def download(req):
        try:
            p = _resolve(req.query.get("path", ""))
            _ensure_exists(p)
            if not p.is_file():
                return web.json_response({"error": "Not a file"}, status=400)
            ct, _ = mimetypes.guess_type(str(p))
            return web.FileResponse(p, headers={
                "Content-Type": ct or "application/octet-stream",
                "Content-Disposition": f'attachment; filename="{p.name}"',
            })
        except Exception as e:
            return web.json_response({"error": str(e)}, status=500)

    # Read text file
    async def read_file(req):
        try:
            p = _resolve(req.query.get("path", ""))
            _ensure_exists(p)
            if not p.is_file():
                return web.json_response({"error": "Not a file"}, status=400)
            if p.suffix.lower() not in TEXT_EXTENSIONS:
                return web.json_response({"error": f"Not a text file: {p.suffix}"}, status=415)
            if p.stat().st_size > MAX_READ_SIZE:
                return web.json_response({"error": "File too large to read"}, status=413)
            content = p.read_text(encoding="utf-8", errors="replace")
            return web.json_response({
                "path": str(p),
                "name": p.name,
                "size": len(content),
                "content": content,
            })
        except Exception as e:
            return web.json_response({"error": str(e)}, status=500)

    # Write text file
    async def write_file(req):
        try:
            data = await req.json()
            p = _resolve(data.get("path", ""))
            content = data.get("content", "")
            p.write_text(content, encoding="utf-8")
            logger.info(f"💾 写入文件: {p} ({len(content)} bytes)")
            db = request.app.get("sensu_db")
            if db:
                db.log_audit(request.get("user", {}).get("username", "?"), "file_write", str(p))
            return web.json_response({"ok": True, "size": len(content)})
        except Exception as e:
            return web.json_response({"error": str(e)}, status=500)

    # File/dir info
    async def file_info(req):
        try:
            p = _resolve(req.query.get("path", ""))
            _ensure_exists(p)
            return web.json_response(_stat(p))
        except Exception as e:
            return web.json_response({"error": str(e)}, status=500)

    # Picker API (for plugins) — returns selected path as JSON
    # GET {prefix}/api/files/picker?mode=file|dir&path=
    async def picker_api(req):
        try:
            mode = req.query.get("mode", "dir")  # file | dir
            path = req.query.get("path", "")
            p = _resolve(path)
            _ensure_exists(p)
            items = []
            if p.is_dir():
                for entry in sorted(p.iterdir(), key=lambda x: (not x.is_dir(), x.name.lower())):
                    if entry.name.startswith('.'):
                        continue
                    if mode == "file" and not entry.is_file():
                        continue
                    if mode == "dir" and not entry.is_dir():
                        continue
                    items.append(_stat(entry))
            return web.json_response({
                "current": str(p),
                "mode": mode,
                "items": items,
            })
        except Exception as e:
            return web.json_response({"error": str(e)}, status=500)

    # ── Register routes ──
    app.router.add_get(f'{prefix}/api/files/list', panel_auth(list_dir))
    app.router.add_post(f'{prefix}/api/files/mkdir', panel_auth(mkdir))
    app.router.add_post(f'{prefix}/api/files/touch', panel_auth(touch))
    app.router.add_post(f'{prefix}/api/files/delete', panel_auth(delete))
    app.router.add_post(f'{prefix}/api/files/rename', panel_auth(rename))
    app.router.add_post(f'{prefix}/api/files/upload', panel_auth(upload))
    app.router.add_get(f'{prefix}/api/files/download', panel_auth(download))
    app.router.add_get(f'{prefix}/api/files/read', panel_auth(read_file))
    app.router.add_post(f'{prefix}/api/files/write', panel_auth(write_file))
    app.router.add_get(f'{prefix}/api/files/info', panel_auth(file_info))
    app.router.add_get(f'{prefix}/api/files/picker', panel_auth(picker_api))