InventoryAllocation

Verifying Webhook HMAC Signatures on Inbound Channel-Manager Events

An unverified webhook endpoint is an open write path into your inventory: anyone who learns the URL can POST a forged availability update and set a room to zero, or replay a captured stop-sell to knock a rate offline. Verifying an HMAC-SHA256 signature on every inbound payload is what turns that endpoint from a trust-anyone door into an authenticated channel, and it is the first thing the receiver must do — before parsing, before enqueuing, before any business logic. This guide implements that verification correctly for a channel-manager webhook, and is the security companion to channel manager webhook integration, where the receiver’s downstream apply path is defined.

Three mistakes make signature checks worthless in practice: comparing against a re-serialized body instead of the exact bytes received, using a non-constant-time string compare that leaks the signature one byte at a time, and omitting a timestamp tolerance so a captured-and-replayed request still validates. This page closes all three.

Prerequisites and environment

The verifier runs at the ASGI boundary, before the request body is deserialized. Pin these so the digest and comparison behave identically across workers:

The signature scheme assumed here is the common one: the sender computes HMAC-SHA256(secret, f"{timestamp}.{raw_body}") and sends the hex digest plus the timestamp in headers (X-CM-Signature, X-CM-Timestamp). Confirm your channel manager’s exact signing string from its docs — whether the timestamp is inside the signed material is the detail that most often breaks a first integration.

The four gates an inbound webhook passes before it is applied An inbound webhook flows left to right through four sequential gates. First, capture the raw body bytes before any JSON parsing. Second, check the timestamp is within a tolerance window to stop replay. Third, recompute HMAC-SHA256 over timestamp dot raw-body and compare it to the header signature using a constant-time comparison. Fourth, only after all three pass, parse the JSON and enqueue the event. Any gate that fails routes to a 401 rejection logged with structured context. Verify before you parse: four ordered gates Raw bytes, then time window, then constant-time HMAC — parse only what is proven authentic. 1 · Capture raw body bytes before JSON parse 2 · Timestamp window reject stale → anti-replay 3 · HMAC compare recompute over ts.body constant-time equality 4 · Parse + enqueue only proven-authentic bytes 401 Unauthorized any gate fails → reject + structured log, no retry requested
Figure 1: Verification is ordered and fail-closed — the JSON is never parsed until the raw bytes have proven authentic and fresh.

Step-by-step implementation

Step 1 — Capture the raw body before anything parses it

The signature covers the exact bytes the sender hashed. If any middleware deserializes the JSON and you re-serialize it to verify, key ordering, whitespace, and Unicode escaping will differ and every signature will fail. Read the raw body first and hold it.

python
from fastapi import FastAPI, Request, Response
import structlog

app = FastAPI()
log = structlog.get_logger()

@app.post("/webhooks/channel-manager")
async def receive(request: Request) -> Response:
    raw_body: bytes = await request.body()     # the exact bytes the sender signed
    signature = request.headers.get("X-CM-Signature", "")
    timestamp = request.headers.get("X-CM-Timestamp", "")

    if not verify_signature(raw_body, signature, timestamp):
        log.warning("webhook.rejected", reason="bad_signature")
        return Response(status_code=401)       # never parse an unverified body
    # ... only now is it safe to json.loads(raw_body) and enqueue
    return Response(status_code=200)

Reading await request.body() before any JSON access is the single most important line: it captures the bytes verbatim, so the digest you compute is over the same material the sender hashed rather than a lossy round-trip through a parser.

Step 2 — Enforce a timestamp tolerance to stop replay

A valid signature proves the payload came from someone holding the secret, but it does not prove the request is recent. An attacker who captures one legitimate signed request can resend it forever. Reject any request whose signed timestamp is outside a tight tolerance — 300 seconds is a common ceiling.

python
import time

REPLAY_TOLERANCE_SECONDS = 300   # 5 minutes; tighten if your clocks are well-synced

def timestamp_is_fresh(timestamp: str) -> bool:
    try:
        sent = int(timestamp)
    except (ValueError, TypeError):
        return False
    # Reject both the distant past (replay) and the far future (clock/forgery).
    return abs(time.time() - sent) <= REPLAY_TOLERANCE_SECONDS

