Stripe Webhooks — Server-Side Payment Confirmation
In this tutorial, you will learn about Stripe Webhooks. We cover key concepts, practical examples, and best practices to help you master this topic.
Stripe webhooks send HTTP requests to your server when events occur, providing reliable server-side confirmation of payments, subscription changes, and other asynchronous operations.
What You'll Learn
By the end of this lesson you will understand how to receive Stripe webhooks, verify signatures, handle key payment events, avoid common Webhook pitfalls, and build idempotent event handlers.
Why It Matters
The success_url is a client-side redirect that can be faked. Webhooks provide cryptographic proof of payment from Stripe's servers -- this is the only reliable way to confirm payments server-side.
Real-World Use
DodaZIP's subscription system relies entirely on webhooks. When Stripe sends checkout.session.completed, DodaZIP activates the subscription. When invoice.payment_failed arrives, DodaZIP notifies the customer to update their payment method.
flowchart LR
S[Stripe] -->|POST /webhook| A[Application]
A -->|Verify Signature| V[Valid?]
V -->|Yes| P[Process Event]
V -->|No| R[Reject 400]
P --> H[Handle Event Type]
H --> SUC[checkout.session.completed]
H --> I[invoice.payment_succeeded]
H --> CAN[customer.subscription.deleted]
H --> F[invoice.payment_failed]
style S fill:#6772e5,color:#fff
Webhook Endpoint Setup
Create a POST endpoint that accepts raw request bodies (not parsed JSON) for signature verification.
# webhook_handler.py
# Stripe webhook handler
import stripe
import json
import os
stripe.api_key = os.environ.get("STRIPE_SECRET_KEY", "sk_test_placeholder")
endpoint_secret = os.environ.get("STRIPE_WEBHOOK_SECRET", "whsec_placeholder")
def handle_webhook(event, context):
payload = event.get("body", "")
sig_header = event.get("headers", {}).get("stripe-signature", "")
try:
webhook_event = stripe.Webhook.construct_event(
payload, sig_header, endpoint_secret
)
except ValueError:
return {"statusCode": 400, "body": json.dumps({"error": "Invalid payload"})}
except stripe.error.SignatureVerificationError:
return {"statusCode": 400, "body": json.dumps({"error": "Invalid signature"})}
event_type = webhook_event["type"]
data = webhook_event["data"]["object"]
handlers = {
"checkout.session.completed": handle_checkout_completed,
"invoice.payment_succeeded": handle_invoice_succeeded,
"invoice.payment_failed": handle_invoice_failed,
"customer.subscription.deleted": handle_subscription_deleted,
}
handler = handlers.get(event_type)
if handler:
handler(data)
return {"statusCode": 200, "body": json.dumps({"received": True})}
def handle_checkout_completed(session):
print(f"Payment completed for customer {session.get('customer')}")
print(f"Subscription: {session.get('subscription')}")
print(f" > Granting access to {session.get('customer_email')}")
def handle_invoice_succeeded(invoice):
print(f"Invoice paid: {invoice.get('id')}")
print(f" Amount: ${invoice.get('amount_paid', 0) / 100:.2f}")
def handle_invoice_failed(invoice):
print(f"Invoice FAILED: {invoice.get('id')}")
print(f" Notifying customer to update payment method")
def handle_subscription_deleted(subscription):
print(f"Subscription canceled: {subscription.get('id')}")
print(f" Revoking access for customer")
stripe.Webhook.construct_event = lambda p, s, e: {"type": "checkout.session.completed", "data": {"object": {"customer": "cus_abc", "subscription": "sub_xyz", "customer_email": "alice@example.com"}}}
os.environ["STRIPE_WEBHOOK_SECRET"] = "whsec_test"
stripe.api_key = "sk_test"
event = {"body": '{"type": "checkout.session.completed"}', "headers": {"stripe-signature": "test_sig"}}
print(json.loads(handle_webhook(event, None)["body"]))
Expected output:
Payment completed for customer cus_abc
Subscription: sub_xyz
> Granting access to alice@example.com
{'received': True}
Signature Verification
Stripe signs webhook payloads with your webhook secret. Always verify the signature before processing.
# signature_verification.py
# Webhook signature verification
import hmac
import hashlib
import json
def verify_webhook_signature(payload, signature, secret):
expected = hmac.new(
secret.encode(),
payload.encode(),
hashlib.sha256
).hexdigest()
if hmac.compare_digest(f"v1={expected}", signature):
print("Signature VERIFIED: Event came from Stripe")
return True
else:
print("Signature INVALID: Event may be fake")
return False
secret = "whsec_test_secret"
payload = '{"type": "checkout.session.completed"}'
valid_sig = "v1=" + hmac.new(secret.encode(), payload.encode(), hashlib.sha256).hexdigest()
invalid_sig = "v1=fake_signature"
verify_webhook_signature(payload, valid_sig, secret)
verify_webhook_signature(payload, invalid_sig, secret)
Expected output:
Signature VERIFIED: Event came from Stripe
Signature INVALID: Event may be fake
Idempotent Processing
Webhook events may be delivered multiple times. Process each event only once.
# idempotent.py
# Idempotent webhook processing
class WebhookProcessor:
def __init__(self):
self.processed_events = set()
def process_event(self, event_id, event_type, data):
if event_id in self.processed_events:
print(f"SKIP: Event {event_id} already processed")
return {"status": "duplicate"}
self.processed_events.add(event_id)
print(f"PROCESSING: {event_type} ({event_id})")
if event_type == "checkout.session.completed":
grant_access(data["customer_email"])
return {"status": "processed"}
def grant_access(email):
print(f" -> Granted access to {email}")
processor = WebhookProcessor()
event = {"id": "evt_001", "type": "checkout.session.completed", "data": {"customer_email": "alice@example.com"}}
processor.process_event("evt_001", "checkout.session.completed", event["data"])
processor.process_event("evt_001", "checkout.session.completed", event["data"]) # Duplicate
Expected output:
PROCESSING: checkout.session.completed (evt_001)
-> Granted access to alice@example.com
SKIP: Event evt_001 already processed
Common Mistakes
Not verifying webhook signatures: Without verification, anyone can POST fake events to your endpoint and grant access without payment.
Using parsed JSON body instead of raw: Signature verification needs the raw body. Express.json() middleware should not be applied to the webhook route.
Returning 200 before processing: Return quickly to acknowledge receipt. Process the event asynchronously to avoid Stripe timeouts.
Not handling all relevant event types: Listen to checkout.session.completed, invoice.payment_succeeded, customer.subscription.deleted, and invoice.payment_failed.
Not idempotent processing: Stripe sends the same event multiple times. Store processed event IDs to prevent duplicate processing.
Practice Questions
What is a Stripe webhook? An HTTP callback from Stripe to your server when events occur, providing reliable server-side notifications.
Why must you verify webhook signatures? To prove the event came from Stripe and not from an attacker. Without verification, anyone can fake payment confirmations.
What is the most important webhook event for payment confirmation? checkout.session.completed confirms a Checkout Session was paid successfully.
How do you handle duplicate webhook deliveries? Store event IDs and check before processing. Return 200 for duplicates without processing.
Challenge: Design a webhook processing system that handles 10 event types, processes events idempotently, and retries failed processing.
FAQ
Mini Project
Create a webhook handler Lambda function that handles checkout.session.completed and invoice.payment_failed.
import json
processed_events = set()
def lambda_handler(event, context):
body = json.loads(event.get("body", "{}"))
sig = event.get("headers", {}).get("stripe-signature", "")
event_id = body.get("id", "unknown")
event_type = body.get("type", "unknown")
if event_id in processed_events:
return {"statusCode": 200, "body": json.dumps({"status": "duplicate"})}
processed_events.add(event_id)
if event_type == "checkout.session.completed":
email = body["data"]["object"].get("customer_email", "unknown")
print(f"Granting access to {email}")
elif event_type == "invoice.payment_failed":
customer = body["data"]["object"].get("customer", "unknown")
print(f"Payment failed for {customer}")
return {"statusCode": 200, "body": json.dumps({"received": True})}
test_event = {"body": json.dumps({"id": "evt_1", "type": "checkout.session.completed", "data": {"object": {"customer_email": "alice@example.com"}}}), "headers": {"stripe-signature": "test"}}
print(lambda_handler(test_event, None)["body"])
What's Next
Next: Customer Management for managing customers.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro