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=<hex>, 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

pip install fastapi uvicorn
ONEHAZEL_SIGNING_SECRET=<whsec signing secret> uvicorn onehazel_payments_intent:app --port 3000
# base URL to register: https://<your host>/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

"""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=<hex HMAC-SHA256(whsec, raw body)>`, 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.