yakpanel-core/YakPanel-server/backend/app/api/files.py

"""YakPanel - File manager API"""
import os
from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, Form
from fastapi.responses import FileResponse
from pydantic import BaseModel

from app.core.config import get_runtime_config
from app.core.utils import read_file, write_file, path_safe_check
from app.api.auth import get_current_user
from app.models.user import User

router = APIRouter(prefix="/files", tags=["files"])


def _resolve_path(path: str) -> str:
    """Resolve and validate path within allowed roots (cross-platform)"""
    cfg = get_runtime_config()
    www_root = os.path.abspath(cfg["www_root"])
    setup_path = os.path.abspath(cfg["setup_path"])
    allowed = [www_root, setup_path]
    if os.name != "nt":
        allowed.append(os.path.abspath("/www"))
    if ".." in path:
        raise HTTPException(status_code=401, detail="Path traversal not allowed")
    norm_path = path.strip().replace("\\", "/").strip("/")
    # Root or www_root-style path
    if not norm_path or norm_path in ("www", "www/wwwroot", "wwwroot"):
        full = www_root
    elif norm_path.startswith("www/wwwroot/"):
        full = os.path.abspath(os.path.join(www_root, norm_path[12:]))
    else:
        full = os.path.abspath(os.path.join(www_root, norm_path))
    if not any(
        full == r or (full + os.sep).startswith(r + os.sep)
        for r in allowed
    ):
        raise HTTPException(status_code=403, detail="Path not allowed")
    return full


@router.get("/list")
async def files_list(
    path: str = "/",
    current_user: User = Depends(get_current_user),
):
    """List directory contents"""
    try:
        full = _resolve_path(path)
    except HTTPException:
        raise
    if not os.path.isdir(full):
        raise HTTPException(status_code=401, detail="Not a directory")
    items = []
    for name in os.listdir(full):
        item_path = os.path.join(full, name)
        try:
            stat = os.stat(item_path)
            items.append({
                "name": name,
                "is_dir": os.path.isdir(item_path),
                "size": stat.st_size if os.path.isfile(item_path) else 0,
            })
        except OSError:
            pass
    return {"path": path, "items": items}


@router.get("/read")
async def files_read(
    path: str,
    current_user: User = Depends(get_current_user),
):
    """Read file content"""
    try:
        full = _resolve_path(path)
    except HTTPException:
        raise
    if not os.path.isfile(full):
        raise HTTPException(status_code=404, detail="Not a file")
    content = read_file(full)
    if content is None:
        raise HTTPException(status_code=500, detail="Failed to read file")
    return {"path": path, "content": content}


@router.get("/download")
async def files_download(
    path: str,
    current_user: User = Depends(get_current_user),
):
    """Download file"""
    try:
        full = _resolve_path(path)
    except HTTPException:
        raise
    if not os.path.isfile(full):
        raise HTTPException(status_code=404, detail="Not a file")
    return FileResponse(full, filename=os.path.basename(full))


@router.post("/upload")
async def files_upload(
    path: str = Form(...),
    file: UploadFile = File(...),
    current_user: User = Depends(get_current_user),
):
    """Upload file to directory"""
    try:
        full = _resolve_path(path)
    except HTTPException:
        raise
    if not os.path.isdir(full):
        raise HTTPException(status_code=400, detail="Path must be a directory")
    filename = file.filename or "upload"
    if ".." in filename or "/" in filename or "\\" in filename:
        raise HTTPException(status_code=400, detail="Invalid filename")
    dest = os.path.join(full, filename)
    content = await file.read()
    if not write_file(dest, content, "wb"):
        raise HTTPException(status_code=500, detail="Failed to write file")
    return {"status": True, "msg": "Uploaded", "path": path + "/" + filename}


class MkdirRequest(BaseModel):
    path: str
    name: str


class RenameRequest(BaseModel):
    path: str
    old_name: str
    new_name: str


class DeleteRequest(BaseModel):
    path: str
    name: str
    is_dir: bool


@router.post("/mkdir")
async def files_mkdir(
    body: MkdirRequest,
    current_user: User = Depends(get_current_user),
):
    """Create directory"""
    try:
        parent = _resolve_path(body.path)
    except HTTPException:
        raise
    if not os.path.isdir(parent):
        raise HTTPException(status_code=400, detail="Parent must be a directory")
    if not body.name or ".." in body.name or "/" in body.name or "\\" in body.name:
        raise HTTPException(status_code=400, detail="Invalid directory name")
    if not path_safe_check(body.name):
        raise HTTPException(status_code=400, detail="Invalid directory name")
    full = os.path.join(parent, body.name)
    if os.path.exists(full):
        raise HTTPException(status_code=400, detail="Already exists")
    try:
        os.makedirs(full, 0o755)
    except OSError as e:
        raise HTTPException(status_code=500, detail=str(e))
    return {"status": True, "msg": "Created", "path": body.path.rstrip("/") + "/" + body.name}


@router.post("/rename")
async def files_rename(
    body: RenameRequest,
    current_user: User = Depends(get_current_user),
):
    """Rename file or directory"""
    try:
        parent = _resolve_path(body.path)
    except HTTPException:
        raise
    if not body.old_name or not body.new_name:
        raise HTTPException(status_code=400, detail="Names required")
    for n in (body.old_name, body.new_name):
        if ".." in n or "/" in n or "\\" in n or not path_safe_check(n):
            raise HTTPException(status_code=400, detail="Invalid name")
    old_full = os.path.join(parent, body.old_name)
    new_full = os.path.join(parent, body.new_name)
    if not os.path.exists(old_full):
        raise HTTPException(status_code=404, detail="Not found")
    if os.path.exists(new_full):
        raise HTTPException(status_code=400, detail="Target already exists")
    try:
        os.rename(old_full, new_full)
    except OSError as e:
        raise HTTPException(status_code=500, detail=str(e))
    return {"status": True, "msg": "Renamed"}


@router.post("/delete")
async def files_delete(
    body: DeleteRequest,
    current_user: User = Depends(get_current_user),
):
    """Delete file or directory"""
    try:
        parent = _resolve_path(body.path)
    except HTTPException:
        raise
    if not body.name or ".." in body.name or "/" in body.name or "\\" in body.name:
        raise HTTPException(status_code=400, detail="Invalid name")
    full = os.path.join(parent, body.name)
    if not os.path.exists(full):
        raise HTTPException(status_code=404, detail="Not found")
    try:
        if body.is_dir:
            import shutil
            shutil.rmtree(full)
        else:
            os.remove(full)
    except OSError as e:
        raise HTTPException(status_code=500, detail=str(e))
    return {"status": True, "msg": "Deleted"}


class WriteFileRequest(BaseModel):
    path: str
    content: str


@router.post("/write")
async def files_write(
    body: WriteFileRequest,
    current_user: User = Depends(get_current_user),
):
    """Write text file content"""
    try:
        full = _resolve_path(body.path)
    except HTTPException:
        raise
    if os.path.isdir(full):
        raise HTTPException(status_code=400, detail="Cannot write to directory")
    if not write_file(full, body.content):
        raise HTTPException(status_code=500, detail="Failed to write file")
    return {"status": True, "msg": "Saved"}