Webhook Signature Verifier With Replay Protection

How I verify Stripe-style webhook signatures and stop someone from re-POSTing yesterday's `invoice.paid`. Stdlib HMAC, a tolerance window, and an idempotency cache that lives in any Redis-shaped store.

Python

Compiler

3 snippets

webhooks

security

authentication

hashing

By @averyperry

November 28, 2025

Updated May 20, 2026

594 views

4.2 (10)

Verify the signature on the raw body

Ask AI

from __future__ import annotations
import hashlib
import hmac
import time

SECRET = b'whsec_replace_me'


def verify_signature(raw_body: bytes, header: str, secret: bytes = SECRET, tolerance_sec: int = 300, now: float = None) -> dict:
    # Header format we mimic: 't=<unix>,v1=<hex>'
    parts = dict(p.split('=', 1) for p in header.split(',') if '=' in p)
    ts_str = parts.get('t')
    sig_hex = parts.get('v1')
    if not ts_str or not sig_hex:
        return {'ok': False, 'reason': 'malformed_header'}
    try:
        ts = int(ts_str)
    except ValueError:
        return {'ok': False, 'reason': 'malformed_timestamp'}
    current = time.time() if now is None else now
    if abs(current - ts) > tolerance_sec:
        return {'ok': False, 'reason': 'outside_tolerance'}
    signed = f'{ts}.'.encode('ascii') + raw_body
    expected = hmac.new(secret, signed, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, sig_hex):
        return {'ok': False, 'reason': 'bad_signature'}
    return {'ok': True, 'timestamp': ts}


if __name__ == '__main__':
    body = b'{"id":"evt_42","type":"invoice.paid"}'
    ts = int(time.time())
    sig = hmac.new(SECRET, f'{ts}.'.encode('ascii') + body, hashlib.sha256).hexdigest()
    header = f't={ts},v1={sig}'
    print('valid:', verify_signature(body, header))

    # Tampered body.
    print('tampered:', verify_signature(body + b' ', header))

    # Stale timestamp (10 minutes old).
    old_ts = ts - 600
    old_sig = hmac.new(SECRET, f'{old_ts}.'.encode('ascii') + body, hashlib.sha256).hexdigest()
    print('stale:', verify_signature(body, f't={old_ts},v1={old_sig}'))

The signed string is <timestamp>.<raw_body>, not the parsed JSON, because re-serializing JSON is the classic way to break a verifier (key order, whitespace, float precision). A 5-minute tolerance is the value Stripe defaults to and is what I keep in my own services because anything tighter starts catching legitimate clock drift between cloud regions. The tolerance check happens before the HMAC check on purpose: an attacker who can spam your endpoint forces you to do real crypto work for free without the timestamp gate. hmac.compare_digest is again the only correct way to compare the two hex strings.

Block replays with an idempotency cache

from __future__ import annotations
import hashlib
import hmac
import time
from typing import Callable

SECRET = b'whsec_replace_me'


class InMemoryStore:
    def __init__(self) -> None:
        self.data: dict = {}

    def setnx(self, key: str, ttl_sec: int) -> bool:
        # Returns True if the key was inserted; False if it already existed.
        now = time.time()
        # Evict expired entries lazily.
        existing = self.data.get(key)
        if existing and existing > now:
            return False
        self.data[key] = now + ttl_sec
        return True


def handle_webhook(
    raw_body: bytes,
    header: str,
    event_id: str,
    store: InMemoryStore,
    secret: bytes = SECRET,
    tolerance_sec: int = 300,
    replay_window_sec: int = 24 * 3600,
) -> dict:
    parts = dict(p.split('=', 1) for p in header.split(',') if '=' in p)
    ts = int(parts['t'])
    sig_hex = parts['v1']
    if abs(time.time() - ts) > tolerance_sec:
        return {'ok': False, 'reason': 'outside_tolerance'}
    expected = hmac.new(secret, f'{ts}.'.encode('ascii') + raw_body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, sig_hex):
        return {'ok': False, 'reason': 'bad_signature'}
    # Replay protection: refuse to process the same event id twice within window.
    if not store.setnx(f'webhook:{event_id}', replay_window_sec):
        return {'ok': False, 'reason': 'already_processed', 'event_id': event_id}
    return {'ok': True, 'event_id': event_id}


