Un webhook è una richiesta HTTP con cui un sistema comunica a un altro che si è verificato un evento. Un pagamento completato, una build terminata o una nuova issue possono arrivare così a un'applicazione senza che questa debba interrogare continuamente il servizio di origine.

Il problema nasce quando il ricevente tratta una POST ben formata come prova di autenticità. Il corpo JSON si può costruire con qualunque client HTTP; anche se proviene davvero dal mittente, la stessa consegna potrebbe arrivare una seconda volta per un timeout o un nuovo tentativo del servizio. Se l'handler crea un ordine, invia una mail o accredita un saldo a ogni richiesta, la duplicazione ha effetti reali.

In questo laboratorio realizziamo un protocollo didattico per webhook interni con HMAC-SHA256, finestra temporale, identificativo dell'evento e inbox SQLite. Il codice usa soltanto la libreria standard di Python 3.10 o successivi. Non implementa il formato di firma di GitHub o Stripe: per quei servizi vanno usati i rispettivi protocolli e, quando disponibili, gli SDK ufficiali.

Indice

Cosa protegge una firma HMAC

HMAC, Hash-based Message Authentication Code, combina una funzione di hash con una chiave segreta condivisa. Chi riceve il messaggio ricalcola il codice e lo confronta con quello fornito dal mittente. Se l'avversario non conosce la chiave, non può produrre una firma valida per un payload modificato con probabilità significativa. Questa è la proprietà descritta dalla RFC 2104.

La verifica HMAC offre autenticazione rispetto alla chiave condivisa e integrità dei dati firmati. Non identifica quale persona abbia avviato l'evento, non impedisce la ripetizione di una richiesta autentica e non garantisce da sola un'elaborazione unica.

Controllo Problema che affronta Limite
HMAC-SHA256 Payload o metadati firmati alterati Una firma autentica si può riutilizzare
Timestamp incluso nella firma Richieste catturate e ripresentate molto tempo dopo I replay entro la finestra restano possibili
Identificativo firmato + chiave univoca Doppia registrazione dello stesso evento Il worker deve gestire anche le proprie ripartenze
TLS Lettura e modifica del traffico in transito Non sostituisce la verifica dell'origine applicativa

La distinzione tra replay e duplicato operativo è utile: il replay può essere tentato intenzionalmente; una ritrasmissione può essere la normale risposta del provider a un errore di rete. In entrambi i casi il risultato desiderato è evitare due effetti applicativi per il medesimo evento.

Dove fallisce un handler ingenuo

Questo frammento rappresenta un errore frequente. Assume un oggetto request fornito dal framework e una funzione applicativa accredita_ordine:

import json

def webhook_insicuro(request):
    evento = json.loads(request.body)
    accredita_ordine(evento["order_id"])
    return 200

Chiunque possa raggiungere l'endpoint può costruire il JSON. Se la richiesta viene inviata due volte, la funzione viene richiamata due volte. Un controllo sul solo header Content-Type: application/json non cambia il problema, perché quell'header è scelto dal client.

Anche firmare soltanto il JSON dopo averlo deserializzato è fragile. Spazi, ordine delle chiavi, codifica Unicode o rappresentazione dei numeri possono essere modificati dal parser o da un serializer. La firma va verificata sui byte originali del corpo HTTP, prima di interpretare il contenuto. È una prescrizione esplicita anche nelle istruzioni di GitHub e Stripe.

Definire il messaggio firmato

Nel nostro protocollo entrambe le applicazioni conoscono una chiave segreta distinta per questa integrazione. Il mittente invia un corpo JSON come byte e tre header:

  • X-Demo-Timestamp: timestamp Unix in secondi, per esempio 1800000000;
  • X-Demo-Delivery: identificativo di 32 caratteri esadecimali minuscoli, stabile anche sui retry dello stesso evento;
  • X-Demo-Signature: stringa sha256= seguita da 64 caratteri esadecimali minuscoli.

La firma copre esattamente questa sequenza di byte:

ASCII(timestamp) + b"." + ASCII(delivery_id) + b"." + raw_body

