jozamudi
202bfa41c3
- one-time import from the legacy JSON (left in place as snapshot); ghost rows (int vs str ep_id) deduped on import, unrepresentable after - history POST is now an upsert: the (user, slug_id, ep_id) PK is the merge key, so the unguarded read-modify-write (lost updates) and the corruption-swallow (JSONDecodeError -> [] -> fresh overwrite) are gone - searches: NOCASE PK upsert + cap 8 - events table + position/search logging (watch-time trajectory), 90-day retention pruned at boot - account delete cascades all four tables - 22/22 tests green; old JSON files untouched
129 lines
4.0 KiB
Python
129 lines
4.0 KiB
Python
"""users.py — multi-user accounts for ani-web (PBKDF2, SQLite via db.py).
|
|
|
|
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 sqlite3
|
|
import time
|
|
|
|
from . import db
|
|
|
|
DATA = db.DATA
|
|
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,
|
|
which db.init() (called after this) absorbs into the tables."""
|
|
cred = DATA / "credentials.json"
|
|
if not (DATA / "users.json").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:
|
|
c = db.connect()
|
|
rows = c.execute("SELECT name, salt, hash, iter, created FROM users").fetchall()
|
|
c.close()
|
|
return {n: {"salt": s, "hash": h, "iter": i, "created": cr} for n, s, h, i, cr in rows}
|
|
|
|
|
|
def save(users: dict):
|
|
"""file-based save, only for migrate_legacy (runs before db.init)."""
|
|
DATA.mkdir(exist_ok=True)
|
|
(DATA / "users.json").write_text(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")
|
|
salt = secrets.token_bytes(16)
|
|
c = db.connect()
|
|
try:
|
|
c.execute("INSERT INTO users VALUES (?,?,?,?,?)",
|
|
(u, salt.hex(),
|
|
hashlib.pbkdf2_hmac("sha256", password.encode(), salt, ITER).hex(),
|
|
ITER, time.time()))
|
|
c.commit()
|
|
except sqlite3.IntegrityError:
|
|
raise ValueError("username already taken")
|
|
finally:
|
|
c.close()
|
|
|
|
|
|
def verify(username: str, password: str) -> bool:
|
|
c = db.connect()
|
|
row = c.execute("SELECT salt, hash, iter FROM users WHERE name=?",
|
|
(username.lower(),)).fetchone()
|
|
c.close()
|
|
if not row:
|
|
return False
|
|
h = hashlib.pbkdf2_hmac("sha256", password.encode(),
|
|
bytes.fromhex(row[0]), row[2] or ITER)
|
|
return secrets.compare_digest(h.hex(), row[1])
|
|
|
|
|
|
def remove(username: str):
|
|
u = username.lower()
|
|
c = db.connect()
|
|
c.execute("DELETE FROM users WHERE name=?", (u,))
|
|
c.execute("DELETE FROM history WHERE user=?", (u,))
|
|
c.execute("DELETE FROM searches WHERE user=?", (u,))
|
|
c.execute("DELETE FROM events WHERE user=?", (u,))
|
|
c.commit()
|
|
c.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import sys
|
|
cmd = sys.argv[1] if len(sys.argv) > 1 else ""
|
|
migrate_legacy() # file-based, runs before db.init absorbs its output
|
|
db.init()
|
|
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)
|