Test webhooks locally

Test authentication and durable receipt without publishing your payloads to an external request-catcher service.

What you will learn

Run a loopback-only Sandbox inbox and replay a synthetic notification.
Confirm harmless duplicate handling and refusal of changed requests.
Move from local fixtures to an HTTPS staging endpoint and inspect deliveries.

Prerequisites

Python 3.11 or newer with standard-library SQLite.
The webhook source archive, extracted locally.
Webhook signature basics.
A private working directory outside any web document root.

Start the local inbox

Work inside the archive's webhooks directory.
Set KDP_WEBHOOK_SECRET through your private process environment to a synthetic test value or a Sandbox endpoint secret.
Set KDP_WEBHOOK_ENVIRONMENT to sandbox.
Set KDP_WEBHOOK_INBOX to a private local SQLite file path outside your web root.
Run the receiver.

bashbash

python3 capture.py

Expected output: Listening on http://127.0.0.1:8108/kdpika; Sandbox notifications only. The process binds loopback, not all network interfaces. It creates its inbox database with owner-only file permissions where the OS supports them.

capture.pypython

#!/usr/bin/env python3
"""Local notification inbox; never a license-validity or payment authority.
Python 3.11+, standard library, verify.py and contract.json in this directory.
Set KDP_WEBHOOK_SECRET, KDP_WEBHOOK_ENVIRONMENT=sandbox and KDP_WEBHOOK_INBOX.
Run: python3 capture.py; POST /kdpika on 127.0.0.1:8108. Expected: accepted JSON.
"""
import hashlib
from http.server import BaseHTTPRequestHandler, HTTPServer
import json
import os
from pathlib import Path
import sqlite3
import time
from verify import verify


class Receiver(BaseHTTPRequestHandler):
    def log_message(self, template, *args):
        # Do not log the raw payload, signature, full URL or secrets.
        return

    def reply(self, status, value):
        body = json.dumps(value, separators=(',', ':')).encode()
        self.send_response(status)
        self.send_header('Content-Type', 'application/json')
        self.send_header('Content-Length', str(len(body)))
        self.send_header('Cache-Control', 'no-store')
        self.send_header('X-Content-Type-Options', 'nosniff')
        self.end_headers()
        self.wfile.write(body)

    def do_POST(self):
        self.connection.settimeout(5)
        if self.path != '/kdpika':
            return self.reply(404, {'accepted': False})
        lengths = self.headers.get_all('Content-Length', [])
        if self.headers.get('Transfer-Encoding') or len(lengths) != 1 or not lengths[0].isdigit():
            return self.reply(400, {'accepted': False})
        length = int(lengths[0])
        if length > 1048576:
            return self.reply(413, {'accepted': False})
        if self.headers.get('Content-Type', '').split(';')[0].strip().lower() != 'application/json':
            return self.reply(415, {'accepted': False})
        names = ('x-lmp-signature', 'x-lmp-event-id', 'x-lmp-event')
        if any(len(self.headers.get_all(name, [])) != 1 for name in names):
            return self.reply(401, {'accepted': False})
        try:
            body = self.rfile.read(length)
            if len(body) != length:
                return self.reply(400, {'accepted': False})
            event = verify(body, {key: self.headers[key] for key in names},
                           self.server.secret, 'sandbox', int(time.time()))
        except (ValueError, TypeError, KeyError, OSError):
            return self.reply(401, {'accepted': False})
        # A duplicate ID is acknowledged only when its authenticated body is identical.
        # The transaction commits before acknowledgement. No API/mail side effects.
        try:
            with sqlite3.connect(self.server.inbox, timeout=3) as database:
                database.execute('BEGIN IMMEDIATE')
                previous = database.execute('SELECT body_hash FROM inbox WHERE event_id=?',
                                            (event['id'],)).fetchone()
                body_hash = hashlib.sha256(body).hexdigest()
                if previous and previous[0] != body_hash:
                    return self.reply(409, {'accepted': False})
                database.execute('INSERT OR IGNORE INTO inbox(event_id,event,body_hash,received_at) VALUES(?,?,?,?)',
                                 (event['id'], event['event'], body_hash, int(time.time())))
                database.commit()
            self.reply(200, {'accepted': True, 'duplicate': previous is not None})
        except sqlite3.Error:
            self.reply(503, {'accepted': False})


