475 lines
19 KiB
Python
475 lines
19 KiB
Python
#!/usr/bin/env python3
|
||
"""Small, dependency-free server for the character preference SPA."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import os
|
||
import secrets
|
||
import sqlite3
|
||
from contextlib import contextmanager
|
||
from difflib import SequenceMatcher
|
||
from datetime import datetime, timedelta, timezone
|
||
from http import HTTPStatus
|
||
from http.cookies import SimpleCookie
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
from pathlib import Path
|
||
from typing import Iterator
|
||
from urllib.parse import urlsplit
|
||
|
||
|
||
ROOT = Path(__file__).resolve().parent
|
||
STATIC = ROOT / "static"
|
||
DB_PATH = Path(os.environ.get("PREFERENCES_DB", ROOT / "data" / "preferences.sqlite3"))
|
||
SESSION_DAYS = 30
|
||
PROPERTIES = [
|
||
{
|
||
"id": "class",
|
||
"group": "WoW Forever Character",
|
||
"name": "Class",
|
||
"categories": [
|
||
"Druid", "Hunter", "Mage", "Paladin", "Priest",
|
||
"Rogue", "Shaman", "Warlock", "Warrior",
|
||
],
|
||
},
|
||
{
|
||
"id": "role",
|
||
"group": "WoW Forever Character",
|
||
"name": "Role",
|
||
"categories": ["Tank", "Healer", "DPS"],
|
||
},
|
||
{
|
||
"id": "content",
|
||
"group": "WoW Forever Character",
|
||
"name": "Content",
|
||
"categories": [
|
||
"Group Questing", "Open World PvP", "Instanced PvP",
|
||
"5-Man Dungeons", "10-Man Raids", "20-Man Raids", "40-Man Raids",
|
||
],
|
||
},
|
||
]
|
||
PROPERTY_BY_ID = {item["id"]: item for item in PROPERTIES}
|
||
CONTENT_CHOICES = {"Yes", "Maybe", "No"}
|
||
def preference_payload(property_id: str, value_json: str, updated_at: str) -> dict:
|
||
field = "choices" if property_id == "content" else "ranking"
|
||
value = json.loads(value_json)
|
||
return {field: value, "updatedAt": updated_at}
|
||
|
||
|
||
def login_key_for(login_username: str) -> str:
|
||
"""Keep login lookup case-insensitive while retaining chosen capitalization."""
|
||
return login_username.strip().casefold()
|
||
|
||
|
||
def valid_account_name(value: str) -> bool:
|
||
return 1 <= len(value) <= 32 and all(ord(char) >= 32 and ord(char) != 127 for char in value)
|
||
|
||
|
||
def user_payload(row: sqlite3.Row) -> dict:
|
||
return {
|
||
"userId": row["user_id"],
|
||
"loginUsername": row["login_username"],
|
||
}
|
||
|
||
|
||
def password_matches(password: str) -> bool:
|
||
expected = os.environ.get("SHARED_PASSWORD", "galnik")
|
||
return hmac.compare_digest(password.encode("utf-8"), expected.encode("utf-8"))
|
||
|
||
|
||
def suggested_usernames(query: str, connection: sqlite3.Connection) -> list[str]:
|
||
key = login_key_for(query)
|
||
if len(key) < 2:
|
||
return []
|
||
matches = []
|
||
for row in connection.execute("SELECT login_key, login_username FROM users"):
|
||
candidate = row["login_key"]
|
||
if candidate == key:
|
||
continue
|
||
similarity = SequenceMatcher(None, key, candidate).ratio()
|
||
if candidate.startswith(key) or key.startswith(candidate) or similarity >= 0.64:
|
||
score = (candidate.startswith(key), similarity)
|
||
matches.append((score, row["login_username"]))
|
||
matches.sort(key=lambda match: (match[0], match[1].casefold()), reverse=True)
|
||
return [username for _, username in matches[:5]]
|
||
|
||
|
||
def now() -> datetime:
|
||
return datetime.now(timezone.utc)
|
||
|
||
|
||
@contextmanager
|
||
def db() -> Iterator[sqlite3.Connection]:
|
||
connection = sqlite3.connect(DB_PATH, timeout=10)
|
||
connection.row_factory = sqlite3.Row
|
||
connection.execute("PRAGMA foreign_keys = ON")
|
||
try:
|
||
with connection:
|
||
yield connection
|
||
finally:
|
||
connection.close()
|
||
|
||
|
||
def init_db() -> None:
|
||
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||
with db() as connection:
|
||
existing = connection.execute("PRAGMA table_info(users)").fetchall()
|
||
if existing and "user_id" not in {row["name"] for row in existing}:
|
||
raise RuntimeError("Existing database uses the old schema. Wipe the preproduction database before starting this version.")
|
||
connection.executescript(
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS users (
|
||
user_id INTEGER PRIMARY KEY,
|
||
login_username TEXT NOT NULL,
|
||
login_key TEXT NOT NULL UNIQUE,
|
||
created_at TEXT NOT NULL
|
||
);
|
||
CREATE TABLE IF NOT EXISTS sessions (
|
||
token_hash TEXT PRIMARY KEY,
|
||
user_id INTEGER NOT NULL REFERENCES users(user_id) ON DELETE CASCADE,
|
||
expires_at TEXT NOT NULL
|
||
);
|
||
CREATE TABLE IF NOT EXISTS preferences (
|
||
user_id INTEGER NOT NULL REFERENCES users(user_id) ON DELETE CASCADE,
|
||
property_id TEXT NOT NULL,
|
||
preference_json TEXT NOT NULL,
|
||
updated_at TEXT NOT NULL,
|
||
PRIMARY KEY (user_id, property_id)
|
||
);
|
||
"""
|
||
)
|
||
if "display_name" in {row["name"] for row in existing}:
|
||
connection.execute("ALTER TABLE users DROP COLUMN display_name")
|
||
|
||
|
||
class Handler(BaseHTTPRequestHandler):
|
||
server_version = "Preferences/1.0"
|
||
|
||
def _json(self, status: int, payload: dict) -> None:
|
||
data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
||
self.send_response(status)
|
||
self.send_header("Content-Type", "application/json; charset=utf-8")
|
||
self.send_header("Content-Length", str(len(data)))
|
||
self.send_header("Cache-Control", "no-store")
|
||
self.send_header("X-Content-Type-Options", "nosniff")
|
||
self.end_headers()
|
||
self.wfile.write(data)
|
||
|
||
def _body(self) -> dict | None:
|
||
if self.headers.get("Content-Type", "").split(";")[0].strip() != "application/json":
|
||
self._json(HTTPStatus.UNSUPPORTED_MEDIA_TYPE, {"error": "Send JSON."})
|
||
return None
|
||
try:
|
||
length = int(self.headers.get("Content-Length", "0"))
|
||
if length < 1 or length > 8192:
|
||
raise ValueError
|
||
value = json.loads(self.rfile.read(length))
|
||
if not isinstance(value, dict):
|
||
raise ValueError
|
||
return value
|
||
except (ValueError, json.JSONDecodeError):
|
||
self._json(HTTPStatus.BAD_REQUEST, {"error": "Invalid request body."})
|
||
return None
|
||
|
||
def _same_site_write(self) -> bool:
|
||
# JSON plus this header prevents simple cross-site form submissions.
|
||
if self.headers.get("X-Requested-With") != "PreferencesSPA":
|
||
self._json(HTTPStatus.FORBIDDEN, {"error": "Invalid request."})
|
||
return False
|
||
return True
|
||
|
||
def _session(self) -> sqlite3.Row | None:
|
||
cookie = SimpleCookie()
|
||
try:
|
||
cookie.load(self.headers.get("Cookie", ""))
|
||
token = cookie["session"].value
|
||
except (KeyError, ValueError):
|
||
return None
|
||
token_hash = hashlib.sha256(token.encode("ascii", "ignore")).hexdigest()
|
||
with db() as connection:
|
||
row = connection.execute(
|
||
"""SELECT u.user_id, u.login_username FROM sessions s
|
||
JOIN users u ON u.user_id = s.user_id
|
||
WHERE s.token_hash = ? AND s.expires_at > ?""",
|
||
(token_hash, now().isoformat()),
|
||
).fetchone()
|
||
return row
|
||
|
||
def _require_session(self) -> sqlite3.Row | None:
|
||
session = self._session()
|
||
if not session:
|
||
self._json(HTTPStatus.UNAUTHORIZED, {"error": "Please sign in."})
|
||
return session
|
||
|
||
def _serve_static(self, path: str) -> None:
|
||
files = {
|
||
"/": ("index.html", "text/html; charset=utf-8"),
|
||
"/app.js": ("app.js", "text/javascript; charset=utf-8"),
|
||
"/styles.css": ("styles.css", "text/css; charset=utf-8"),
|
||
"/icon.webp": ("icon.webp", "image/webp"),
|
||
}
|
||
if path not in files:
|
||
self.send_error(HTTPStatus.NOT_FOUND)
|
||
return
|
||
filename, content_type = files[path]
|
||
data = (STATIC / filename).read_bytes()
|
||
self.send_response(HTTPStatus.OK)
|
||
self.send_header("Content-Type", content_type)
|
||
self.send_header("Content-Length", str(len(data)))
|
||
self.send_header("X-Content-Type-Options", "nosniff")
|
||
self.end_headers()
|
||
self.wfile.write(data)
|
||
|
||
def do_GET(self) -> None:
|
||
path = urlsplit(self.path).path
|
||
if path == "/api/bootstrap":
|
||
session = self._session()
|
||
if not session:
|
||
self._json(HTTPStatus.OK, {"user": None, "properties": PROPERTIES})
|
||
return
|
||
user_id = session["user_id"]
|
||
with db() as connection:
|
||
own_rows = connection.execute(
|
||
"SELECT property_id, preference_json, updated_at FROM preferences WHERE user_id = ?",
|
||
(user_id,),
|
||
).fetchall()
|
||
people_rows = connection.execute(
|
||
"""SELECT u.user_id, u.login_username,
|
||
p.property_id, p.preference_json, p.updated_at
|
||
FROM users u LEFT JOIN preferences p ON p.user_id = u.user_id
|
||
ORDER BY u.login_username COLLATE NOCASE"""
|
||
).fetchall()
|
||
own = {
|
||
row["property_id"]: preference_payload(row["property_id"], row["preference_json"], row["updated_at"])
|
||
for row in own_rows
|
||
}
|
||
people: dict[int, dict] = {}
|
||
for row in people_rows:
|
||
person = people.setdefault(row["user_id"], {**user_payload(row), "preferences": {}})
|
||
if row["property_id"]:
|
||
person["preferences"][row["property_id"]] = preference_payload(
|
||
row["property_id"], row["preference_json"], row["updated_at"]
|
||
)
|
||
self._json(HTTPStatus.OK, {
|
||
"user": user_payload(session),
|
||
"properties": PROPERTIES,
|
||
"preferences": own,
|
||
"people": list(people.values()),
|
||
})
|
||
return
|
||
if path.startswith("/api/"):
|
||
self._json(HTTPStatus.NOT_FOUND, {"error": "Not found."})
|
||
return
|
||
self._serve_static(path)
|
||
|
||
def do_POST(self) -> None:
|
||
path = urlsplit(self.path).path
|
||
if not self._same_site_write():
|
||
return
|
||
if path == "/api/username-suggestions":
|
||
body = self._body()
|
||
if body is None:
|
||
return
|
||
query = body.get("loginUsername")
|
||
if not isinstance(query, str):
|
||
self._json(HTTPStatus.BAD_REQUEST, {"error": "Enter a username."})
|
||
return
|
||
if len(query) > 32:
|
||
self._json(HTTPStatus.BAD_REQUEST, {"error": "Username must be at most 32 characters."})
|
||
return
|
||
with db() as connection:
|
||
suggestions = suggested_usernames(query, connection)
|
||
self._json(HTTPStatus.OK, {"suggestions": suggestions})
|
||
return
|
||
if path == "/api/login":
|
||
body = self._body()
|
||
if body is None:
|
||
return
|
||
username = body.get("loginUsername")
|
||
password = body.get("password")
|
||
if not isinstance(username, str) or not isinstance(password, str):
|
||
self._json(HTTPStatus.BAD_REQUEST, {"error": "Enter a username and password."})
|
||
return
|
||
username = username.strip()
|
||
if not valid_account_name(username):
|
||
self._json(HTTPStatus.BAD_REQUEST, {"error": "Username must be 1–32 printable characters."})
|
||
return
|
||
if not password_matches(password):
|
||
self._json(HTTPStatus.UNAUTHORIZED, {"error": "Incorrect password."})
|
||
return
|
||
key = login_key_for(username)
|
||
token = secrets.token_urlsafe(32)
|
||
token_hash = hashlib.sha256(token.encode("ascii")).hexdigest()
|
||
expiry = now() + timedelta(days=SESSION_DAYS)
|
||
with db() as connection:
|
||
existing = connection.execute(
|
||
"SELECT user_id FROM users WHERE login_key = ?", (key,)
|
||
).fetchone()
|
||
if not existing and body.get("createNew") is not True:
|
||
suggestions = suggested_usernames(username, connection)
|
||
if suggestions:
|
||
self._json(HTTPStatus.CONFLICT, {
|
||
"error": "Similar usernames already exist. Choose one or confirm a new account.",
|
||
"suggestions": suggestions,
|
||
})
|
||
return
|
||
connection.execute(
|
||
"INSERT OR IGNORE INTO users (login_username, login_key, created_at) VALUES (?, ?, ?)",
|
||
(username, key, now().isoformat()),
|
||
)
|
||
user = connection.execute(
|
||
"SELECT user_id, login_username FROM users WHERE login_key = ?", (key,)
|
||
).fetchone()
|
||
connection.execute(
|
||
"INSERT INTO sessions (token_hash, user_id, expires_at) VALUES (?, ?, ?)",
|
||
(token_hash, user["user_id"], expiry.isoformat()),
|
||
)
|
||
connection.execute("DELETE FROM sessions WHERE expires_at <= ?", (now().isoformat(),))
|
||
data = json.dumps({"user": user_payload(user)}).encode("utf-8")
|
||
self.send_response(HTTPStatus.OK)
|
||
secure = "; Secure" if os.environ.get("COOKIE_SECURE") == "1" else ""
|
||
self.send_header("Set-Cookie", f"session={token}; HttpOnly; SameSite=Lax; Path=/; Max-Age={SESSION_DAYS * 86400}{secure}")
|
||
self.send_header("Content-Type", "application/json; charset=utf-8")
|
||
self.send_header("Content-Length", str(len(data)))
|
||
self.send_header("Cache-Control", "no-store")
|
||
self.end_headers()
|
||
self.wfile.write(data)
|
||
return
|
||
if path == "/api/logout":
|
||
cookie = SimpleCookie()
|
||
try:
|
||
cookie.load(self.headers.get("Cookie", ""))
|
||
token = cookie["session"].value
|
||
token_hash = hashlib.sha256(token.encode("ascii", "ignore")).hexdigest()
|
||
with db() as connection:
|
||
connection.execute("DELETE FROM sessions WHERE token_hash = ?", (token_hash,))
|
||
except (KeyError, ValueError):
|
||
pass
|
||
self.send_response(HTTPStatus.NO_CONTENT)
|
||
self.send_header("Set-Cookie", "session=; HttpOnly; SameSite=Lax; Path=/; Max-Age=0")
|
||
self.send_header("Cache-Control", "no-store")
|
||
self.end_headers()
|
||
return
|
||
self._json(HTTPStatus.NOT_FOUND, {"error": "Not found."})
|
||
|
||
def do_DELETE(self) -> None:
|
||
path = urlsplit(self.path).path
|
||
if not self._same_site_write():
|
||
return
|
||
if path != "/api/me":
|
||
self._json(HTTPStatus.NOT_FOUND, {"error": "Not found."})
|
||
return
|
||
session = self._require_session()
|
||
if not session:
|
||
return
|
||
body = self._body()
|
||
if body is None:
|
||
return
|
||
if body.get("confirmLoginUsername") != session["login_username"]:
|
||
self._json(HTTPStatus.BAD_REQUEST, {"error": "Type your login username exactly to confirm deletion."})
|
||
return
|
||
with db() as connection:
|
||
connection.execute("DELETE FROM users WHERE user_id = ?", (session["user_id"],))
|
||
self.send_response(HTTPStatus.NO_CONTENT)
|
||
self.send_header("Set-Cookie", "session=; HttpOnly; SameSite=Lax; Path=/; Max-Age=0")
|
||
self.send_header("Cache-Control", "no-store")
|
||
self.end_headers()
|
||
|
||
def do_PUT(self) -> None:
|
||
path = urlsplit(self.path).path
|
||
if not self._same_site_write():
|
||
return
|
||
session = self._require_session()
|
||
if not session:
|
||
return
|
||
if path == "/api/me":
|
||
body = self._body()
|
||
if body is None:
|
||
return
|
||
login_username = body.get("loginUsername")
|
||
if not isinstance(login_username, str):
|
||
self._json(HTTPStatus.BAD_REQUEST, {"error": "Enter a login username."})
|
||
return
|
||
login_username = login_username.strip()
|
||
if not valid_account_name(login_username):
|
||
self._json(HTTPStatus.BAD_REQUEST, {"error": "Username must be 1–32 printable characters."})
|
||
return
|
||
try:
|
||
with db() as connection:
|
||
connection.execute(
|
||
"""UPDATE users SET login_username = ?, login_key = ?
|
||
WHERE user_id = ?""",
|
||
(login_username, login_key_for(login_username), session["user_id"]),
|
||
)
|
||
user = connection.execute(
|
||
"SELECT user_id, login_username FROM users WHERE user_id = ?",
|
||
(session["user_id"],),
|
||
).fetchone()
|
||
except sqlite3.IntegrityError:
|
||
self._json(HTTPStatus.CONFLICT, {"error": "That login username is already in use."})
|
||
return
|
||
if not user:
|
||
self._json(HTTPStatus.UNAUTHORIZED, {"error": "Please sign in."})
|
||
return
|
||
self._json(HTTPStatus.OK, {"user": user_payload(user)})
|
||
return
|
||
if not path.startswith("/api/preferences/"):
|
||
self._json(HTTPStatus.NOT_FOUND, {"error": "Not found."})
|
||
return
|
||
property_id = path.removeprefix("/api/preferences/")
|
||
property_ = PROPERTY_BY_ID.get(property_id)
|
||
if not property_:
|
||
self._json(HTTPStatus.NOT_FOUND, {"error": "Unknown property."})
|
||
return
|
||
body = self._body()
|
||
if body is None:
|
||
return
|
||
categories = property_["categories"]
|
||
if property_id == "content":
|
||
value = body.get("choices")
|
||
if not isinstance(value, dict) or set(value) != set(categories) or any(
|
||
not isinstance(choice, str) or choice not in CONTENT_CHOICES for choice in value.values()
|
||
):
|
||
self._json(HTTPStatus.BAD_REQUEST, {"error": "Choose Yes, Maybe, or No for every content category."})
|
||
return
|
||
else:
|
||
value = body.get("ranking")
|
||
if (
|
||
not isinstance(value, list)
|
||
or any(not isinstance(item, str) for item in value)
|
||
or len(value) != len(set(value))
|
||
or not set(value).issubset(categories)
|
||
):
|
||
self._json(HTTPStatus.BAD_REQUEST, {"error": "Rank each chosen category once; leave other categories unranked."})
|
||
return
|
||
updated_at = now().isoformat()
|
||
with db() as connection:
|
||
connection.execute(
|
||
"""INSERT INTO preferences (user_id, property_id, preference_json, updated_at)
|
||
VALUES (?, ?, ?, ?)
|
||
ON CONFLICT(user_id, property_id) DO UPDATE SET
|
||
preference_json = excluded.preference_json, updated_at = excluded.updated_at""",
|
||
(session["user_id"], property_id, json.dumps(value), updated_at),
|
||
)
|
||
self._json(HTTPStatus.OK, preference_payload(property_id, json.dumps(value), updated_at))
|
||
|
||
|
||
def main() -> None:
|
||
init_db()
|
||
host = os.environ.get("HOST", "127.0.0.1")
|
||
port = int(os.environ.get("PORT", "8000"))
|
||
server = ThreadingHTTPServer((host, port), Handler)
|
||
print(f"Character preferences running at http://{host}:{port}", flush=True)
|
||
try:
|
||
server.serve_forever()
|
||
except KeyboardInterrupt:
|
||
pass
|
||
finally:
|
||
server.server_close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|