#!/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 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 _ALLOWED_ROOTS = [ Path("/"), Path("/media/sd"), # Android shared storage Path("/mnt"), # WSL mounts ] _ALLOWED_ROOTS.append(_PROJECT_ROOT) # 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}") 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)") 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', list_dir) app.router.add_post(f'{prefix}/api/files/mkdir', mkdir) app.router.add_post(f'{prefix}/api/files/touch', touch) app.router.add_post(f'{prefix}/api/files/delete', delete) app.router.add_post(f'{prefix}/api/files/rename', rename) app.router.add_post(f'{prefix}/api/files/upload', upload) app.router.add_get(f'{prefix}/api/files/download', download) app.router.add_get(f'{prefix}/api/files/read', read_file) app.router.add_post(f'{prefix}/api/files/write', write_file) app.router.add_get(f'{prefix}/api/files/info', file_info) app.router.add_get(f'{prefix}/api/files/picker', picker_api)