Webhook Event Ingestion

This guide is part of the Data Ingestion & Grant Parsing Workflows reference. Webhook Event Ingestion is the push-driven intake gate that accepts funder and grant-portal callbacks — award status changes, submission receipts, deadline revisions — proves each callback is authentic and non-replayed, collapses duplicate retries, orders events that arrive out of sequence, and emits a single canonical ingestion record for downstream validation.

The scope is strictly confined to the receive-verify-normalize boundary for inbound HTTP callbacks. This gate does not poll or pull anything on a schedule — scheduled acquisition from grant-portal APIs belongs to API Polling & Rate Limiting. It does not fan work out across concurrent workers; bounded concurrent dispatch of accepted events belongs to Async Batch Processing Pipelines. It does not adjudicate funder or federal rules against the payload — that evaluation belongs to the Core Architecture & Compliance Mapping reference, which consumes what this gate emits. Two mechanics are deep enough to warrant their own guides: the cryptographic detail of signature verification lives in Verifying HMAC signatures on grant portal webhooks, and the retry-collapse detail lives in Deduplicating webhook retries with idempotency keys. Every code path here terminates at exactly one of four outcomes: an accepted canonical event handed to validation, or a deterministic reject on a bad signature, a stale timestamp, or a duplicate event id.

Prerequisites #

This gate targets Python 3.11+ for the standard-library hmac.compare_digest constant-time primitive and modern pydantic v2 typing ergonomics. Pin dependencies so signature verification and payload coercion are byte-for-byte reproducible across funder integrations and across audit re-runs — an unpinned pydantic bump can change coercion behavior and quietly alter what the canonical contract accepts.

# requirements.txt — pinned for reproducible ingestion
fastapi==0.115.6
pydantic==2.9.2
httpx==0.27.2
starlette==0.41.3

The receiver itself is framework-agnostic (it takes raw bytes and a header mapping), but a thin ASGI edge is expected in production; fastapi/starlette supply that edge and httpx is used only in the verification test harness to replay signed fixtures. Environment variables consumed by this stage:

Upstream stage dependency: unlike a pull ingestion gate, this stage has no upstream stage inside the pipeline — the funder’s grant portal is the producer, and the callback arrives unsolicited over the public edge. That inversion is exactly why authentication happens before anything else: there is no prior stage that has already vouched for the artifact. The scheduled, pull-based counterpart to this gate is API Polling & Rate Limiting; the two are complementary intake modes, not substitutes, and a resilient integration usually runs both so that a missed callback is eventually reconciled by a scheduled pull.

The receiver contract #

A webhook receiver must be callable from any HTTP edge without dragging a framework into the core. The contract is deliberately narrow: it accepts the raw request body as bytes (never a re-serialized dict — re-serialization changes byte order and breaks the signature) and a case-insensitive mapping of headers, and it returns a structured ReceiveResult naming the terminal outcome. The four checks run in a fixed order so that an unauthenticated caller can never advance the dedupe or mapping machinery.

1. Signature verification. Recompute HMAC-SHA256 over the exact raw body and compare against the funder-supplied header in constant time.
2. Replay-window check. Reject any callback whose signed timestamp is outside WEBHOOK_REPLAY_WINDOW_SECONDS, which bounds how long a captured-and-replayed request stays useful.
3. Idempotent de-duplication. Collapse retries by recording the event id in a shared seen-store; a repeat is acknowledged but never re-processed.
4. Canonical mapping. Translate the funder-specific payload into the canonical ingestion record before handoff.

Core Implementation #

The receiver below is a single callable. It routes every terminal outcome through one structured ReceiveResult and emits an append-only audit line per decision, so a compliance officer can reconstruct why any callback was accepted or rejected from the log alone. Deep verification detail is intentionally delegated — see Verifying HMAC signatures on grant portal webhooks — so this class stays focused on orchestration.

import hashlib
import hmac
import logging
import time
from dataclasses import dataclass
from typing import Any, Dict, Mapping, Optional, Protocol

logger = logging.getLogger("grant_webhook.receiver")


class SeenStore(Protocol):
    """Shared idempotency store; first-write-wins semantics required."""

    def add_if_absent(self, event_id: str, ttl_seconds: int) -> bool:
        """Return True if the event id was newly recorded, False if already present."""


