--- title: "payments.intent sample: Python + FastAPI" description: "Operator Contract v1.4.1 payments.intent module, Python + FastAPI. Generated from the contract." --- # payments.intent: Python + FastAPI The endpoint OneHazel delivers payment intent status changes to. OneHazel never holds funds: the PSP settles, OneHazel relays the status, and this module credits a settled deposit once per `intent_id`. You create intents by calling OneHazel's `POST /payments/intents` with an `oh_live_*` key. Generated from Operator Contract v1.4.1. Key `python/fastapi/payments.intent`; the `connect_samples` MCP tool returns this page for that key. ## What it implements | Operation | Purpose | Idempotency key | |---|---|---| | `POST /payments/webhook` | Receive a payment intent status change. | no | Every delivery is checked before it is parsed: `X-OneHazel-Signature` is `sha256=`, HMAC-SHA256 of the RAW body keyed by the `whsec_` signing secret, compared in constant time. There is no timestamp and no bearer, so deduplicating on `deliveryId` is the replay defence. `X-OneHazel-Env` is then required: `test` traffic goes to a separate test ledger and never touches the live one. A bad signature answers `401` with the flat `{ code, message, retryable }` envelope. ## Run it ```bash pip install fastapi uvicorn ONEHAZEL_SIGNING_SECRET= uvicorn onehazel_payments_intent:app --port 3000 # base URL to register: https:///onehazel/v1 (ONEHAZEL_BASE_PATH changes the path) ``` Environment: `ONEHAZEL_SIGNING_SECRET` is the `whsec_` secret `connect_register_endpoint` showed you once for `payments.intent`. During a rotation set `ONEHAZEL_SIGNING_SECRET_PREVIOUS` to the old one for the overlap hour. ## Code `onehazel_payments_intent.py` ```python """OneHazel payments.intent 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": "sha256", "bearer": false, "envHeader": true }, "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": "/payments/webhook", "operationId": "paymentIntentStatusReceive", "summary": "Receive a payment intent status change.", "idempotent": false, "fields": [ { "name": "type", "type": "string", "required": true, "enum": [ "payment.intent.created", "payment.intent.authorised", "payment.intent.settled", "payment.intent.failed", "payment.intent.cancelled" ] }, { "name": "data", "type": "object", "required": true, "fields": [ { "name": "intent_id", "type": "string", "required": true, "minLength": 1 }, { "name": "status", "type": "string", "required": true, "enum": [ "created", "authorised", "settled", "failed", "cancelled" ] }, { "name": "direction", "type": "string", "required": true, "enum": [ "deposit", "withdraw" ] }, { "name": "connection_id", "type": "string", "required": true }, { "name": "player_external_id", "type": "string", "required": true }, { "name": "amount_minor", "type": "integer", "required": false, "minimum": 0 }, { "name": "amount_minor_string", "type": "string", "required": false, "pattern": "^[1-9][0-9]{15,39}$" }, { "name": "currency", "type": "string", "required": true, "pattern": "^[A-Z][A-Z0-9]{2,9}$" }, { "name": "exponent", "type": "integer", "required": true, "minimum": 0 }, { "name": "method", "type": "string", "required": true }, { "name": "redirect_url", "type": "string", "required": false }, { "name": "idempotency_key", "type": "string", "required": false, "minLength": 1, "maxLength": 128 }, { "name": "duplicate", "type": "boolean", "required": false }, { "name": "failure_reason", "type": "string", "required": false }, { "name": "provider_state", "type": "string", "required": false, "enum": [ "no_adapter" ] }, { "name": "created_at", "type": "string", "required": true }, { "name": "updated_at", "type": "string", "required": true } ] }, { "name": "operatorId", "type": "string", "required": true }, { "name": "timestamp", "type": "string", "required": true }, { "name": "deliveryId", "type": "string", "required": true } ], "closed": false } ] }''') # ───────────────────────────── 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) # ───────────────────────────── delivery authentication ───────────────────────────── # # Payment status deliveries use the OneHazel webhook engine's format: # `X-OneHazel-Signature: sha256=`, no timestamp and no # bearer. Unlike webhook.receive they also carry `X-OneHazel-Env` (test or live), checked # after the signature. Verify over the RAW bytes, before the body is parsed. def verify_webhook_signature(raw_body: bytes, header: Optional[str], keys: list[str]) -> bool: match = re.fullmatch(r"sha256=([0-9a-f]{64})", (header or "").strip()) if not match: return False ok = False for key in keys: expected = hmac.new(key.encode(), raw_body, hashlib.sha256).hexdigest() # No early exit: every secret is compared. ok = hmac.compare_digest(expected, match.group(1)) or ok return ok def authenticate_webhook(header: Header, raw_body: bytes) -> str: if not verify_webhook_signature(raw_body, header("x-onehazel-signature"), SIGNING_SECRETS): raise ContractError("signature_invalid", "bad 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 # ───────────────────────────── payment status ───────────────────────────── # # OneHazel never holds funds: the PSP settles, OneHazel records the intent and relays each # status here. Deliveries are at-least-once, so deduplicate on deliveryId AND apply each # status idempotently per intent_id: a settled deposit is credited once, whatever the number # of deliveries. The state change is applied before acknowledging (it is a dict update, well # inside the 5 s budget), so an acknowledged credit is never lost. `test` traffic (every # conformance run) only ever touches the test ledger. Everything lives in memory here; use # your database, with the credit and the intent update in one transaction. TERMINAL = ("settled", "failed", "cancelled") SEEN: set[str] = set() INTENTS: dict[str, dict[str, dict[str, Any]]] = {"test": {}, "live": {}} # Balances per (player, currency): minor units of different currencies never add up. Python ints # are exact at any size, so an amount_minor_string above 2^53 - 1 needs nothing special. LEDGER: dict[str, dict[tuple[str, str], int]] = {"test": {}, "live": {}} # Intents a status arrived for before we knew them: read each with GET /payments/intents/{intent_id}. TO_RECONCILE: set[str] = set() _LOCK = threading.Lock() def amount_minor(intent: dict[str, Any]) -> int: """The exact amount, whichever of amount_minor / amount_minor_string carries it.""" has_int = "amount_minor" in intent if has_int == ("amount_minor_string" in intent): raise ContractError("invalid_request", "exactly one of amount_minor and amount_minor_string is required") return int(intent["amount_minor"]) if has_int else int(intent["amount_minor_string"]) def apply_status(env: str, intent: dict[str, Any]) -> None: intents = INTENTS[env] known = intents.get(intent["intent_id"]) if known is None and intent["status"] != "created": TO_RECONCILE.add(intent["intent_id"]) # A terminal status never changes again; a late or replayed delivery is a no-op. if known is not None and known["status"] in TERMINAL: return record = known or {"status": intent["status"], "credited": False} record["status"] = intent["status"] if intent["status"] == "settled" and intent["direction"] == "deposit" and not record["credited"]: ledger = LEDGER[env] key = (intent["player_external_id"], intent["currency"]) ledger[key] = ledger.get(key, 0) + amount_minor(intent) record["credited"] = True # A withdrawal you held on create is released on failed or cancelled; a settled one is final. intents[intent["intent_id"]] = record 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_webhook(header, raw_body) event = check_fields(parse_body(raw_body), op["fields"], op["closed"], "") if header("x-onehazel-delivery") != event["deliveryId"] or header("x-onehazel-event") != event["type"]: raise ContractError("invalid_request", "X-OneHazel-Delivery and X-OneHazel-Event must match the body") if event["type"] != "payment.intent." + event["data"]["status"]: raise ContractError("invalid_request", "type must match data.status") amount_minor(event["data"]) # reject a malformed amount before the delivery is marked seen with _LOCK: if event["deliveryId"] in SEEN: return 200, {"received": True, "duplicate": True} SEEN.add(event["deliveryId"]) apply_status(env, event["data"]) return 200, {"received": True, "duplicate": False} except ContractError as e: return error_reply(e.code, e.message) except Exception as e: # noqa: BLE001 print(f"payment status 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. - Credit the player and record the intent in one database transaction, keyed by `intent_id`, so a settled deposit is credited exactly once. - Run conformance (`connect_run_conformance`) until every scenario passes.