wuwiandClaude Opus 5
20a742fbf3
Nothing secret is in it — the .env never enters the image and the client is public JS regardless — but handing out server.py and the asset-pipeline scripts to anyone who guesses a filename publishes the exact validation rules for free. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
160 lines
7.3 KiB
Python
160 lines
7.3 KiB
Python
#!/usr/bin/env python3
|
|
"""The whole backend: static files plus a two-route high-score API, on the standard library.
|
|
|
|
It replaced the nginx image because the leaderboard needs ONE writable process and nginx is not
|
|
one. Everything nginx was doing here is four lines of headers, so a second container (and a second
|
|
Coolify resource, and a proxy rule) would have bought separation we have no use for at this size.
|
|
|
|
GET /api/scores -> {"scores": [{"n","k","g","t"}, ...]} best first
|
|
POST /api/scores <- {"n","k","g"} -> {"ok", "rank", "scores"}
|
|
|
|
Scores live in ONE json file under SPLAT_DATA (default /data, which must be a persistent volume —
|
|
anything written inside the container image is gone on the next deploy). There is no auth: the
|
|
board is a wall for writing your name on, and the client is public JS that could be made to post
|
|
anything regardless. What IS enforced is that nothing here can be made to crash, hang or grow
|
|
without bound by a hostile request — see clean() and the caps below.
|
|
"""
|
|
import json, os, re, threading, time
|
|
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
DATA = os.environ.get('SPLAT_DATA') or ('/data' if os.path.isdir('/data') else HERE)
|
|
FILE = os.path.join(DATA, 'scores.json')
|
|
PORT = int(os.environ.get('PORT', 80))
|
|
|
|
KEEP = 200 # rows kept on disk; the board only ever shows the top slice of this
|
|
MAX_BODY = 2048 # a submission is ~60 bytes. Anything larger is not one.
|
|
NAME_MAX = 12
|
|
lock = threading.Lock()
|
|
|
|
|
|
def load():
|
|
try:
|
|
with open(FILE) as f:
|
|
rows = json.load(f)
|
|
return rows if isinstance(rows, list) else []
|
|
except (OSError, ValueError):
|
|
return [] # missing, empty or corrupt — an empty board, never a 500
|
|
|
|
|
|
def save(rows):
|
|
os.makedirs(DATA, exist_ok=True)
|
|
tmp = FILE + '.tmp' # write-then-rename: a torn file here is a wiped leaderboard
|
|
with open(tmp, 'w') as f:
|
|
json.dump(rows, f)
|
|
os.replace(tmp, FILE)
|
|
|
|
|
|
def clean(payload):
|
|
"""A submission, or None. Every field is re-derived — nothing from the request reaches the file
|
|
unexamined, because that file is read back and rendered into the page."""
|
|
if not isinstance(payload, dict):
|
|
return None
|
|
name = re.sub(r'[^A-Z0-9 ._-]', '', str(payload.get('n', '')).upper()).strip()[:NAME_MAX]
|
|
if not name:
|
|
return None
|
|
try:
|
|
kills, games = int(payload['k']), int(payload['g'])
|
|
except (KeyError, TypeError, ValueError):
|
|
return None
|
|
if not (0 <= kills <= 1_000_000 and 0 <= games <= 10_000):
|
|
return None # bounds keep the sort honest and the column narrow
|
|
who = re.sub(r'[^a-z0-9]', '', str(payload.get('id', '')))[:40]
|
|
return {'n': name, 'k': kills, 'g': games, 't': int(time.time()), 'id': who}
|
|
|
|
|
|
def public(rows):
|
|
"""The board as anyone may see it. `id` is the ONLY thing standing between a player and someone
|
|
else typing their name, so it must never leave the server."""
|
|
return [{k: v for k, v in r.items() if k != 'id'} for r in rows[:100]]
|
|
|
|
|
|
def rank_of(rows, name):
|
|
return next((i + 1 for i, r in enumerate(rows) if r.get('n') == name), len(rows))
|
|
|
|
|
|
class Handler(SimpleHTTPRequestHandler):
|
|
def __init__(self, *a, **kw):
|
|
super().__init__(*a, directory=HERE, **kw)
|
|
|
|
def log_message(self, *a):
|
|
pass # one line per sprite request is not a log, it is noise
|
|
|
|
# ---- the API ----
|
|
def reply(self, code, obj):
|
|
body = json.dumps(obj).encode()
|
|
self.send_response(code)
|
|
self.send_header('Content-Type', 'application/json')
|
|
self.send_header('Content-Length', str(len(body)))
|
|
self.send_header('Cache-Control', 'no-store')
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def do_GET(self):
|
|
path = self.path.split('?')[0]
|
|
if path == '/api/scores':
|
|
with lock:
|
|
return self.reply(200, {'scores': public(load())})
|
|
# Don't hand out the source. Nothing secret lives in it — the .env is not in the image and
|
|
# the client is public JS regardless — but serving server.py and the asset-pipeline scripts
|
|
# to anyone who guesses the name is untidy, and it publishes the exact validation rules.
|
|
if path.endswith(('.py', '.pyc')) or '__pycache__' in path:
|
|
return self.send_error(404)
|
|
return super().do_GET()
|
|
|
|
def do_POST(self):
|
|
if self.path.split('?')[0] != '/api/scores':
|
|
return self.send_error(404)
|
|
try:
|
|
n = int(self.headers.get('Content-Length') or 0)
|
|
except ValueError:
|
|
return self.reply(400, {'error': 'bad length'})
|
|
if not 0 < n <= MAX_BODY:
|
|
return self.reply(400, {'error': 'bad length'})
|
|
try:
|
|
row = clean(json.loads(self.rfile.read(n)))
|
|
except ValueError:
|
|
row = None
|
|
if not row:
|
|
return self.reply(400, {'error': 'bad score'})
|
|
with lock:
|
|
rows = load()
|
|
mine = [r for r in rows if r.get('n') == row['n']]
|
|
# A NAME BELONGS TO WHOEVER TOOK IT FIRST. The client mints a random id once and keeps
|
|
# it; the row remembers it, and a submission under someone else's name with a different
|
|
# id is refused. That is the whole of the "security" here — enough that a stranger
|
|
# cannot overwrite your row, and not one line more. Clear your browser storage and you
|
|
# lose the name, which is the honest cost of having no accounts.
|
|
if mine and mine[0].get('id') and mine[0]['id'] != row['id']:
|
|
return self.reply(409, {'error': 'name taken'})
|
|
# EXACTLY ONE ROW PER NAME, their best. Without this the board is one player's session:
|
|
# every personal best posts again and a good run buries everyone else under itself.
|
|
# The client only posts on a personal best, but its idea of "best" is per-device — the
|
|
# same player on a second phone starts from zero and posts worse runs, so a weaker
|
|
# submission has to be dropped HERE rather than merely not sorted to the top.
|
|
if row['k'] > max((r.get('k', 0) for r in mine), default=-1):
|
|
rows = [r for r in rows if r.get('n') != row['n']]
|
|
rows.append(row)
|
|
rows.sort(key=lambda r: (-r.get('k', 0), -r.get('g', 0), r.get('t', 0)))
|
|
rows = rows[:KEEP]
|
|
save(rows)
|
|
return self.reply(200, {'ok': True, 'rank': rank_of(rows, row['n']), 'scores': public(rows)})
|
|
|
|
# ---- static, with the headers the nginx conf used to set ----
|
|
def end_headers(self):
|
|
if not self.path.startswith('/api/'):
|
|
self.send_header('Cache-Control', 'no-cache, must-revalidate')
|
|
super().end_headers()
|
|
|
|
def guess_type(self, path):
|
|
# Chrome refuses to install a homescreen app whose manifest arrives as octet-stream, and
|
|
# mimetypes has no .webmanifest entry.
|
|
if path.endswith('.webmanifest'):
|
|
return 'application/manifest+json'
|
|
return super().guess_type(path)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
print(f'splatter-tennis on :{PORT}, scores in {FILE}', flush=True)
|
|
ThreadingHTTPServer(('', PORT), Handler).serve_forever()
|