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=<unix seconds>,v1=<hex>, HMAC-SHA256 of "<t>." + 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=<callback key> ONEHAZEL_SIGNING_SECRET=<signing secret> uvicorn onehazel_player:app --port 3000
# base URL to register: https://<your host>/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=<unix seconds>,v1=<hex HMAC-SHA256(secret, "<t>." + 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.

player: Python + FastAPI ​

What it implements ​

Run it ​

Code ​

Before you go live ​

player: Python + FastAPI

What it implements

Run it

Code

Before you go live