Upload 3 files

Dockerfile

@@ -0,0 +1,16 @@
+# Read the doc: https://huggingface.co/docs/hub/spaces-sdks-docker
+# you will also find guides on how best to write your Dockerfile
+
+FROM python:3.9
+
+RUN useradd -m -u 1000 user
+USER user
+ENV PATH="/home/user/.local/bin:$PATH"
+
+WORKDIR /app
+
+COPY --chown=user ./requirements.txt requirements.txt
+RUN pip install --no-cache-dir --upgrade -r requirements.txt
+
+COPY --chown=user . /app
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860"]

app.py

@@ -0,0 +1,145 @@
+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}

requirements.txt

@@ -0,0 +1,3 @@
+fastapi==0.111.0
+uvicorn[standard]==0.30.1
+requests==2.32.3