--- title: "player sample: Python + FastAPI" description: "Operator Contract v1.4.1 player module, Python + FastAPI. Generated from the contract." --- # player: Python + FastAPI The player lookup OneHazel calls for currency, country, status and active restrictions. Generated from Operator Contract v1.4.1. Key `python/fastapi/player`; the `connect_samples` MCP tool returns this page for that key. ## What it implements | Operation | Purpose | Idempotency key | |---|---|---| | `POST /player/lookup` | Read a player's currency, country, status and active restrictions. | no | Every request is checked in this order: the `Authorization: Bearer` callback key, then `X-OneHazel-Signature` (`t=,v1=`, HMAC-SHA256 of `"." + raw body`, 300 s tolerance, constant-time compare) over the RAW body before it is parsed, then `X-OneHazel-Env`. `test` traffic goes to a separate test ledger and never touches the live one. Errors use the flat `{ code, message, retryable }` envelope; money is integer minor units. ## Run it ```bash pip install fastapi uvicorn ONEHAZEL_CALLBACK_KEY= ONEHAZEL_SIGNING_SECRET= uvicorn onehazel_player:app --port 3000 # base URL to register: https:///onehazel/v1 (ONEHAZEL_BASE_PATH changes the path) ``` Environment: `ONEHAZEL_CALLBACK_KEY` and `ONEHAZEL_SIGNING_SECRET` are the values `connect_register_endpoint` showed you once. During a secret rotation set `ONEHAZEL_SIGNING_SECRET_PREVIOUS` to the old secret for the overlap hour. `ONEHAZEL_TEST_PLAYER_ID` (default `test-player-1`) is the funded player on the test ledger. ## Code `onehazel_player.py` ```python """OneHazel player module, Operator Contract v1.4.1. Generated; regenerate rather than edit.""" from __future__ import annotations import hashlib import hmac import json import os import re import secrets import sys import threading import time from typing import Any, Callable, Optional from fastapi import FastAPI, Request from fastapi.responses import JSONResponse # Generated from the OneHazel Operator Contract v1.4.1: operations, request fields, # error codes with their HTTP status, and the signature tolerance. Do not hand-edit; regenerate. CONTRACT: dict[str, Any] = json.loads(r'''{ "version": "1.4.1", "auth": { "scheme": "timestamped", "bearer": true, "envHeader": true }, "toleranceSec": 300, "errorStatus": { "unauthorized": 401, "signature_invalid": 401, "idempotency_key_conflict": 409, "launch_token_consumed": 409, "round_closed": 409, "currency_unsupported": 422, "currency_mismatch": 422, "player_blocked": 422, "rollback_amount_mismatch": 422, "insufficient_funds": 402, "internal_error": 500, "unavailable": 500, "invalid_request": 400, "session_unknown": 404, "session_expired": 404, "launch_token_unknown": 404, "launch_token_expired": 404, "player_unknown": 404, "game_unknown": 404, "transaction_unknown": 404, "freeround_unknown": 404 }, "retryable": [ "internal_error", "unavailable" ], "ops": [ { "path": "/player/lookup", "operationId": "playerLookup", "summary": "Read a player's currency, country, status and active restrictions.", "idempotent": false, "fields": [ { "name": "player_external_id", "type": "string", "required": true, "minLength": 1 } ], "closed": true } ] }''') # ───────────────────────────── configuration ───────────────────────────── def _required_env(name: str) -> str: value = os.environ.get(name, "") if not value: # Never print the value of a secret, only its name. print(f"{name} is not set", file=sys.stderr) sys.exit(1) return value # During a rotation OneHazel signs with the new secret; the old one stays valid for 1 hour. SIGNING_SECRETS = [s for s in (_required_env("ONEHAZEL_SIGNING_SECRET"), os.environ.get("ONEHAZEL_SIGNING_SECRET_PREVIOUS", "")) if s] BASE_PATH = os.environ.get("ONEHAZEL_BASE_PATH", "/onehazel/v1").rstrip("/") Reply = tuple[int, dict[str, Any]] Header = Callable[[str], Optional[str]] # ───────────────────────────── errors ───────────────────────────── class ContractError(Exception): """Raised anywhere in a handler; becomes the flat {code, message, retryable} envelope.""" def __init__(self, code: str, message: str) -> None: super().__init__(message) self.code = code self.message = message def error_reply(code: str, message: str) -> Reply: return CONTRACT["errorStatus"][code], {"code": code, "message": message, "retryable": code in CONTRACT["retryable"]} # ───────────────────────────── request validation ───────────────────────────── _TYPES: dict[str, Callable[[Any], bool]] = { "integer": lambda v: isinstance(v, int) and not isinstance(v, bool), "string": lambda v: isinstance(v, str), "boolean": lambda v: isinstance(v, bool), "array": lambda v: isinstance(v, list), "object": lambda v: isinstance(v, dict), } def check_fields(value: Any, fields: list[dict[str, Any]], closed: bool, where: str) -> dict[str, Any]: if not isinstance(value, dict): raise ContractError("invalid_request", f"{where or 'body'} must be a JSON object") if closed: known = {f["name"] for f in fields} for key in value: if key not in known: raise ContractError("invalid_request", f"unknown field {where}{key}") for f in fields: name = where + f["name"] v = value.get(f["name"]) if v is None: if f["required"]: raise ContractError("invalid_request", f"{name} is required") continue if not _TYPES.get(f["type"], lambda _v: True)(v): raise ContractError("invalid_request", f"{name} must be {f['type']}") if isinstance(v, int) and not isinstance(v, bool): if "minimum" in f and v < f["minimum"]: raise ContractError("invalid_request", f"{name} is below {f['minimum']}") if "exclusiveMinimum" in f and v <= f["exclusiveMinimum"]: raise ContractError("invalid_request", f"{name} must be greater than {f['exclusiveMinimum']}") if isinstance(v, str): if "minLength" in f and len(v) < f["minLength"]: raise ContractError("invalid_request", f"{name} is too short") if "maxLength" in f and len(v) > f["maxLength"]: raise ContractError("invalid_request", f"{name} is too long") if "pattern" in f and not re.search(f["pattern"], v): raise ContractError("invalid_request", f"{name} is malformed") if "enum" in f and v not in f["enum"]: raise ContractError("invalid_request", f"{name} is not an allowed value") if "fields" in f: check_fields(v, f["fields"], False, f"{name}.") return value def parse_body(raw_body: bytes) -> Any: try: return json.loads(raw_body) except ValueError: raise ContractError("invalid_request", "body is not valid JSON") from None def find_op(path: str) -> Optional[dict[str, Any]]: return next((o for o in CONTRACT["ops"] if o["path"] == path), None) # ───────────────────────────── OneHazel-to-operator call authentication ───────────────────────────── CALLBACK_KEY = _required_env("ONEHAZEL_CALLBACK_KEY") TEST_PLAYER_ID = os.environ.get("ONEHAZEL_TEST_PLAYER_ID", "test-player-1") TEST_PLAYER_BALANCE = int(os.environ.get("ONEHAZEL_TEST_PLAYER_BALANCE", "100000")) TEST_PLAYER_CURRENCY = os.environ.get("ONEHAZEL_TEST_PLAYER_CURRENCY", "EUR") def verify_signature(raw_body: bytes, header: Optional[str], keys: list[str], tolerance_sec: int, now: Optional[int] = None) -> bool: """X-OneHazel-Signature: t=,v1=." + raw body)>. Runs over the RAW bytes, before the body is parsed. Constant-time compare. """ if not header: return False t: Optional[int] = None macs: list[str] = [] for part in header.split(","): k, _, v = part.strip().partition("=") if k == "t" and v.isdigit(): t = int(v) if k == "v1" and re.fullmatch(r"[0-9a-f]{64}", v): macs.append(v) now = int(time.time()) if now is None else now if t is None or abs(now - t) > tolerance_sec or not macs: return False for key in keys: expected = hmac.new(key.encode(), f"{t}.".encode() + raw_body, hashlib.sha256).hexdigest() if any(hmac.compare_digest(expected, mac) for mac in macs): return True return False def authenticate(header: Header, raw_body: bytes) -> str: """Bearer, then signature over the raw body, then X-OneHazel-Env. Returns "test" or "live".""" if not hmac.compare_digest((header("authorization") or "").encode(), f"Bearer {CALLBACK_KEY}".encode()): raise ContractError("unauthorized", "missing or wrong callback key") if not verify_signature(raw_body, header("x-onehazel-signature"), SIGNING_SECRETS, int(CONTRACT["toleranceSec"])): raise ContractError("signature_invalid", "bad or stale X-OneHazel-Signature") env = header("x-onehazel-env") if env not in ("test", "live"): raise ContractError("invalid_request", "X-OneHazel-Env must be test or live") return env # ───────────────────────────── players ───────────────────────────── # # One directory per X-OneHazel-Env. Conformance runs always send `test`. Replace # `live` with a read from your player database; return only these fields. RESTRICTED_TEST_PLAYER_ID = os.environ.get("ONEHAZEL_RESTRICTED_TEST_PLAYER_ID", "test-player-restricted") DIRECTORIES: dict[str, dict[str, dict[str, Any]]] = { "test": { TEST_PLAYER_ID: {"player_external_id": TEST_PLAYER_ID, "currency": TEST_PLAYER_CURRENCY, "country": "MT", "status": "active", "restrictions": []}, RESTRICTED_TEST_PLAYER_ID: { "player_external_id": RESTRICTED_TEST_PLAYER_ID, "currency": TEST_PLAYER_CURRENCY, "country": "MT", "status": "active", "restrictions": [{"kind": "deposit_limit", "detail": "Test restriction: 100.00 per day"}], }, }, # Replace with your player database. "live": {}, } def handle(path: str, header: Header, raw_body: bytes) -> Reply: """The whole module: one call per request, framework-agnostic.""" try: op = find_op(path) if op is None: return 404, {"code": "invalid_request", "message": "unknown operation", "retryable": False} env = authenticate(header, raw_body) body = check_fields(parse_body(raw_body), op["fields"], op["closed"], "") player = DIRECTORIES[env].get(body["player_external_id"]) # Unknown is a 404, never a 200 with empty fields. if player is None: raise ContractError("player_unknown", "unknown player") return 200, player except ContractError as e: return error_reply(e.code, e.message) except Exception as e: # noqa: BLE001 print(f"player handler failed: {type(e).__name__}", file=sys.stderr) return error_reply("internal_error", "internal error") # ───────────────────────────── FastAPI wiring ───────────────────────────── app = FastAPI() @app.post(BASE_PATH + "/{operation:path}") async def onehazel(operation: str, request: Request) -> JSONResponse: # Read the RAW body: the signature is computed over the exact bytes OneHazel sent. raw = await request.body() status, body = handle("/" + operation, lambda name: request.headers.get(name), raw) return JSONResponse(body, status_code=status) ``` ## Before you go live - Replace the in-memory (or file) stores with your database. The idempotency store must survive restarts. - Point the `live` ledger at your real wallet; keep `test` traffic on the test ledger. - Run conformance (`connect_run_conformance`) until every scenario passes.