Build a durable local inbox
This single-subscription receiver verifies exact raw bytes with the SDK, rejects invalid or expired signatures, and atomically records the event ID and one local alert row in SQLite. It acknowledges duplicates without another alert. webhook.test is verified and recorded but never creates an alert. Unknown event types are acknowledged without business effects. The example does not send email or mutate an external system.
Install and configure
Use a separate empty directory. Node.js 22.13+ includes node:sqlite; Python 3.11+ includes sqlite3. Set WEBHOOK_SECRET privately to the signing secret returned at subscription creation, not your API key. WEBHOOK_DB must name a persistent writable local file. One database and secret set belong to one subscription; do not mix customer tenants. For local verification, use a disposable database and a local-only secret of at least 32 characters.
uv venv
uv pip install https://hintadata.fi/sdk/price_compare-3.1.0.tar.gzSave the receiver
Save as receiver.py. The listener binds only to loopback on port 8080.
import os
import sqlite3
from http.server import BaseHTTPRequestHandler, HTTPServer
from price_compare.webhooks import verify_webhook
secrets = [value for value in (os.environ.get("WEBHOOK_SECRET"), os.environ.get("WEBHOOK_PREVIOUS_SECRET")) if value]
if not secrets or any(len(value) < 32 for value in secrets):
raise ValueError("Configure the webhook signing secret")
db = sqlite3.connect(os.environ.get("WEBHOOK_DB", "receiver.sqlite"), timeout=5)
db.executescript("CREATE TABLE IF NOT EXISTS received (id TEXT PRIMARY KEY); CREATE TABLE IF NOT EXISTS alerts (event_id TEXT PRIMARY KEY, product_id TEXT NOT NULL);")
class Receiver(BaseHTTPRequestHandler):
def log_message(self, *args):
pass # Do not log incoming headers or payloads.
def reply(self, status):
self.send_response(status)
self.send_header("Content-Length", "0")
self.end_headers()
def do_POST(self):
self.connection.settimeout(10)
if self.path != "/events":
self.reply(404)
return
# The TLS proxy must forward an unmodified, bounded body with Content-Length.
try:
length = int(self.headers.get("Content-Length", "-1"))
if self.headers.get("Transfer-Encoding") or not 0 <= length <= 262144:
self.reply(413)
return
raw = self.rfile.read(length)
if len(raw) != length:
self.reply(400)
return
except (ValueError, OSError):
self.reply(400)
return
event = None
for secret in secrets:
try:
event = verify_webhook(raw, self.headers.get("X-Webhook-Signature", ""), secret)
break
except (ValueError, TypeError):
continue
if event is None:
self.reply(400)
return
if event["type"] not in ("watchlist.triggered", "webhook.test"):
self.reply(204)
return
product = event["data"].get("canonical_product_id")
if event["type"] == "watchlist.triggered" and (not isinstance(product, str) or not product):
self.reply(400)
return
try:
with db:
db.execute("BEGIN IMMEDIATE")
inserted = db.execute("INSERT OR IGNORE INTO received (id) VALUES (?)", (event["id"],))
if inserted.rowcount and event["type"] == "watchlist.triggered":
# The local inbox row is the effect; no external network call.
db.execute("INSERT INTO alerts (event_id, product_id) VALUES (?, ?)", (event["id"], product))
except sqlite3.Error:
self.reply(503)
return
self.reply(204)
if __name__ == "__main__":
server = HTTPServer(("127.0.0.1", int(os.environ.get("PORT", "8080"))), Receiver)
try:
server.serve_forever()
except KeyboardInterrupt:
pass
finally:
server.server_close()
db.close()
uv run --no-project python receiver.pyVerify before exposing it
Using a local-only signing secret, send a signed watchlist.triggered event with a canonical_product_id. Expect 204 and one row in both received and alerts. Send the exact event again, then restart the receiver with the same database and send it again: both replies must be 204 and both row counts must remain one. Change one raw body byte without signing again: expect 400 and unchanged counts. Sign with a timestamp older than five minutes: expect 400. A signed webhook.test must return 204 without an alert. If durable storage fails, expect 503, not 204. These checks are also run against this exact published code in the repository.
Place behind your HTTPS service
The local listener is not a public production server. Put it behind a managed HTTPS reverse proxy on port 443 that preserves raw bytes, enforces a 256 KiB body limit and request deadlines, and forwards Content-Length (the Python example does not accept chunked transfer). Run one process for this example with persistent storage, restricted file permissions, backup and supervised restart. Register the public /events URL only after deploying it; localhost and private URLs cannot be registered. Never expose the database or log secrets and request bodies.
Extend without losing duplicate protection
The alert row is deliberately the whole business effect. For email or another external action, write a durable outbox entry in the same transaction and let a worker deliver it with an idempotency key. Do not mark an event received and then make an unprotected external call. Events may arrive out of order: this example appends notifications, it does not overwrite current watch state. Retain received IDs for the entire replay horizon, at least seven days; the example does not delete them automatically.
Accept two keys only during rotation
During the documented prepare/activate rotation, configure one secret as WEBHOOK_SECRET and the other as WEBHOOK_PREVIOUS_SECRET, then restart. Both are accepted, allowing the new key to be installed before activation. After activation and at least 60 seconds of overlap, keep only the new key as WEBHOOK_SECRET and restart. Never retain the old key indefinitely. A queued test is not delivery proof: verify delivered status and the received row.