Skip to main content

Verifying signatures

Every delivery carries an HMAC-SHA256 signature over the timestamp + raw body. Verifying it on your server is non-optional — without it, anyone who learns your URL can forge events.

The canonical string​

canonical = f"{X-CPG-Timestamp}\n".encode("utf-8") + <raw body bytes>

The HMAC input is the byte concatenation of the timestamp string, a single newline, and the exact request body bytes. Sign / verify against the raw body — don't parse JSON and re-serialise; any whitespace or key-order change breaks the signature.

The signature​

X-CPG-Signature = hmac_sha256(signing_secret, canonical).hexdigest()   # lowercase hex

What your receiver must do​

On every request:

  1. Reject if |now - X-CPG-Timestamp| > 300 seconds (5-minute replay window).
  2. Recompute the HMAC over f"{X-CPG-Timestamp}\n".encode() + raw_body with your stored signing secret.
  3. Constant-time compare the recomputed signature to X-CPG-Signature. Use hmac.compare_digest (Python) / crypto.timingSafeEqual (Node) / hash_equals (PHP) — never ==. A naive comparison leaks timing information that an attacker can exploit to forge signatures byte-by-byte.

If any of the three fails, respond 401 Unauthorized and do not process the payload.

Recipes​

import hmac
import hashlib
import time
from typing import Mapping

REPLAY_WINDOW_SECONDS = 300


def verify_webhook(body: bytes, headers: Mapping[str, str], signing_secret: str) -> bool:
    """Return True iff the signature is valid and the timestamp is fresh."""
    ts = headers.get("X-CPG-Timestamp") or headers.get("x-cpg-timestamp")
    sig = headers.get("X-CPG-Signature") or headers.get("x-cpg-signature")
    if not ts or not sig:
        return False
    try:
        if abs(int(time.time()) - int(ts)) > REPLAY_WINDOW_SECONDS:
            return False
    except ValueError:
        return False
    canonical = f"{ts}\n".encode("utf-8") + body
    expected = hmac.new(signing_secret.encode("utf-8"), canonical, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected.lower(), sig.lower())

FastAPI integration:

from fastapi import FastAPI, Header, HTTPException, Request

app = FastAPI()
SIGNING_SECRET = "<your stored secret>"


@app.post("/webhooks/cpg")
async def cpg_webhook(
    request: Request,
    x_cpg_timestamp: str | None = Header(default=None),
    x_cpg_signature: str | None = Header(default=None),
):
    raw_body = await request.body()
    headers = {"X-CPG-Timestamp": x_cpg_timestamp, "X-CPG-Signature": x_cpg_signature}
    if not verify_webhook(raw_body, headers, SIGNING_SECRET):
        raise HTTPException(status_code=401, detail="Invalid signature")

    # Dedupe by envelope.id, enqueue for async processing, return 2xx fast.
    return {"received": True}

Common verification mistakes​

  1. Parsing then re-serialising the body. JSON.parse(req.body) followed by JSON.stringify(parsed) produces a different byte string than what was signed. The signature is over the raw bytes — never re-serialise before verifying.
  2. Treating the signature as base64. It's lowercase hex (a1b2c3...), not base64. Don't base64_decode() before comparing.
  3. Case-sensitive header lookup. HTTP header names are case-insensitive. Frameworks may normalise to lowercase (x-cpg-signature) or preserve as sent — your verifier must check both casings or canonicalise.
  4. String comparison with ==. Vulnerable to timing attacks. Always use the framework's constant-time compare (hmac.compare_digest, crypto.timingSafeEqual, hash_equals).
  5. Forgetting the timestamp window. Without the 300-second freshness check, an attacker who captures one valid delivery can replay it forever. The signature itself doesn't expire — only the timestamp check makes it freshness-bound.
  6. Body parsing middleware running before the verifier. In Express, putting app.use(express.json()) before the webhook route means req.body is the parsed object — the raw bytes are gone. Use express.raw() on the webhook route specifically (as in the example).