Initial commit
This commit is contained in:
Eric Meehan -X (ermeehan - ACTALENT SERVICES LLC at Cisco)
commit
fb7a34a95e
@@ -0,0 +1,495 @@
|
||||
"""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"),
|
||||
}
|
||||
|
||||
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_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'; 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"],
|
||||
"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()
|
||||
Reference in New Issue
Block a user