@dataclass(frozen=True)
class ReceiveResult:
    outcome: str  # ACCEPTED | DUPLICATE | ERR_BAD_SIGNATURE | ERR_STALE_TIMESTAMP
    http_status: int
    event_id: Optional[str] = None
    canonical: Optional[Dict[str, Any]] = None


def audit_log(event: str, event_id: Optional[str], metadata: Dict[str, Any]) -> None:
    """Append-only structured audit hook for compliance traceability."""
    logger.info(
        "%s | event_id=%s | framework=NIST_SP_800-53_AU-2 | metadata=%s",
        event, event_id, metadata,
    )


class WebhookReceiver:
    def __init__(
        self,
        signing_secret: bytes,
        seen_store: SeenStore,
        replay_window_seconds: int,
        dedupe_ttl_seconds: int,
    ) -> None:
        self._secret = signing_secret
        self._seen = seen_store
        self._replay_window = replay_window_seconds
        self._dedupe_ttl = dedupe_ttl_seconds

    def _valid_signature(self, raw_body: bytes, provided_hex: str) -> bool:
        expected = hmac.new(self._secret, raw_body, hashlib.sha256).hexdigest()
        # Constant-time compare defeats timing side channels; never use ==.
        return hmac.compare_digest(expected, provided_hex)

    def receive(self, raw_body: bytes, headers: Mapping[str, str]) -> ReceiveResult:
        lower = {k.lower(): v for k, v in headers.items()}
        provided_sig = lower.get("x-grant-signature", "")
        signed_ts = lower.get("x-grant-timestamp", "")

        # Stage 1 — authenticate before touching any stateful machinery.
        if not provided_sig or not self._valid_signature(raw_body, provided_sig):
            audit_log("webhook_rejected", None, {"reason": "ERR_BAD_SIGNATURE"})
            return ReceiveResult("ERR_BAD_SIGNATURE", http_status=401)

        # Stage 2 — bound replay usefulness by clock skew.
        try:
            skew = abs(time.time() - float(signed_ts))
        except ValueError:
            audit_log("webhook_rejected", None, {"reason": "ERR_STALE_TIMESTAMP", "ts": signed_ts})
            return ReceiveResult("ERR_STALE_TIMESTAMP", http_status=401)
        if skew > self._replay_window:
            audit_log("webhook_rejected", None, {"reason": "ERR_STALE_TIMESTAMP", "skew": skew})
            return ReceiveResult("ERR_STALE_TIMESTAMP", http_status=401)

        canonical = map_to_canonical(raw_body)
        event_id = canonical["event_id"]

        # Stage 3 — idempotent de-dup; a retry acks 200 but never re-processes.
        if not self._seen.add_if_absent(event_id, self._dedupe_ttl):
            audit_log("webhook_duplicate", event_id, {"outcome": "DUPLICATE"})
            return ReceiveResult("DUPLICATE", http_status=200, event_id=event_id)

        # Stage 4 — accepted; hand the canonical record to validation.
        audit_log("webhook_accepted", event_id, {"event_type": canonical["event_type"]})
        return ReceiveResult("ACCEPTED", http_status=202, event_id=event_id, canonical=canonical)

The order is load-bearing. Signature verification precedes timestamp parsing so that a forged request never influences control flow through a crafted timestamp; de-duplication precedes mapping side effects so that a retry is cheap; and mapping happens last so that only authentic, fresh, first-seen events are ever normalized. The retry-collapse mechanics — TTL selection, first-write-wins races, and store durability — are treated in depth in Deduplicating webhook retries with idempotency keys.

Canonical Mapping & Schema Contract #

A funder payload is raw portal vocabulary — every grant portal names its fields differently. Before an event can be trusted downstream it must be projected onto one canonical ingestion record, so validation and rule evaluation never see portal-specific keys. This gate emits raw values keyed to canonical names; organization-specific synonym dictionaries and cross-portal aliasing belong to Field Mapping & Normalization. The alias table below resolves the event envelopes seen across the common grant-portal webhook formats.

The contract is enforced with a pydantic v2 model so coercion and validation happen at one boundary and emit structured errors rather than silently accepting a malformed envelope:

import json
import logging
from datetime import datetime
from enum import Enum
from typing import Any, Dict

from pydantic import BaseModel, ValidationError, field_validator

logger = logging.getLogger("grant_webhook.contract")


class EventType(str, Enum):
    AWARD_STATUS_CHANGED = "award_status_changed"
    SUBMISSION_RECEIVED = "submission_received"
    DEADLINE_UPDATED = "deadline_updated"


