multi-user auth (CLI-created accounts), per-user history, test suite
- 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
This commit is contained in:
1 parent
76c45a8f29
commit
c697500c7d
12 files changed
+595
-122
No files matched your search
+116
@@ -0,0 +1,116 @@
|
||||
"""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)
|
||||
Reference in new issue
Block a user