Runtime: Python 3.9+
pip install requests # hmac and hashlib are standard library
# havala_sign.py — the whole of Havala gateway auth, in one function.
import hashlib
import hmac
import time
WRITE_METHODS = {"POST", "PUT", "PATCH", "DELETE"}
def sign_request(secret, method, path, body="", idempotency_key=None):
"""Return the X-Signature header value: t={unix_seconds},v1={hex}.
secret is the merchant point apiSecret (64 hex chars), never the API key.
path is the path alone: the server signs request.path, so a query string
must be left out. body is the exact text you will send.
"""
method = method.upper()
# t is checked against server time with a +/-300s window, so sign
# immediately before sending. A signature parked in a queue expires there.
t = int(time.time())
# sha256 over the raw body bytes. No body hashes the empty string, which is
# always e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855.
body_hash = hashlib.sha256(body.encode("utf-8")).hexdigest()
# {t}\n{METHOD}\n{path}\n[{idempotency_key}\n]{sha256(body)}
# The idempotency line is a write-method line. Add it to a GET, or drop it
# from a POST, and the two sides hash a different number of lines.
parts = [str(t), method, path]
if method in WRITE_METHODS and idempotency_key:
parts.append(idempotency_key)
parts.append(body_hash)
v1 = hmac.new(
secret.encode("utf-8"),
"\n".join(parts).encode("utf-8"),
hashlib.sha256,
).hexdigest()
return f"t={t},v1={v1}"
import hashlib
import hmac
import json
import os
import time
import uuid
import requests
BASE_URL = "https://api2.havala.io"
API_KEY = os.environ["HAVALA_API_KEY"] # merchant point apiKey, 32 hex chars
API_SECRET = os.environ["HAVALA_API_SECRET"] # merchant point apiSecret, 64 hex chars
WRITE_METHODS = {"POST", "PUT", "PATCH", "DELETE"}
def _sign(method, path, body, idempotency_key=None):
t = int(time.time())
parts = [str(t), method, path]
if method in WRITE_METHODS and idempotency_key:
parts.append(idempotency_key)
parts.append(hashlib.sha256(body.encode("utf-8")).hexdigest())
v1 = hmac.new(
API_SECRET.encode("utf-8"),
"\n".join(parts).encode("utf-8"),
hashlib.sha256,
).hexdigest()
return f"t={t},v1={v1}"
def call(method, path, payload=None, idempotency_key=None):
# Serialise once: the signature covers these exact bytes. separators= keeps
# the string compact and, more usefully, stable if you ever log it.
body = "" if payload is None else json.dumps(payload, separators=(",", ":"))
headers = {
"Content-Type": "application/json",
"X-API-Key": API_KEY,
"X-Signature": _sign(method, path, body, idempotency_key),
}
if idempotency_key:
headers["Idempotency-Key"] = idempotency_key
# data=, never json=: requests would re-encode the dict and the bytes on the
# wire would stop matching the ones you hashed.
response = requests.request(
method,
BASE_URL + path,
headers=headers,
data=body.encode("utf-8") if body else None,
timeout=30,
)
envelope = response.json()
if not response.ok:
raise RuntimeError(f"{envelope['error']['code']} - {envelope['error']['message']}")
# Every gateway response is wrapped: {"success", "data", "timestamp", "requestId"}.
return envelope["data"]
# Pin the key to the operation, not to the attempt: store it beside the order and
# resend it on every retry. A fresh key opens a second invoice; the same key with
# a different body is refused with IDEMPOTENCY_KEY_CONFLICT (409).
idempotency_key = str(uuid.uuid4())
invoice = call(
"POST",
"/api/v1/invoices",
{
"orderId": "order-8814",
"amount": "12500", # minor units of currency — 12500 is USD 125.00
"currency": "USD", # what you price in; the buyer picks the crypto at checkout
"description": "Pro plan, annual billing",
"customerEmail": "buyer@example.com",
"returnUrl": "https://merchant.example.com/checkout/success",
"cancelUrl": "https://merchant.example.com/checkout/cancel",
"expiresInMinutes": 60, # 15-1440; expiresAt is stamped from this
},
idempotency_key,
)
print(invoice["id"], invoice["referenceId"], invoice["status"]) # -> clyv3n8x… INV-260826-041C OPEN
# Read it back. Nothing to hash, and no idempotency line on a GET.
fresh = call("GET", "/api/v1/invoices/" + invoice["id"])
# payments[] is included on this route and on no other: create, checkout and
# cancel return the invoice without that key at all.
print(fresh["status"], fresh.get("payments"))
import hashlib
import hmac
import json
import os
import time
import uuid
import requests
# Every failure shares one envelope, built by the gateway's exception filter:
#
# {
# "success": false,
# "error": {"code": "SIGNATURE_INVALID", "message": "Invalid request signature"},
# "timestamp": "2026-08-26T09:14:22.481Z",
# "requestId": "3f1c0b7d-4a29-4e63-9b0c-8d17a4e92f6b"
# }
#
# error.details is present on VALIDATION_ERROR, where it carries
# details.errors[] of {field, message, value}. requestId is the correlation id
# that also appears in Havala's logs — quote it when you ask why a call failed.
BASE_URL = "https://api2.havala.io"
WRITE_METHODS = {"POST", "PUT", "PATCH", "DELETE"}
class HavalaError(Exception):
def __init__(self, code, message, status, request_id, details=None):
super().__init__(f"{code}: {message}")
self.code = code
self.status = status
self.request_id = request_id
self.details = details or {}
def call(method, path, payload=None, idempotency_key=None):
body = "" if payload is None else json.dumps(payload, separators=(",", ":"))
t = int(time.time())
parts = [str(t), method, path]
if method in WRITE_METHODS and idempotency_key:
parts.append(idempotency_key)
parts.append(hashlib.sha256(body.encode("utf-8")).hexdigest())
v1 = hmac.new(
os.environ["HAVALA_API_SECRET"].encode("utf-8"),
"\n".join(parts).encode("utf-8"),
hashlib.sha256,
).hexdigest()
headers = {
"Content-Type": "application/json",
"X-API-Key": os.environ["HAVALA_API_KEY"],
"X-Signature": f"t={t},v1={v1}",
}
if idempotency_key:
headers["Idempotency-Key"] = idempotency_key
response = requests.request(
method, BASE_URL + path, headers=headers,
data=body.encode("utf-8") if body else None, timeout=30,
)
envelope = response.json()
if not response.ok:
error = envelope["error"]
raise HavalaError(
error["code"], error["message"], response.status_code,
envelope.get("requestId"), error.get("details"),
)
return envelope["data"]
idempotency_key = str(uuid.uuid4())
try:
call("POST", "/api/v1/invoices",
{"orderId": "order-8814", "amount": "12500", "currency": "USD"},
idempotency_key)
except HavalaError as err:
if err.code == "API_KEY_INVALID":
# 401. Unknown key, or the merchant point was deactivated. Nothing about
# this request will succeed on a retry.
pass
elif err.code == "SIGNATURE_MISSING":
# 401. X-Signature never arrived. Usually a proxy or a CDN dropping
# unknown headers rather than a bug in your signer.
pass
elif err.code == "SIGNATURE_INVALID":
# 401. Either the header did not match ^t=\d+,v1=[a-f0-9]{64}$ or the HMAC
# disagreed; the guard will not say which. Check the canonical string
# before you suspect the secret — a query string left in the signed path
# and a re-serialised body both land here.
pass
elif err.code == "SIGNATURE_EXPIRED":
# 401. |now - t| exceeded 300s: your clock, not your key. Re-sign with a
# fresh t and resend with the SAME idempotency key, or you create a
# second invoice for the same order.
pass
elif err.code == "IDEMPOTENCY_KEY_REQUIRED":
# 400. A write method without the header. The guard rejects before the
# handler runs, so nothing was created and nothing needs unwinding.
pass
elif err.code == "IDEMPOTENCY_KEY_CONFLICT":
# 409. This key was already used with a different method, path or body
# hash — or the first attempt is still in flight. Read the original
# invoice rather than reissuing.
pass
elif err.code == "VALIDATION_ERROR":
# 422. The DTO layer runs with forbidNonWhitelisted, so an unrecognised
# property fails the request instead of being dropped quietly.
print(err.details.get("errors"))
else:
# 5xx and anything unmapped: retry with backoff, reusing the same key.
raise
except requests.RequestException:
# Never reached the server, or never read the answer. The write may still
# have landed: retry with the same idempotency key and Havala replays the
# first response rather than creating a second invoice.
raise
import hashlib
import hmac
import json
import os
import time
from http.server import BaseHTTPRequestHandler, HTTPServer
# The signing key is the merchant point apiSecret — the same value that keys
# X-Signature on the gateway API. Rotating it in the dashboard invalidates both
# at the same instant; there is no overlap window.
SECRET = os.environ["HAVALA_API_SECRET"].encode("utf-8")
class HavalaWebhookHandler(BaseHTTPRequestHandler):
def do_POST(self):
if self.path != "/havala/webhooks":
return self._reply(404)
# Read the untouched bytes. Havala signs the string its dispatcher put on
# the wire, so json.loads followed by json.dumps yields a different
# string and a different digest. Flask: request.get_data(). Django:
# request.body. FastAPI: await request.body(). Never the parsed object.
raw = self.rfile.read(int(self.headers.get("Content-Length", 0)))
expected = "sha256=" + hmac.new(SECRET, raw, hashlib.sha256).hexdigest()
provided = self.headers.get("X-Webhook-Signature", "")
# compare_digest is constant-time and tolerates a length mismatch; ==
# short-circuits on the first differing byte and leaks the prefix length.
if not hmac.compare_digest(provided, expected):
return self._reply(401, b"bad signature")
# Only the body is signed — no timestamp, method or path is folded in —
# so a captured delivery replays forever unless you bound its age
# yourself. X-Webhook-Timestamp is MILLISECONDS since epoch, not seconds.
sent_at = int(self.headers.get("X-Webhook-Timestamp", 0))
if abs(time.time() * 1000 - sent_at) > 300_000:
return self._reply(401, b"stale delivery")
event = json.loads(raw)
if event["event"] == "payment.completed":
# The envelope is camelCase; the object inside data is snake_case.
payment = event["data"]["payment"]
fulfil_order(payment["id"], payment["transaction_hash"], payment["received_amount"])
# Havala allows 10 seconds and counts only 2xx as delivered. Acknowledge
# first, work after.
return self._reply(200, b"ok")
def _reply(self, status, body=b""):
self.send_response(status)
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def fulfil_order(payment_id, tx_hash, received_amount):
"""Hand off to your own queue; this handler must return inside 10 seconds.
payment.completed is at-least-once: the confirmation tracker re-enters the
notify path when a job re-runs against an already COMPLETED payment. Dedupe
on payment_id + tx_hash before you release anything.
received_amount is a decimal string in the crypto's own unit ("149.500000"),
not minor units and not a float — parse it with decimal.Decimal.
The envelope is a notification, not proof of settlement. Confirm against
GET /api/v1/payments/{id}/status before shipping anything expensive.
"""
HTTPServer(("", 3000), HavalaWebhookHandler).serve_forever()