Checking abs(now - sent) rather than only now - sent rejects far-future timestamps too, which matters because a forged timestamp set ahead of your clock would otherwise sail through a one-sided window and reopen the replay hole you just closed.

Step 3 — Recompute the HMAC and compare in constant time

Compute HMAC-SHA256 over the signed string (timestamp.raw_body) with the shared secret, then compare against the header using hmac.compare_digest. A naive == on strings short-circuits on the first differing byte, and the timing difference leaks the correct signature to an attacker who measures it — a constant-time compare closes that side channel.

python
import hmac
import hashlib

def verify_signature(raw_body: bytes, signature: str, timestamp: str) -> bool:
    if not signature or not timestamp_is_fresh(timestamp):
        return False
    secret = load_channel_secret()             # from secrets manager, bytes
    signed_payload = timestamp.encode() + b"." + raw_body
    expected = hmac.new(secret, signed_payload, hashlib.sha256).hexdigest()
    # Constant-time: comparison cost does not depend on where the strings differ.
    return hmac.compare_digest(expected, signature)

Using hmac.compare_digest instead of expected == signature makes the comparison take the same time whether the mismatch is in the first byte or the last, so a remote attacker cannot binary-search the valid digest by timing thousands of forged requests.

Step 4 — Fold verification into a reusable dependency

To guarantee no future route skips the check, express it as a dependency (or middleware) that runs before the handler and rejects with 401 on failure, returning the verified raw body to the handler so it is not read twice.

python
from fastapi import Depends, HTTPException

async def verified_body(request: Request) -> bytes:
    raw_body = await request.body()
    if not verify_signature(
        raw_body,
        request.headers.get("X-CM-Signature", ""),
        request.headers.get("X-CM-Timestamp", ""),
    ):
        raise HTTPException(status_code=401, detail="signature verification failed")
    return raw_body

@app.post("/webhooks/channel-manager/inventory")
async def inventory_hook(body: bytes = Depends(verified_body)) -> Response:
    payload = json_loads_inventory(body)       # safe: bytes are authenticated
    await enqueue(payload)
    return Response(status_code=200)

Expressing verification as a dependency means a new endpoint added next year cannot accidentally omit the check — the type signature forces every handler to receive an already-verified body rather than a raw request it might forget to authenticate.

Gotchas and production notes

Verification snippet

Confirm the two security-critical behaviours — a tampered body is rejected and a stale timestamp is rejected — with assertions rather than a live sender.

python
import hmac, hashlib, time

_SECRET = b"test-secret"

def _sign(body: bytes, ts: str) -> str:
    return hmac.new(_SECRET, ts.encode() + b"." + body, hashlib.sha256).hexdigest()

def test_valid_signature_passes(monkeypatch):
    monkeypatch.setattr("__main__.load_channel_secret", lambda: _SECRET)
    body, ts = b'{"available_units":3}', str(int(time.time()))
    assert verify_signature(body, _sign(body, ts), ts) is True

def test_tampered_body_is_rejected(monkeypatch):
    monkeypatch.setattr("__main__.load_channel_secret", lambda: _SECRET)
    ts = str(int(time.time()))
    sig = _sign(b'{"available_units":3}', ts)
    assert verify_signature(b'{"available_units":0}', sig, ts) is False  # body changed

def test_stale_timestamp_is_rejected(monkeypatch):
    monkeypatch.setattr("__main__.load_channel_secret", lambda: _SECRET)
    old = str(int(time.time()) - 3600)         # one hour old → outside window
    body = b'{"available_units":3}'
    assert verify_signature(body, _sign(body, old), old) is False

The tampered-body case is the one that matters most: it proves that flipping a single field — available_units from 3 to 0, a forged stop-sell — invalidates the signature, so an attacker cannot alter the inventory an authentic-looking request carries. In production, also assert that the 401 path emits exactly one structured webhook.rejected log per forged request so probing attempts are observable.

← Back to Channel Manager Webhook Integration