jozamudi
c697500c7d
- server/users.py replaces credentials.py: users.json (0600), PBKDF2, one-time migration of the legacy single password to user admin (history.json -> history/admin.json) - accounts are CLI-managed only: python -m server.users add|rm|list (no self-signup on an internet-exposed app); /signup route removed, login page back to username + password - per-user 30-day signed session cookie (payload carries the username), per-user lockout, GET /api/me, POST /account/delete - per-user watch history: server/data/history/<user>.json; ANIWEB_DATA env override for the data dir - header chip: logged-in user, logout, delete account - tests/: throwaway-server suite (port 8399, temp data dir) — 9 HTTP tests + 2 Playwright browser tests (login UI, real video playback behind auth); runs via python tests/run.py
117 lines
3.6 KiB
Python
117 lines
3.6 KiB
Python
"""users.py — multi-user accounts for ani-web (PBKDF2, single json file).
|
|
|
|
CLI: .venv/bin/python -m server.users add <username> (password via prompt)
|
|
.venv/bin/python -m server.users rm <username>
|
|
.venv/bin/python -m server.users list
|
|
"""
|
|
import getpass
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import secrets
|
|
import time
|
|
from pathlib import Path
|
|
|
|
DATA = Path(os.environ.get("ANIWEB_DATA", Path(__file__).parent / "data"))
|
|
USERS = DATA / "users.json"
|
|
SECRET = DATA / "secret.key"
|
|
ITER = 210_000 # OWASP 2023 minimum for PBKDF2-SHA256
|
|
NAME_RE = re.compile(r"^[a-z0-9_-]{3,32}$")
|
|
|
|
|
|
def get_secret() -> bytes:
|
|
if SECRET.exists():
|
|
return SECRET.read_bytes()
|
|
s = secrets.token_bytes(32)
|
|
DATA.mkdir(exist_ok=True)
|
|
fd = os.open(SECRET, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
|
|
with os.fdopen(fd, "wb") as f:
|
|
f.write(s)
|
|
return s
|
|
|
|
|
|
def migrate_legacy():
|
|
"""one-time: single-password era -> users.json + per-user history files."""
|
|
cred = DATA / "credentials.json"
|
|
if not USERS.exists() and cred.exists():
|
|
d = json.loads(cred.read_text())
|
|
save({"admin": {"salt": d["salt"], "hash": d["hash"], "created": time.time()}})
|
|
cred.unlink()
|
|
old_hist = DATA / "history.json"
|
|
if not (DATA / "history" / "admin.json").exists() and old_hist.exists():
|
|
(DATA / "history").mkdir(exist_ok=True)
|
|
old_hist.rename(DATA / "history" / "admin.json")
|
|
|
|
|
|
def load() -> dict:
|
|
try:
|
|
return json.loads(USERS.read_text())
|
|
except (FileNotFoundError, json.JSONDecodeError):
|
|
return {}
|
|
|
|
|
|
def save(users: dict):
|
|
DATA.mkdir(exist_ok=True)
|
|
# ponytail: mode 0600 on create; won't tighten an existing file
|
|
fd = os.open(USERS, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
|
|
with os.fdopen(fd, "w") as f:
|
|
json.dump(users, f, indent=1)
|
|
|
|
|
|
def add(username: str, password: str):
|
|
u = username.lower()
|
|
if not NAME_RE.match(u):
|
|
raise ValueError("username: 3-32 chars, only a-z 0-9 _ -")
|
|
if len(password) < 8:
|
|
raise ValueError("password: min 8 chars")
|
|
users = load()
|
|
if u in users:
|
|
raise ValueError("username already taken")
|
|
salt = secrets.token_bytes(16)
|
|
users[u] = {"salt": salt.hex(),
|
|
"hash": hashlib.pbkdf2_hmac("sha256", password.encode(), salt, ITER).hex(),
|
|
"created": time.time()}
|
|
save(users)
|
|
|
|
|
|
def verify(username: str, password: str) -> bool:
|
|
d = load().get(username.lower())
|
|
if not d:
|
|
return False
|
|
h = hashlib.pbkdf2_hmac("sha256", password.encode(),
|
|
bytes.fromhex(d["salt"]), d.get("iter", ITER))
|
|
return secrets.compare_digest(h.hex(), d["hash"])
|
|
|
|
|
|
def remove(username: str):
|
|
u = username.lower()
|
|
users = load()
|
|
users.pop(u, None)
|
|
save(users)
|
|
(DATA / "history" / f"{u}.json").unlink(missing_ok=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
cmd = sys.argv[1] if len(sys.argv) > 1 else ""
|
|
if cmd == "add" and len(sys.argv) > 2:
|
|
name = sys.argv[2]
|
|
pw = getpass.getpass(f"Password for {name} (min 8 chars): ")
|
|
if pw != getpass.getpass("Again: "):
|
|
raise SystemExit("passwords did not match")
|
|
try:
|
|
add(name, pw)
|
|
except ValueError as e:
|
|
raise SystemExit(str(e))
|
|
print(f"added user {name.lower()}")
|
|
elif cmd == "rm" and len(sys.argv) > 2:
|
|
remove(sys.argv[2])
|
|
print(f"removed {sys.argv[2].lower()} (incl. history)")
|
|
elif cmd == "list":
|
|
for u in sorted(load()):
|
|
print(u)
|
|
else:
|
|
print(__doc__)
|
|
raise SystemExit(2)
|