"""The Button - a single-file, dependency-free game server (Python 3.9+).""" import csv import copy import hashlib import hmac import json import math import os import re import secrets import sqlite3 import threading import time from contextlib import contextmanager from datetime import datetime, timezone from functools import lru_cache from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path ROOT = Path(__file__).resolve().parent STATIC_DIR = ROOT / "static" DATA_FILE = Path(os.environ.get("BUTTON_DATA_FILE", ROOT / "data.json")) DATABASE_FILE = Path(os.environ.get("BUTTON_DATABASE_FILE", DATA_FILE.with_suffix(".sqlite3"))) MAILING_LIST_FILE = Path(os.environ.get("BUTTON_MAILING_LIST_FILE", ROOT / "mailing_list.csv")) MAILING_LIST_FIELDS = ["subscribed_at", "player", "first_name", "last_name", "email", "phone"] EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$") PHONE_RE = re.compile(r"^\+?[0-9 ()\-.]{7,25}$") HOST = os.environ.get("BUTTON_HOST", "127.0.0.1") PORT = int(os.environ.get("BUTTON_PORT", "8000")) DURATION_MS = int(float(os.environ.get("BUTTON_DURATION_SECONDS", 24 * 60 * 60)) * 1000) MAX_BODY_BYTES = 1024 MAX_NAME_LEN = 24 MIN_PASSWORD_LEN = 8 MAX_PASSWORD_LEN = 128 PBKDF2_ITERATIONS = 600_000 MAX_SESSIONS_PER_PLAYER = 10 LOGIN_MAX_FAILURES = 5 LOGIN_LOCKOUT_MS = 15 * 60 * 1000 MAX_CHAT_LENGTH = 200 CHAT_HISTORY_LIMIT = 100 CHAT_COOLDOWN_MS = 2000 PRESENCE_TTL_SECONDS = 45 MAX_PRESENCE_TABS_PER_SESSION = 10 # Ordered from a freshly reset countdown to an almost-expired one. # "share" is each zone's relative slice of the timer, so high-value zones are brief. ZONES = [ {"name": "green", "points": 0, "share": 32}, {"name": "blue", "points": 2, "share": 16}, {"name": "yellow", "points": 4, "share": 8}, {"name": "red", "points": 8, "share": 4}, {"name": "black", "points": 16, "share": 2}, {"name": "white", "points": 32, "share": 1}, ] _total_share = sum(z["share"] for z in ZONES) _upper = DURATION_MS for _z in ZONES: _z["remaining_from_ms"] = _upper _upper = round(_upper - DURATION_MS * _z["share"] / _total_share) _z["remaining_to_ms"] = max(_upper, 0) ZONES[-1]["remaining_to_ms"] = 0 STATIC_FILES = { **{f"/favicon-{zone['name']}.svg": (f"favicon-{zone['name']}.svg", "image/svg+xml") for zone in ZONES}, "/": ("index.html", "text/html; charset=utf-8"), "/index.html": ("index.html", "text/html; charset=utf-8"), "/app.js": ("app.js", "text/javascript; charset=utf-8"), "/style.css": ("style.css", "text/css; charset=utf-8"), "/favicon.ico": ("favicon.ico", "image/x-icon"), "/poster.png": ("poster.png", "image/png"), } lock = threading.Lock() def now_ms(): return int(time.time() * 1000) @lru_cache(maxsize=64) def season_boundaries(year): """UTC midnights of astronomical events (Meeus, Astronomical Algorithms ch. 27). Uses the 2000-era equinox/solstice polynomials and periodic correction, then a Delta-T estimate to convert terrestrial time to UTC before taking the date. Dates, not the event's hour, define the game boundary. Supports 2000-2100; 2026/2027 UTC dates are regression checked against USNO data. """ if not 1999 <= year <= 2101: raise ValueError("Season calendar supports years 2000-2100") y = (year - 2000) / 1000 coefficients = ( (2451623.80984, 365242.37404, .05169, -.00411, -.00057), (2451716.56767, 365241.62603, .00325, .00888, -.00030), (2451810.21715, 365242.01767, -.11575, .00337, .00078), (2451900.05952, 365242.74049, -.06223, -.00823, .00032), ) terms = ( (485, 324.96, 1934.136), (203, 337.23, 32964.467), (199, 342.08, 20.186), (182, 27.85, 445267.112), (156, 73.14, 45036.886), (136, 171.52, 22518.443), (77, 222.54, 65928.934), (74, 296.72, 3034.906), (70, 243.58, 9037.513), (58, 119.81, 33718.147), (52, 297.17, 150.678), (50, 21.02, 2281.226), (45, 247.54, 29929.562), (44, 325.15, 31555.956), (29, 60.93, 4443.417), (18, 155.12, 67555.328), (17, 288.79, 4562.452), (16, 198.04, 62894.029), (14, 199.76, 31436.921), (12, 95.39, 14577.848), (12, 287.11, 31931.756), (12, 320.81, 34777.259), (9, 227.73, 1222.114), (8, 15.45, 16859.074), ) result = [] for index, coefficients_for_event in enumerate(coefficients): jde = sum(c * y ** power for power, c in enumerate(coefficients_for_event)) t = (jde - 2451545) / 36525 w = math.radians(35999.373 * t - 2.47) correction = sum(a * math.cos(math.radians(b + c * t)) for a, b, c in terms) jde += .00001 * correction / (1 + .0334 * math.cos(w) + .0007 * math.cos(2 * w)) decimal_year = year + (index * 3 + 2.5) / 12 u = decimal_year - 2000 if decimal_year < 2005: delta_t = 63.86 + .3345 * u - .060374 * u**2 + .0017275 * u**3 + .000651814 * u**4 + .00002373599 * u**5 elif decimal_year < 2050: delta_t = 62.92 + .32217 * u + .005589 * u**2 else: delta_t = -20 + 32 * ((decimal_year - 1820) / 100)**2 - .5628 * (2150 - decimal_year) event = datetime.fromtimestamp((jde - 2440587.5) * 86400 - delta_t, timezone.utc) midnight = event.replace(hour=0, minute=0, second=0, microsecond=0) result.append({"id": midnight.strftime("%Y-%m-%d"), "name": f"{('March equinox', 'June solstice', 'September equinox', 'December solstice')[index]} {year}", "starts_at": int(midnight.timestamp() * 1000)}) return tuple(result) def season_for_time(at): year = datetime.fromtimestamp(at / 1000, timezone.utc).year if not 2000 <= year <= 2100: raise ValueError("Season calendar supports years 2000-2100") boundaries = [entry for y in (year - 1, year, year + 1) for entry in season_boundaries(y)] index = max(i for i, entry in enumerate(boundaries) if entry["starts_at"] <= at) return {**boundaries[index], "ends_at": boundaries[index + 1]["starts_at"]} def ensure_season(connection, at=None): """Adopt the existing game into the current season without deleting its history.""" if connection.execute("SELECT 1 FROM season_clock WHERE singleton = 1").fetchone() is None: season = season_for_time(now_ms() if at is None else at) connection.execute( "INSERT INTO season_clock (singleton, id, name, starts_at, ends_at) VALUES (1, ?, ?, ?, ?)", (season["id"], season["name"], season["starts_at"], season["ends_at"]), ) connection.execute("""INSERT OR IGNORE INTO season_numbering (singleton, origin_at) SELECT 1, MIN(starts_at) FROM ( SELECT starts_at FROM season_clock UNION ALL SELECT starts_at FROM season_results )""") def roman_numeral(number): if number < 1: raise ValueError("Season numbers must be positive") result = [] for value, symbol in ((1000, "M"), (900, "CM"), (500, "D"), (400, "CD"), (100, "C"), (90, "XC"), (50, "L"), (40, "XL"), (10, "X"), (9, "IX"), (5, "V"), (4, "IV"), (1, "I")): count, number = divmod(number, value) result.append(symbol * count) return "".join(result) def numbered_season(connection, season): origin = connection.execute("SELECT origin_at FROM season_numbering WHERE singleton = 1").fetchone()[0] def ordinal(at): date = datetime.fromtimestamp(at / 1000, timezone.utc) return date.year * 4 + (date.month - 3) // 3 number = ordinal(season["starts_at"]) - ordinal(origin) + 1 return {**season, "number": number, "name": f"Season {roman_numeral(number)}"} def current_season(connection): row = dict(connection.execute("SELECT * FROM season_clock WHERE singleton = 1").fetchone()) del row["singleton"] return numbered_season(connection, row) def hash_token(token): return hashlib.sha256(token.encode("utf-8")).hexdigest() def hash_password(password, salt=None, iterations=PBKDF2_ITERATIONS): salt = salt or secrets.token_hex(16) digest = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), bytes.fromhex(salt), iterations).hex() return f"pbkdf2_sha256${iterations}${salt}${digest}" def verify_password(password, stored): _, iterations, salt, _ = stored.split("$") return hmac.compare_digest(hash_password(password, salt, int(iterations)), stored) # Verified against when the name doesn't exist so response timing doesn't reveal valid names. DUMMY_PASSWORD_HASH = hash_password(secrets.token_hex(16)) def zone_for_remaining(remaining_ms): for zone in ZONES: if remaining_ms > zone["remaining_to_ms"]: return zone return ZONES[-1] def new_game(): t = now_ms() return { "duration_ms": DURATION_MS, "started_at": t, "deadline": t + DURATION_MS, "ended": False, "ended_at": None, "players": {}, # id -> {name, password_hash, token_hashes, created_at} "presses": [], # {player_id, at, remaining_ms, zone, points} } @contextmanager def database(): """Use a separate connection per operation, with atomic writes and FK checks.""" connection = sqlite3.connect(DATABASE_FILE) connection.row_factory = sqlite3.Row try: connection.execute("PRAGMA foreign_keys = ON") with connection: yield connection finally: connection.close() def write_state(connection, state): """Persist a complete snapshot inside the caller's transaction.""" connection.execute("DELETE FROM presses") connection.execute("DELETE FROM sessions") connection.execute("DELETE FROM players") connection.executemany( """INSERT INTO players (id, name, password_hash, created_at, subscribed, is_admin, is_banned, is_deleted) VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", [(pid, p["name"], p.get("password_hash"), p["created_at"], p.get("subscribed", False), p.get("is_admin", False), p.get("is_banned", False), p.get("is_deleted", False)) for pid, p in state["players"].items()], ) connection.executemany( "INSERT INTO sessions (player_id, position, token_hash) VALUES (?, ?, ?)", [(pid, position, token_hash) for pid, p in state["players"].items() for position, token_hash in enumerate(p["token_hashes"])], ) connection.executemany( "INSERT INTO presses (position, player_id, at, remaining_ms, zone, points) VALUES (?, ?, ?, ?, ?, ?)", [(position, p["player_id"], p["at"], p["remaining_ms"], p["zone"], p["points"]) for position, p in enumerate(state["presses"])], ) connection.execute( """INSERT INTO game (id, duration_ms, started_at, deadline, ended, ended_at) VALUES (1, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET duration_ms = excluded.duration_ms, started_at = excluded.started_at, deadline = excluded.deadline, ended = excluded.ended, ended_at = excluded.ended_at""", (state["duration_ms"], state["started_at"], state["deadline"], state["ended"], state["ended_at"]), ) def milestone_status(game_state, at): """Twelve fixed community goals; elapsed survival stops at the deadline.""" until = min(at, game_state["ended_at"] if game_state["ended_at"] is not None else game_state["deadline"]) elapsed = max(0, until - game_state["started_at"]) players = len(game_state["players"]) presses = len(game_state["presses"]) scores = {} visited = set() for press in game_state["presses"]: scores[press["player_id"]] = scores.get(press["player_id"], 0) + press["points"] visited.add(press["zone"]) goals = [ ("day_one", "One day together", "The community kept the game alive for one day!", elapsed, 86400000), ("one_week", "A week of nerve", "Seven days together. The clock is still part of our story!", elapsed, 7 * 86400000), ("one_month", "Thirty days strong", "The game survived 30 days. What a community!", elapsed, 30 * 86400000), ("ten_players", "A growing crowd", "Ten players have joined the experiment!", players, 10), ("fifty_players", "Fifty strong", "Fifty players, one shared clock. Welcome aboard!", players, 50), ("hundred_presses", "A hundred second chances", "Together we've pressed the button 100 times!", presses, 100), ("thousand_presses", "A thousand moments", "1,000 presses. A thousand decisions to keep us going!", presses, 1000), ("first_photo_finish", "Our first photo finish", "A community first: someone pressed with less than ten seconds left!", int(any(0 < p["remaining_ms"] < 10000 for p in game_state["presses"])), 1), ("shared_spectrum", "Every shade together", "The community has pressed in every scoring zone this season!", len(visited & {z["name"] for z in ZONES}), len(ZONES)), ("ten_contenders", "A race worth watching", "Ten players have earned points on this season's leaderboard!", sum(points > 0 for points in scores.values()), 10), ("thousand_points", "A thousand points together", "The seasonal leaderboard has reached 1,000 combined points!", sum(scores.values()), 1000), ("ten_thousand_points", "A season to remember", "The community has earned 10,000 points this season!", sum(scores.values()), 10000), ] return [{"id": key, "name": name, "message": message, "progress": min(progress, target), "target": target} for key, name, message, progress, target in goals] def sync_milestones(connection, game_state, at=None): """Commit unlocks with game changes. Existing goals are silently baselined once.""" at = now_ms() if at is None else at for goal in milestone_status(game_state, at): existed = connection.execute("SELECT 1 FROM community_milestones WHERE id = ?", (goal["id"],)).fetchone() is not None connection.execute("INSERT OR IGNORE INTO community_milestones (id) VALUES (?)", (goal["id"],)) if goal["progress"] >= goal["target"]: connection.execute( """UPDATE community_milestones SET reached_at = ?, announced_at = ?, message = ? WHERE id = ? AND reached_at IS NULL""", (at, at if existed else None, f"Milestone: {goal['name']}. {goal['message']}", goal["id"]), ) def load_state(): if DATABASE_FILE.resolve() == DATA_FILE.resolve(): raise ValueError("BUTTON_DATABASE_FILE must differ from the legacy BUTTON_DATA_FILE") DATABASE_FILE.parent.mkdir(parents=True, exist_ok=True) with database() as connection: connection.executescript(""" CREATE TABLE IF NOT EXISTS game ( id INTEGER PRIMARY KEY CHECK (id = 1), duration_ms INTEGER NOT NULL, started_at INTEGER NOT NULL, deadline INTEGER NOT NULL, ended INTEGER NOT NULL CHECK (ended IN (0, 1)), ended_at INTEGER ); CREATE TABLE IF NOT EXISTS players ( id TEXT PRIMARY KEY, name TEXT NOT NULL, password_hash TEXT, created_at INTEGER NOT NULL, subscribed INTEGER NOT NULL CHECK (subscribed IN (0, 1)) ); CREATE TABLE IF NOT EXISTS sessions ( player_id TEXT NOT NULL REFERENCES players(id), position INTEGER NOT NULL, token_hash TEXT NOT NULL UNIQUE, PRIMARY KEY (player_id, position) ); CREATE TABLE IF NOT EXISTS presses ( position INTEGER PRIMARY KEY, player_id TEXT NOT NULL REFERENCES players(id), at INTEGER NOT NULL, remaining_ms INTEGER NOT NULL, zone TEXT NOT NULL, points INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS chat_messages ( id INTEGER PRIMARY KEY, player_id TEXT NOT NULL REFERENCES players(id) DEFERRABLE INITIALLY DEFERRED, name TEXT NOT NULL, at INTEGER NOT NULL, message TEXT NOT NULL CHECK (length(message) BETWEEN 1 AND 200) ); CREATE INDEX IF NOT EXISTS chat_player_messages ON chat_messages (player_id, id); CREATE TABLE IF NOT EXISTS community_milestones ( id TEXT PRIMARY KEY, reached_at INTEGER, announced_at INTEGER, message TEXT ); CREATE TABLE IF NOT EXISTS season_clock ( singleton INTEGER PRIMARY KEY CHECK (singleton = 1), id TEXT NOT NULL, name TEXT NOT NULL, starts_at INTEGER NOT NULL, ends_at INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS season_results ( id TEXT PRIMARY KEY, name TEXT NOT NULL, starts_at INTEGER NOT NULL, ends_at INTEGER NOT NULL, played_from INTEGER NOT NULL, finished_at INTEGER NOT NULL, reason TEXT NOT NULL, leaderboard TEXT NOT NULL, presses TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS season_milestone_messages ( id TEXT PRIMARY KEY, at INTEGER NOT NULL, message TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS season_numbering ( singleton INTEGER PRIMARY KEY CHECK (singleton = 1), origin_at INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS chat_moderation ( message_id TEXT NOT NULL, at INTEGER NOT NULL, PRIMARY KEY (message_id, at) ); CREATE TABLE IF NOT EXISTS admin_audit ( id INTEGER PRIMARY KEY, at INTEGER NOT NULL, actor_id TEXT NOT NULL, action TEXT NOT NULL, target_id TEXT NOT NULL ); """) player_columns = {row["name"] for row in connection.execute("PRAGMA table_info(players)")} for field in ("is_admin", "is_banned", "is_deleted"): if field not in player_columns: # Fixed internal identifiers only; never interpolate request input. connection.execute(f"ALTER TABLE players ADD COLUMN {field} INTEGER NOT NULL DEFAULT 0 CHECK ({field} IN (0, 1))") columns = {row["name"] for row in connection.execute("PRAGMA table_info(chat_messages)")} if "kind" not in columns: connection.execute("ALTER TABLE chat_messages ADD COLUMN kind TEXT NOT NULL DEFAULT 'message'") # Serialize initialization. A committed game row is the migration marker, # not merely the existence of a possibly empty or interrupted database. connection.execute("BEGIN IMMEDIATE") game = connection.execute("SELECT * FROM game WHERE id = 1").fetchone() if game is None: if DATA_FILE.exists(): with DATA_FILE.open("r", encoding="utf-8") as f: loaded = json.load(f) for player in loaded["players"].values(): # Legacy JSON is not an administrator provisioning mechanism. player.update(is_admin=False, is_banned=False, is_deleted=False) if "token_hash" in player: # migrate pre-login data as well player["token_hashes"] = [player.pop("token_hash")] else: loaded = new_game() # Any read/validation/SQL failure aborts startup without consuming the # JSON source or committing a partial game. The next start can retry. write_state(connection, loaded) sync_milestones(connection, loaded) ensure_season(connection) return loaded loaded = dict(game) del loaded["id"] loaded["ended"] = bool(loaded["ended"]) loaded["players"] = {} for row in connection.execute("SELECT * FROM players ORDER BY rowid"): player = dict(row) pid = player.pop("id") player["subscribed"] = bool(player["subscribed"]) for field in ("is_admin", "is_banned", "is_deleted"): player[field] = bool(player[field]) player["token_hashes"] = [] loaded["players"][pid] = player for row in connection.execute("SELECT * FROM sessions ORDER BY player_id, position"): loaded["players"][row["player_id"]]["token_hashes"].append(row["token_hash"]) loaded["presses"] = [] for row in connection.execute("SELECT * FROM presses ORDER BY position"): press = dict(row) del press["position"] loaded["presses"].append(press) sync_milestones(connection, loaded) ensure_season(connection) return loaded def save_state(state): with database() as connection: write_state(connection, state) sync_milestones(connection, state) state = load_state() token_index = {th: pid for pid, p in state["players"].items() for th in p["token_hashes"]} failed_logins = {} # lowercase name -> (failure count, first failure ms) presence = {} # (token hash, tab id) -> (monotonic heartbeat seconds, viewing chat) def prune_presence(): """Must be called with the lock held. Presence is never persisted.""" cutoff = time.monotonic() - PRESENCE_TTL_SECONDS for key, (seen, _) in list(presence.items()): player = state["players"].get(token_index.get(key[0]), {}) if seen <= cutoff or key[0] not in token_index or player.get("is_banned") or player.get("is_deleted"): del presence[key] def public_presence(): """Public aggregates and opted-in chat names only; no session identifiers.""" prune_presence() online = {token_index[th] for th, _ in presence} chatting = {token_index[th] for (th, _), (_, in_chat) in presence.items() if in_chat} return { "online_count": len(online), "chat_players": sorted((state["players"][pid]["name"] for pid in chatting), key=str.lower), } def chat_messages(): with database() as connection: rows = connection.execute( """SELECT id, name, at, message, kind FROM ( SELECT id, name, at, message, kind, 0 AS source, id AS position FROM chat_messages UNION ALL SELECT 'milestone:' || id, 'The Button', announced_at, message, 'milestone', 1, rowid FROM community_milestones WHERE announced_at IS NOT NULL UNION ALL SELECT id, 'The Button', at, message, 'milestone', 1, rowid FROM season_milestone_messages ) AS feed WHERE NOT EXISTS ( SELECT 1 FROM chat_moderation m WHERE m.message_id = CAST(feed.id AS TEXT) AND m.at = feed.at ) ORDER BY at DESC, source DESC, position DESC LIMIT ?""", (CHAT_HISTORY_LIMIT,), ).fetchall() return [dict(row) for row in reversed(rows)] def find_player_by_name(name): for pid, p in state["players"].items(): if not p.get("is_deleted") and p["name"].lower() == name.lower(): return pid return None def issue_token(pid): """Must be called with the lock held.""" token = secrets.token_urlsafe(32) th = hash_token(token) hashes = state["players"][pid]["token_hashes"] hashes.append(th) token_index[th] = pid while len(hashes) > MAX_SESSIONS_PER_PLAYER: token_index.pop(hashes.pop(0), None) save_state(state) return token def validate_credentials(data): """Returns (name, password, error).""" name, password = data.get("name"), data.get("password") if not isinstance(name, str) or not (1 <= len(name.strip()) <= MAX_NAME_LEN): return None, None, f"Name must be 1-{MAX_NAME_LEN} characters" name = " ".join(name.split()) if not name.isprintable(): return None, None, "Name contains invalid characters" if not isinstance(password, str) or not (MIN_PASSWORD_LEN <= len(password) <= MAX_PASSWORD_LEN): return None, None, f"Password must be {MIN_PASSWORD_LEN}-{MAX_PASSWORD_LEN} characters" return name, password, None def archive_season(connection, season, game_state, finished_at, reason): """Immutable score/history snapshots have no FK to mutable player rows.""" if connection.execute("SELECT 1 FROM season_results WHERE id = ?", (season["id"],)).fetchone(): return scores = {pid: {"name": p["name"], "total_points": 0, "press_count": 0} for pid, p in game_state["players"].items()} for press in game_state["presses"]: scores[press["player_id"]]["total_points"] += press["points"] scores[press["player_id"]]["press_count"] += 1 standings = sorted(scores.values(), key=lambda p: (-p["total_points"], p["name"].lower())) connection.execute( """INSERT OR IGNORE INTO season_results (id, name, starts_at, ends_at, played_from, finished_at, reason, leaderboard, presses) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", (season["id"], season["name"], season["starts_at"], season["ends_at"], game_state["started_at"], max(game_state["started_at"], finished_at), reason, json.dumps(standings), json.dumps(game_state["presses"])), ) def check_ended(at=None): """Advance season and game atomically; caller must hold the game lock.""" t = now_ms() if at is None else at replacement = None with database() as connection: connection.execute("BEGIN IMMEDIATE") season = current_season(connection) while t >= season["ends_at"]: previous = replacement if replacement is not None else state finish = min(previous["ended_at"] if previous["ended_at"] is not None else previous["deadline"], season["ends_at"]) archive_season(connection, season, previous, finish, "countdown" if finish < season["ends_at"] else "boundary") # Preserve old milestone announcements before resetting seasonal goals. connection.execute( """INSERT OR IGNORE INTO season_milestone_messages (id, at, message) SELECT ? || id, announced_at, message FROM community_milestones WHERE announced_at IS NOT NULL""", (f"milestone:{season['id']}:",), ) connection.execute( """INSERT OR IGNORE INTO chat_moderation (message_id, at) SELECT ? || c.id, c.announced_at FROM community_milestones c JOIN chat_moderation m ON m.message_id = 'milestone:' || c.id AND m.at = c.announced_at WHERE c.announced_at IS NOT NULL""", (f"milestone:{season['id']}:",), ) connection.execute("DELETE FROM community_milestones") season = season_for_time(season["ends_at"]) connection.execute( "UPDATE season_clock SET id = ?, name = ?, starts_at = ?, ends_at = ? WHERE singleton = 1", (season["id"], season["name"], season["starts_at"], season["ends_at"]), ) replacement = {**previous, "duration_ms": DURATION_MS, "started_at": season["starts_at"], "deadline": season["starts_at"] + DURATION_MS, "ended": False, "ended_at": None, "presses": []} current = replacement if replacement is not None else state if not current["ended"] and current["deadline"] > season["ends_at"]: current = {**current, "deadline": season["ends_at"]} replacement = current if not current["ended"] and t >= current["deadline"]: current = {**current, "ended": True, "ended_at": current["deadline"]} replacement = current if replacement is not None: write_state(connection, current) sync_milestones(connection, current, min(t, season["ends_at"])) if current["ended"]: archive_season(connection, season, current, current["ended_at"], "boundary" if current["ended_at"] == season["ends_at"] else "countdown") if replacement is not None: state.clear() state.update(replacement) def seasonal_leaderboard(): with database() as connection: season = current_season(connection) rows = connection.execute("""SELECT id, name, starts_at, ends_at, played_from, finished_at, reason, leaderboard FROM season_results ORDER BY starts_at DESC""").fetchall() rows = [numbered_season(connection, dict(row)) for row in rows] archives = [] for row in rows: standings = json.loads(row["leaderboard"]) participants = [player for player in standings if player["press_count"] > 0] top = max((p["total_points"] for p in participants), default=None) archives.append({ "id": row["id"], "name": row["name"], "number": row["number"], "starts_at": row["starts_at"], "ends_at": row["ends_at"], "played_from": row["played_from"], "finished_at": row["finished_at"], "duration_ms": max(0, row["finished_at"] - row["played_from"]), "reason": row["reason"], "winners": [p for p in participants if p["total_points"] == top], "leaderboard": standings, }) return {"current": {**season, "played_from": state["started_at"], "ended": state["ended"], "leaderboard": public_leaderboard()}, "seasons": archives} def season_summary(): with database() as connection: return {**current_season(connection), "played_from": state["started_at"], "ended": state["ended"]} def empty_zone_counts(): return {z["name"]: 0 for z in ZONES} def seasonal_achievements(pid): """Lifetime badges from immutable archived presses, keyed by player ID, not names.""" with database() as connection: exists = connection.execute("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'season_results'").fetchone() rows = connection.execute("SELECT finished_at, presses FROM season_results ORDER BY finished_at, starts_at").fetchall() if exists else [] finished = 0 finisher_at = veteran_at = podium_at = champion_at = None for row in rows: scores = {} for press in json.loads(row["presses"]): scores[press["player_id"]] = scores.get(press["player_id"], 0) + press["points"] if pid not in scores: continue finished += 1 if finisher_at is None: finisher_at = row["finished_at"] if finished >= 2 and veteran_at is None: veteran_at = row["finished_at"] # Competition ranks: tied scores share rank; never use alphabetical order for awards. rank = 1 + sum(points > scores[pid] for points in scores.values()) if rank <= 3 and podium_at is None: podium_at = row["finished_at"] if rank == 1 and champion_at is None: champion_at = row["finished_at"] rules = [ ("season_finisher", "Part of the story", "Participate with a press in a completed season.", finisher_at, min(finished, 1), 1), ("season_veteran", "Back for another", "Participate in two completed seasons.", veteran_at, min(finished, 2), 2), ("season_podium", "Podium finish", "Finish a season in the top three by points. Ties share rank.", podium_at, int(podium_at is not None), 1), ("season_champion", "Season champion", "Finish a season with the highest score. Joint winners qualify.", champion_at, int(champion_at is not None), 1), ] return [{"id": key, "name": name, "description": description, "unlocked": at is not None, "unlocked_at": at, "progress": progress, "target": target, "scope": "lifetime"} for key, name, description, at, progress, target in rules] def player_achievements(pid): """Derive private achievements from recorded history, including legacy games.""" quiet_target = max(1, state["duration_ms"] // 2) close_at = spectrum_at = quiet_at = None visited = set() longest_quiet = 0 previous_at = state["started_at"] zone_names = {zone["name"] for zone in ZONES} extra_rules = [ ("red_alert", "Red alert", "Press in the red zone.", "red", None), ("into_the_dark", "Into the dark", "Press in the black zone.", "black", None), ("white_knuckle", "White knuckle", "Press in the white zone.", "white", None), ("photo_finish", "Photo finish", "Press with less than ten seconds remaining.", None, 10000), ("last_second", "Last second", "Press with less than one second remaining.", None, 1000), ] extra_unlocks = {rule[0]: None for rule in extra_rules} for press in state["presses"]: quiet_ms = max(0, press["at"] - previous_at) previous_at = press["at"] # measure the global gap, not this player's gap if press["player_id"] != pid: continue for achievement_id, _, _, zone, threshold in extra_rules: qualifies = press["zone"] == zone if zone is not None else 0 < press["remaining_ms"] < threshold if extra_unlocks[achievement_id] is None and qualifies: extra_unlocks[achievement_id] = press["at"] if close_at is None and 0 < press["remaining_ms"] < 60000: close_at = press["at"] if press["zone"] in zone_names: visited.add(press["zone"]) if spectrum_at is None and visited == zone_names: spectrum_at = press["at"] longest_quiet = max(longest_quiet, quiet_ms) if quiet_at is None and quiet_ms >= quiet_target: quiet_at = press["at"] return [ {"id": "close_call", "name": "Close call", "description": "Press with less than one minute remaining.", "unlocked": close_at is not None, "unlocked_at": close_at, "progress": int(close_at is not None), "target": 1}, {"id": "full_spectrum", "name": "Full spectrum", "description": "Press in all six scoring zones.", "unlocked": spectrum_at is not None, "unlocked_at": spectrum_at, "progress": len(visited), "target": len(zone_names)}, {"id": "keeping_us_alive", "name": "Keeping us alive", "description": "Press after nobody has pressed for at least half the game's countdown.", "unlocked": quiet_at is not None, "unlocked_at": quiet_at, "progress": min(longest_quiet, quiet_target), "target": quiet_target}, ] + [ {"id": achievement_id, "name": name, "description": description, "unlocked": extra_unlocks[achievement_id] is not None, "unlocked_at": extra_unlocks[achievement_id], "progress": int(extra_unlocks[achievement_id] is not None), "target": 1} for achievement_id, name, description, _, _ in extra_rules ] + seasonal_achievements(pid) def player_summary(pid): presses = [p for p in state["presses"] if p["player_id"] == pid] counts = empty_zone_counts() for p in presses: counts[p["zone"]] += 1 standings = ranked_players() position = next(i for i, (player_id, _) in enumerate(standings) if player_id == pid) points = standings[position][1]["total_points"] rival = standings[position - 1][1] if position else None def record(press): return {key: press[key] for key in ("at", "remaining_ms", "zone", "points")} if press else None closest = min(presses, key=lambda p: (p["remaining_ms"], p["at"]), default=None) best = min(presses, key=lambda p: (-p["points"], p["remaining_ms"], p["at"]), default=None) return { "name": state["players"][pid]["name"], "is_admin": bool(state["players"][pid].get("is_admin", False)), "subscribed": state["players"][pid].get("subscribed", False), "total_players": len(state["players"]), "total_presses": len(state["presses"]), "total_points": sum(p["points"] for p in presses), "press_count": len(presses), "zone_counts": counts, "best_remaining_ms": min((p["remaining_ms"] for p in presses), default=None), "personal_bests": {"closest_call": record(closest), "highest_scoring_press": record(best)}, "achievements": player_achievements(pid), "ranking": { "position": position + 1, "total_players": len(standings), "tied_for_lead": len(standings) > 1 and points == standings[0][1]["total_points"] and sum(score["total_points"] == points for _, score in standings) > 1, "rival": { "name": rival["name"], "total_points": rival["total_points"], "points_to_tie": rival["total_points"] - points, "points_to_pass": rival["total_points"] - points + 1, } if rival else None, }, "presses": [ {k: p[k] for k in ("at", "remaining_ms", "zone", "points")} for p in presses ], } def public_press(p): return { "name": state["players"][p["player_id"]]["name"], "at": p["at"], "remaining_ms": p["remaining_ms"], "zone": p["zone"], "points": p["points"], } def ranked_players(): """Internal standings retain IDs to identify players even in legacy data.""" scores = { pid: {"name": player["name"], "total_points": 0, "press_count": 0} for pid, player in state["players"].items() } for press in state["presses"]: score = scores[press["player_id"]] score["total_points"] += press["points"] score["press_count"] += 1 return sorted(scores.items(), key=lambda item: (-item[1]["total_points"], item[1]["name"].lower())) def public_leaderboard(): """Public score totals only; do not expose sessions or individual presses.""" return [score for _, score in ranked_players()] def global_stats(): counts = empty_zone_counts() for p in state["presses"]: counts[p["zone"]] += 1 leaderboard = [] for pid in state["players"]: s = player_summary(pid) del s["presses"], s["subscribed"], s["personal_bests"], s["ranking"], s["achievements"], s["is_admin"] leaderboard.append(s) leaderboard.sort(key=lambda s: (-s["total_points"], s["name"].lower())) presses = state["presses"] lowest = min(presses, key=lambda p: p["remaining_ms"], default=None) return { "started_at": state["started_at"], "ended_at": state["ended_at"], "total_players": len(state["players"]), "total_presses": len(presses), "zone_counts": counts, "leaderboard": leaderboard, "last_press": public_press(presses[-1]) if presses else None, "lowest_press": public_press(lowest) if lowest else None, "presses": [public_press(p) for p in presses], } def active_admin(player): return bool(player.get("is_admin") and not player.get("is_banned") and not player.get("is_deleted")) def revoke_player_sessions(pid): for th, owner in list(token_index.items()): if owner == pid: del token_index[th] prune_presence() def admin_overview(actor): scores = dict(ranked_players()) with database() as connection: audit = [dict(row) for row in connection.execute( "SELECT at, actor_id, action, target_id FROM admin_audit ORDER BY id DESC LIMIT 50")] total_messages = connection.execute("SELECT COUNT(*) FROM chat_messages").fetchone()[0] return { "actor_id": actor, "season": season_summary(), "game": {"ended": state["ended"], "deadline": state["deadline"], "total_presses": len(state["presses"]), "stored_chat_messages": total_messages}, "players": [ {"id": pid, "name": p["name"], "created_at": p["created_at"], "is_admin": bool(p.get("is_admin")), "is_banned": bool(p.get("is_banned")), "sessions": len(p["token_hashes"]), **{k: scores[pid][k] for k in ("total_points", "press_count")}} for pid, p in state["players"].items() if not p.get("is_deleted") ], "messages": chat_messages(), "audit": audit, } class AdminActionError(ValueError): def __init__(self, status, message): super().__init__(message) self.status = status def admin_action(actor, data): """Caller holds lock and has checked admin authority; publish memory after commit only.""" action, target = data.get("action"), data.get("target_id") if action not in ("set_admin", "set_banned", "delete_player", "delete_message"): raise AdminActionError(HTTPStatus.BAD_REQUEST, "Unknown admin action") if not isinstance(target, str) or not target or len(target) > 150: raise AdminActionError(HTTPStatus.BAD_REQUEST, "Invalid target") if action in ("set_admin", "set_banned") and not isinstance(data.get("value"), bool): raise AdminActionError(HTTPStatus.BAD_REQUEST, "Expected a boolean value") replacement = None original_name = None with database() as connection: connection.execute("BEGIN IMMEDIATE") if action == "delete_message": at = data.get("at") if not isinstance(at, int) or isinstance(at, bool): raise AdminActionError(HTTPStatus.BAD_REQUEST, "Message timestamp required") # Work with the same bounded, visible feed shown to the administrator. message = next((m for m in chat_messages() if str(m["id"]) == target and m["at"] == at), None) if message is None: raise AdminActionError(HTTPStatus.NOT_FOUND, "Message not found in recent chat") connection.execute("INSERT OR IGNORE INTO chat_moderation (message_id, at) VALUES (?, ?)", (target, at)) if target.isdecimal(): # Keep row/ordering/cooldown intact without retaining the removed text. connection.execute("UPDATE chat_messages SET message = '[Message removed]' WHERE id = ? AND at = ?", (int(target), at)) else: connection.execute("UPDATE community_milestones SET message = '[Message removed]' WHERE 'milestone:' || id = ? AND announced_at = ?", (target, at)) connection.execute("UPDATE season_milestone_messages SET message = '[Message removed]' WHERE id = ? AND at = ?", (target, at)) else: player = state["players"].get(target) if player is None or player.get("is_deleted"): raise AdminActionError(HTTPStatus.NOT_FOUND, "Player not found") harms_admin = action == "delete_player" or (action == "set_admin" and not data["value"]) or (action == "set_banned" and data["value"]) if target == actor and harms_admin: raise AdminActionError(HTTPStatus.CONFLICT, "You cannot remove your own access") if active_admin(player) and harms_admin and sum(active_admin(p) for p in state["players"].values()) <= 1: raise AdminActionError(HTTPStatus.CONFLICT, "The last active administrator must be preserved") if action == "set_admin" and data["value"] and player.get("is_banned"): raise AdminActionError(HTTPStatus.CONFLICT, "Unban this player before granting admin access") replacement = copy.deepcopy(state) changed = replacement["players"][target] if action == "set_admin": changed["is_admin"] = data["value"] # Reauthentication is required after any privilege change. changed["token_hashes"] = [] elif action == "set_banned": changed["is_banned"] = data["value"] changed["token_hashes"] = [] else: original_name = player["name"] changed.update(name="Deleted player", password_hash=None, token_hashes=[], subscribed=False, is_admin=False, is_banned=True, is_deleted=True) # Preserve scored presses and FK integrity with an inert anonymized tombstone. connection.execute("UPDATE chat_messages SET name = 'Deleted player', message = '[Account deleted]' WHERE player_id = ? AND kind = 'message'", (target,)) connection.execute("UPDATE chat_messages SET message = '[Press by deleted player]' WHERE player_id = ? AND kind = 'press'", (target,)) # Legacy archive standings lack IDs. Exact-name matches are anonymized conservatively. for row in connection.execute("SELECT id, leaderboard FROM season_results").fetchall(): standings = json.loads(row["leaderboard"]) for score in standings: if score["name"] == original_name: score["name"] = "Deleted player" connection.execute("UPDATE season_results SET leaderboard = ? WHERE id = ?", (json.dumps(standings), row["id"])) write_state(connection, replacement) connection.execute("INSERT INTO admin_audit (at, actor_id, action, target_id) VALUES (?, ?, ?, ?)", (now_ms(), actor, action, target)) if replacement is not None: state.clear() state.update(replacement) revoke_player_sessions(target) if original_name is not None: failed_logins.pop(original_name.lower(), None) class Handler(BaseHTTPRequestHandler): server_version = "TheButton/1.0" def log_message(self, fmt, *args): pass def send_json(self, status, payload): body = json.dumps(payload).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Cache-Control", "no-store") self.send_common_headers() self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def send_common_headers(self): self.send_header("X-Content-Type-Options", "nosniff") self.send_header("X-Frame-Options", "DENY") self.send_header("Referrer-Policy", "no-referrer") self.send_header( "Content-Security-Policy", "default-src 'self'; script-src 'self' https://js.stripe.com; " "frame-src 'self' https://js.stripe.com; frame-ancestors 'none'; " "base-uri 'none'; form-action 'self'", ) def read_json(self): length = int(self.headers.get("Content-Length") or 0) if length > MAX_BODY_BYTES: raise ValueError("Request body too large") raw = self.rfile.read(length) if length else b"{}" data = json.loads(raw.decode("utf-8")) if not isinstance(data, dict): raise ValueError("Expected a JSON object") return data def bearer_hash(self): auth = self.headers.get("Authorization", "") if not auth.startswith("Bearer "): return None return hash_token(auth[7:].strip()) def authed_player(self): th = self.bearer_hash() pid = token_index.get(th) if th else None player = state["players"].get(pid) return pid if player and not player.get("is_banned") and not player.get("is_deleted") else None def require_admin(self): pid = self.authed_player() if pid is None: self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "Log in as an administrator"}) return None if not active_admin(state["players"][pid]): self.send_json(HTTPStatus.FORBIDDEN, {"error": "Administrator access required"}) return None return pid def do_GET(self): path = self.path.split("?", 1)[0] if path == "/api/admin": with lock: actor = self.require_admin() if actor is None: return check_ended() return self.send_json(HTTPStatus.OK, admin_overview(actor)) if path == "/api/state": with lock: check_ended() return self.send_json(HTTPStatus.OK, { "now": now_ms(), "started_at": state["started_at"], "deadline": state["deadline"], "duration_ms": DURATION_MS, "ended": state["ended"], "total_players": len(state["players"]), "total_presses": len(state["presses"]), "zones": ZONES, "season": season_summary(), }) if path == "/api/seasons": with lock: check_ended() return self.send_json(HTTPStatus.OK, seasonal_leaderboard()) if path == "/api/leaderboard": with lock: check_ended() return self.send_json(HTTPStatus.OK, {"leaderboard": public_leaderboard()}) if path == "/api/chat": return self.send_json(HTTPStatus.OK, {"messages": chat_messages()}) if path == "/api/presence": with lock: return self.send_json(HTTPStatus.OK, public_presence()) if path == "/api/me": with lock: check_ended() pid = self.authed_player() if pid is None: return self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "Not registered"}) return self.send_json(HTTPStatus.OK, player_summary(pid)) if path == "/api/stats": with lock: check_ended() if not state["ended"]: return self.send_json(HTTPStatus.FORBIDDEN, {"error": "Stats are published when the game ends"}) return self.send_json(HTTPStatus.OK, global_stats()) if path in STATIC_FILES: filename, ctype = STATIC_FILES[path] body = (STATIC_DIR / filename).read_bytes() self.send_response(HTTPStatus.OK) self.send_header("Content-Type", ctype) self.send_common_headers() self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) return self.send_json(HTTPStatus.NOT_FOUND, {"error": "Not found"}) def do_POST(self): path = self.path.split("?", 1)[0] if path.startswith("/api/admin"): # Authorization precedes parsing/lookup; unknown admin paths also deny by default. with lock: actor = self.require_admin() if actor is None: return if path != "/api/admin/action": return self.send_json(HTTPStatus.NOT_FOUND, {"error": "Not found"}) try: data = self.read_json() if self.headers.get("Sec-Fetch-Site") == "cross-site": return self.send_json(HTTPStatus.FORBIDDEN, {"error": "Cross-site admin requests are not allowed"}) admin_action(actor, data) except AdminActionError as error: return self.send_json(error.status, {"error": str(error)}) except sqlite3.Error: return self.send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "Admin action could not be saved; no changes were applied"}) except (ValueError, UnicodeDecodeError): return self.send_json(HTTPStatus.BAD_REQUEST, {"error": "Invalid request body"}) return self.send_json(HTTPStatus.OK, {"ok": True}) try: data = self.read_json() except (ValueError, UnicodeDecodeError): return self.send_json(HTTPStatus.BAD_REQUEST, {"error": "Invalid JSON body"}) if path == "/api/presence": with lock: th = self.bearer_hash() if self.authed_player() is None: return self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "Log in to share your presence"}) tab = data.get("tab_id") active, in_chat = data.get("active"), data.get("in_chat", False) if (not isinstance(tab, str) or re.fullmatch(r"[a-f0-9]{32}", tab) is None or not isinstance(active, bool) or not isinstance(in_chat, bool)): return self.send_json(HTTPStatus.BAD_REQUEST, {"error": "Invalid presence update"}) prune_presence() key = (th, tab) if not active: presence.pop(key, None) else: if key not in presence and sum(k[0] == th for k in presence) >= MAX_PRESENCE_TABS_PER_SESSION: return self.send_json(HTTPStatus.TOO_MANY_REQUESTS, {"error": "Too many active tabs"}) presence[key] = (time.monotonic(), in_chat) return self.send_json(HTTPStatus.OK, public_presence()) if path == "/api/chat": with lock: pid = self.authed_player() if pid is None: return self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "Log in to send messages"}) message = data.get("message") if not isinstance(message, str) or not (1 <= len(message.strip()) <= MAX_CHAT_LENGTH): return self.send_json(HTTPStatus.BAD_REQUEST, {"error": f"Message must be 1-{MAX_CHAT_LENGTH} characters"}) message = message.strip() if not message.isprintable(): return self.send_json(HTTPStatus.BAD_REQUEST, {"error": "Message contains invalid characters"}) with database() as connection: connection.execute("BEGIN IMMEDIATE") t = now_ms() last = connection.execute( "SELECT at FROM chat_messages WHERE player_id = ? AND kind = 'message' ORDER BY id DESC LIMIT 1", (pid,), ).fetchone() if last is not None and t - last["at"] < CHAT_COOLDOWN_MS: return self.send_json(HTTPStatus.TOO_MANY_REQUESTS, {"error": "Please wait two seconds between messages"}) name = state["players"][pid]["name"] cursor = connection.execute( "INSERT INTO chat_messages (player_id, name, at, message) VALUES (?, ?, ?, ?)", (pid, name, t, message), ) posted = {"id": cursor.lastrowid, "name": name, "at": t, "message": message, "kind": "message"} return self.send_json(HTTPStatus.CREATED, posted) if path == "/api/register": name, password, error = validate_credentials(data) if error: return self.send_json(HTTPStatus.BAD_REQUEST, {"error": error}) password_hash = hash_password(password) # slow; done outside the lock with lock: check_ended() if state["ended"]: return self.send_json(HTTPStatus.CONFLICT, {"error": "The game is over"}) if find_player_by_name(name) is not None: return self.send_json(HTTPStatus.CONFLICT, {"error": "Name already taken"}) pid = secrets.token_hex(8) state["players"][pid] = { "name": name, "password_hash": password_hash, "token_hashes": [], "created_at": now_ms(), } token = issue_token(pid) return self.send_json(HTTPStatus.CREATED, {"name": name, "token": token}) if path == "/api/login": name, password = data.get("name"), data.get("password") if not isinstance(name, str) or not isinstance(password, str) or len(password) > MAX_PASSWORD_LEN: return self.send_json(HTTPStatus.BAD_REQUEST, {"error": "Name and password required"}) name = " ".join(name.split()) key = name.lower() with lock: count, first = failed_logins.get(key, (0, 0)) if count >= LOGIN_MAX_FAILURES and now_ms() - first < LOGIN_LOCKOUT_MS: return self.send_json(HTTPStatus.TOO_MANY_REQUESTS, {"error": "Too many failed attempts; try again later"}) pid = find_player_by_name(name) stored = state["players"][pid].get("password_hash") if pid else None ok = verify_password(password, stored or DUMMY_PASSWORD_HASH) and stored is not None with lock: current_player = state["players"].get(pid) ok = bool(ok and current_player and not current_player.get("is_banned") and not current_player.get("is_deleted") and current_player.get("password_hash") == stored) if not ok: count, first = failed_logins.get(key, (0, 0)) if now_ms() - first >= LOGIN_LOCKOUT_MS: count, first = 0, now_ms() failed_logins[key] = (count + 1, first) return self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "Invalid name or password"}) failed_logins.pop(key, None) token = issue_token(pid) return self.send_json(HTTPStatus.OK, {"name": state["players"][pid]["name"], "token": token}) if path == "/api/logout": with lock: th = self.bearer_hash() pid = token_index.pop(th, None) if th else None prune_presence() if pid is not None: state["players"][pid]["token_hashes"].remove(th) save_state(state) return self.send_json(HTTPStatus.OK, {"ok": True}) if path == "/api/subscribe": with lock: pid = self.authed_player() player_name = state["players"][pid]["name"] if pid else None already_subscribed = bool(pid and state["players"][pid].get("subscribed")) if pid is None: return self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "Log in to join the mailing list"}) if already_subscribed: return self.send_json(HTTPStatus.OK, {"ok": True}) entry, error = validate_subscription(data) if error: return self.send_json(HTTPStatus.BAD_REQUEST, {"error": error}) entry["player"] = player_name with mailing_lock: # Same response for repeat emails so the endpoint doesn't reveal who is subscribed. if entry["email"].lower() not in subscribed_emails: add_subscriber(entry) with lock: if self.authed_player() != pid: return self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "Log in to join the mailing list"}) state["players"][pid]["subscribed"] = True save_state(state) return self.send_json(HTTPStatus.OK, {"ok": True}) if path == "/api/press": with lock: pid = self.authed_player() if pid is None: return self.send_json(HTTPStatus.UNAUTHORIZED, {"error": "Not registered"}) t = now_ms() check_ended(t) if state["ended"]: return self.send_json(HTTPStatus.CONFLICT, {"error": "The game is over"}) remaining = state["deadline"] - t zone = zone_for_remaining(remaining) press = { "player_id": pid, "at": t, "remaining_ms": remaining, "zone": zone["name"], "points": zone["points"], } state["presses"].append(press) with database() as connection: season_end = current_season(connection)["ends_at"] state["deadline"] = min(t + DURATION_MS, season_end) hours, remainder = divmod(max(0, remaining), 3600000) minutes, remainder = divmod(remainder, 60000) seconds, milliseconds = divmod(remainder, 1000) time_left = f"{hours:02d}:{minutes:02d}:{seconds:02d}.{milliseconds:03d}" name = state["players"][pid]["name"] points = press["points"] announcement = f"{name} pushed the button with {time_left} remaining and received {points} {'point' if points == 1 else 'points'}." try: with database() as connection: write_state(connection, state) connection.execute( "INSERT INTO chat_messages (player_id, name, at, message, kind) VALUES (?, ?, ?, ?, 'press')", (pid, "The Button", t, announcement), ) sync_milestones(connection, state, t) except Exception: # Keep memory consistent with the rolled-back database snapshot. state["presses"].pop() state["deadline"] = t + remaining raise return self.send_json(HTTPStatus.OK, { "press": {k: press[k] for k in ("at", "remaining_ms", "zone", "points")}, "deadline": state["deadline"], "now": t, }) self.send_json(HTTPStatus.NOT_FOUND, {"error": "Not found"}) mailing_lock = threading.Lock() def load_subscribed_emails(): if not MAILING_LIST_FILE.exists(): return set() with MAILING_LIST_FILE.open("r", newline="", encoding="utf-8") as f: return {row["email"].lower() for row in csv.DictReader(f)} subscribed_emails = load_subscribed_emails() def clean_text(value, max_len): """Returns a whitespace-normalized string, "" for missing values, or None if invalid.""" if value is None: return "" if not isinstance(value, str): return None value = " ".join(value.split()) if len(value) > max_len or not value.isprintable(): return None return value def validate_subscription(data): first = clean_text(data.get("first_name"), 50) last = clean_text(data.get("last_name"), 50) email = clean_text(data.get("email"), 254) phone = clean_text(data.get("phone"), 25) if not first: return None, "First name is required (max 50 characters)" if last is None: return None, "Last name must be at most 50 characters" if not email or not EMAIL_RE.match(email): return None, "A valid email address is required" if phone is None or (phone and not PHONE_RE.match(phone)): return None, "Phone number is invalid" return {"first_name": first, "last_name": last, "email": email, "phone": phone}, None def csv_safe(value): # Stops spreadsheet apps from interpreting user input as a formula. return "'" + value if value[:1] in ("=", "+", "-", "@") else value def add_subscriber(entry): """Must be called with mailing_lock held.""" is_new_file = not MAILING_LIST_FILE.exists() row = {k: csv_safe(v) for k, v in entry.items()} row["subscribed_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) with MAILING_LIST_FILE.open("a", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=MAILING_LIST_FIELDS) if is_new_file: writer.writeheader() writer.writerow(row) subscribed_emails.add(entry["email"].lower()) def end_watcher(): while True: with lock: check_ended() with database() as connection: next_boundary = current_season(connection)["ends_at"] wake_at = next_boundary if state["ended"] else min(state["deadline"], next_boundary) wait = (wake_at - now_ms()) / 1000 time.sleep(min(max(wait, 0.05), 60)) if __name__ == "__main__": threading.Thread(target=end_watcher, daemon=True).start() print(f"The Button running on http://{HOST}:{PORT} (timer: {DURATION_MS / 1000:.0f}s)") ThreadingHTTPServer((HOST, PORT), Handler).serve_forever()