Webhooks are HTTP endpoints that receive data from external services. Anyone who discovers your webhook URL can send requests to it. That’s a problem. Here’s how to secure them properly.
The Threat Model#
Your webhook endpoint faces several threats:
- Spoofing: Attacker sends fake payloads pretending to be Stripe/GitHub/etc.
- Replay attacks: Attacker captures a legitimate request and resends it
- Tampering: Attacker intercepts and modifies payloads in transit
- Enumeration: Attacker discovers your webhook URLs through guessing
- Denial of service: Attacker floods your endpoint with requests
Signature Verification#
Most webhook providers sign their payloads. Always verify:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
| import hmac
import hashlib
def verify_stripe_signature(payload: bytes, signature: str, secret: str) -> bool:
"""Verify Stripe webhook signature."""
# Stripe format: t=timestamp,v1=signature
parts = dict(p.split("=") for p in signature.split(","))
timestamp = parts.get("t")
expected_sig = parts.get("v1")
if not timestamp or not expected_sig:
return False
# Construct signed payload
signed_payload = f"{timestamp}.{payload.decode()}"
# Compute HMAC
computed = hmac.new(
secret.encode(),
signed_payload.encode(),
hashlib.sha256
).hexdigest()
return hmac.compare_digest(computed, expected_sig)
@app.post("/webhooks/stripe")
async def stripe_webhook(request: Request):
payload = await request.body()
signature = request.headers.get("Stripe-Signature", "")
if not verify_stripe_signature(payload, signature, STRIPE_WEBHOOK_SECRET):
raise HTTPException(401, "Invalid signature")
# Process verified payload
event = json.loads(payload)
handle_stripe_event(event)
return {"status": "ok"}
|
Critical: Use hmac.compare_digest() for timing-safe comparison. Regular == leaks information through timing differences.
GitHub Webhook Verification#
1
2
3
4
5
6
7
8
9
10
11
12
13
14
| def verify_github_signature(payload: bytes, signature: str, secret: str) -> bool:
"""Verify GitHub webhook signature (sha256)."""
if not signature.startswith("sha256="):
return False
expected = signature[7:] # Remove "sha256=" prefix
computed = hmac.new(
secret.encode(),
payload,
hashlib.sha256
).hexdigest()
return hmac.compare_digest(computed, expected)
|
Timestamp Validation#
Prevent replay attacks by checking timestamps:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| import time
MAX_AGE_SECONDS = 300 # 5 minutes
def verify_with_timestamp(payload: bytes, signature: str, timestamp: str, secret: str) -> bool:
# Check timestamp freshness
try:
ts = int(timestamp)
except ValueError:
return False
now = int(time.time())
if abs(now - ts) > MAX_AGE_SECONDS:
return False # Too old or too far in future
# Then verify signature
return verify_signature(payload, signature, secret)
|
Why 5 minutes? Clock skew happens. Too tight and legitimate requests fail. Too loose and replay attacks work.
Idempotency for Webhooks#
Webhooks can be delivered multiple times. Handle duplicates:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| def handle_webhook(event_id: str, payload: dict):
# Check if we've seen this event
if redis.sismember("processed_webhooks", event_id):
logger.info(f"Duplicate webhook {event_id}, skipping")
return {"status": "already_processed"}
# Process the event
try:
process_event(payload)
# Mark as processed (expire after 24 hours)
redis.sadd("processed_webhooks", event_id)
redis.expire("processed_webhooks", 86400)
return {"status": "ok"}
except Exception as e:
logger.error(f"Failed to process {event_id}: {e}")
raise
|
Most providers include an event ID (event.id for Stripe, X-GitHub-Delivery header for GitHub). Use it.
URL Obscurity#
Don’t use predictable webhook URLs:
1
2
3
4
5
6
7
8
| # BAD: Predictable
/webhooks/stripe
/webhooks/github
/api/webhook
# BETTER: Random path component
/webhooks/stripe/a8f3b2c1d4e5
/hooks/7f8a9b0c1d2e3f4a5b6c
|
Generate random suffixes and store them securely:
1
2
3
4
5
| import secrets
def generate_webhook_url(provider: str) -> str:
token = secrets.token_urlsafe(16)
return f"/webhooks/{provider}/{token}"
|
Note: This is defense in depth. Signatures are still required—obscurity alone isn’t security.
IP Allowlisting#
Some providers publish their IP ranges. Verify them:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
| import ipaddress
# GitHub's webhook IPs (check their docs for current list)
GITHUB_IPS = [
ipaddress.ip_network("192.30.252.0/22"),
ipaddress.ip_network("185.199.108.0/22"),
# ... add all ranges
]
def verify_github_ip(request_ip: str) -> bool:
ip = ipaddress.ip_address(request_ip)
return any(ip in network for network in GITHUB_IPS)
@app.post("/webhooks/github/{token}")
async def github_webhook(request: Request, token: str):
client_ip = request.client.host
if not verify_github_ip(client_ip):
logger.warning(f"Webhook from non-GitHub IP: {client_ip}")
raise HTTPException(403, "Forbidden")
# Continue with signature verification...
|
Warning: IP ranges change. Fetch them dynamically or update regularly.
Rate Limiting#
Protect against floods:
1
2
3
4
5
6
7
8
9
10
| from slowapi import Limiter
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address)
@app.post("/webhooks/stripe/{token}")
@limiter.limit("100/minute") # Adjust based on expected volume
async def stripe_webhook(request: Request, token: str):
# Process webhook
pass
|
Set limits based on expected legitimate traffic plus headroom.
Async Processing#
Don’t process webhooks synchronously:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| @app.post("/webhooks/stripe/{token}")
async def stripe_webhook(request: Request, token: str):
payload = await request.body()
# Verify signature first
if not verify_signature(payload, request.headers):
raise HTTPException(401, "Invalid signature")
event = json.loads(payload)
# Queue for async processing
await queue.enqueue("process_stripe_event", event)
# Return immediately
return {"status": "queued"}
|
Why?
- Providers expect fast responses (often <30 seconds)
- Long processing blocks their retry queue
- Failures during processing shouldn’t return errors to provider
Logging and Alerting#
Log everything, but carefully:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
| @app.post("/webhooks/{provider}/{token}")
async def webhook(request: Request, provider: str, token: str):
# Log metadata, not full payload (might contain PII)
logger.info(
"Webhook received",
extra={
"provider": provider,
"event_type": request.headers.get("X-Event-Type"),
"delivery_id": request.headers.get("X-Delivery-ID"),
"client_ip": request.client.host,
}
)
try:
result = await process_webhook(request)
logger.info("Webhook processed", extra={"status": "success"})
return result
except SignatureError:
logger.warning("Invalid webhook signature", extra={"client_ip": request.client.host})
raise HTTPException(401)
|
Alert on:
- Signature verification failures (possible attack)
- Unusual volume (possible DoS or integration issues)
- Processing failures (bugs in your code)
Security Checklist#
Before deploying a webhook endpoint:
Webhooks are inbound API calls from the internet. Treat them with the same suspicion you’d treat any public endpoint—because that’s exactly what they are.