Webhook Signature Verification: Securing Inbound Events for AI Agent Systems

Why Webhook Security Is Non-Negotiable

AI agent systems often receive events from external services — a payment processed via Stripe, a commit pushed to GitHub, a ticket created in Jira. These events arrive as HTTP POST requests to your webhook endpoint. Without verification, an attacker can send fabricated events to trigger agent actions: fake payment confirmations, spoofed deployment triggers, or forged customer messages.

Webhook signature verification ensures that every inbound event genuinely originated from the expected sender and has not been modified in transit. This is a foundational security requirement for any AI agent that takes actions based on external events.

How HMAC Signatures Work

The sender and receiver share a secret key. When the sender dispatches a webhook, it computes an HMAC (Hash-based Message Authentication Code) over the request body using the shared secret and includes the resulting signature in a header. The receiver recomputes the HMAC using the same secret and compares the signatures. If they match, the payload is authentic and unmodified.

The standard algorithm is HMAC-SHA256, which provides both authentication (the sender knows the secret) and integrity (the payload has not been altered).

Building the Verification Module

Here is a reusable webhook signature verification module:

# webhooks/verification.py
import hmac
import hashlib
import time
from fastapi import HTTPException, Request

MAX_TIMESTAMP_AGE_SECONDS = 300  # 5 minutes


def compute_signature(payload: bytes, secret: str, timestamp: str) -> str:
    """Compute HMAC-SHA256 signature over timestamp + payload."""
    message = f"{timestamp}.".encode() + payload
    return hmac.new(
        secret.encode(),
        message,
        hashlib.sha256,
    ).hexdigest()


def verify_signature(
    payload: bytes,
    secret: str,
    received_signature: str,
    timestamp: str,
) -> bool:
    """Verify webhook signature with timing-safe comparison."""
    expected = compute_signature(payload, secret, timestamp)
    return hmac.compare_digest(expected, received_signature)

Two critical details in this code. First, the timestamp is included in the signed message, binding the signature to a specific moment in time. Second, hmac.compare_digest performs a constant-time comparison that prevents timing attacks — an attacker cannot deduce the correct signature by measuring response times.

Timestamp Validation to Prevent Replay Attacks

Even with valid signatures, an attacker who intercepts a webhook can replay it later. Timestamp validation prevents this by rejecting events that are too old:

def validate_timestamp(timestamp: str) -> None:
    """Reject webhooks with timestamps older than the threshold."""
    try:
        event_time = int(timestamp)
    except (ValueError, TypeError):
        raise HTTPException(status_code=400, detail="Invalid timestamp format")

    current_time = int(time.time())
    age = abs(current_time - event_time)

    if age > MAX_TIMESTAMP_AGE_SECONDS:
        raise HTTPException(
            status_code=403,
            detail=f"Webhook timestamp too old: {age}s exceeds {MAX_TIMESTAMP_AGE_SECONDS}s limit",
        )

FastAPI Dependency for Webhook Verification

Wrap the verification logic into a reusable FastAPI dependency:

from fastapi import Depends, Header
from typing import Annotated


class WebhookVerifier:
    def __init__(self, secret_env_var: str):
        import os
        self.secret = os.environ[secret_env_var]

    async def __call__(
        self,
        request: Request,
        x_webhook_signature: Annotated[str, Header()],
        x_webhook_timestamp: Annotated[str, Header()],
    ) -> bytes:
        # Read the raw body
        body = await request.body()

        # Validate timestamp
        validate_timestamp(x_webhook_timestamp)

        # Verify signature
        if not verify_signature(body, self.secret, x_webhook_signature, x_webhook_timestamp):
            raise HTTPException(
                status_code=403,
                detail="Invalid webhook signature",
            )

        return body


# Create verifiers for each provider
verify_stripe = WebhookVerifier("STRIPE_WEBHOOK_SECRET")
verify_github = WebhookVerifier("GITHUB_WEBHOOK_SECRET")

Using the Verifier in Agent Webhook Endpoints

Apply the dependency to any webhook handler:

import json
from fastapi import APIRouter, Depends

router = APIRouter(prefix="/webhooks")


@router.post("/stripe")
async def handle_stripe_webhook(
    body: bytes = Depends(verify_stripe),
):
    event = json.loads(body)
    event_type = event.get("type")

    if event_type == "invoice.paid":
        await agent_billing.process_payment(event["data"]["object"])
    elif event_type == "customer.subscription.deleted":
        await agent_provisioning.deactivate_tenant(event["data"]["object"])

    return {"status": "processed"}


@router.post("/github")
async def handle_github_webhook(
    body: bytes = Depends(verify_github),
):
    event = json.loads(body)
    action = event.get("action")

    if action == "opened" and "pull_request" in event:
        await code_review_agent.review_pr(event["pull_request"])

    return {"status": "processed"}

Idempotency for Webhook Processing

Webhook providers retry on failure, which means your endpoint may receive the same event multiple times. Use an idempotency key to ensure each event is processed exactly once:

async def process_webhook_idempotently(
    event_id: str, processor, event_data: dict,
):
    # Check if already processed
    cache_key = f"webhook_processed:{event_id}"
    already_processed = await redis_client.get(cache_key)
    if already_processed:
        return {"status": "already_processed"}

    # Process the event
    result = await processor(event_data)

    # Mark as processed with a TTL (e.g., 72 hours)
    await redis_client.setex(cache_key, 72 * 3600, "1")
    return result

Sending Signed Webhooks from Your Platform

When your AI agent platform sends webhooks to customers, sign them the same way:

import httpx


async def send_webhook(url: str, payload: dict, secret: str):
    body = json.dumps(payload).encode()
    timestamp = str(int(time.time()))
    signature = compute_signature(body, secret, timestamp)

    async with httpx.AsyncClient() as client:
        response = await client.post(
            url,
            content=body,
            headers={
                "Content-Type": "application/json",
                "X-Webhook-Signature": signature,
                "X-Webhook-Timestamp": timestamp,
            },
            timeout=10.0,
        )
    return response.status_code

FAQ

Why include the timestamp in the signature instead of just signing the body?

Signing the body alone means the signature is valid forever. An attacker who intercepts a legitimate webhook can replay it at any time — days, weeks, or months later. By including the timestamp in the signed message, the signature is bound to a specific time window. Even if intercepted, the event can only be replayed within the tolerance window (typically five minutes).

How do I handle webhook signature verification for providers like Stripe that use their own format?

Major providers use slightly different signing schemes. Stripe uses whsec_ prefixed secrets and a specific header format. GitHub uses X-Hub-Signature-256. Write provider-specific verifier classes that inherit from a base verifier but override the header names and signature computation. Most providers document their signing algorithm, so adaptation is straightforward.

What should I do if webhook verification fails?

Return an appropriate HTTP error (401 or 403) with a generic message — never reveal which part of the verification failed. Log the failure with the source IP, headers, and timestamp for security monitoring. If you see repeated verification failures from the same source, consider rate limiting or blocking that IP. Alert your security team if failure rates spike, as it may indicate an attack.

---

#Webhooks #HMAC #Security #FastAPI #AIAgents #EventDriven #AgenticAI #LearnAI #AIEngineering