the-button/server.py

"""The Button - a single-file, dependency-free game server (Python 3.9+)."""

import csv
import hashlib
import hmac
import json
import os
import re
import secrets
import threading
import time
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"))
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

# 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 = {
    "/": ("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)


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}
    }


def load_state():
    if DATA_FILE.exists():
        with DATA_FILE.open("r", encoding="utf-8") as f:
            return json.load(f)
    state = new_game()
    save_state(state)
    return state


def save_state(state):
    tmp = DATA_FILE.with_suffix(".tmp")
    with tmp.open("w", encoding="utf-8") as f:
        json.dump(state, f)
    os.replace(tmp, DATA_FILE)


state = load_state()
for _p in state["players"].values():
    if "token_hash" in _p:  # migrate pre-login data
        _p["token_hashes"] = [_p.pop("token_hash")]
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)


def find_player_by_name(name):
    for pid, p in state["players"].items():
        if 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 check_ended():
    """Must be called with the lock held."""
    if not state["ended"] and now_ms() >= state["deadline"]:
        state["ended"] = True
        state["ended_at"] = state["deadline"]
        save_state(state)


def empty_zone_counts():
    return {z["name"]: 0 for z in ZONES}


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
    return {
        "name": state["players"][pid]["name"],
        "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),
        "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 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"]
        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],
    }


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()
        return token_index.get(th) if th else None

    def do_GET(self):
        path = self.path.split("?", 1)[0]
        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,
                })
        if path == "/api/me":
            with lock:
                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]
        try:
            data = self.read_json()
        except (ValueError, UnicodeDecodeError):
            return self.send_json(HTTPStatus.BAD_REQUEST, {"error": "Invalid JSON body"})

        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:
                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
                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:
                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"})
                check_ended()
                if state["ended"]:
                    return self.send_json(HTTPStatus.CONFLICT, {"error": "The game is over"})
                t = now_ms()
                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)
                state["deadline"] = t + DURATION_MS
                save_state(state)
                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()
            if state["ended"]:
                return
            wait = (state["deadline"] - 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()