if __name__ == '__main__':
    body = b'{"id":"evt_42","type":"invoice.paid"}'
    ts = int(time.time())
    sig = hmac.new(SECRET, f'{ts}.'.encode('ascii') + body, hashlib.sha256).hexdigest()
    header = f't={ts},v1={sig}'
    store = InMemoryStore()

    print('first delivery:', handle_webhook(body, header, 'evt_42', store))
    print('replay:', handle_webhook(body, header, 'evt_42', store))
    print('different event:', handle_webhook(body, header, 'evt_43', store))

Signature verification stops a stranger from forging a webhook, but it does nothing about the legitimate sender retrying. A network timeout on the provider's side means your endpoint will see the same evt_42 twice and, if your handler is not idempotent, charges or status flips can double up. The pattern I ship is setnx(event_id, ttl) against any key-value store: the first writer wins, the second sees False and returns 200 immediately so the provider stops retrying. The TTL is set to the provider's longest retry window plus a margin (Stripe is 3 days; I use 7 to be safe).

Run it as a Flask handler

from __future__ import annotations
import hashlib
import hmac
import json
import time

SECRET = b'whsec_replace_me'


class InMemoryStore:
    def __init__(self) -> None:
        self.data: dict = {}

    def setnx(self, key: str, ttl_sec: int) -> bool:
        now = time.time()
        if self.data.get(key, 0) > now:
            return False
        self.data[key] = now + ttl_sec
        return True


store = InMemoryStore()


def webhook_endpoint(raw_body: bytes, signature_header: str) -> tuple:
    # Verify signature.
    parts = dict(p.split('=', 1) for p in signature_header.split(',') if '=' in p)
    ts_str = parts.get('t', '')
    sig_hex = parts.get('v1', '')
    if not ts_str or not sig_hex:
        return 400, {'error': 'bad_header'}
    ts = int(ts_str)
    if abs(time.time() - ts) > 300:
        return 400, {'error': 'outside_tolerance'}
    expected = hmac.new(SECRET, f'{ts}.'.encode('ascii') + raw_body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, sig_hex):
        return 400, {'error': 'bad_signature'}
    # Parse AFTER verification; never decode untrusted JSON before the HMAC check.
    payload = json.loads(raw_body)
    event_id = payload.get('id')
    if not event_id:
        return 400, {'error': 'missing_id'}
    # Replay-safe handler.
    if not store.setnx(f'webhook:{event_id}', 7 * 24 * 3600):
        # Return 200 so provider stops retrying; we already saw this one.
        return 200, {'status': 'duplicate'}
    # ... do the real work here, idempotent at the database level too ...
    return 200, {'status': 'processed', 'event_id': event_id}


if __name__ == '__main__':
    body = b'{"id":"evt_99","type":"invoice.paid"}'
    ts = int(time.time())
    sig = hmac.new(SECRET, f'{ts}.'.encode('ascii') + body, hashlib.sha256).hexdigest()
    header = f't={ts},v1={sig}'

    print('first:', webhook_endpoint(body, header))
    print('retry:', webhook_endpoint(body, header))
    print('forged sig:', webhook_endpoint(body, f't={ts},v1=' + 'aa' * 32))

The order of operations matters more than any individual check: header parse, timestamp tolerance, HMAC compare, and only then json.loads on the raw body. JSON parsing is not free and accepts a wide variety of inputs, so doing it on a forged 100MB body before signature verification is a denial-of-service vector. Returning 200 for a known-duplicate event is the part most homegrown handlers get wrong; if you return 4xx, the provider will keep retrying on its schedule for hours. In production I swap InMemoryStore for Redis with SET key value NX EX ttl and the rest of the code is unchanged.

Back to Code Snippets