626 lines
25 KiB
Python
626 lines
25 KiB
Python
"""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 sqlite3
|
|
import threading
|
|
import time
|
|
from contextlib import contextmanager
|
|
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
|
|
|
|
# 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}
|
|
}
|
|
|
|
|
|
@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) VALUES (?, ?, ?, ?, ?)",
|
|
[(pid, p["name"], p.get("password_hash"), p["created_at"], p.get("subscribed", 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 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
|
|
);
|
|
""")
|
|
# 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():
|
|
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)
|
|
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"])
|
|
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)
|
|
return loaded
|
|
|
|
|
|
def save_state(state):
|
|
with database() as connection:
|
|
write_state(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)
|
|
|
|
|
|
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 public_leaderboard():
|
|
"""Public score totals only; do not expose sessions or individual presses."""
|
|
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.values(), key=lambda p: (-p["total_points"], p["name"].lower()))
|
|
|
|
|
|
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/leaderboard":
|
|
with lock:
|
|
return self.send_json(HTTPStatus.OK, {"leaderboard": public_leaderboard()})
|
|
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()
|