jozamudi
aae5857dba
- /stream/{token}: tokens are plain base64; allowlist https://hls.anidb.app
so a session can't use the proxy for arbitrary URLs (new test)
- X-Aniweb:1 required on POST/DELETE except /login — SameSite=Lax is
still sent on cross-site form POSTs, which could hit /account/delete
(new test; SPA + tests send the header)
- tmp+os.replace writes for users/history/searches: a crash mid-write
could corrupt the JSON -> load() returned {} -> silent data loss
- resume positions now read from full history (?all=1), not top-5
- pagehide bound once; savePos reads currentEp (no stale closures)
- account delete also removes the searches file; episodes TTL 5m
- ani-web.service: supervised restart (not installed; see header)
- gitignore pi-session dumps + .pytest_cache
126 lines
3.9 KiB
Python
126 lines
3.9 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 atomic_write(path, text):
|
|
"""tmp + os.replace: a crash mid-write leaves the previous file intact
|
|
(in-place writes corrupted the file -> load() returned {} -> accounts
|
|
or history silently vanished)."""
|
|
tmp = path.parent / (path.name + ".tmp")
|
|
fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
|
|
with os.fdopen(fd, "w") as f:
|
|
f.write(text)
|
|
os.replace(tmp, path)
|
|
|
|
|
|
def save(users: dict):
|
|
DATA.mkdir(exist_ok=True)
|
|
atomic_write(USERS, json.dumps(users, 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)
|
|
(DATA / "searches" / 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)
|