"""Receive and verify Bitnob webhooks in Python.
No third-party packages needed (Python 3.8+).
How Bitnob signs a webhook
--------------------------
Every webhook is a POST with a JSON body shaped like:
{"data": {...}, "event": "payouts.withdrawal.success", "event_id": "..."}
and these headers:
x-bitnob-signature hex HMAC-SHA512 of the raw request body,
keyed with your webhook secret
x-bitnob-event-id the event id, for deduplicating retries
Always verify against the RAW request body bytes. Do not parse the JSON and
re-serialise it (json.dumps, JSON.stringify): spacing, key order and escaping
(Bitnob sends "<", ">" and "&" as \\u003c, \\u003e, \\u0026) will differ, and
the signature will not match.
Bitnob treats any 2xx as delivered. 400, 404, 410, 422 and 451 are final and
never retried; anything else (5xx, 401, timeouts...) is retried. The same event
can arrive more than once, so deduplicate on the event id. If your handler
fails temporarily, respond 5xx (this receiver returns 500) so Bitnob retries.
Usage
-----
python bitnob_webhook.py --self-test # check everything works
BITNOB_WEBHOOK_SECRET=... python bitnob_webhook.py --port 8000
The receiver listens on POST /webhooks/bitnob.
Using a web framework instead? Only verify_signature() is needed:
# Flask
raw = request.get_data()
ok = verify_signature(raw, request.headers.get("x-bitnob-signature"), SECRET)
# Django
ok = verify_signature(request.body, request.headers.get("x-bitnob-signature"), SECRET)
# FastAPI
raw = await request.body()
ok = verify_signature(raw, request.headers.get("x-bitnob-signature"), SECRET)
Reject the request with 401 when ok is False.
"""
from __future__ import annotations
import argparse
import hashlib
import hmac
import json
import os
import sys
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Optional
SIGNATURE_HEADER = "x-bitnob-signature"
EVENT_ID_HEADER = "x-bitnob-event-id"
WEBHOOK_PATH = "/webhooks/bitnob"
MAX_BODY_BYTES = 1024 * 1024
def compute_signature(raw_body: bytes, secret: str) -> str:
"""Hex HMAC-SHA512 of the raw body, keyed with the webhook secret."""
return hmac.new(secret.encode("utf-8"), raw_body, hashlib.sha512).hexdigest()
def verify_signature(raw_body: bytes, signature: Optional[str], secret: str) -> bool:
"""True only when the signature header matches the body.
Uses a constant-time comparison so the check can't be timed.
"""
if not secret:
raise ValueError("webhook secret is not configured")
if not signature:
return False
expected = compute_signature(raw_body, secret)
return hmac.compare_digest(expected, signature.strip().lower())
# --- Your business logic -----------------------------------------------------
def handle_event(event: dict) -> None:
"""Called once per verified, non-duplicate event. Replace with your logic."""
name = event.get("event", "")
data = event.get("data") or {}
if name == "deposit.success":
print(f"deposit confirmed: reference={data.get('reference')} amount={data.get('amount')}", flush=True)
elif name.startswith("withdrawal."):
print(f"{name}: reference={data.get('reference')} status={data.get('status')}", flush=True)
else:
print(f"unhandled event: {name}", flush=True)
# --- HTTP receiver (standard library only) -----------------------------------
class EventStore:
"""Remembers processed event ids. Use your database in production.
An event counts as processed only after handle_event succeeds. If handling
fails, the claim is released so Bitnob's retry is processed, not skipped.
"""
def __init__(self) -> None:
self._done = set()
self._in_progress = set()
self._lock = threading.Lock()
def claim(self, event_id: str) -> bool:
"""False if the event was already processed or is being processed now."""
with self._lock:
if event_id in self._done or event_id in self._in_progress:
return False
self._in_progress.add(event_id)
return True
def complete(self, event_id: str) -> None:
with self._lock:
self._in_progress.discard(event_id)
self._done.add(event_id)
def release(self, event_id: str) -> None:
with self._lock:
self._in_progress.discard(event_id)
def make_handler(secret: str, store: EventStore, on_event=handle_event):
class WebhookHandler(BaseHTTPRequestHandler):
def do_POST(self) -> None:
if self.path != WEBHOOK_PATH:
return self._reply(404)
try:
length = int(self.headers.get("Content-Length", "0"))
except ValueError:
return self._reply(400)
if length <= 0 or length > MAX_BODY_BYTES:
return self._reply(400)
raw_body = self.rfile.read(length)
if not verify_signature(raw_body, self.headers.get(SIGNATURE_HEADER), secret):
return self._reply(401)
try:
event = json.loads(raw_body)
except json.JSONDecodeError:
return self._reply(400)
event_id = self.headers.get(EVENT_ID_HEADER) or event.get("event_id") or ""
if event_id and not store.claim(event_id):
return self._reply(200) # duplicate delivery: acknowledge, do nothing
try:
on_event(event)
except Exception as exc: # non-200 makes Bitnob retry later
if event_id:
store.release(event_id) # so the retry is processed, not skipped
print(f"failed to handle event {event_id}: {exc}", file=sys.stderr)
return self._reply(500)
if event_id:
store.complete(event_id)
return self._reply(200)
def _reply(self, status: int) -> None:
self.send_response(status)
self.send_header("Content-Length", "0")
self.end_headers()
def log_message(self, fmt, *args) -> None:
sys.stderr.write(f"[bitnob-webhook] {self.address_string()} {fmt % args}\n")
return WebhookHandler
def serve(secret: str, host: str, port: int) -> None:
server = ThreadingHTTPServer((host, port), make_handler(secret, EventStore()))
print(f"Listening on http://{host}:{port}{WEBHOOK_PATH}")
try:
server.serve_forever()
except KeyboardInterrupt:
pass
finally:
server.server_close()
# --- Self-test ---------------------------------------------------------------
# A sample payload signed the way Bitnob signs, with the secret
# "test_secret". The reference contains <, > and & to cover Bitnob's escaping.
VECTOR_SECRET = "test_secret"
VECTOR_BODY = (
b'{"data":{"amount":"9000000","chain":"tron","reference":"ref\\u003c1\\u003e\\u0026x"},'
b'"event":"payouts.withdrawal.success","event_id":"evt_1"}'
)
VECTOR_SIGNATURE = (
"2367e131d723576925cd841dd00490f5f0f6717f7caaa9130f28543dfb641f4e"
"58b3e6c240dee32066f27b757ba767d97a39808c7b03669b500088c9c391a0aa"
)
def self_test() -> int:
import urllib.error
import urllib.request
failures = []
def check(name: str, condition: bool) -> None:
print(f" {'PASS' if condition else 'FAIL'} {name}")
if not condition:
failures.append(name)
print("Signature checks")
check("Bitnob-signed body verifies", verify_signature(VECTOR_BODY, VECTOR_SIGNATURE, VECTOR_SECRET))
check("uppercase signature header verifies",
verify_signature(VECTOR_BODY, VECTOR_SIGNATURE.upper(), VECTOR_SECRET))
check("wrong secret is rejected", not verify_signature(VECTOR_BODY, VECTOR_SIGNATURE, "wrong"))
check("tampered amount is rejected",
not verify_signature(VECTOR_BODY.replace(b"9000000", b"9900000"), VECTOR_SIGNATURE, VECTOR_SECRET))
check("missing signature is rejected", not verify_signature(VECTOR_BODY, None, VECTOR_SECRET))
check("re-serialised JSON does not verify (use the raw body!)",
not verify_signature(json.dumps(json.loads(VECTOR_BODY)).encode(), VECTOR_SIGNATURE, VECTOR_SECRET))
print("HTTP receiver checks")
handled = []
fail_next = []
def record(event: dict) -> None:
if fail_next:
fail_next.pop()
raise RuntimeError("simulated failure, e.g. database down")
handled.append(event)
server = ThreadingHTTPServer(("127.0.0.1", 0),
make_handler(VECTOR_SECRET, EventStore(), record))
server.RequestHandlerClass.log_message = lambda *a, **k: None # quiet during the test
threading.Thread(target=server.serve_forever, daemon=True).start()
url = f"http://127.0.0.1:{server.server_address[1]}{WEBHOOK_PATH}"
def post(body: bytes, signature: Optional[str], event_id: str = "evt_1") -> int:
headers = {"Content-Type": "application/json", EVENT_ID_HEADER: event_id}
if signature is not None:
headers[SIGNATURE_HEADER] = signature
req = urllib.request.Request(url, data=body, headers=headers, method="POST")
try:
with urllib.request.urlopen(req, timeout=5) as resp:
return resp.status
except urllib.error.HTTPError as err:
return err.code
try:
check("valid webhook -> 200", post(VECTOR_BODY, VECTOR_SIGNATURE) == 200)
check("event handed to handle_event once", len(handled) == 1 and handled[0]["event_id"] == "evt_1")
check("duplicate delivery -> 200, not handled again",
post(VECTOR_BODY, VECTOR_SIGNATURE) == 200 and len(handled) == 1)
check("bad signature -> 401", post(VECTOR_BODY, "deadbeef", "evt_2") == 401)
check("missing signature -> 401", post(VECTOR_BODY, None, "evt_3") == 401)
check("rejected events are not handled", len(handled) == 1)
fail_next.append(True)
check("handler failure -> 500 so Bitnob retries", post(VECTOR_BODY, VECTOR_SIGNATURE, "evt_4") == 500)
check("retry after a failure is processed, not skipped",
post(VECTOR_BODY, VECTOR_SIGNATURE, "evt_4") == 200 and len(handled) == 2)
finally:
server.shutdown()
server.server_close()
print()
if failures:
print(f"{len(failures)} check(s) failed")
return 1
print("All checks passed")
return 0
def main() -> int:
parser = argparse.ArgumentParser(description="Bitnob webhook receiver")
parser.add_argument("--self-test", action="store_true", help="run the built-in checks and exit")
parser.add_argument("--host", default="0.0.0.0")
parser.add_argument("--port", type=int, default=8000)
args = parser.parse_args()
if args.self_test:
return self_test()
secret = os.environ.get("BITNOB_WEBHOOK_SECRET", "")
if not secret:
print("Set BITNOB_WEBHOOK_SECRET to your Bitnob webhook secret", file=sys.stderr)
return 2
serve(secret, args.host, args.port)
return 0
if __name__ == "__main__":
sys.exit(main())
