Add admin console for managing users; default account is admin

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

server.py

@@ -26,6 +26,7 @@ import time
 import urllib.error
 import urllib.request
 from http.cookies import SimpleCookie
+from urllib.parse import unquote
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
 HERE = os.path.dirname(os.path.abspath(__file__))
@@ -86,6 +87,13 @@ def oa_login():
     return os.environ.get("OPENAUDIBLE_USER") or cfg.get("user"), os.environ.get("OPENAUDIBLE_PASSWORD") or cfg.get("password", "")
 
 
+USERNAME = re.compile(r"^[a-z0-9_.-]{1,32}$")
+
+
+def revoke_sessions(name):
+    write(F_SESSIONS, {k: v for k, v in read(F_SESSIONS, {}).items() if v["user"] != name}, private=True)
+
+
 def sha(token):
     return hashlib.sha256(token.encode()).hexdigest()
 
@@ -115,6 +123,8 @@ def cli(argv):
         if name in users:
             sys.exit("That user already exists (use passwd).")
         users[name] = hash_pw(ask_password())
+        if len(users) == 1 or "--admin" in argv:
+            users[name]["admin"] = True
         write(F_USERS, users, private=True)
         if len(users) == 1 and os.path.exists(F_LEGACY):  # carry over progress from before accounts
             legacy = read(F_LEGACY, {})
@@ -126,9 +136,9 @@ def cli(argv):
         name = argv[1].lower()
         if name not in users:
             sys.exit("No such user.")
-        users[name] = hash_pw(ask_password())
+        users[name] = {**hash_pw(ask_password()), **({"admin": True} if users[name].get("admin") else {})}
         write(F_USERS, users, private=True)
-        write(F_SESSIONS, {k: v for k, v in read(F_SESSIONS, {}).items() if v["user"] != name}, private=True)
+        revoke_sessions(name)
         print("Password changed; that user was signed out of all devices.")
     elif cmd == "deluser":
         name = argv[1].lower()
@@ -137,7 +147,7 @@ def cli(argv):
         write(F_SESSIONS, {k: v for k, v in read(F_SESSIONS, {}).items() if v["user"] != name}, private=True)
         print(f"Removed '{name}'.")
     elif cmd == "users":
-        print("\n".join(sorted(users)) or "(no users yet; run: python3 server.py adduser NAME)")
+        print("\n".join(n + ("  (admin)" if users[n].get("admin") else "") for n in sorted(users)) or "(no users yet; run: python3 server.py adduser NAME)")
 
 
 # ---------- login page ----------
@@ -214,7 +224,14 @@ class Handler(BaseHTTPRequestHandler):
             with open(os.path.join(HERE, "index.html"), "rb") as f:
                 return self.send_body(200, f.read(), "text/html; charset=utf-8", [("Cache-Control", "no-cache")])
         if path == "/me":
-            return self.json(200, {"user": user})
+            return self.json(200, {"user": user, "admin": bool(read(F_USERS, {}).get(user, {}).get("admin"))})
+        if path in ("/admin", "/admin/users"):
+            if not read(F_USERS, {}).get(user, {}).get("admin"):
+                return self.json(403, {"error": "admins only"}) if path == "/admin/users" else self.send_body(403, "Admins only.", "text/plain")
+            if path == "/admin":
+                with open(os.path.join(HERE, "admin.html"), "rb") as f:
+                    return self.send_body(200, f.read(), "text/html; charset=utf-8", [("Cache-Control", "no-cache")])
+            return self.json(200, self.user_list())
         if path == "/sync/progress":
             return self.json(200, read(F_PROGRESS, {}).get(user, {}))
         if any(p.match(path) for p in ALLOWED):
@@ -235,6 +252,13 @@ class Handler(BaseHTTPRequestHandler):
         user = self.user()
         if not user:
             return self.deny(path)