I due campi precedenti al corpo hanno una sintassi vincolata che esclude il punto: la concatenazione è quindi interpretabile senza ambiguità. Il mittente deve generarli, firmarli e inviarli insieme allo stesso corpo. La chiave deve avere almeno 32 byte generati casualmente con un generatore crittograficamente sicuro, essere conservata fuori dal repository e trasmessa al ricevente su un canale adeguato. La lunghezza minima controllata dal codice non prova da sola che la chiave sia imprevedibile.

La finestra temporale dell'esempio è di 300 secondi in entrambe le direzioni rispetto all'orologio del server. È una scelta del laboratorio, non una proprietà universale di HMAC. Il mittente e il ricevente devono avere orologi sincronizzati, e un ritentativo effettuato dopo la scadenza deve avere un nuovo timestamp e una nuova firma ma mantenere l'identificativo logico dell'evento.

Implementare verifica e deduplicazione

Salva il file seguente come webhook_security.py. make_signature rappresenta anche la funzione da usare lato mittente di questo protocollo; il ricevente applica tutti i controlli prima di accettare l'evento.

"""Protocollo didattico per webhook interni, non compatibile con GitHub/Stripe."""
import hashlib
import hmac
import re
import sqlite3
import time

MAX_BODY_BYTES = 64 * 1024
CLOCK_TOLERANCE_SECONDS = 300


def make_signature(secret: bytes, body: bytes, timestamp: str, delivery_id: str) -> str:
    if len(secret) < 32:
        raise ValueError("chiave segreta troppo corta")
    signed = timestamp.encode("ascii") + b"." + delivery_id.encode("ascii") + b"." + body
    return "sha256=" + hmac.new(secret, signed, hashlib.sha256).hexdigest()


def verify_webhook(
    secret: bytes, body: bytes, headers: dict[str, str], now: int | None = None
) -> str:
    if len(body) > MAX_BODY_BYTES:
        raise ValueError("payload troppo grande")

    timestamp = headers.get("X-Demo-Timestamp", "")
    delivery_id = headers.get("X-Demo-Delivery", "")
    provided = headers.get("X-Demo-Signature", "")

    if not re.fullmatch(r"[0-9]{10,12}", timestamp):
        raise ValueError("timestamp non valido")
    if not re.fullmatch(r"[0-9a-f]{32}", delivery_id):
        raise ValueError("identificativo non valido")
    if not re.fullmatch(r"sha256=[0-9a-f]{64}", provided):
        raise ValueError("firma mancante o malformata")

    current_time = int(time.time()) if now is None else now
    if abs(current_time - int(timestamp)) > CLOCK_TOLERANCE_SECONDS:
        raise ValueError("timestamp fuori finestra")

    expected = make_signature(secret, body, timestamp, delivery_id)
    if not hmac.compare_digest(expected, provided):
        raise ValueError("firma non valida")
    return delivery_id


def record_once(conn: sqlite3.Connection, delivery_id: str, body: bytes) -> bool:
    """Inserisce atomicamente nella inbox; True solo per la prima consegna."""
    with conn:
        cur = conn.execute(
            "INSERT OR IGNORE INTO webhook_inbox (delivery_id, payload) VALUES (?, ?)",
            (delivery_id, body),
        )
    return cur.rowcount == 1

La regex sulla firma elimina input malformati prima del confronto, mentre hmac.compare_digest() evita il confronto carattere per carattere con uscita anticipata che potrebbe introdurre differenze temporali osservabili. Il limite di 64 KiB va applicato anche a livello di web server o reverse proxy, per impedire che il framework carichi in memoria richieste molto più grandi prima dell'esecuzione di questa funzione.

L'inbox impedisce il doppio inserimento

Dopo la verifica crittografica, registriamo l'evento in una inbox, una tabella persistente che conserva le consegne ricevute prima dell'elaborazione asincrona. L'unicità della chiave viene fatta rispettare dal database, invece di affidarsi a un controllo del tipo «cerco il record e, se manca, lo inserisco». Quest'ultima sequenza è vulnerabile alle condizioni di gara tra richieste concorrenti.

CREATE TABLE webhook_inbox (
    delivery_id TEXT PRIMARY KEY,
    payload BLOB NOT NULL,
    status TEXT NOT NULL DEFAULT 'pending',
    received_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);

