""" tactical-shooter — Master Server API ====================================== Standalone REST service that receives heartbeat pings from game servers and serves the server list to client browsers. Endpoints: POST /api/v1/heartbeat — Register/refresh a game server GET /api/v1/servers — List all active servers GET /api/v1/servers/:id — Get a single server's details GET /api/v1/status — Health check + stats Data: SQLite with TTL-based expiry (servers older than heartbeat_ttl are pruned). """ from __future__ import annotations import hashlib import json import logging import os import time import uuid from collections.abc import AsyncIterator from contextlib import asynccontextmanager from datetime import datetime, timezone from pathlib import Path import aiosqlite import uvicorn from fastapi import FastAPI, HTTPException, Request, UploadFile, File, Form from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse, FileResponse from pydantic import BaseModel, Field # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- CONFIG_PATH = Path(__file__).resolve().parent / "master_config.json" with open(CONFIG_PATH) as f: _cfg = json.load(f) HOST: str = _cfg.get("host", "0.0.0.0") PORT: int = _cfg.get("port", 28961) DB_PATH: str = _cfg.get("database", "master_server.db") HEARTBEAT_TTL: int = _cfg.get("heartbeat_ttl_seconds", 180) MAX_SERVERS: int = _cfg.get("max_servers", 1000) RATE_LIMIT: int = _cfg.get("rate_limit_per_ip", 30) RATE_LIMIT_WINDOW: int = _cfg.get("rate_limit_window_seconds", 60) DB_PATH = os.environ.get("MASTER_DB_PATH", DB_PATH) log = logging.getLogger("master_server") log.setLevel(_cfg.get("log_level", "info").upper()) # --------------------------------------------------------------------------- # Schemas # --------------------------------------------------------------------------- class HeartbeatPayload(BaseModel): """Payload sent by game servers via POST /heartbeat.""" server_id: str | None = None # auto-generated if missing name: str = "Unnamed Server" map: str = "unknown" players: int = 0 max_players: int = 16 game_mode: str = "deathmatch" version: str = "1.0.0" tags: str = "" password: bool = False host: str = "" port: int = 0 steam_id: str = "" class ServerInfo(BaseModel): """Public server info returned to clients.""" server_id: str name: str map: str players: int max_players: int game_mode: str version: str tags: list[str] password: bool host: str port: int steam_id: str last_seen: str # ISO-8601 uptime_seconds: int class ServerListResponse(BaseModel): count: int servers: list[ServerInfo] class StatusResponse(BaseModel): status: str = "ok" uptime_seconds: int active_servers: int total_heartbeats: int version: str = "1.0.0" # --------------------------------------------------------------------------- # Map Registry Models # --------------------------------------------------------------------------- # Directory where uploaded .pck files are stored MAPS_DIR: str = _cfg.get("maps_dir", "workshop_maps") os.makedirs(MAPS_DIR, exist_ok=True) class MapInfo(BaseModel): """Public map metadata returned to clients.""" name: str size: int = 0 version: int = 1 description: str = "" scene: str = "" author: str = "" checksum_sha256: str = "" uploaded_at: str = "" download_url: str = "" class MapListResponse(BaseModel): count: int maps: list[MapInfo] # --------------------------------------------------------------------------- # Database # --------------------------------------------------------------------------- CREATE_TABLE_SQL = """ CREATE TABLE IF NOT EXISTS servers ( server_id TEXT PRIMARY KEY, name TEXT NOT NULL, map TEXT NOT NULL DEFAULT 'unknown', players INTEGER NOT NULL DEFAULT 0, max_players INTEGER NOT NULL DEFAULT 16, game_mode TEXT NOT NULL DEFAULT 'deathmatch', version TEXT NOT NULL DEFAULT '1.0.0', tags TEXT NOT NULL DEFAULT '', password INTEGER NOT NULL DEFAULT 0, host TEXT NOT NULL DEFAULT '', port INTEGER NOT NULL DEFAULT 0, steam_id TEXT NOT NULL DEFAULT '', first_seen REAL NOT NULL, last_seen REAL NOT NULL, heartbeat_count INTEGER NOT NULL DEFAULT 1 ); """ CREATE_INDEX_SQL = "CREATE INDEX IF NOT EXISTS idx_servers_last_seen ON servers(last_seen);" CREATE_MAPS_SQL = """ CREATE TABLE IF NOT EXISTS workshop_maps ( name TEXT PRIMARY KEY, size INTEGER NOT NULL DEFAULT 0, version INTEGER NOT NULL DEFAULT 1, description TEXT NOT NULL DEFAULT '', scene TEXT NOT NULL DEFAULT '', author TEXT NOT NULL DEFAULT '', checksum_sha256 TEXT NOT NULL DEFAULT '', filename TEXT NOT NULL, uploaded_at REAL NOT NULL ); """ async def get_db() -> AsyncIterator[aiosqlite.Connection]: async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row await db.execute("PRAGMA journal_mode=WAL") await db.execute("PRAGMA synchronous=NORMAL") await db.execute(CREATE_TABLE_SQL) await db.execute(CREATE_INDEX_SQL) await db.commit() yield db async def _prune_expired(db: aiosqlite.Connection) -> int: """Remove servers whose heartbeat has expired. Returns count pruned.""" cutoff = time.time() - HEARTBEAT_TTL db.row_factory = aiosqlite.Row cursor = await db.execute("DELETE FROM servers WHERE last_seen < ?", (cutoff,)) await db.commit() return cursor.rowcount def _row_to_server(row: aiosqlite.Row) -> ServerInfo: now = time.time() tags_list = [t.strip() for t in row["tags"].split(",") if t.strip()] if row["tags"] else [] return ServerInfo( server_id=row["server_id"], name=row["name"], map=row["map"], players=row["players"], max_players=row["max_players"], game_mode=row["game_mode"], version=row["version"], tags=tags_list, password=bool(row["password"]), host=row["host"], port=row["port"], steam_id=row["steam_id"], last_seen=datetime.fromtimestamp(row["last_seen"], tz=timezone.utc).isoformat(), uptime_seconds=int(max(0, now - row["first_seen"])), ) # --------------------------------------------------------------------------- # App # --------------------------------------------------------------------------- _start_time = time.time() @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncIterator[None]: """Startup / shutdown — ensure DB schema exists.""" async with aiosqlite.connect(DB_PATH) as db: await db.execute("PRAGMA journal_mode=WAL") await db.execute("PRAGMA synchronous=NORMAL") await db.execute(CREATE_TABLE_SQL) await db.execute(CREATE_INDEX_SQL) await db.execute(CREATE_MAPS_SQL) await db.commit() # Ensure maps directory exists os.makedirs(MAPS_DIR, exist_ok=True) log.info("Master server ready on %s:%s", HOST, PORT) log.info("Maps directory: %s (absolute: %s)", MAPS_DIR, os.path.abspath(MAPS_DIR)) yield app = FastAPI( title="Tactical Shooter Master Server", version="1.0.0", lifespan=lifespan, ) # Allow cross-origin from any game client / browser app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=False, allow_methods=["*"], allow_headers=["*"], ) # In-memory rate limiter: {ip: [timestamps]} _rate_buckets: dict[str, list[float]] = {} def _check_rate_limit(ip: str) -> None: now = time.time() window_start = now - RATE_LIMIT_WINDOW timestamps = _rate_buckets.get(ip, []) # Drop old entries timestamps = [t for t in timestamps if t > window_start] if len(timestamps) >= RATE_LIMIT: raise HTTPException( status_code=429, detail=f"Rate limit exceeded: max {RATE_LIMIT} requests per {RATE_LIMIT_WINDOW}s", ) timestamps.append(now) _rate_buckets[ip] = timestamps # --------------------------------------------------------------------------- # Routes # --------------------------------------------------------------------------- @app.get("/api/v1/status", response_model=StatusResponse) async def get_status(): """Health check and server stats.""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row cursor = await db.execute("SELECT COUNT(*) AS cnt FROM servers") row = await cursor.fetchone() active = row["cnt"] if row else 0 cursor = await db.execute("SELECT COALESCE(SUM(heartbeat_count), 0) AS total FROM servers") row = await cursor.fetchone() total_heartbeats = row["total"] if row else 0 return StatusResponse( uptime_seconds=int(time.time() - _start_time), active_servers=active, total_heartbeats=total_heartbeats, ) @app.post("/api/v1/heartbeat") async def heartbeat(payload: HeartbeatPayload, request: Request): """Register or refresh a game server. If server_id is not provided (or empty), one is auto-generated in the response. The server should store the returned id and send it on subsequent heartbeats. """ ip = request.client.host if request.client else "unknown" _check_rate_limit(ip) # Generate id for new servers server_id = payload.server_id or str(uuid.uuid4()) now = time.time() async with aiosqlite.connect(DB_PATH) as db: # Check if we're at capacity db.row_factory = aiosqlite.Row cursor = await db.execute("SELECT COUNT(*) AS cnt FROM servers") row = await cursor.fetchone() count = row["cnt"] if row else 0 if count >= MAX_SERVERS: # Try to prune expired first await _prune_expired(db) cursor = await db.execute("SELECT COUNT(*) AS cnt FROM servers") row = await cursor.fetchone() count = row["cnt"] if row else 0 if count >= MAX_SERVERS: raise HTTPException(status_code=503, detail="Server registry full") # Upsert await db.execute( """ INSERT INTO servers (server_id, name, map, players, max_players, game_mode, version, tags, password, host, port, steam_id, first_seen, last_seen, heartbeat_count) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1) ON CONFLICT(server_id) DO UPDATE SET name = excluded.name, map = excluded.map, players = excluded.players, max_players = excluded.max_players, game_mode = excluded.game_mode, version = excluded.version, tags = excluded.tags, password = excluded.password, host = excluded.host, port = excluded.port, steam_id = excluded.steam_id, last_seen = excluded.last_seen, heartbeat_count = servers.heartbeat_count + 1 """, ( server_id, payload.name, payload.map, payload.players, payload.max_players, payload.game_mode, payload.version, payload.tags, 1 if payload.password else 0, payload.host or ip, payload.port, payload.steam_id, now, now, ), ) await db.commit() log.info("Heartbeat from %s (%s) — %s/%s players on %s", payload.name, server_id[:8], payload.players, payload.max_players, payload.map) return { "status": "ok", "server_id": server_id, "registered": True, } @app.get("/api/v1/servers", response_model=ServerListResponse) async def list_servers(request: Request, tags: str | None = None): """List all active servers, optionally filtered by comma-separated tags.""" _check_rate_limit(request.client.host if request.client else "unknown") async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row await _prune_expired(db) if tags: # Filter servers whose tags field contains any of the requested tags tag_list = [t.strip() for t in tags.split(",") if t.strip()] placeholders = ",".join("?" for _ in tag_list) query = f""" SELECT * FROM servers WHERE last_seen > ? AND ({' OR '.join(f"tags LIKE ?" for _ in tag_list)}) ORDER BY players DESC """ params = [time.time() - HEARTBEAT_TTL] for tg in tag_list: params.append(f"%{tg}%") cursor = await db.execute(query, params) else: cursor = await db.execute( "SELECT * FROM servers WHERE last_seen > ? ORDER BY players DESC", (time.time() - HEARTBEAT_TTL,), ) rows = await cursor.fetchall() servers = [_row_to_server(r) for r in rows] return ServerListResponse(count=len(servers), servers=servers) @app.get("/api/v1/servers/{server_id}", response_model=ServerInfo) async def get_server(server_id: str): """Get details for a single server.""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row cursor = await db.execute("SELECT * FROM servers WHERE server_id = ?", (server_id,)) row = await cursor.fetchone() if not row: raise HTTPException(status_code=404, detail="Server not found") now = time.time() if row["last_seen"] < now - HEARTBEAT_TTL: raise HTTPException(status_code=410, detail="Server has expired (no recent heartbeat)") return _row_to_server(row) # --------------------------------------------------------------------------- # Map Registry — Workshop API # --------------------------------------------------------------------------- def _compute_sha256(filepath: str) -> str: """Compute SHA-256 checksum of a file.""" try: h = hashlib.sha256() with open(filepath, "rb") as f: while True: chunk = f.read(65536) if not chunk: break h.update(chunk) return h.hexdigest() except OSError: return "" def _row_to_map_info(row: aiosqlite.Row) -> MapInfo: """Convert a DB row to a MapInfo response model.""" return MapInfo( name=row["name"], size=row["size"], version=row["version"], description=row["description"], scene=row["scene"], author=row["author"], checksum_sha256=row["checksum_sha256"], uploaded_at=datetime.fromtimestamp(row["uploaded_at"], tz=timezone.utc).isoformat(), download_url=f"/api/v1/maps/{row['name']}.pck", ) @app.get("/api/v1/maps", response_model=MapListResponse) async def list_workshop_maps(): """List all community maps available for download.""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row cursor = await db.execute( "SELECT * FROM workshop_maps ORDER BY name ASC" ) rows = await cursor.fetchall() maps = [_row_to_map_info(r) for r in rows] return MapListResponse(count=len(maps), maps=maps) @app.get("/api/v1/maps/{map_name}") async def get_workshop_map(map_name: str): """Get metadata for a specific workshop map.""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row cursor = await db.execute( "SELECT * FROM workshop_maps WHERE name = ?", (map_name,) ) row = await cursor.fetchone() if not row: raise HTTPException(status_code=404, detail=f"Map '{map_name}' not found") return _row_to_map_info(row) @app.get("/api/v1/maps/{map_name}.pck") async def download_workshop_map(map_name: str): """Download a workshop map .pck file.""" async with aiosqlite.connect(DB_PATH) as db: db.row_factory = aiosqlite.Row cursor = await db.execute( "SELECT * FROM workshop_maps WHERE name = ?", (map_name,) ) row = await cursor.fetchone() if not row: raise HTTPException(status_code=404, detail=f"Map '{map_name}' not found") filepath = os.path.join(MAPS_DIR, row["filename"]) if not os.path.exists(filepath): raise HTTPException(status_code=410, detail="Map file has been removed from storage") return FileResponse( path=filepath, filename=row["filename"], media_type="application/octet-stream", headers={ "Access-Control-Allow-Origin": "*", "X-Checksum-SHA256": row["checksum_sha256"], }, ) @app.post("/api/v1/maps/upload") async def upload_workshop_map( request: Request, file: UploadFile = File(...), name: str = Form(...), description: str = Form(""), scene: str = Form(""), author: str = Form(""), version: int = Form(1), ): """Upload a new workshop map .pck file.""" ip = request.client.host if request.client else "unknown" _check_rate_limit(ip) # Validate file extension if not file.filename or not file.filename.endswith(".pck"): raise HTTPException(status_code=400, detail="Only .pck files are accepted") # Save file to disk filename = f"{name}.pck" filepath = os.path.join(MAPS_DIR, filename) content = await file.read() with open(filepath, "wb") as f: f.write(content) file_size = os.path.getsize(filepath) checksum = _compute_sha256(filepath) now = time.time() async with aiosqlite.connect(DB_PATH) as db: await db.execute("PRAGMA journal_mode=WAL") await db.execute("PRAGMA synchronous=NORMAL") await db.execute( """INSERT OR REPLACE INTO workshop_maps (name, size, version, description, scene, author, checksum_sha256, filename, uploaded_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", (name, file_size, version, description, scene, author, checksum, filename, now), ) await db.commit() log.info("Map uploaded: %s (%s, %d bytes, v%d)", name, filename, file_size, version) return { "status": "ok", "map": name, "size": file_size, "version": version, "checksum_sha256": checksum, } # --------------------------------------------------------------------------- # Entrypoint # --------------------------------------------------------------------------- if __name__ == "__main__": logging.basicConfig( level=_cfg.get("log_level", "info").upper(), format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", ) uvicorn.run( "main:app", host=HOST, port=PORT, log_level=_cfg.get("log_level", "info"), reload=False, )