Validate Stripe Webhook Signatures using request raw body

const crypto = require("crypto");

const {
  context: { request },
} = yepcode;

const webhookSecret = process.env.STRIPE_WEBHOOK_SECRET;

// Get the Stripe signature from headers
const stripeSignature = request.headers["stripe-signature"];

if (!stripeSignature) {
  return {
    status: 400,
    body: { error: "Missing Stripe signature header" },
  };
}

// Verify the webhook signature
try {
  // Parse the signature header (format: t=timestamp,v1=signature)
  const signatures = stripeSignature.split(",").reduce((acc, pair) => {
    const [key, value] = pair.split("=");
    acc[key] = value;
    return acc;
  }, {});

  const timestamp = signatures.t;
  const receivedSignature = signatures.v1;

  if (!timestamp || !receivedSignature) {
    console.error("Missing timestamp or signature in Stripe header");
    return {
      status: 401,
      body: { error: "Invalid signature format" },
    };
  }

  // Check timestamp to prevent replay attacks (allow up to 5 minutes)
  const currentTime = Math.floor(Date.now() / 1000);
  const webhookTime = parseInt(timestamp);
  const tolerance = 300; // 5 minutes in seconds

  if (Math.abs(currentTime - webhookTime) > tolerance) {
    console.error("Webhook timestamp too old, possible replay attack");
    return {
      status: 401,
      body: { error: "Request timestamp too old" },
    };
  }

  // Create the signed payload (timestamp + raw body)
  const signedPayload = timestamp + "." + request.rawBody;

  const expectedSignature = crypto
    .createHmac("sha256", webhookSecret)
    .update(signedPayload)
    .digest("hex");

  if (receivedSignature !== expectedSignature) {
    console.error("Stripe webhook signature verification failed");
    return {
      status: 401,
      body: { error: "Invalid signature" },
    };
  }

  console.log("✅ Stripe webhook signature verified successfully");

  // Parse the webhook payload
  const event = JSON.parse(request.rawBody);
  console.log(`Processing event: ${event.type}`);

  // Your business logic here

  return {
    status: 200,
    body: { received: true },
  };
} catch (error) {
  console.error("Error processing Stripe webhook:", error);
  return {
    status: 500,
    body: { error: "Internal server error" },
  };
}

import hmac
import hashlib
import json
import os


def main():
    request = yepcode.context.request

    webhook_secret = os.getenv("STRIPE_WEBHOOK_SECRET")

    # Get the Stripe signature from headers
    stripe_signature = request.get("headers", {}).get("stripe-signature", "")

    if not stripe_signature:
        return {"status": 400, "body": {"error": "Missing Stripe signature header"}}

    # Verify the webhook signature
    try:
        # Parse the signature header (format: t=timestamp,v1=signature)
        signatures = {}
        for pair in stripe_signature.split(","):
            key, value = pair.split("=", 1)
            signatures[key] = value

        timestamp = signatures.get("t", "")
        received_signature = signatures.get("v1", "")

        if not timestamp or not received_signature:
            print("Missing timestamp or signature in Stripe header")
            return {"status": 401, "body": {"error": "Invalid signature format"}}

        # Check timestamp to prevent replay attacks (allow up to 5 minutes)
        import time
        current_time = int(time.time())
        webhook_time = int(timestamp)
        tolerance = 300  # 5 minutes in seconds

        if abs(current_time - webhook_time) > tolerance:
            print("Webhook timestamp too old, possible replay attack")
            return {"status": 401, "body": {"error": "Request timestamp too old"}}

        # Create the signed payload (timestamp + raw body)
        raw_body = request.get("raw_body", "")
        signed_payload = timestamp + "." + raw_body

        expected_signature = hmac.new(
            webhook_secret.encode(),
            signed_payload.encode("utf-8"),
            hashlib.sha256,
        ).hexdigest()

        if received_signature != expected_signature:
            print("Stripe webhook signature verification failed")
            return {"status": 401, "body": {"error": "Invalid signature"}}

        print("✅ Stripe webhook signature verified successfully")

        # Parse the webhook payload
        event = json.loads(request.get("raw_body", ""))
        print(f"Processing event: {event.get('type', '')}")

        # Your business logic here

        return {"status": 200, "body": {"received": True}}

    except Exception as error:
        print(f"Error processing Stripe webhook: {error}")
        return {"status": 500, "body": {"error": "Internal server error"}}