Verifying Webhook Signatures from Serialization Partners

Any HTTP endpoint that accepts inbound ObjectEvent webhooks from trading partners is a forgery surface: if the URL leaks or is guessed, anyone can POST a fabricated commissioning or decommissioning record and have it land in your EPCIS repository as if a real partner sent it. This is the exact gap this guide closes, as a code-first companion within API Polling & Webhook Integration — the area governing how the ingestion side of Serialization Data Ingestion & EPCIS Event Sync pulls and receives partner data. A webhook payload carrying GTIN (01), serial (21), lot (10), and expiry (17) fields is only as trustworthy as the cryptographic proof that it originated from the partner who holds the shared secret, arrived recently, and has not been replayed. Without HMAC-SHA256 verification, a constant-time comparison, and a timestamp/nonce anti-replay guard, an unauthenticated handler cannot give you non-repudiation for the audit trail — and DSCSA audit readiness depends on being able to prove who sent what, and when.

Prerequisites #

• Python 3.10+ — snippets use X | Y union types and standard-library hmac/hashlib, no third-party crypto dependency.
• FastAPI (or plain ASGI) — fastapi, starlette, and an ASGI server (uvicorn) to receive the raw request body before any JSON parsing occurs.
• redis (async client) or an equivalent key-value store — to persist nonces with a TTL so a replayed request is rejected even if the signature and timestamp both still check out.
• A per-partner shared secret — provisioned during trading-partner onboarding and scoped to the partner’s GLN, never reused across partners, and stored in a secrets manager rather than environment files.
• DSCSA data prerequisites — the webhook body carries an EPCIS ObjectEvent (or AggregationEvent) referencing GTIN (01), serial (21), lot (10), and expiry (17); these fields only enter the schema validation stage once the signature gate has already accepted the request.

Before writing the handler, agree with each partner on the exact string that gets signed — typically f"{timestamp}.{raw_body}" — and on the header names carrying the signature and timestamp (this guide uses X-Signature-256 and X-Signature-Timestamp). A mismatch in what is signed versus what is verified is the single most common cause of legitimate webhooks failing verification.

Step-by-Step Solution #

Step 1 — Compute and compare the HMAC-SHA256 signature in constant time #

Never compare signatures with ==; a naive string comparison short-circuits on the first mismatched byte, leaking timing information an attacker can use to forge a valid signature byte-by-byte. hmac.compare_digest runs in constant time regardless of where the strings diverge.

import hashlib
import hmac


def compute_signature(secret: bytes, signed_payload: bytes) -> str:
    """Return the hex-encoded HMAC-SHA256 digest of the signed payload."""
    return hmac.new(secret, signed_payload, hashlib.sha256).hexdigest()


def verify_signature(secret: bytes, signed_payload: bytes, provided_signature: str) -> bool:
    """Constant-time comparison of a provided signature against the expected one."""
    expected = compute_signature(secret, signed_payload)
    return hmac.compare_digest(expected, provided_signature)

DSCSA/GS1 note: the signed_payload must be the exact raw bytes the partner signed — read the request body as bytes before any JSON decoding or re-serialization, since re-encoding whitespace or key order invalidates the digest and produces false rejections of genuine ObjectEvent records.

Step 2 — Enforce a timestamp tolerance window against replay #

A valid signature alone does not prove freshness — an attacker who captures a signed request can replay it verbatim. Binding the signature to a timestamp, and rejecting requests whose timestamp is too old or in the future, bounds the window in which a captured request remains useful.

import time


REPLAY_TOLERANCE_SECONDS = 300  # 5 minutes; matches partner onboarding agreement


def timestamp_is_fresh(header_timestamp: str, tolerance: int = REPLAY_TOLERANCE_SECONDS) -> bool:
    """Reject requests whose timestamp is stale or implausibly far in the future."""
    try:
        sent_at = int(header_timestamp)
    except ValueError:
        return False
    now = int(time.time())
    return abs(now - sent_at) <= tolerance