record_once() esegue un INSERT OR IGNORE transazionale. La prima chiamata restituisce True; le successive, anche da connessioni concorrenti verso lo stesso database SQLite, restituiscono False. L'evento rimane in stato pending finché un worker non lo prende in carico.

Attenzione alla semantica: l'inserimento unico non equivale a un'esecuzione esattamente una volta. Un processo può interrompersi dopo aver chiamato un servizio esterno e prima di segnare il job come completato. Un worker affidabile deve adottare lock o claim transazionali, retry e un'ulteriore chiave di idempotenza nell'operazione di business. Per pagamenti e altre operazioni sensibili è preferibile un vincolo unico sull'identificativo dell'operazione anche nella tabella applicativa o un'idempotency key supportato dall'API a valle.

Eseguire i test

Salva test_webhook_security.py nella stessa directory. L'intera suite non richiede servizi esterni, webhook reali né richieste verso domini di terzi.

import hashlib
import hmac
import os
import sqlite3
import tempfile
import unittest
from concurrent.futures import ThreadPoolExecutor

from webhook_security import (
    MAX_BODY_BYTES, make_signature, record_once, verify_webhook,
)


DDL = """CREATE TABLE webhook_inbox (
    delivery_id TEXT PRIMARY KEY,
    payload BLOB NOT NULL,
    status TEXT NOT NULL DEFAULT 'pending',
    received_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
)"""


class WebhookTests(unittest.TestCase):
    SECRET = b"chiave-di-test-solo-per-laboratorio"
    BODY = b'{"type":"order.paid","order_id":42}'
    NOW = 1_800_000_000
    DELIVERY_ID = "a" * 32

    def headers(self, body=None, timestamp=None, delivery_id=None):
        body = self.BODY if body is None else body
        timestamp = str(self.NOW if timestamp is None else timestamp)
        delivery_id = self.DELIVERY_ID if delivery_id is None else delivery_id
        return {
            "X-Demo-Timestamp": timestamp,
            "X-Demo-Delivery": delivery_id,
            "X-Demo-Signature": make_signature(self.SECRET, body, timestamp, delivery_id),
        }

    def test_secret_not_configured(self):
        with self.assertRaises(ValueError):
            self.headers()["X-Demo-Signature"] = make_signature(b"", self.BODY, str(self.NOW), self.DELIVERY_ID)

    def test_valid_signature(self):
        self.assertEqual(
            verify_webhook(self.SECRET, self.BODY, self.headers(), self.NOW),
            self.DELIVERY_ID,
        )

    def test_tampered_body(self):
        with self.assertRaises(ValueError):
            verify_webhook(self.SECRET, self.BODY + b"x", self.headers(), self.NOW)

    def test_missing_signature(self):
        headers = self.headers()
        del headers["X-Demo-Signature"]
        with self.assertRaises(ValueError):
            verify_webhook(self.SECRET, self.BODY, headers, self.NOW)

    def test_stale_and_future_timestamps(self):
        for value in (self.NOW - 301, self.NOW + 301):
            with self.subTest(value=value), self.assertRaises(ValueError):
                verify_webhook(self.SECRET, self.BODY, self.headers(timestamp=value), self.NOW)

    def test_tampered_delivery_id(self):
        headers = self.headers()
        headers["X-Demo-Delivery"] = "b" * 32
        with self.assertRaises(ValueError):
            verify_webhook(self.SECRET, self.BODY, headers, self.NOW)

    def test_oversized_body(self):
        with self.assertRaises(ValueError):
            verify_webhook(self.SECRET, b"x" * (MAX_BODY_BYTES + 1), self.headers(), self.NOW)

    def test_duplicate_insert(self):
        conn = sqlite3.connect(":memory:")
        try:
            conn.execute(DDL)
            self.assertTrue(record_once(conn, self.DELIVERY_ID, self.BODY))
            self.assertFalse(record_once(conn, self.DELIVERY_ID, self.BODY))
            self.assertEqual(conn.execute("SELECT COUNT(*) FROM webhook_inbox").fetchone()[0], 1)
        finally:
            conn.close()

    def test_concurrent_duplicate_insert(self):
        with tempfile.TemporaryDirectory() as folder:
            path = os.path.join(folder, "inbox.sqlite")
            conn = sqlite3.connect(path)
            conn.execute(DDL)
            conn.close()

            def insert(_):
                db = sqlite3.connect(path, timeout=3)
                try:
                    return record_once(db, self.DELIVERY_ID, self.BODY)
                finally:
                    db.close()

            with ThreadPoolExecutor(max_workers=4) as executor:
                result = list(executor.map(insert, range(4)))
            self.assertEqual(result.count(True), 1)
            self.assertEqual(result.count(False), 3)

    def test_hmac_official_github_vector(self):
        secret = b"It's a Secret to Everybody"
        signature = hmac.new(secret, b"Hello, World!", hashlib.sha256).hexdigest()
        self.assertEqual(signature, "757107ea0eb2509fc211221cce984b8a37570b6d7586c22c46f4379c8b043e17")


