← All Articles

Webhook Security: Validating Inbound Requests to Your AI Pipeline

Signature checks, atomic idempotency, and fast acknowledgment: what actually holds up when a webhook feeds an LLM, verified against the provider docs.

An attacker doesn't need to intercept anything to abuse your webhook endpoint. They need the URL. It's in your Stripe dashboard, in your CloudWatch logs, in the Slack message where someone pasted it during setup. With the URL, they POST a hand-written charge.succeeded event and your pipeline does what it was built to do: generates an invoice, emails a customer, and hands the attacker's text to an LLM that decides what happens next.

That last step is why webhook security matters more when an AI pipeline is downstream. A forged webhook against a traditional integration creates a bad database row. A forged webhook against an AI pipeline puts attacker-controlled text into a prompt that has tools attached, and every replay burns tokens on your bill.

The good news: the defenses are standard and the providers document them well. The bad news: most hand-rolled implementations get two or three of them subtly wrong. This article covers what actually holds up, verified against the Stripe, GitHub, Shopify, and AWS docs as of September 2026.

Use the provider's SDK for signature verification

Every major webhook sender signs requests with HMAC-SHA256 and ships a library function that verifies the signature correctly. Use it.

For Stripe, that's stripe.Webhook.construct_event. It parses the Stripe-Signature header, handles multiple signatures (Stripe sends more than one while you're rotating a secret), enforces a timestamp tolerance, and uses a constant-time comparison. Writing that yourself is where bugs creep in. The most common one: grabbing the first v1= value with a string split and silently breaking the day you roll your signing secret.

import base64 import os import stripe # Fetch from Secrets Manager once per cold start, not once per request. SIGNING_SECRET = load_secret(os.environ["STRIPE_WEBHOOK_SECRET_ARN"]) def verify(event: dict) -> stripe.Event: """Verify a Stripe webhook delivered through API Gateway -> Lambda.""" raw = event["body"].encode() if event.get("isBase64Encoded"): raw = base64.b64decode(event["body"]) # HTTP APIs lowercase header names; REST APIs preserve case. Handle both. headers = {k.lower(): v for k, v in event.get("headers", {}).items()} sig_header = headers.get("stripe-signature", "") return stripe.Webhook.construct_event( payload=raw, sig_header=sig_header, secret=SIGNING_SECRET, tolerance=300, )

construct_event raises ValueError on a malformed body and stripe.SignatureVerificationError on a bad signature. Return a 400 for either and move on.

GitHub's equivalent is documented under validating webhook deliveries (header X-Hub-Signature-256). Shopify's is under verify webhook deliveries (header X-Shopify-Hmac-Sha256, base64-encoded). Providers that follow the Standard Webhooks spec, including OpenAI's newer event webhooks, use webhook-id, webhook-timestamp, and webhook-signature headers and ship their own verifier.

If you must verify by hand (say, a provider with no Python library), follow the provider's manual steps exactly. For Stripe that means: split the header on commas, collect every v1= value, compute HMAC-SHA256(secret, f"{timestamp}.{raw_body}"), and accept if any received signature matches using hmac.compare_digest. The manual verification steps are on the same docs page as the library approach.

Verify the raw bytes, never the parsed body

This is the single most common cause of signature failures in production, and it has nothing to do with cryptography.

The signature covers the exact bytes the provider sent. If anything between the provider and your verification code re-serializes the body, the signature will never match. Common culprits:

Stripe's docs put it plainly: any manipulation of the raw body causes verification to fail. Verify first, on the bytes. Parse second.

Timestamps are provider-specific, so know what yours sends

It's tempting to describe "timestamp validation" as a universal second layer. It isn't.

Stripe includes a timestamp inside the signed payload (t= in the Stripe-Signature header). Their libraries reject anything older than five minutes by default, and they warn against setting the tolerance to zero, which disables the check entirely. Standard Webhooks providers do the same thing through webhook-timestamp.

GitHub and Shopify don't include a timestamp in their signatures. There's nothing to validate. For those providers, replay protection comes entirely from the delivery ID and your idempotency store (next section).

Be honest about what a timestamp check buys you even where it exists. It bounds the replay window. It does not prevent a replay inside that window, and the realistic replay vector is a raw request body that landed in a log somewhere, not intercepted TLS traffic. Two consequences follow: don't log raw webhook bodies, and treat idempotency as the layer that actually stops duplicates.

Idempotency: claim the key atomically

Providers retry. Stripe retries a failed delivery for up to three days in live mode. Shopify retries 19 times over 48 hours. GitHub doesn't retry automatically, but lets you redeliver anything from the past three days by hand. And Stripe's docs say outright that endpoints "might occasionally receive the same event more than once" even without a failure.

