# Receive a Referral Reward Webhook in Python with Flask

> Receive a signed WarpLink referral webhook in Python with Flask: verify the HMAC signature on the raw body, dedupe by event id in SQLite, and test locally.

_Last updated: 2026-10-26_

To receive a WarpLink referral webhook in Python, read the raw bytes with `request.get_data()`, verify the `X-WarpLink-Signature` HMAC-SHA256 over `timestamp.body` with the whole `whsec_` secret, insert the event id so a retry does nothing, and grant the reward. This guide gives a complete Flask receiver you can run and test locally.

You get a server that verifies each delivery, stores handled event ids in SQLite, handles `referral.claimed`, `referral.flagged`, and `referral.updated`, and answers well inside the 10-second limit. A script signs a sample event so you can test without a phone. The Node.js version is [here](https://warplink.app/blog/referral-reward-webhook-node).

## What arrives

Every delivery is a `POST` with `Content-Type: application/json` and three headers:

| Header | Value |
|--------|-------|
| `X-WarpLink-Signature` | One or more lowercase hex signatures, separated by commas |
| `X-WarpLink-Timestamp` | The send time in seconds since the Unix epoch |
| `X-WarpLink-Event-Id` | The event id, the same value as `id` in the body |

The body has four fields: `id`, `type`, `created`, and `data`. For a referral event, `data` holds `referral_id`, `app_id`, `link` (`id`, `slug`, `url`), `installer_id`, `platform`, `match` (`method`, `exact`, `confidence`), `tapped_at`, `installed_at`, `claimed_at`, `claimed_via`, and `flags`. The [webhooks guide](https://warplink.app/docs/guides/webhooks) lists the complete payloads.

The signature is the lowercase hex HMAC-SHA256 of `${timestamp}.${rawBody}`, keyed with the whole signing secret including its `whsec_` prefix. For 24 hours after a rotation the header holds two signatures, separated by a comma.

## Step 1: Create the endpoint and copy the secret

Open the app in the dashboard, go to **Referral handoff**, choose the **Webhook** tab, and add an endpoint with a public `https` address. Copy the signing secret, which starts with `whsec_` and is shown in full once.

```bash
export WARPLINK_WEBHOOK_SECRET="whsec_your_secret_here"
```

## Step 2: Set up the project

```bash
mkdir warplink-webhook && cd warplink-webhook
python3 -m venv .venv
source .venv/bin/activate
pip install flask
```

## Step 3: Write the verifier

The order follows the docs: check the timestamp is a plain integer within 300 seconds, compute the HMAC over the raw bytes, and compare in constant time against every listed signature. Save this as `verify.py`:

```python
import hashlib
import hmac
import re
import time

TOLERANCE_SECONDS = 300

def verify_signature(
    secret: str, raw_body: bytes, signature_header: str, timestamp: str
) -> bool:
    if not re.fullmatch(r"\d{1,12}", timestamp):
        return False
    if abs(time.time() - int(timestamp)) > TOLERANCE_SECONDS:
        return False

    signed = timestamp.encode() + b"." + raw_body
    expected = hmac.new(secret.encode(), signed, hashlib.sha256).hexdigest()

    return any(
        hmac.compare_digest(signature.strip().lower(), expected)
        for signature in signature_header.split(",")
    )
```

`hmac.compare_digest` runs in constant time. The `any` call accepts a match on any listed signature, which is what makes a secret rotation painless.

## Step 4: Write the receiver

Save this as `app.py`. It verifies the raw bytes first, records the event id in SQLite with `INSERT OR IGNORE`, and answers fast:

```python
import json
import os
import sqlite3

from flask import Flask, request

from verify import verify_signature

SECRET = os.environ["WARPLINK_WEBHOOK_SECRET"]  # whsec_...

db = sqlite3.connect("webhooks.db", check_same_thread=False)
db.execute("CREATE TABLE IF NOT EXISTS events (id TEXT PRIMARY KEY)")
db.execute(
    "CREATE TABLE IF NOT EXISTS rewards ("
    "installer_id TEXT PRIMARY KEY, link_id TEXT, event_id TEXT)"
)

app = Flask(__name__)

def first_time(event_id: str) -> bool:
    cursor = db.execute("INSERT OR IGNORE INTO events (id) VALUES (?)", (event_id,))
    db.commit()
    return cursor.rowcount == 1

def set_reward(event: dict) -> None:
    data = event["data"]
    db.execute(
        "INSERT OR REPLACE INTO rewards (installer_id, link_id, event_id) VALUES (?, ?, ?)",
        (data["installer_id"], data["link"]["id"], event["id"]),
    )
    db.commit()

def handle_event(event: dict) -> None:
    if not first_time(event["id"]):
        return  # a retry or a resend

    kind = event["type"]
    if kind == "referral.claimed":
        set_reward(event)
        print(f"reward for installer {event['data']['installer_id']}")
    elif kind == "referral.updated":
        set_reward(event)  # the credit moves to the new link
        print(f"credit moved from {event['data']['previous']['link']['id']}")
    elif kind == "referral.flagged":
        print(f"late claim {event['data']['referral_id']}: {event['data']['flags']}")
    elif kind == "webhook.test":
        print("test event received")

@app.post("/webhooks/warplink")
def warplink_webhook():
    raw_body = request.get_data()
    valid = verify_signature(
        SECRET,
        raw_body,
        request.headers.get("X-WarpLink-Signature", ""),
        request.headers.get("X-WarpLink-Timestamp", ""),
    )
    if not valid:
        return "invalid signature", 400

    handle_event(json.loads(raw_body))
    return "", 204
```

Start it:

```bash
flask --app app run --port 3000
```

Four choices are worth naming:

- **`request.get_data()`, not `request.json`.** The signature covers the exact bytes we sent. Parse the JSON from those bytes after the check.
- **`400` for a bad signature.** Any `4xx` ends the event as failed, so a forged request is never retried. Only a `5xx` or a timeout retries.
- **Dedupe in the database.** Delivery is at least once. The `PRIMARY KEY` on the event id makes `INSERT OR IGNORE` an atomic "have I seen this?" check, even with several workers.
- **Keep the handler short.** We wait 10 seconds for a reply. For slow reward systems, enqueue the event and return at once.

## Step 5: Test without a phone

The dashboard's **Send test** button reaches a public address. For a local server, sign a sample event yourself. Save this as `send_test.py`:

```python
import hashlib
import hmac
import json
import os
import time
import urllib.error
import urllib.request

SECRET = os.environ["WARPLINK_WEBHOOK_SECRET"]

event = {
    "id": "evt_local_0001",
    "type": "referral.claimed",
    "created": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
    "data": {
        "referral_id": "ref_local_0001",
        "app_id": "3f6c1c0e-6a7e-4d0e-9b1e-0a5a8f0c2b11",
        "link": {
            "id": "b7a1f2e4-52c8-4a38-8f64-1d7a0c9e3a55",
            "slug": "invite-ana",
            "url": "https://acme.aplnk.to/invite-ana",
        },
        "installer_id": "d1c2f9a0-7b43-4f0e-a6d5-3c2b8e1f7a90",
        "platform": "ios",
        "match": {"method": "install_token", "exact": True, "confidence": 1},
        "tapped_at": "2026-10-19T15:24:41Z",
        "installed_at": "2026-10-19T15:29:55Z",
        "claimed_at": "2026-10-19T15:30:01Z",
        "claimed_via": "sdk",
        "flags": [],
    },
}

raw_body = json.dumps(event).encode()
timestamp = str(int(time.time()))
signature = hmac.new(
    SECRET.encode(), timestamp.encode() + b"." + raw_body, hashlib.sha256
).hexdigest()

def send() -> None:
    request = urllib.request.Request(
        "http://localhost:3000/webhooks/warplink",
        data=raw_body,
        method="POST",
        headers={
            "Content-Type": "application/json",
            "X-WarpLink-Signature": signature,
            "X-WarpLink-Timestamp": timestamp,
            "X-WarpLink-Event-Id": event["id"],
        },
    )
    try:
        with urllib.request.urlopen(request) as response:
            print(response.status)
    except urllib.error.HTTPError as error:
        print(error.code)

send()
send()  # the second delivery is a repeat and grants nothing
```

Run it in a second terminal with the same secret:

```bash
python3 send_test.py
```

You see `204` twice, and the server prints one `reward for installer ...` line. The second `204` is the dedupe working: same event id, no second reward. Change a byte of `raw_body` after signing and the answer is `400`.

To reach your machine from the dashboard, expose port 3000 through any `https` tunnel, add that address as the endpoint, and press **Send test**. The test event has an id that starts `evt_test_`, goes out once, and is never retried.

## Step 6: Move to production

- **Use your own database.** Put the event insert and the reward write in one transaction, so a crash between them cannot lose or double a reward.
- **Run behind a production server** such as gunicorn, and keep the endpoint on public `https`.
- **Handle `referral.updated`.** It carries `previous`, the old link and match. Take the credit from the previous link's owner and give it to the new one. See [exact versus estimated matches](https://warplink.app/blog/referral-attribution-exact-vs-estimated).
- **Rotate without downtime.** The old secret keeps signing beside the new one for 24 hours. The verifier already accepts either signature.
- **Resend after an outage.** A failed event can be sent again with the same id for 7 days, from the Referrals report or with `POST /v1/webhook-events/{id}/resend`.

For the rules that decide whether a claim should be paid, read [verify a referral before you grant the reward](https://warplink.app/blog/verify-referral-before-reward).

## Frequently Asked Questions

### How do I verify a WarpLink webhook signature in Python?

Read the raw bytes with request.get_data(), build the timestamp, a period, and the body, and compute hmac.new with the whole whsec_ secret, the signed bytes, and hashlib.sha256. Compare the hex digest to each signature in X-WarpLink-Signature with hmac.compare_digest, after checking the timestamp is within 300 seconds.

### Should I use request.json or request.get_data() in Flask?

Use request.get_data(). The signature covers the exact bytes we sent, and a parsed and re-serialized body does not match. Parse the JSON from the same bytes after the check passes.

### How do I stop a retry from paying a reward twice?

Insert the event id into a table with the id as primary key before granting the reward. If the insert adds no row, the event is a repeat. The id is the same on every retry and every resend.

### Does the secret include the whsec_ prefix?

Yes. The HMAC key is the full secret string including the whsec_ prefix, encoded as UTF-8.

### How many times will WarpLink retry a failed delivery?

Up to 5 attempts over 20 minutes, at once and then after 30 seconds, 2 minutes, 8 minutes, and 20 minutes from the first. A 5xx answer or a timeout retries. A 4xx or 3xx answer ends the event as failed.

## Related Guides

- **Reference:** [Receive referral webhooks](https://warplink.app/docs/guides/webhooks) and the [Webhooks API](https://warplink.app/docs/api/webhooks).
- **Same receiver in Node.js:** [Receive a referral reward webhook in Node.js](https://warplink.app/blog/referral-reward-webhook-node).
- **The whole path:** [Referrals](https://warplink.app/docs/guides/referrals) and [WarpLink SDK 1.2.0](https://warplink.app/blog/warplink-sdk-1-2-0).
- **Launch pass:** [A referral link checklist for launch](https://warplink.app/blog/referral-link-checklist-for-launch).
- **Use case:** [App referral programs](https://warplink.app/use-cases/app-referral-programs).

## How WarpLink Helps

We do the matching, keep the record, and send a signed event the moment an install is credited. Your receiver stays small because the hard part, proving which tap led to which sign-up, is already done.

One workflow, three pillars. Deep linking carries the invitation, install attribution credits the link that produced the new user, and real-time analytics shows referrals beside every other link. Sub-10ms redirects on the way in, one SDK on the way out.

[Create a free WarpLink account](https://dash.warplink.app/auth/signup) and get deep linking, install attribution, and real-time analytics in one SDK, with 10,000 clicks a month on the free plan.