This guide is part of the Webhook Event Ingestion section within the broader Data Ingestion & Grant Parsing Workflows framework, and it solves one narrow problem: how do you prove that a POST to your /webhooks/grants endpoint actually came from the funder portal — and not from anyone who guessed the URL — before you let its payload touch your pipeline?
A public webhook endpoint is an unauthenticated write into your system. The portal signs each request with a shared secret and an HMAC-SHA256 digest; your job is to recompute that digest and admit only requests whose signatures match, within a bounded time window, using a comparison that leaks no timing information. Get any one of those three wrong — the byte-exact body, the constant-time compare, or the replay window — and the check is decorative.
When to Use This Approach
Reach for signature verification on every inbound webhook where all three conditions hold:
- The endpoint is internet-reachable and effect-bearing. A grant-status webhook that flips an application to
awardedor enqueues a disbursement is a state change an attacker would love to forge. Signature verification is the transmission-integrity control that NIST SP 800-53 SC-8 requires for data in transit, and the accept/reject decision is exactly the kind of event NIST SP 800-53 AU-2 expects you to log. - The sender publishes a signing scheme. Grants.gov, Fluxx, Submittable, and most foundation portals document a header pair — a signature and a timestamp — and a canonical signing string, almost always
timestamp + "." + raw_body. If the portal only offers Basic Auth or an IP allowlist, this guide does not apply; use their scheme. - You control the web framework’s body handling. The single hardest requirement is capturing the raw request bytes before any JSON middleware parses and re-serializes them. If your stack forces early parsing you must disable it for this route, or the signature will never match.
Payload schema validation, retry deduplication, and downstream routing are explicitly out of scope here. A verified-but-replayed delivery is deduplicated by Deduplicating Webhook Retries with Idempotency Keys; outbound polling that pulls the same events belongs to API Polling & Rate Limiting. This stage answers one question only — is this request authentic and fresh? — and returns 401 when the answer is no.
Step-by-Step Implementation
The reference implementation targets Python 3.10+ and uses only the standard library — hmac, hashlib, time, and logging. No third-party crypto is needed or wanted; the stdlib primitives are the correct tools and carry the constant-time comparison you must have.
# stdlib only — no pip install required
# python >= 3.10
One contract to internalize before writing any code: the signature is computed over bytes you must not modify. The portal signs the exact octets it put on the wire. The instant a JSON library parses that body and you re-serialize it — reordering keys, changing whitespace, normalizing unicode — the digest changes and every request fails verification. Capture the raw body first, verify, then parse.
Step 1: Capture the raw request body before any parsing
This is the pitfall that breaks more webhook integrations than all others combined. Read the raw bytes off the request and keep them; do not hand the request to a JSON parser and try to re-serialize its output. The helper below is framework-agnostic — it takes the already-read raw bytes plus the header mapping, so your route handler is responsible only for pulling request.body (Django), await request.body() (FastAPI/Starlette), or request.get_data() (Flask) before anything else touches it.
import logging
from dataclasses import dataclass
from typing import Mapping
AUDIT_LOGGER = logging.getLogger("grant_webhook.audit")
AUDIT_LOGGER.setLevel(logging.INFO)
_handler = logging.StreamHandler()
_handler.setFormatter(logging.Formatter("%(asctime)s | %(levelname)s | %(message)s"))
AUDIT_LOGGER.addHandler(_handler)
@dataclass(frozen=True)
class InboundWebhook:
"""Immutable capture of a webhook exactly as received, before JSON parsing."""
raw_body: bytes # the untouched request octets — never re-serialized
signature_hex: str # hex digest sent by the portal
timestamp: str # signed timestamp header, seconds since epoch
key_id: str # selects which shared secret signed this request
def capture_webhook(raw_body: bytes, headers: Mapping[str, str]) -> InboundWebhook:
"""Bind raw bytes and signing headers into one immutable record."""
try:
return InboundWebhook(
raw_body=raw_body,
signature_hex=headers["X-Grant-Signature"].strip(),
timestamp=headers["X-Grant-Timestamp"].strip(),
key_id=headers.get("X-Grant-Key-Id", "default").strip(),
)
except KeyError as exc:
AUDIT_LOGGER.warning("Webhook rejected: missing signing header %s", exc)
raise ValueError(f"missing required signing header: {exc}") from exc
Freezing the dataclass guarantees nothing downstream mutates raw_body between capture and digest. The key_id header defaults to "default" so single-secret portals still work, while multi-secret portals during rotation select explicitly.
Step 2: Select the active secret by key_id
Grant portals rotate signing secrets on a schedule, and rotation is only safe if both the old and new secret verify during the overlap window. Model the keyring as a mapping from key_id to a raw-bytes secret, and resolve the incoming key_id against it. An unknown key_id is a hard reject, not a fall-through to a default — that would let an attacker downgrade to a retired key.
from typing import Dict
class KeyResolutionError(Exception):
"""Raised when a webhook names a key_id that is not currently active."""
def resolve_secret(key_id: str, keyring: Dict[str, bytes]) -> bytes:
"""Return the active secret for key_id, or reject an unknown key outright."""
secret = keyring.get(key_id)
if secret is None:
AUDIT_LOGGER.warning("Webhook rejected: unknown key_id %r", key_id)
raise KeyResolutionError(f"unknown key_id: {key_id!r}")
return secret
During a rotation you populate keyring with {"2026-q2": OLD_SECRET, "2026-q3": NEW_SECRET}; once the portal confirms it has cut over, you drop the retired entry and the old key stops verifying immediately. Load these secrets from a secret manager, never a source file — the access boundary for that store is covered by Data Security & Access Boundaries.
Step 3: Rebuild the exact signing string and compute the digest
Concatenate the timestamp, a literal ., and the raw body — matching the sender’s canonical form byte-for-byte — then compute HMAC-SHA256 over it with the resolved secret. Everything here operates on bytes; encoding the timestamp and joining with a b"." separator avoids any implicit unicode step that would diverge from what the portal signed.
import hashlib
import hmac
def compute_expected_signature(webhook: InboundWebhook, secret: bytes) -> str:
"""Rebuild 'timestamp.raw_body' and return the HMAC-SHA256 hex digest."""
signing_string = webhook.timestamp.encode("ascii") + b"." + webhook.raw_body
digest = hmac.new(secret, signing_string, hashlib.sha256).hexdigest()
return digest
The signing string ordering — timestamp first, then body — is what binds the timestamp into the signature. Because the timestamp is signed, an attacker cannot replay an old body under a fresh timestamp: changing the timestamp changes the digest, so the freshness check in Step 5 cannot be bypassed without the secret.
Step 4: Compare digests in constant time
Compare the expected digest to the one the portal sent using hmac.compare_digest, which takes the same time regardless of where the first differing byte is. A plain == short-circuits on the first mismatched character, and that timing difference is enough for a remote attacker to reconstruct a valid signature byte by byte. This is the non-negotiable line of the whole guide.
def signatures_match(expected_hex: str, provided_hex: str) -> bool:
"""Constant-time signature comparison — resists timing side channels."""
# compare_digest tolerates str inputs of equal length; it never short-circuits
return hmac.compare_digest(expected_hex, provided_hex)
hmac.compare_digest accepts either str (ASCII only) or bytes; the hex digests here are ASCII, so passing them directly is safe. Never substitute expected_hex == provided_hex, and never log the two values on mismatch — an emitted expected digest hands an attacker the answer.
Step 5: Enforce a timestamp tolerance window to stop replay
Even a perfectly valid signature can be captured and replayed. Reject any request whose signed timestamp is more than a small tolerance — 300 seconds is the common default — away from your clock. This bounds the replay window to the tolerance and is what turns “authentic” into “authentic and fresh.”
import time
class ReplayWindowError(Exception):
"""Raised when a signed timestamp falls outside the freshness tolerance."""
def assert_fresh(timestamp: str, tolerance_seconds: int = 300) -> None:
"""Reject deliveries whose signed timestamp is outside the tolerance window."""
try:
sent_at = int(timestamp)
except ValueError as exc:
raise ReplayWindowError(f"non-integer timestamp: {timestamp!r}") from exc
skew = abs(time.time() - sent_at)
if skew > tolerance_seconds:
AUDIT_LOGGER.warning("Webhook rejected: timestamp skew %.0fs exceeds window", skew)
raise ReplayWindowError(f"timestamp outside {tolerance_seconds}s window")
Use abs() so both stale replays and clock-ahead forgeries are caught. Keep tolerance tight; widening it to tolerate sloppy clocks proportionally widens the window an attacker has to replay a captured delivery. Genuine duplicate deliveries inside the window are the retry problem, handled by Deduplicating Webhook Retries with Idempotency Keys.
Step 6: Compose the gate and log the AU-2 decision
Wire the steps into one verifier that returns a clean pass/fail and emits exactly one audit line per decision. Every accept and every reject is a security-relevant event under NIST SP 800-53 AU-2, so log the key_id, the outcome, and a truncated digest — never the full expected signature or the secret.
def verify_webhook(webhook: InboundWebhook, keyring: Dict[str, bytes],
tolerance_seconds: int = 300) -> bool:
"""Full gate: authenticate then check freshness, logging the AU-2 decision."""
try:
secret = resolve_secret(webhook.key_id, keyring)
expected = compute_expected_signature(webhook, secret)
if not signatures_match(expected, webhook.signature_hex):
AUDIT_LOGGER.warning("DENY signature | key_id=%s", webhook.key_id)
return False
assert_fresh(webhook.timestamp, tolerance_seconds)
except (KeyResolutionError, ReplayWindowError) as exc:
AUDIT_LOGGER.warning("DENY | key_id=%s | %s", webhook.key_id, exc)
return False
AUDIT_LOGGER.info("ALLOW | key_id=%s | sig=%s...", webhook.key_id, webhook.signature_hex[:10])
return True
Only after verify_webhook returns True do you parse the JSON and admit the event. A False return is a 401 and a stop — the payload is never inspected.
Verification
Confirm the gate behaves correctly with four checks:
- A genuine signature passes. Sign
timestamp + "." + raw_bodywith the shared secret, set the current timestamp, and assertverify_webhookreturnsTrueand emits oneALLOWaudit line. - A tampered body fails. Keep the signature but flip one byte of
raw_body; assertverify_webhookreturnsFalsewith aDENY signatureline. This proves the digest covers the exact bytes. - A stale but validly-signed request is replay-rejected. Sign correctly but set the timestamp to
now - 3600; assert the result isFalsewith a timestamp-windowDENY. This proves the freshness window is enforced independently of the signature. - An unknown key_id is rejected outright. Point
key_idat a value absent from the keyring and assertFalsewith anunknown key_idline — never a silent fall-through to a default secret.
The end-to-end assertion below exercises the happy path and the tamper path against one keyring, which is the minimum an auditor accepts as proof the control is live.
secret = b"portal-shared-secret-2026-q3"
keyring = {"2026-q3": secret}
body = b'{"event":"grant.awarded","application_id":"APP-4471"}'
ts = str(int(time.time()))
sig = hmac.new(secret, ts.encode("ascii") + b"." + body, hashlib.sha256).hexdigest()
good = capture_webhook(body, {"X-Grant-Signature": sig, "X-Grant-Timestamp": ts, "X-Grant-Key-Id": "2026-q3"})
assert verify_webhook(good, keyring) is True
tampered = capture_webhook(body + b" ", {"X-Grant-Signature": sig, "X-Grant-Timestamp": ts, "X-Grant-Key-Id": "2026-q3"})
assert verify_webhook(tampered, keyring) is False
Common Errors & Fixes
| Error | Cause | Fix |
|---|---|---|
Every valid request returns 401 |
JSON middleware parsed and re-serialized the body, so the digest is computed over different bytes | Capture request.body raw bytes before any parser runs; verify, then parse. Disable early JSON parsing on this route. |
| Signatures match in tests but fail in production | Signing string assembled with the wrong separator or order (e.g. body-first, or + on str) |
Match the portal’s documented canonical form byte-for-byte: timestamp.encode() + b"." + raw_body, timestamp first. |
| Intermittent verification failures under load | Using == to compare digests, exposing a timing side channel — or a truncated/uppercased hex mismatch |
Always use hmac.compare_digest; normalize hex case and strip whitespace from the header before comparing. |
| Replayed deliveries accepted hours later | No timestamp freshness check, or tolerance set too wide | Enforce assert_fresh with a tight tolerance (300s); reject on abs(skew) > tolerance. |
| Rotation breaks all webhooks at cutover | Single-secret keyring with no overlap window | Keep old and new secrets both active in the keyring during rotation; drop the retired key_id only after the portal confirms cutover. |
Related
- Parent section: Webhook Event Ingestion
- What happens to a verified duplicate delivery: Deduplicating Webhook Retries with Idempotency Keys
- The pull-based counterpart to push webhooks: API Polling & Rate Limiting
- Where the signing secrets are stored and access-bounded: Data Security & Access Boundaries