class CanonicalEvent(BaseModel):
    event_id: str
    event_type: EventType
    occurred_at: datetime
    sequence: int
    resource_id: str

    @field_validator("event_id", "resource_id")
    @classmethod
    def non_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("identifier fields must be non-empty")
        return v.strip()


_ALIASES = {
    "event_id": ("event_id", "id", "eventId", "delivery_id"),
    "event_type": ("event_type", "type", "event", "topic"),
    "occurred_at": ("occurred_at", "created", "timestamp", "event_time"),
    "sequence": ("sequence", "seq", "version", "revision"),
    "resource_id": ("resource_id", "award_id", "submission_id"),
}


def _first_present(payload: Dict[str, Any], keys: tuple[str, ...]) -> Any:
    for key in keys:
        if key in payload and payload[key] is not None:
            return payload[key]
    return None


def map_to_canonical(raw_body: bytes) -> Dict[str, Any]:
    """Project a funder payload onto the canonical contract or raise a structured error."""
    try:
        payload = json.loads(raw_body)
    except json.JSONDecodeError as exc:
        logger.warning("canonical_map_failed | reason=ERR_MALFORMED_JSON | detail=%s", exc)
        raise ValueError("ERR_MALFORMED_JSON") from exc

    projected = {field: _first_present(payload, keys) for field, keys in _ALIASES.items()}
    try:
        event = CanonicalEvent(**projected)
    except ValidationError as exc:
        failed = [str(err["loc"][0]) for err in exc.errors()]
        logger.warning("canonical_map_failed | reason=ERR_CONTRACT_VIOLATION | failed=%s", failed)
        raise ValueError("ERR_CONTRACT_VIOLATION") from exc

    return event.model_dump(mode="json")

Event ordering under out-of-order retries #

Retries and multi-region delivery mean callbacks routinely arrive out of order: a deadline_updated at sequence=7 can land before the sequence=5 that preceded it. Ordering is resolved per resource_id using the monotonic sequence, never wall-clock arrival time. The rule is last-writer-by-sequence-wins: an event is applied only if its sequence exceeds the highest already applied for that resource; a lower or equal sequence is a stale echo and is acknowledged without effect. This keeps the accepted state convergent regardless of delivery order and requires no locking beyond a compare-and-set on the per-resource high-water mark.

import logging
from typing import Dict

logger = logging.getLogger("grant_webhook.ordering")


def should_apply(resource_id: str, sequence: int, high_water: Dict[str, int]) -> bool:
    """Apply an event only if it advances the per-resource sequence high-water mark."""
    current = high_water.get(resource_id, -1)
    if sequence <= current:
        logger.info(
            "event_superseded | resource_id=%s | seq=%d | high_water=%d",
            resource_id, sequence, current,
        )
        return False
    high_water[resource_id] = sequence
    return True

Validation & Testing #

Determinism is only credible if it is asserted. The following pytest cases pin the four terminal outcomes — accepted, bad signature, stale timestamp, and duplicate — and confirm the audit trail records each decision. A signed-fixture helper reproduces exactly what a funder edge sends, so a dependency bump that changes byte handling fails loudly.

import hashlib
import hmac
import json
import time

import pytest

from grant_webhook.receiver import ReceiveResult, WebhookReceiver

SECRET = b"whsec_test_secret"


class InMemorySeenStore:
    def __init__(self) -> None:
        self._seen: set[str] = set()

    def add_if_absent(self, event_id: str, ttl_seconds: int) -> bool:
        if event_id in self._seen:
            return False
        self._seen.add(event_id)
        return True


def _signed(payload: dict, ts: float | None = None) -> tuple[bytes, dict]:
    body = json.dumps(payload).encode("utf-8")
    signed_ts = str(ts if ts is not None else time.time())
    sig = hmac.new(SECRET, body, hashlib.sha256).hexdigest()
    return body, {"X-Grant-Signature": sig, "X-Grant-Timestamp": signed_ts}


@pytest.fixture
def receiver() -> WebhookReceiver:
    return WebhookReceiver(SECRET, InMemorySeenStore(), replay_window_seconds=300, dedupe_ttl_seconds=86400)


def _event() -> dict:
    return {"id": "evt_001", "type": "award_status_changed", "created": time.time(),
            "sequence": 3, "award_id": "AWD-7781"}