def main():
    secret = os.environ.get('KDP_WEBHOOK_SECRET', '')
    if not secret or os.environ.get('KDP_WEBHOOK_ENVIRONMENT', '').lower() != 'sandbox':
        raise SystemExit('Set the Sandbox endpoint secret and environment first.')
    configured = os.environ.get('KDP_WEBHOOK_INBOX', '')
    if not configured:
        raise SystemExit('Set KDP_WEBHOOK_INBOX to a private local file outside your web root.')
    inbox = Path(configured).resolve()
    os.umask(0o077)
    inbox.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
    with sqlite3.connect(inbox) as database:
        database.execute('CREATE TABLE IF NOT EXISTS inbox(event_id TEXT PRIMARY KEY,event TEXT NOT NULL,body_hash TEXT NOT NULL,received_at INTEGER NOT NULL)')
    os.chmod(inbox, 0o600)
    with HTTPServer(('127.0.0.1', 8108), Receiver) as server:
        server.secret = secret
        server.inbox = str(inbox)
        print('Listening on http://127.0.0.1:8108/kdpika; Sandbox notifications only.', flush=True)
        server.serve_forever()


if __name__ == '__main__':
    main()

This program is a local QA inbox, not a production HTTP server or payment fulfiller. It verifies before inserting and stores only event ID/name, body hash and receipt time. It does not save the raw customer payload, call a licensing API, email a key or decide license validity. Its transaction commits before a successful acknowledgement. It intentionally has no public inbox-inspection route.

Generate and replay a request

Open a second terminal with the same private environment values.
Generate a freshly signed synthetic request.
Replay it to the local receiver.

bashbash

python3 make-test.py events/test-ping.json private/request-1
python3 replay.py private/request-1/body.json private/request-1/headers.json

Expected result: the first replay returns status=200, accepted=true and duplicate=false. The signature covers the saved raw bytes. The replay tool can send only to the loopback inbox and prints no request signature or secret.

The generator and verifiers share this file. Event names come from the sender; the size/freshness settings are local receiver policy.

contract.jsonjson

{
  "events": [
    "license.created",
    "license.activated",
    "license.deactivated",
    "license.suspended",
    "license.restored",
    "license.revoked",
    "license.renewed",
    "license.expiring",
    "license.expired",
    "device.reset",
    "test.ping"
  ],
  "environments": ["sandbox", "live"],
  "body_limit": 1048576,
  "tolerance_seconds": 300
}

make-test.pypython

#!/usr/bin/env python3
"""Sign a captured synthetic Sandbox envelope with the configured TEST secret.
Python 3.11+, standard library. Run: python3 make-test.py events/test-ping.json private/request
Writes body.json/headers.json to the private output directory; never sends a request.
Expected: Created private test request files. No signature or secret is printed.
"""
import hashlib
import hmac
import json
import os
from pathlib import Path
import sys
import time


def main():
    if len(sys.argv) != 3:
        raise SystemExit('Supply a captured JSON file and a private output directory.')
    secret = os.environ.get('KDP_WEBHOOK_SECRET', '')
    if not secret or os.environ.get('KDP_WEBHOOK_ENVIRONMENT', '').lower() != 'sandbox':
        raise SystemExit('Configure a Sandbox test secret first.')
    body = Path(sys.argv[1]).read_bytes()
    event = json.loads(body)
    if event.get('environment') != 'sandbox':
        raise SystemExit('Only Sandbox test envelopes are accepted.')
    stamp = str(int(time.time()))
    headers = {'x-lmp-signature': 't=' + stamp + ',v1=' + hmac.new(secret.encode(), stamp.encode() + b'.' + body, hashlib.sha256).hexdigest(),
               'x-lmp-event-id': event['id'], 'x-lmp-event': event['event']}
    os.umask(0o077)
    directory = Path(sys.argv[2])
    directory.mkdir(parents=True, mode=0o700, exist_ok=True)
    for name, content in [('body.json', body), ('headers.json', json.dumps(headers).encode())]:
        path = directory / name
        if path.exists():
            raise SystemExit('Request file already exists; use a new private directory.')
        with path.open('xb') as stream:
            stream.write(content)
        path.chmod(0o600)
    print('Created private test request files.')


if __name__ == '__main__':
    main()

replay.pypython

#!/usr/bin/env python3
"""Replay a private signed test request to the loopback capture.py inbox only.
Python 3.11+, standard library. Run: python3 replay.py private/request-1/body.json private/request-1/headers.json
Add --tamper to append a body byte. Expected: accepted JSON/exit 0, or refusal/exit 1.
This tool cannot send to a public URL and does not print request secrets.
"""
import argparse
from http.client import HTTPConnection
import json
from pathlib import Path


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('body')
    parser.add_argument('headers')
    parser.add_argument('--tamper', action='store_true')
    args = parser.parse_args()
    body = Path(args.body).read_bytes()
    headers = json.loads(Path(args.headers).read_text())
    if args.tamper:
        body += b' '
    headers['Content-Type'] = 'application/json'
    connection = HTTPConnection('127.0.0.1', 8108, timeout=5)
    try:
        connection.request('POST', '/kdpika', body=body, headers=headers)
        response = connection.getresponse()
        content = json.loads(response.read(4096))
        print(json.dumps({'status': response.status, **content}))
        return 0 if 200 <= response.status < 300 else 1
    except (OSError, ValueError):
        print('{"accepted":false,"error":"Local receiver unavailable"}')
        return 1
    finally:
        connection.close()


