Python helpers
Preview, copy, and understand every helper function.
Downloading is optional: preview and copy the complete code below. Download the file.
Copy this complete module into vantra_client.py beside your Python server code. It uses the Python 3 standard library. FastAPI is only needed if you choose the FastAPI receiver example; it is not required by this helper.
Function reference
sign_request
sign_request(secret, timestamp, body="")
Returns lowercase HMAC-SHA256 of timestamp + dot + body. Timestamp is Unix seconds; body is the exact UTF-8 JSON string to transmit. Method, path, and idempotency_key remain optional compatibility arguments but are not signed. This function does not send a request.
gateway_request
gateway_request(origin, api_key, secret, path, data=None, idempotency_key="", method="POST")
Makes a synchronous signed request with a 30-second timeout and returns the response’s data field. Omit data for GET. The idempotency_key argument is optional, including for creation. Supply orderId in data for duplicate-order checks. HTTP failures raise urllib.error.HTTPError; use error.read() to retain the response for reconciliation. Network errors and decoding errors propagate. No automatic retries occur. The internal NoRedirect class prevents following redirects. In async servers, run this blocking helper in a worker thread rather than blocking the event loop.
verify_webhook
verify_webhook(raw_body, headers, secret, environment=None, now=None)
raw_body is the original bytes; headers must support case-insensitive/lowercase lookup (such as Starlette Headers). environment is optional; omit it to verify with the webhook signing secret alone and read event["environment"]. If supplied, TEST or LIVE restricts accepted events. Both payload environment fields must agree and contain TEST or LIVE. Match the verified environment to the saved payment before fulfillment. now, if supplied, is Unix seconds—not JavaScript milliseconds. Returns the parsed event after signature and consistency checks. Invalid input raises an exception. It does not implement persistence or order fulfillment; your receiver must store/deduplicate the event before acknowledging.
Complete source code
"""Python 3 server-side helper. Uses only the standard library."""
import hashlib
import hmac
import json
import time
import urllib.request
import urllib.error
from urllib.parse import urlsplit, urljoin
def sign_request(secret, timestamp, method=None, path=None, idempotency_key="", body=""):
# method/path/idempotency_key are retained as optional compatibility arguments.
payload = f"{timestamp}.{body}"
return hmac.new(secret.encode(), payload.encode(), hashlib.sha256).hexdigest()
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl):
return None
def gateway_request(origin, api_key, secret, path, data=None,
idempotency_key="", method="POST"):
if not origin or not api_key or not secret:
raise ValueError("Configure the gateway origin and server credentials")
if not path.startswith("/api/v1/"):
raise ValueError("Use an /api/v1/ path")
url = urljoin(origin, path)
parsed = urlsplit(url)
body = "" if data is None else json.dumps(data, separators=(",", ":"), ensure_ascii=False)
timestamp = str(int(time.time()))
signed_path = parsed.path + ("?" + parsed.query if parsed.query else "")
headers = {"Content-Type": "application/json", "x-api-key": api_key,
"x-timestamp": timestamp,
"x-signature": sign_request(secret, timestamp, method, signed_path, idempotency_key, body)}
if idempotency_key:
headers["Idempotency-Key"] = idempotency_key
request = urllib.request.Request(url, data=body.encode() if body else None,
headers=headers, method=method)
# HTTPError retains response bytes for reconciliation. Do not automatically retry.
with urllib.request.build_opener(NoRedirect).open(request, timeout=30) as response:
return json.load(response)["data"]
def verify_webhook(raw_body, headers, secret, environment=None, now=None):
"""headers must support lowercase names (e.g. Starlette Headers)."""
import re
if not secret or (environment is not None and environment not in ("TEST", "LIVE")):
raise ValueError("Webhook configuration missing")
timestamp = headers.get("x-webhook-timestamp", "")
signature = headers.get("x-webhook-signature", "")
if not re.fullmatch(r"[0-9]+", timestamp) or not re.fullmatch(r"[a-f0-9]{64}", signature):
raise ValueError("Invalid headers")
if abs(int(time.time() if now is None else now) - int(timestamp)) > 300:
raise ValueError("Stale webhook")
expected = hmac.new(secret.encode(), timestamp.encode() + b"." + raw_body, hashlib.sha256).hexdigest()
if not hmac.compare_digest(signature, expected):
raise ValueError("Invalid signature")
event = json.loads(raw_body)
data = event.get("data", {})
if (not isinstance(event.get("id"), str) or not event["id"]
or event["id"] != headers.get("x-webhook-id")
or event.get("environment") not in ("TEST", "LIVE")
or data.get("environment") != event.get("environment")
or (environment is not None and event.get("environment") != environment)
or not isinstance(event.get("transactionId"), str)
or event["transactionId"] != data.get("transactionId")
or event.get("event") != headers.get("x-webhook-event")
or event.get("event") != "PAYMENT_" + str(data.get("status"))):
raise ValueError("Invalid event")
return event