#!/usr/bin/env python3 """Python 3.10+ / standard library. Run on a trusted server, never in a browser.""" from __future__ import annotations import argparse import http.client import json import os import re import socket import sys import threading import time from urllib.parse import urlencode, urlsplit class ExampleError(Exception): """Only fixed, credential-free messages may reach stderr.""" def api_origin(local_base: str | None, key: str) -> str: if local_base is None: return "https://woksteamapi.com" parsed = urlsplit(local_base) try: port = parsed.port except ValueError: raise ExampleError("Invalid local test address") from None if (parsed.scheme != "http" or parsed.hostname not in {"127.0.0.1", "::1"} or port is None or not 1 <= port <= 65535 or parsed.username is not None or parsed.password is not None or parsed.path not in {"", "/"} or parsed.query or parsed.fragment): raise ExampleError("--local-base requires an HTTP loopback IP and explicit port") if not key.startswith("local-test-"): raise ExampleError("Local fixtures require a dummy WOK_API_KEY starting local-test-") return local_base.rstrip("/") def read_json(origin: str, key: str, path: str, params: dict | None = None, *, allow_missing: bool = False) -> dict | None: """One GET, no redirects or retries, 35-second deadline and 1 MiB body cap.""" parsed = urlsplit(origin) connection_type = (http.client.HTTPSConnection if parsed.scheme == "https" else http.client.HTTPConnection) connection = connection_type(parsed.hostname, parsed.port, timeout=5) deadline = time.monotonic() + 35 timer = None response = None try: connection.connect() transport = connection.sock if transport is None: raise ExampleError("Connection unavailable") def expire(): try: transport.shutdown(socket.SHUT_RDWR) except OSError: pass timer = threading.Timer(max(0.001, deadline - time.monotonic()), expire) timer.daemon = True timer.start() # Also bounds a response that trickles headers or individual bytes. transport.settimeout(max(0.001, deadline - time.monotonic())) target = path + ("?" + urlencode(params) if params else "") connection.request("GET", target, headers={ "Authorization": "Bearer " + key, "Accept": "application/json", }) response = connection.getresponse() if allow_missing and response.status == 404: return None if response.status != 200: raise ExampleError(f"WOK HTTP {response.status}; stopped without retry") if response.getheader("Content-Type", "").split(";")[0].strip().lower() != "application/json": raise ExampleError("Expected a JSON response") parts = [] size = 0 while True: remaining = deadline - time.monotonic() if remaining <= 0: raise ExampleError("Request deadline exceeded; stopped without retry") # read1() may close its socket after consuming Content-Length. # The absolute timer bounds reads without touching that closed socket. chunk = response.read1(65536) if not chunk: break size += len(chunk) if size > 1024 * 1024: raise ExampleError("Response exceeded the 1 MiB example limit") parts.append(chunk) try: def reject_constant(_value): raise ValueError("Non-finite JSON number") value = json.loads(b"".join(parts).decode("utf-8"), parse_constant=reject_constant) except (UnicodeError, ValueError, RecursionError): raise ExampleError("Malformed JSON response") from None if not isinstance(value, dict) or "error" in value: raise ExampleError("Unexpected JSON response") return value except (OSError, http.client.HTTPException, socket.timeout): raise ExampleError("Network failure or timeout; stopped without retry") from None finally: if timer is not None: timer.cancel() if response is not None: response.close() connection.close() def main() -> int: parser = argparse.ArgumentParser(description="Read a Steam profile, then its FACEIT CS2 profile.") parser.add_argument("steam_id", help="17-digit SteamID64") parser.add_argument("--stats", action="store_true", help="Also fetch FACEIT CS2 lifetime statistics") parser.add_argument("--local-base", help="Offline fixture only: http://127.0.0.1:PORT with a dummy key") args = parser.parse_args() try: if not re.fullmatch(r"[0-9]{17}", args.steam_id): raise ExampleError("Supply a 17-digit SteamID64") key = os.environ.get("WOK_API_KEY", "") if not key or len(key) > 4096 or any(ord(char) < 33 or ord(char) > 126 for char in key): raise ExampleError("Set a valid WOK_API_KEY in the server environment") origin = api_origin(args.local_base, key) steam = read_json(origin, key, "/v1/profile", {"steam_id": args.steam_id}) if not isinstance(steam, dict) or steam.get("steamid") != args.steam_id: raise ExampleError("Steam profile identifier did not match the requested SteamID64") output = {"steam_id": args.steam_id, "steam": steam, "faceit": {"status": "unavailable", "profile": None}} profile = read_json(origin, key, "/faceit/data/v4/players", { "game": "cs2", "game_player_id": args.steam_id, }, allow_missing=True) if profile is not None: player_id = profile.get("player_id") if not isinstance(player_id, str) or not re.fullmatch(r"[A-Za-z0-9-]{1,64}", player_id): raise ExampleError("FACEIT response did not contain a valid player_id") if not isinstance(profile.get("games"), dict): raise ExampleError("FACEIT response did not contain a games object") faceit = {"status": "available", "profile": profile} if args.stats: stats = read_json(origin, key, f"/faceit/data/v4/players/{player_id}/stats/cs2", allow_missing=True) if stats is not None and not isinstance(stats.get("lifetime"), dict): raise ExampleError("FACEIT response did not contain lifetime statistics") faceit["lifetime"] = None if stats is None else stats["lifetime"] faceit["lifetime_status"] = ("unavailable" if stats is None else "available" if stats["lifetime"] else "empty") output["faceit"] = faceit print(json.dumps(output, ensure_ascii=True, indent=2, allow_nan=False)) return 0 except ExampleError as exc: print(str(exc), file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())