"""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": "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 real payload and signature produced by Bitnob's signer, 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":"withdrawal.success","event_id":"evt_1"}' ) VECTOR_SIGNATURE = ( "380e5000acdb890ba06f7727c9667bcc7a1bb9d8270963d015926209c150e53b" "51a11a05912372eb4e64f0fb91942b7b63fe646b63d669d0f06a800e483b4fdb" ) 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())