+        if path.startswith("/admin/users"):
+            if not read(F_USERS, {}).get(user, {}).get("admin"):
+                return self.json(403, {"error": "admins only"})
+            if not self.headers.get("Content-Type", "").startswith("application/json"):
+                return self.json(415, {"error": "json only"})  # blocks cross-site form posts
+            with lock:
+                return self.admin_api(user, path, body)
         if path == "/sync/progress":
             with lock:
                 allp = read(F_PROGRESS, {})
@@ -246,6 +270,62 @@ class Handler(BaseHTTPRequestHandler):
             return self.json(200, mine)
         self.send_error(404)
 
+    def user_list(self):
+        prog = read(F_PROGRESS, {})
+        out = []
+        for name, rec in sorted(read(F_USERS, {}).items()):
+            p = prog.get(name, {})
+            out.append({"name": name, "admin": bool(rec.get("admin")), "books": len(p), "last": max((v.get("t", 0) for v in p.values()), default=0)})
+        return out
+
+    def admin_api(self, me, path, body):
+        users = read(F_USERS, {})
+        bad = lambda msg, code=400: self.json(code, {"error": msg})
+        if path == "/admin/users":  # create
+            name = str(body.get("username", "")).strip().lower()
+            pw = str(body.get("password", ""))
+            if not USERNAME.match(name):
+                return bad("Username: 1-32 characters, letters, numbers, . _ -")
+            if name in users:
+                return bad("That username is taken.")
+            if len(pw) < 6:
+                return bad("Password must be at least 6 characters.")
+            users[name] = hash_pw(pw)
+            if body.get("admin"):
+                users[name]["admin"] = True
+            write(F_USERS, users, private=True)
+            return self.json(200, {"ok": True})
+        m = re.match(r"^/admin/users/([^/]+)/(password|admin|delete)$", path)
+        if not m:
+            return self.send_error(404)
+        name, action = unquote(m.group(1)).lower(), m.group(2)
+        if name not in users:
+            return bad("No such user.", 404)
+        others_admin = any(r.get("admin") for n, r in users.items() if n != name)
+        if action == "password":
+            pw = str(body.get("password", ""))
+            if len(pw) < 6:
+                return bad("Password must be at least 6 characters.")
+            users[name] = {**hash_pw(pw), **({"admin": True} if users[name].get("admin") else {})}
+            write(F_USERS, users, private=True)
+            revoke_sessions(name)  # signs them out everywhere
+        elif action == "admin":
+            flag = bool(body.get("admin"))
+            if not flag and not others_admin:
+                return bad("There must be at least one admin.")
+            users[name]["admin"] = flag
+            write(F_USERS, users, private=True)
+        else:  # delete
+            if name == me:
+                return bad("You can't delete your own account.")
+            users.pop(name)
+            write(F_USERS, users, private=True)
+            revoke_sessions(name)
+            prog = read(F_PROGRESS, {})
+            if prog.pop(name, None) is not None:
+                write(F_PROGRESS, prog)
+        self.json(200, {"ok": True})
+
     def login(self, body):
         ip = self.client_address[0]
         now = time.time()
@@ -327,6 +407,14 @@ def serve(argv):
             if legacy and all(isinstance(v, dict) and "pos" in v for v in legacy.values()):
                 write(F_PROGRESS, {du: legacy})
         print(f"Created default account '{du}'.")
+    if du in users and not users[du].get("admin"):  # the default account is always an admin
+        users[du]["admin"] = True
+        write(F_USERS, users, private=True)
+    if users and not any(r.get("admin") for r in users.values()):  # accounts from before roles existed
+        first = next(iter(users))
+        users[first]["admin"] = True
+        write(F_USERS, users, private=True)
+        print(f"No admin found; made '{first}' an admin.")
     if not users:
         sys.exit("No accounts yet. Run: python3 server.py adduser NAME (or set PLAYER_USER and PLAYER_PASSWORD)")
     if not oa_login()[0]: