import hashlib
import hmac
import time
class SignatureError(Exception):
pass
def parse_signature_header(header):
parts = {}
for item in header.split(","):
key, _, value = item.partition("=")
parts.setdefault(key.strip(), value.strip())
if "t" not in parts or "v1" not in parts:
raise SignatureError("malformed signature header")
return parts["t"], parts["v1"]
def verify_signature(payload, header, secret, tolerance=300):
timestamp, provided = parse_signature_header(header)
try:
age = time.time() - int(timestamp)
except ValueError:
raise SignatureError("invalid timestamp")
if age > tolerance:
raise SignatureError("timestamp outside tolerance window")
signed = b"%s.%s" % (timestamp.encode("utf-8"), payload)
expected = hmac.new(secret.encode("utf-8"), signed, hashlib.sha256).hexdigest()
if not hmac.compare_digest(expected, provided):
raise SignatureError("signature mismatch")
return True
from flask import Blueprint, current_app, jsonify, request
import json
from signing import verify_signature, SignatureError
bp = Blueprint("webhook", __name__)
def already_processed(event_id):
# Replace with a real store (Redis SETNX, unique DB row, etc.)
return False
@bp.route("/webhooks/payments", methods=["POST"])
def payments_webhook():
raw = request.get_data()
header = request.headers.get("Stripe-Signature", "")
secret = current_app.config["WEBHOOK_SECRET"]
try:
verify_signature(raw, header, secret)
except SignatureError as exc:
current_app.logger.warning("rejected webhook: %s", exc)
return jsonify(error=str(exc)), 400
event = json.loads(raw)
if already_processed(event["id"]):
return jsonify(status="duplicate"), 200
if event["type"] == "payment_intent.succeeded":
handle_payment_succeeded(event["data"]["object"])
elif event["type"] == "charge.refunded":
handle_refund(event["data"]["object"])
else:
current_app.logger.info("ignoring event type %s", event["type"])
return jsonify(status="ok"), 200
def handle_payment_succeeded(intent):
current_app.logger.info("payment succeeded for %s", intent["id"])
def handle_refund(charge):
current_app.logger.info("refund processed for %s", charge["id"])
import os
from flask import Flask
from webhook import bp as webhook_bp
def create_app():
app = Flask(__name__)
app.config["WEBHOOK_SECRET"] = os.environ["WEBHOOK_SECRET"]
app.config["MAX_CONTENT_LENGTH"] = 1 * 1024 * 1024
app.register_blueprint(webhook_bp)
return app
if __name__ == "__main__":
create_app().run(port=8000)
A webhook endpoint is a public HTTP handler that acts on events sent by a third party, so it must never trust the request body until the sender is proven authentic. This snippet shows the Stripe-style approach: the sender signs the raw payload with a shared secret and includes a timestamp, and the receiver recomputes the HMAC to confirm both authenticity and freshness before doing any work.
In signing.py, parse_signature_header splits the Stripe-Signature header into its t (timestamp) and v1 (signature) components. verify_signature reconstructs the signed payload as "{timestamp}.{body}", computes an HMAC-SHA256 over it with the endpoint secret, and compares it against the header value using hmac.compare_digest. That constant-time comparison matters: a naive == can leak how many leading bytes matched through timing, which is enough to forge a signature over many attempts. The function also enforces a tolerance window so a captured request cannot be replayed hours later — an old timestamp is rejected even if the signature itself is valid.
A critical detail is that verification runs against the exact bytes received. In webhook.py the handler reads request.get_data() rather than request.json, because re-serializing parsed JSON would reorder keys or change whitespace and break the HMAC. Parsing happens only after the signature checks out.
The webhook.py blueprint wires this together: it pulls the secret from config, calls verify_signature, and returns 400 on a bad or stale signature so the sender knows to retry or stop. Only then does it parse the event and dispatch on event['type']. It also guards against duplicate delivery via already_processed(event['id']), since webhook providers guarantee at-least-once delivery and will resend on timeouts. Making handlers idempotent by event id is essential to avoid double-charging or duplicate fulfillment.
app.py shows the minimal factory that registers the blueprint and loads the secret from the environment. Together the files separate pure crypto (signing.py) from transport concerns (webhook.py), which keeps the verification logic unit-testable without spinning up Flask. The pattern generalizes to any signed-webhook provider that uses a timestamped HMAC scheme.
Related snips
payload = {
sub: user.id,
iss: 'https://auth.example.com',
aud: 'codesnips-api',
exp: 15.minutes.from_now.to_i,
iat: Time.now.to_i,JWT issuance and verification without common footguns
timestamp = request.headers.fetch('X-Signature-Timestamp')
signature = request.headers.fetch('X-Signature')
payload = request.raw_post
data = "#{timestamp}.#{payload}"
expected = OpenSSL::HMAC.hexdigest('SHA256', ENV.fetch('WEBHOOK_SECRET'), data)HMAC signed API requests for webhook and partner integrity
#!/usr/bin/env bash
set -euo pipefail
export VAULT_ADDR="https://vault.internal:8200"
export VAULT_TOKEN="${VAULT_TOKEN:?missing VAULT_TOKEN}"Secrets management with environment isolation and Vault
import crypto from 'node:crypto';
import jwt from 'jsonwebtoken';
export function signAccessToken(payload: object) {
return jwt.sign(payload, process.env.JWT_SECRET!, { expiresIn: '15m' });
}JWT access + refresh token rotation (conceptual)
package files
import (
"context"
"time"Presigned S3 upload URLs (AWS SDK v2)
module EmailNormalization
extend ActiveSupport::Concern
included do
attr_accessor :soft_warningsSoft Validation: Normalize + Validate Email
Share this code
Here's the card — post it anywhere.