DSCSA/GS1 note: the tolerance window should be generous enough to absorb normal clock skew between your server and the partner’s, but never so wide that a stolen signature stays useful for hours — a mismatch here undermines the same non-repudiation guarantee that the immutable, hash-chained audit trail depends on for EPCIS eventTime integrity.

Step 3 — Track nonces so a request cannot be replayed inside the tolerance window #

Timestamp freshness alone still permits replay for the full length of the tolerance window. A nonce store — the signature itself, or a partner-supplied request ID — makes each request usable exactly once, with a TTL matching the tolerance window so the store never grows unbounded.

import redis.asyncio as redis


async def check_and_consume_nonce(
    redis_client: redis.Redis, nonce: str, ttl_seconds: int = REPLAY_TOLERANCE_SECONDS
) -> bool:
    """Return True if this nonce has not been seen before; False if it is a replay."""
    key = f"webhook:nonce:{nonce}"
    # SET ... NX is atomic: only the first caller for a given nonce gets True
    was_set = await redis_client.set(key, "1", nx=True, ex=ttl_seconds)
    return bool(was_set)

DSCSA/GS1 note: using the signature itself as the nonce avoids requiring partners to mint and track their own request IDs, while still guaranteeing each signed request is consumed exactly once before its corresponding ObjectEvent reaches the ingestion queue.

Step 4 — Wire the checks into a FastAPI webhook handler #

The handler must read the raw body first, run all three checks in order, and only then hand the parsed payload to the ingestion pipeline. Any failure short-circuits to a rejection response without ever constructing an EPCIS record from unverified bytes.

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

PARTNER_SECRETS: dict[str, bytes] = {
    "urn:epc:id:sgln:0312345.00000.0": b"replace-with-secret-from-vault",
}


@app.post("/webhooks/epcis-events")
async def receive_epcis_webhook(request: Request):
    partner_gln = request.headers.get("X-Partner-GLN", "")
    signature = request.headers.get("X-Signature-256", "")
    timestamp = request.headers.get("X-Signature-Timestamp", "")
    raw_body = await request.body()

    secret = PARTNER_SECRETS.get(partner_gln)
    if secret is None or not signature or not timestamp:
        await log_verification_result(partner_gln, "rejected", "missing_credentials")
        return JSONResponse({"error": "unauthenticated"}, status_code=401)

    signed_payload = f"{timestamp}.".encode() + raw_body
    if not verify_signature(secret, signed_payload, signature):
        await log_verification_result(partner_gln, "rejected", "bad_signature")
        return JSONResponse({"error": "invalid signature"}, status_code=403)

    if not timestamp_is_fresh(timestamp):
        await log_verification_result(partner_gln, "rejected", "stale_timestamp")
        return JSONResponse({"error": "timestamp outside tolerance"}, status_code=403)

    if not await check_and_consume_nonce(request.app.state.redis, signature):
        await log_verification_result(partner_gln, "rejected", "replayed_signature")
        return JSONResponse({"error": "duplicate request"}, status_code=403)

    await log_verification_result(partner_gln, "accepted", "ok")
    await request.app.state.ingestion_queue.put(raw_body)
    return JSONResponse({"status": "accepted"}, status_code=202)

DSCSA/GS1 note: accepting the request only enqueues the raw bytes for parsing — it does not itself commit an ObjectEvent. That separation keeps the signature gate a pure authentication boundary, leaving GTIN/serial/lot/expiry structural correctness to the schema-validation stage that runs next.

Step 5 — Log every verification decision to an immutable audit trail #

Both accepted and rejected requests must be recorded, because a spike in rejected signatures from one partner GLN is itself a security signal, and because non-repudiation requires proof of what was rejected and why, not just what was ultimately committed.

import json
from datetime import datetime, timezone