So you need to process each event exactly once, and the obvious implementation is wrong:

# DON'T DO THIS if table.get_item(Key={"event_id": event_id}).get("Item"): return cached_result do_the_work() table.put_item(Item={"event_id": event_id})

Two concurrent Lambda invocations for the same event both pass the get_item check and both do the work. The fix is to claim the key with a conditional write before doing anything:

import time import boto3 from botocore.exceptions import ClientError table = boto3.resource("dynamodb").Table("webhook-events") # Longer than any provider's retry window. Stripe: 3 days. Shopify: 48 hours. RETENTION_SECONDS = 7 * 86400 def claim_event(event_id: str) -> bool: """Return True if this is the first delivery of event_id we've accepted.""" try: table.put_item( Item={ "event_id": event_id, "status": "queued", "claimed_at": int(time.time()), "ttl": int(time.time()) + RETENTION_SECONDS, }, ConditionExpression="attribute_not_exists(event_id)", ) return True except ClientError as e: if e.response["Error"]["Code"] == "ConditionalCheckFailedException": return False raise def release_event(event_id: str) -> None: """Undo a claim when we failed to hand the event off. Lets the provider's retry succeed.""" table.delete_item(Key={"event_id": event_id})

DynamoDB evaluates the condition atomically, so exactly one caller wins. The loser returns 200 (the event is already handled or in flight) and does nothing.

Three details that matter here:

Which ID to use. Stripe: event.id. Stripe also notes that two distinct Event objects can be generated for the same underlying change, so for a belt-and-suspenders check, key on data.object.id plus event.type as well. GitHub: the X-GitHub-Delivery header. Shopify: X-Shopify-Webhook-Id identifies the delivery and X-Shopify-Event-Id identifies the underlying event. Don't use Stripe's created timestamp for dedup. Their docs say not to, because distinct events can share a second.

TTL must exceed the retry window. A 24-hour TTL means a Stripe retry on day two gets processed as a brand-new event. Seven days covers every provider listed above with margin.

Handle failure after the claim. If you claim the key and then crash before the work is durably handed off, that event is lost: the provider's retry hits the claim and gets dropped. Either release the claim on failure (as above) or track a status field and alert on anything stuck in queued for longer than your processing SLO.

Return 2xx fast, do the work in a queue

Every provider wants a quick acknowledgment. GitHub terminates the connection after 10 seconds. Shopify allows one second to connect and five for the whole request. Stripe doesn't publish a number, but lists "Timed out" as a delivery failure and says to return a 2xx before any complex logic. All three recommend an asynchronous queue.

For an AI pipeline this is non-negotiable. An LLM call can take longer than any of those limits, and a spike (every subscription renewing on the first of the month) will overwhelm a synchronous handler.

The ingress handler should do exactly four things:

import json import os import boto3 import stripe sqs = boto3.client("sqs") QUEUE_URL = os.environ["QUEUE_URL"] def handler(event, context): # 1. Verify the signature on the raw bytes. try: evt = verify(event) except (ValueError, stripe.SignatureVerificationError): emit_metric("SignatureFailure", 1) return {"statusCode": 400} # 2. Claim the event ID. Duplicate -> 200, nothing else. if not claim_event(evt["id"]): emit_metric("DuplicateDelivery", 1) return {"statusCode": 200} # 3. Hand off the IDs only. The consumer fetches the real object. try: sqs.send_message( QueueUrl=QUEUE_URL, MessageBody=json.dumps({ "event_id": evt["id"], "type": evt["type"], "object_id": evt["data"]["object"]["id"], }), ) except Exception: release_event(evt["id"]) raise # 5xx -> provider retries, and the claim is free again # 4. Acknowledge. return {"statusCode": 200}

Put a dead-letter queue on the SQS queue and alarm on its depth. Processing failures now land somewhere you can inspect and replay, instead of turning into three days of provider retries against a bug that fails the same way every time.

Notice what the message contains: IDs, not the payload. That's deliberate, and it's the next point.

Treat the webhook as a notification, not the truth

Stripe's own guidance is to use the event as a signal and then retrieve the current object from the API. Their thin-event SDK pattern makes this explicit with fetch_related_object(). The reasons hold for every provider:

For an AI pipeline, this is also your defense against prompt injection through the webhook. A customer's name, a memo field, a commit message: whatever free text flows from the webhook into your prompt is attacker-influenced. Fetch it from the provider, pass it to the model as clearly delimited data rather than instructions, and never let it decide which tools the model can call.

Network-layer restrictions, and what AWS actually supports

Stripe recommends IP allowlisting alongside signature verification, and publishes its webhook IP addresses. It's worth doing. It cuts noise and stops probing before it reaches your code. It is not a substitute for signature verification.