if __name__ == '__main__':
    raise SystemExit(main())

Check duplicates and failure paths

Replay the same request again within the freshness window.
Confirm the result has status=200, accepted=true and duplicate=true.
Send a changed byte with the original signature.

bashbash

python3 replay.py private/request-1/body.json private/request-1/headers.json --tamper

Expected result: status=401, accepted=false and exit 1. No additional inbox row is accepted. A correctly re-signed different body with an existing event ID returns 409 rather than applying a second meaning to that ID. Oversized bodies return 413; unsupported content type returns 415. If durable SQLite storage is temporarily unavailable, the inbox returns 503 instead of claiming successful receipt.

Generate another private request after changing only the local test secret and confirm it is refused by the receiver still using the old test value.
Restore matching local configuration before the next positive test.
Restart the receiver with the same inbox file and replay an accepted ID with a fresh valid timestamp.

Expected result: deduplication survives restart. Freshness checks apply to attempt timestamps; a new signature is needed after the 300-second window. There is no in-memory-only success claim.

Inspect safely

Use a local SQLite viewer to read event_id, event, body_hash and received_at from the private inbox table. The body hash detects a conflicting authenticated duplicate without retaining its full body. Do not expose that file, generated headers or request directories through a web server, Git repository or public support attachment.

For real platform delivery, open Webhooks and match the authenticated event ID to Delivery history. The page shows the latest delivery row state and attempt count, not every historic request/response body. The delivery reference describes retained data and the six-attempt limit.

Use HTTPS staging or an optional tunnel

The platform refuses loopback/private targets; entering http://127.0.0.1:8108/kdpika in the dashboard is not a platform-delivery test.

Prefer a receiver on your own HTTPS staging host with valid TLS and raw-body forwarding.
If local networking requires a tunnel, choose and operate it explicitly; a tunnel is an optional testing transport, never a runtime dependency for licensing.
Forward only the dedicated notification route to the local inbox, preserving body bytes and all three signature/event headers.
Verify that the public HTTPS URL resolves only to public addresses and does not redirect.
Create a separate Sandbox endpoint at that final URL.
Configure its one-time secret on the receiver.
Select Send test and match the event ID at both ends.
Stop public forwarding when testing ends and disable that temporary endpoint.

Expected result: a real signed test delivery receives a 2xx acknowledgement without exposing an administration page, private inbox or general development server. A tunnel operator can carry your test bytes; do not send real customer payloads through an unreviewed service. No particular tunnel account, external catcher or fixed IP range is required by the licensing platform.

Expected result

You have reproducible evidence that valid recent requests are durably accepted, identical duplicates are harmless, altered requests fail and delivery-history status matches the receiver acknowledgement.

Common mistakes

Disabling signature validation to make a local test pass.
Exposing the entire development server instead of one notification route.
Resetting the inbox before testing replay after restart.
Calling a license renewal during a notification-only capture test.

Security notes

Use synthetic Sandbox data. Keep signatures and keys out of screenshots and ticket text. The local receiver has no production authentication-management or operations UI; put a reviewed production HTTP adapter in front of a durable worker for real deployment. Never turn a locally re-signed fixture into a payment confirmation.

Troubleshooting

Use the dedicated webhook troubleshooting guide for receiver, routing, timing and retry symptoms.

Symptom	Cause	Fix
Connection refused	Local receiver stopped or port unavailable	Start capture.py and check the loopback port
Exit 1 with 401	Changed bytes, secret, environment or stale timestamp	Regenerate the private request with matching configuration
Duplicate=true on the first visible replay	The durable inbox already contains that captured ID	Inspect the existing record; do not erase evidence to manufacture a first receipt

Next steps

Review delivery and acknowledgement rules, then test your real backend with a separate Sandbox endpoint.

Test webhooks locally

What you will learn

Prerequisites

Start the local inbox

Generate and replay a request

Check duplicates and failure paths

Inspect safely

Use HTTPS staging or an optional tunnel

Expected result

Common mistakes

Security notes

Troubleshooting

Next steps

Was this page helpful?

Search documentation

Test webhooks locally

What you will learn

Prerequisites

Start the local inbox

Generate and replay a request

Check duplicates and failure paths

Inspect safely

Use HTTPS staging or an optional tunnel

Expected result

Common mistakes

Security notes

Troubleshooting

Next steps

Related

Was this page helpful?