if __name__ == "__main__":
    unittest.main()

Esegui:

python3 -m unittest -v test_webhook_security.py

La suite verifica la firma valida, un corpo modificato, la firma assente, timestamp vecchi e futuri, identificativi alterati, payload eccessivi e la deduplicazione sia sequenziale sia concorrente. L'ultimo test usa anche il vettore HMAC pubblico delle istruzioni di GitHub: questo controlla indipendentemente il calcolo SHA-256 utilizzato nel laboratorio, senza simulare falsamente il protocollo completo di GitHub.

Se tutti i test terminano con OK, la logica dimostrativa supera i casi coperti. Non significa che un'applicazione integrata in produzione sia già sicura: occorre verificare body, header, limiti e persistence layer nella configurazione reale.

Integrare i controlli in un'applicazione

In Django, per esempio, il corpo originale è disponibile tramite request.body. Nel controller dedicato ai webhook la sequenza corretta è questa: accettare solo POST, leggere i byte una sola volta, applicare il limite di dimensione, costruire la mappa dei tre header, chiamare verify_webhook(), validare la struttura dell'evento e infine registrarlo con record_once().

Un endpoint che usa HMAC come autenticazione non può dipendere da un token CSRF destinato alle form del browser. Se si usa csrf_exempt per quella singola route, la verifica della firma deve restare obbligatoria e avvenire prima di qualunque effetto o inserimento nella coda.

Per lo stato HTTP si può adottare questa convenzione: 401 se la firma è assente o errata; 400 per JSON o metadati malformati; 413 per un corpo oltre la dimensione massima; 202 quando l'evento è stato registrato in modo durevole; 200 per una consegna già registrata. Se il database non è disponibile, rispondere con un codice 5xx permette al mittente che supporta i retry di ritentare. Il contratto del provider va verificato prima di adottare questi codici, perché non tutti interpretano gli errori allo stesso modo.

Altri controlli che restano necessari

Nel deploy, la verifica non deve fidarsi dell'indirizzo IP come unica forma di autenticazione. Servono HTTPS, timeout, rate limiting, segmentazione delle credenziali e log strutturati privi di segreti o payload completi. La rotazione della chiave richiede un periodo controllato in cui si accettano vecchia e nuova chiave, seguito dalla revoca effettiva di quella precedente.

Un identificativo evento deve rimanere stabile nei retry e non essere riutilizzato per eventi differenti. Se il provider offre soltanto la firma sul body, come nello schema GitHub documentato con X-Hub-Signature-256, non si devono inventare timestamp o metadati firmati che il provider non produce. Si verifica la firma secondo le sue regole e si implementa la deduplicazione usando l'ID di consegna fornito, valutando il perimetro di fiducia del relativo header. Per GitHub, inoltre, X-GitHub-Delivery non è coperto dall’HMAC del corpo: è utile per riconoscere le normali ritrasmissioni, ma da solo non costituisce una protezione crittografica contro chi ripresenta un payload firmato alterando quell’header. Per Stripe occorre seguire la costruzione del payload firmato e le indicazioni sul timestamp della sua documentazione, preferibilmente tramite SDK.

Un ultimo test significativo consiste nel far arrivare due consegne valide quasi contemporaneamente e controllare l'effetto sul database applicativo, non soltanto sul codice HTTP restituito. È lì che l'idempotenza diventa verificabile.

Risorse