The catch on AWS: API Gateway HTTP APIs (the aws_apigatewayv2_* resources) do not support AWS WAF. The REST vs HTTP API comparison lists WAF as REST-only. If you want WAF in front of a webhook endpoint you need a REST API, or CloudFront in front of whatever you have.

Here's the REST API version:

resource "aws_wafv2_ip_set" "stripe" { name = "stripe-webhook-ips" scope = "REGIONAL" ip_address_version = "IPV4" addresses = var.stripe_webhook_cidrs # refresh from docs.stripe.com/ips } resource "aws_wafv2_web_acl" "webhooks" { name = "webhook-ip-restrict" scope = "REGIONAL" default_action { block {} } rule { name = "allow-stripe" priority = 1 action { allow {} } statement { ip_set_reference_statement { arn = aws_wafv2_ip_set.stripe.arn } } visibility_config { cloudwatch_metrics_enabled = true metric_name = "allow-stripe" sampled_requests_enabled = true } } visibility_config { cloudwatch_metrics_enabled = true metric_name = "webhook-ip-restrict" sampled_requests_enabled = true } } resource "aws_wafv2_web_acl_association" "webhooks" { resource_arn = aws_api_gateway_stage.webhooks.arn web_acl_arn = aws_wafv2_web_acl.webhooks.arn }

Two operational notes. Stripe's IP list changes, so var.stripe_webhook_cidrs needs a scheduled job that pulls the published list and opens a PR, or you'll block legitimate traffic one day without warning. And a default-block ACL is only appropriate on an API that serves nothing but this webhook.

The strongest option on AWS skips the public endpoint entirely. Stripe can deliver events directly to Amazon EventBridge as an event destination. No URL to leak, no signature to verify, no WAF to maintain. If you're building new and your provider supports it, start there.

What to monitor

Signature failures, as a count. A spike means someone is probing, or you rotated a secret and missed a deployment. Both are worth a page.

Duplicate deliveries, as a count. Establish your own baseline in the first week rather than trusting a number from a blog post. A sustained rise usually means your handler is timing out and the provider is retrying.

Queue age and dead-letter depth. These are your real processing-health signals once the ingress handler is doing nothing but verify, claim, enqueue.

Stuck claims. Anything in the idempotency table with status = queued older than your processing SLO means an event was accepted and then lost.

from datetime import datetime, timezone import boto3 cloudwatch = boto3.client("cloudwatch") def emit_metric(name: str, value: float, unit: str = "Count") -> None: cloudwatch.put_metric_data( Namespace="Webhooks", MetricData=[{ "MetricName": name, "Value": value, "Unit": unit, "Timestamp": datetime.now(timezone.utc), }], )

Also watch the provider's side. Stripe's Workbench shows delivery status and pending retries per endpoint. GitHub shows recent deliveries with response codes under the webhook's settings. If your metrics say everything is fine and the provider's dashboard says otherwise, the provider is right.

Mistakes that survive code review

Verifying a parsed-and-re-serialized body instead of the raw bytes. This one produces a signature failure on every request, which at least fails loudly.

Comparing signatures with ==. Use hmac.compare_digest, or use the SDK, which already does.

Reading the signing secret from an environment variable that was populated from a .tfvars file that was committed. Secrets Manager, fetched once per cold start.

Parsing only the first v1= signature. Works until the day you rotate the secret.

Setting the idempotency TTL shorter than the provider's retry window.

Checking for a duplicate with a read and then writing. Claim with a conditional write.

Doing the work before returning 2xx. Times out under load, and the retries make the load worse.

Feeding webhook payload text into an LLM prompt as if it were trusted. Fetch the object from the provider and delimit the data.

None of this is exotic. It's the same discipline you apply to any untrusted input: authenticate it, deduplicate it, acknowledge it, then go get the real data before you act. The AI pipeline downstream just raises the cost of skipping a step.

Want a second set of eyes on your webhook handlers? Book a call and we'll walk through what you have against this list.

Get the free AI Readiness Checklist

15 questions to diagnose your team’s AI readiness, where you’ll see ROI fastest, and what to tackle first.

✓ Takes 5 minutes ✓ Actionable next steps ✓ No sales pitch

No spam. Unsubscribe anytime.

or

Ready to build AI that actually works?

Let’s talk about how SRE discipline transforms AI from a risky experiment into a reliable business system.

Start the Conversation

About the author

Charles Harvey is the founder of Three Moons Network and a site reliability engineer who builds production-grade AI automation for small businesses — monitoring, cost visibility, and documentation included. He writes about his hands-on AI experiments at floggingclaude.com. Connect on LinkedIn or see the code on GitHub.