def test_authentic_event_is_accepted(receiver: WebhookReceiver) -> None:
    body, headers = _signed(_event())
    result = receiver.receive(body, headers)
    assert result.outcome == "ACCEPTED"
    assert result.http_status == 202


def test_tampered_body_fails_signature(receiver: WebhookReceiver) -> None:
    body, headers = _signed(_event())
    result = receiver.receive(body + b" ", headers)  # one byte changes the HMAC
    assert result.outcome == "ERR_BAD_SIGNATURE"
    assert result.http_status == 401


def test_stale_timestamp_is_rejected(receiver: WebhookReceiver) -> None:
    body, headers = _signed(_event(), ts=time.time() - 3600)
    result = receiver.receive(body, headers)
    assert result.outcome == "ERR_STALE_TIMESTAMP"


def test_retry_is_deduplicated(receiver: WebhookReceiver) -> None:
    body, headers = _signed(_event())
    first = receiver.receive(body, headers)
    second = receiver.receive(body, headers)
    assert first.outcome == "ACCEPTED"
    assert second.outcome == "DUPLICATE"
    assert second.http_status == 200

Performance & Scale Considerations #

Nonprofit-scale webhook traffic is low-volume but latency-sensitive: funder edges expect a fast acknowledgement and will retry aggressively if the receiver is slow, which amplifies duplicate load exactly when the system is already struggling. Tune for a fast, cheap acknowledgement rather than throughput.

• Acknowledge fast, process later. The receiver should verify, dedupe, and persist the canonical record, then return 202 immediately. Heavy downstream work must not run inside the request; dispatch it onto bounded worker pools governed by Async Batch Processing Pipelines.
• Seen-store sizing. The idempotency set grows with event volume; bound it with WEBHOOK_DEDUPE_TTL_SECONDS (24 h comfortably covers a funder’s retry schedule) so memory stays flat. A Redis set with per-key TTL is the pragmatic default at nonprofit scale.
• Signature cost. HMAC-SHA256 over a few kilobytes is sub-millisecond; it is never the bottleneck. Do not cache verification results — recompute per request so a replayed body with a stale signature cannot slip through a cache.
• Retry-storm ceiling. When a downstream store is briefly unavailable, funders will resend; a bounded seen-store plus a fast 202 keeps a retry storm from becoming a re-processing storm. Sustained failures should shed to a dead-letter path, not block the edge.

Failure Modes & Troubleshooting #

Every rejection resolves to a machine-readable reason code and an audit event — never a silent drop and never a 500 that invites an infinite retry loop. A sustained shift in one reason code is the primary operational signal; a surge of ERR_BAD_SIGNATURE usually means the funder rotated the signing secret, which is fixed by updating WEBHOOK_SIGNING_SECRET, not by loosening verification.

Transient faults in the seen-store or the downstream queue are not this stage’s concern to retry — bounded retries and backoff belong to Pipeline Fallback & Retry Logic, and rule evaluation against the accepted event belongs to the Core Architecture & Compliance Mapping reference.

Compliance Alignment #

This gate satisfies specific integrity, internal-control, and audit-event obligations rather than generic “compliance.” Every accepted event and every rejection is auditable evidence, anchored to an immutable event_id and a signed timestamp, so the intake decision for any callback is reconstructable from the append-only log.

Compliance officers can reconstruct the exact intake decision for any callback by querying the append-only audit logs by event_id, satisfying the traceability requirement of 2 CFR §200.302 and the audit-event requirement of NIST SP 800-53 AU-2. Under the 2 CFR §200.334 three-year minimum, retained envelopes are records and must be preserved (longer where a funder or state statute extends it). Audit artifacts emitted here conform to the conventions defined in Compliance Metadata Standards, and access to the seen-store and signing secret is governed by Data Security & Access Boundaries.

Related #

• Parent: Data Ingestion & Grant Parsing Workflows
• Verifying HMAC signatures on grant portal webhooks — the constant-time signature-verification detail this gate defers to
• Deduplicating webhook retries with idempotency keys — the retry-collapse and TTL mechanics behind the dedupe stage
• API Polling & Rate Limiting — the scheduled pull counterpart to this push-based gate
• Async Batch Processing Pipelines — bounded concurrent dispatch of accepted events
• Data Security & Access Boundaries — controls the signing secret and seen-store access this gate depends on