| 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 |
|
|
|
|
| |
| |
| 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} |
|
|
