single-user login screen
- server/credentials.py: PBKDF2 password + session secret, 0600 json, one-time setup command (server refuses to start without it) - app.py: signed 30-day cookie (itsdangerous), middleware 401s API / redirects pages to /login?next=, 5-fail/15-min lockout, /logout - static/login.html: dark-themed form, inline error display - app.js: api() 401 -> redirect to /login preserving current URL - docs: README quick start + features, SPEC auth section
This commit is contained in:
1 parent
2c203990bc
commit
76c45a8f29
7 files changed
+181
-6
No files matched your search
+61
-3
@@ -8,17 +8,75 @@ import base64
|
||||
import json
|
||||
import time
|
||||
from pathlib import Path
|
||||
from urllib.parse import urljoin
|
||||
from urllib.parse import quote, urljoin
|
||||
|
||||
from fastapi import Body, FastAPI, HTTPException, Query
|
||||
from fastapi.responses import Response
|
||||
from fastapi.responses import FileResponse, JSONResponse, RedirectResponse, Response
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from itsdangerous import BadSignature, TimestampSigner
|
||||
|
||||
from . import upstream
|
||||
from . import credentials, upstream
|
||||
|
||||
ROOT = Path(__file__).parent.parent
|
||||
app = FastAPI(title="ani-web")
|
||||
|
||||
# ---------- auth: single user, signed cookie ----------
|
||||
_SIGNER = TimestampSigner(credentials.load()["secret"]) # exits with instructions if unset
|
||||
_SESSION = "aniweb_session"
|
||||
_SESSION_AGE = 30 * 86400
|
||||
_PUBLIC = {("/login", "GET"), ("/login", "POST")}
|
||||
|
||||
# ponytail: in-memory lockout, single process — 5 fails lock for 15 min
|
||||
_lockout = {"fails": 0, "until": 0.0}
|
||||
|
||||
|
||||
def _auth_ok(request) -> bool:
|
||||
try:
|
||||
return _SIGNER.unsign(request.cookies.get(_SESSION, ""),
|
||||
max_age=_SESSION_AGE) == b"ok"
|
||||
except BadSignature:
|
||||
return False
|
||||
|
||||
|
||||
@app.middleware("http")
|
||||
async def require_auth(request, call_next):
|
||||
if (request.url.path, request.method) in _PUBLIC or _auth_ok(request):
|
||||
return await call_next(request)
|
||||
if request.url.path.startswith("/api/"):
|
||||
return JSONResponse({"detail": "not authenticated"}, status_code=401)
|
||||
next_ = request.url.path + (f"?{request.url.query}" if request.url.query else "")
|
||||
return RedirectResponse(f"/login?next={quote(next_, safe='')}", status_code=302)
|
||||
|
||||
|
||||
@app.get("/login", include_in_schema=False)
|
||||
def login_page():
|
||||
return FileResponse(ROOT / "static" / "login.html")
|
||||
|
||||
|
||||
@app.post("/login")
|
||||
async def login(payload: dict = Body(...)):
|
||||
if time.time() < _lockout["until"]:
|
||||
return JSONResponse({"detail": "too many attempts, try later"}, status_code=429)
|
||||
if credentials.check(payload.get("password", "")):
|
||||
_lockout.update(fails=0, until=0.0)
|
||||
resp = JSONResponse({"ok": True})
|
||||
resp.set_cookie(_SESSION, _SIGNER.sign(b"ok").decode(),
|
||||
max_age=_SESSION_AGE, httponly=True, samesite="lax")
|
||||
return resp
|
||||
_lockout["fails"] += 1
|
||||
if _lockout["fails"] >= 5:
|
||||
_lockout.update(fails=0, until=time.time() + 900)
|
||||
time.sleep(0.5) # slow down brute force
|
||||
return JSONResponse({"detail": "wrong password"}, status_code=401)
|
||||
|
||||
|
||||
@app.post("/logout")
|
||||
def logout():
|
||||
resp = JSONResponse({"ok": True})
|
||||
resp.delete_cookie(_SESSION)
|
||||
return resp
|
||||
|
||||
|
||||
HIST_FILE = Path(__file__).parent / "data" / "history.json"
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
"""credentials.py — single-user password + session secret for ani-web.
|
||||
|
||||
Run once: .venv/bin/python -m server.credentials
|
||||
Stores a PBKDF2 password hash and a random session-signing secret in
|
||||
server/data/credentials.json (mode 0600). No env vars, no DB.
|
||||
"""
|
||||
import getpass
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import secrets
|
||||
from pathlib import Path
|
||||
|
||||
FILE = Path(__file__).parent / "data" / "credentials.json"
|
||||
ITER = 210_000 # OWASP 2023 minimum for PBKDF2-SHA256
|
||||
|
||||
|
||||
def set_password():
|
||||
pw = getpass.getpass("Choose a password (min 8 chars): ")
|
||||
if len(pw) < 8:
|
||||
raise SystemExit("password too short")
|
||||
if pw != getpass.getpass("Again: "):
|
||||
raise SystemExit("passwords did not match")
|
||||
salt = secrets.token_bytes(16)
|
||||
data = {
|
||||
"v": 1,
|
||||
"salt": salt.hex(),
|
||||
"hash": hashlib.pbkdf2_hmac("sha256", pw.encode(), salt, ITER).hex(),
|
||||
"secret": secrets.token_hex(32),
|
||||
}
|
||||
FILE.parent.mkdir(exist_ok=True)
|
||||
# ponytail: mode 0600 on create; won't tighten an existing file
|
||||
fd = os.open(FILE, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
|
||||
with os.fdopen(fd, "w") as f:
|
||||
json.dump(data, f)
|
||||
print(f"credentials saved to {FILE}")
|
||||
|
||||
|
||||
def load():
|
||||
if not FILE.exists():
|
||||
raise SystemExit(
|
||||
"no credentials yet — set a password first:\n"
|
||||
" .venv/bin/python -m server.credentials")
|
||||
return json.loads(FILE.read_text())
|
||||
|
||||
|
||||
def check(password: str) -> bool:
|
||||
d = load()
|
||||
h = hashlib.pbkdf2_hmac("sha256", password.encode(),
|
||||
bytes.fromhex(d["salt"]), d.get("iter", ITER))
|
||||
return secrets.compare_digest(h.hex(), d["hash"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
set_password()
|
||||
Reference in new issue
Block a user