app.py

import hashlib
import hmac
import os
import shutil
import tempfile
from pathlib import Path

from fastapi import Depends, FastAPI, Header, HTTPException, Query, Request
from fastapi.responses import FileResponse


# Replace this value with sha256(FS_PASSWORD). Example:
# python3 -c 'import hashlib; print(hashlib.sha256(b"your-password").hexdigest())'
key = '25c20cbbe2eaf5f6dbf349965a04597c51df059dcaec38e6dcaae4e96bdc7c3e'


app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)


def _storage_root() -> Path:
    configured = os.environ.get("FS_ROOT")
    if configured:
        root = Path(configured)
    elif Path("/data").exists():
        root = Path("/data/fs")
    else:
        root = Path("/app/data")

    root.mkdir(parents=True, exist_ok=True)
    return root.resolve()


ROOT = _storage_root()


def _sha256(value: str) -> str:
    return hashlib.sha256(value.encode("utf-8")).hexdigest()


def _require_password(x_fs_password: str = Header(default="")) -> None:
    if key == "REPLACE_WITH_SHA256_OF_FS_PASSWORD":
        raise HTTPException(status_code=503, detail="server password hash is not configured")

    if not x_fs_password or not hmac.compare_digest(_sha256(x_fs_password), key):
        raise HTTPException(status_code=401, detail="unauthorized")


def _safe_path(path: str) -> Path:
    if not path or path == ".":
        return ROOT

    if "\x00" in path:
        raise HTTPException(status_code=400, detail="invalid path")

    raw = Path(path)
    if raw.is_absolute() or any(part in ("..", "") for part in raw.parts):
        raise HTTPException(status_code=400, detail="invalid path")

    resolved = (ROOT / raw).resolve()
    if ROOT != resolved and ROOT not in resolved.parents:
        raise HTTPException(status_code=400, detail="invalid path")
    return resolved


@app.get("/")
async def root() -> dict:
    raise HTTPException(status_code=404, detail="not found")


@app.put("/file", dependencies=[Depends(_require_password)])
async def upload(request: Request, path: str = Query(..., min_length=1)) -> dict:
    target = _safe_path(path)
    if target == ROOT:
        raise HTTPException(status_code=400, detail="path must be a file")

    target.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp_name = tempfile.mkstemp(prefix=f".{target.name}.", suffix=".upload", dir=str(target.parent))
    bytes_written = 0

    try:
        with os.fdopen(fd, "wb") as tmp_file:
            async for chunk in request.stream():
                if chunk:
                    tmp_file.write(chunk)
                    bytes_written += len(chunk)
            tmp_file.flush()
            os.fsync(tmp_file.fileno())

        os.replace(tmp_name, target)
    except Exception:
        try:
            os.unlink(tmp_name)
        except FileNotFoundError:
            pass
        raise

    return {"path": str(target.relative_to(ROOT)), "bytes": bytes_written}


@app.get("/file", dependencies=[Depends(_require_password)])
async def download(path: str = Query(..., min_length=1)) -> FileResponse:
    target = _safe_path(path)
    if not target.is_file():
        raise HTTPException(status_code=404, detail="file not found")

    return FileResponse(target, media_type="application/octet-stream", filename=target.name)


@app.get("/ls", dependencies=[Depends(_require_password)])
async def ls(path: str = ".") -> list[str]:
    target = _safe_path(path)
    if not target.exists():
        raise HTTPException(status_code=404, detail="path not found")

    if target.is_file():
        return [target.name]

    entries = []
    for child in target.iterdir():
        name = child.name + ("/" if child.is_dir() else "")
        entries.append(name)
    return sorted(entries)


@app.delete("/rm", dependencies=[Depends(_require_password)])
async def rm(path: str = ".") -> dict:
    target = _safe_path(path)
    if target == ROOT:
        removed = 0
        for child in ROOT.iterdir():
            if child.is_dir():
                shutil.rmtree(child)
            else:
                child.unlink()
            removed += 1
        return {"removed": removed}

    if not target.exists():
        return {"removed": 0}

    if target.is_dir():
        shutil.rmtree(target)
    else:
        target.unlink()
    return {"removed": 1}