async def log_verification_result(partner_gln: str, outcome: str, reason: str) -> None:
    """Append a structured, timestamped verification decision to the audit sink."""
    entry = {
        "partner_gln": partner_gln or "unknown",
        "outcome": outcome,          # "accepted" | "rejected"
        "reason": reason,
        "logged_at": datetime.now(timezone.utc).isoformat(),
    }
    # In production this appends to a hash-chained, append-only audit store.
    print(json.dumps(entry))

DSCSA/GS1 note: retaining every verification decision — not only successful ones — is what turns webhook authentication into evidence, satisfying the same non-repudiation expectation that governs Verification Router Service Architecture logging on the outbound side of partner queries.

Verification #

Confirm the gate behaves correctly with table-driven tests that exercise all four outcomes: a valid request, a tampered body, a stale timestamp, and a replayed signature.

import time

import pytest

SECRET = b"test-secret"


def _sign(body: bytes, ts: str) -> str:
    return compute_signature(SECRET, f"{ts}.".encode() + body)


def test_valid_signature_is_accepted():
    body = b'{"epc": "urn:epc:id:sgtin:0312345.011111.SERIAL001"}'
    ts = str(int(time.time()))
    sig = _sign(body, ts)
    assert verify_signature(SECRET, f"{ts}.".encode() + body, sig)


def test_tampered_body_is_rejected():
    body = b'{"epc": "urn:epc:id:sgtin:0312345.011111.SERIAL001"}'
    ts = str(int(time.time()))
    sig = _sign(body, ts)
    tampered = body.replace(b"SERIAL001", b"SERIAL999")
    assert not verify_signature(SECRET, f"{ts}.".encode() + tampered, sig)


def test_stale_timestamp_is_rejected():
    old_ts = str(int(time.time()) - 3600)
    assert not timestamp_is_fresh(old_ts)


@pytest.mark.asyncio
async def test_replayed_nonce_is_rejected(fake_redis):
    first = await check_and_consume_nonce(fake_redis, "sig-abc123")
    second = await check_and_consume_nonce(fake_redis, "sig-abc123")
    assert first is True
    assert second is False

Run these against a fake or ephemeral Redis instance in CI, then confirm in a staging environment that a real partner’s signed request round-trips end to end: send it once and expect 202, replay the identical request and expect 403 duplicate request. Finally, inspect the audit sink and confirm a log entry exists for both outcomes with the correct partner_gln and reason.

Gotchas & Edge Cases #

• Signing the wrong bytes. If your handler parses JSON and re-serializes it before hashing, differences in key order or whitespace break every signature. Always verify against the untouched raw request body captured before parsing.
• Nonce TTL shorter than the timestamp tolerance. If the nonce key expires before the timestamp tolerance window closes, a captured request can be replayed successfully in the gap. Set the nonce TTL greater than or equal to the tolerance window, never less.
• Clock skew masquerading as an attack. A partner’s server clock running a few minutes fast or slow will trip timestamp_is_fresh even for a legitimate request. Log the skew you observe per partner GLN so you can distinguish a systemic clock problem from an actual stale replay.
• Secret rotation without downtime. Rotating a partner’s secret invalidates every in-flight signature computed with the old one. Accept both the current and previous secret for a bounded overlap window, and reject only once the old secret is fully retired.
• Conflating signature failures with payload failures. A 403 invalid signature and a downstream GTIN check-digit failure are different problems with different remediation paths; don’t let leading-zero GTIN or NDC formatting issues get triaged as if they were an authentication defect — the signature gate and the schema-validation gate must report distinct error codes.

Related #

• Up to the parent topic: API Polling & Webhook Integration
• Handling Rate Limits on FDA Verification APIs — the polling-side counterpart to this inbound webhook path
• Schema Validation & Error Handling — where an accepted, authenticated payload is checked for structural correctness next
• Serialization Data Ingestion & EPCIS Event Sync — the pipeline this verification gate protects