File size: 4,292 Bytes
a37fc18 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 | 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}
|