Verifying webhook signatures

const http = require('http');
const crypto = require('crypto');

const SECRET_KEYS = [process.env.LOOP_WEBHOOK_SECRET]; // add more keys during rotation

function verifyLoopWebhook(rawBody, signatureHeader) {
  if (!signatureHeader) return false;

  const receivedSignatures = signatureHeader
    .split(',')
    .map(s => s.trim())
    .filter(s => s.startsWith('v1='));

  for (const secret of SECRET_KEYS) {
    const expected = 'v1=' + crypto
      .createHmac('sha256', secret)
      .update(rawBody, 'utf8')
      .digest('hex');

    for (const received of receivedSignatures) {
      const expectedBuf = Buffer.from(expected);
      const receivedBuf = Buffer.from(received);
      if (
        expectedBuf.length === receivedBuf.length &&
        crypto.timingSafeEqual(expectedBuf, receivedBuf)
      ) {
        return true;
      }
    }
  }

  return false;
}

const server = http.createServer((req, res) => {
  if (req.method !== 'POST') {
    res.writeHead(405).end();
    return;
  }

  const chunks = [];
  req.on('data', chunk => chunks.push(chunk));
  req.on('end', () => {
    const rawBody = Buffer.concat(chunks).toString('utf8');
    const signatureHeader = req.headers['x-loop-webhook-signature'];

    if (!verifyLoopWebhook(rawBody, signatureHeader)) {
      res.writeHead(401, { 'Content-Type': 'application/json' });
      res.end(JSON.stringify({ error: 'Invalid signature' }));
      return;
    }

    // Safe to parse and process the body
    const payload = JSON.parse(rawBody);
    console.log('Verified webhook:', payload.metaData?.source);

    res.writeHead(200).end();
  });
});

server.listen(3000, () => console.log('Listening on :3000'));

const express = require('express');
const crypto = require('crypto');

const app = express();

const SECRET_KEYS = [process.env.LOOP_WEBHOOK_SECRET]; // add more keys during rotation

function verifyLoopWebhook(rawBody, signatureHeader) {
  if (!signatureHeader) return false;

  const receivedSignatures = signatureHeader
    .split(',')
    .map(s => s.trim())
    .filter(s => s.startsWith('v1='));

  for (const secret of SECRET_KEYS) {
    const expected = 'v1=' + crypto
      .createHmac('sha256', secret)
      .update(rawBody) // rawBody is already a Buffer when using express.raw()
      .digest('hex');

    for (const received of receivedSignatures) {
      const expectedBuf = Buffer.from(expected);
      const receivedBuf = Buffer.from(received);
      if (
        expectedBuf.length === receivedBuf.length &&
        crypto.timingSafeEqual(expectedBuf, receivedBuf)
      ) {
        return true;
      }
    }
  }

  return false;
}

// Use express.raw() to preserve the raw body bytes
app.post('/webhooks/loop', express.raw({ type: 'application/json' }), (req, res) => {
  const signatureHeader = req.headers['x-loop-webhook-signature'];

  if (!verifyLoopWebhook(req.body, signatureHeader)) {
    return res.status(401).json({ error: 'Invalid signature' });
  }

  // Safe to parse now
  const payload = JSON.parse(req.body.toString('utf8'));
  console.log('Verified webhook topic:', req.headers['x-loop-webhook-topic']);

  res.sendStatus(200);
});

app.listen(3000, () => console.log('Listening on :3000'));

import hashlib
import hmac
import os
from flask import Flask, request, abort

app = Flask(__name__)

# Add more keys during key rotation
SECRET_KEYS = [os.environ['LOOP_WEBHOOK_SECRET'].encode('utf-8')]


def verify_loop_webhook(raw_body: bytes, signature_header: str) -> bool:
    if not signature_header:
        return False

    received_signatures = [
        s.strip() for s in signature_header.split(',')
        if s.strip().startswith('v1=')
    ]

    for secret in SECRET_KEYS:
        expected = 'v1=' + hmac.new(secret, raw_body, hashlib.sha256).hexdigest()

        for received in received_signatures:
            if hmac.compare_digest(expected, received):
                return True

    return False


@app.route('/webhooks/loop', methods=['POST'])
def loop_webhook():
    raw_body = request.get_data()  # raw bytes, before any parsing
    signature_header = request.headers.get('X-Loop-Webhook-Signature', '')

    if not verify_loop_webhook(raw_body, signature_header):
        abort(401)

    payload = request.get_json(force=True)
    print('Verified webhook topic:', request.headers.get('X-Loop-Webhook-Topic'))

    return '', 200


if __name__ == '__main__':
    app.run(port=3000)

# views.py
import hashlib
import hmac
import json
import os

from django.http import HttpResponse, HttpResponseForbidden
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST

# Add more keys during key rotation
SECRET_KEYS = [os.environ['LOOP_WEBHOOK_SECRET'].encode('utf-8')]


def verify_loop_webhook(raw_body: bytes, signature_header: str) -> bool:
    if not signature_header:
        return False

    received_signatures = [
        s.strip() for s in signature_header.split(',')
        if s.strip().startswith('v1=')
    ]

    for secret in SECRET_KEYS:
        expected = 'v1=' + hmac.new(secret, raw_body, hashlib.sha256).hexdigest()

        for received in received_signatures:
            if hmac.compare_digest(expected, received):
                return True

    return False


@csrf_exempt
@require_POST
def loop_webhook(request):
    raw_body = request.body  # Django exposes the raw body here
    signature_header = request.headers.get('X-Loop-Webhook-Signature', '')

    if not verify_loop_webhook(raw_body, signature_header):
        return HttpResponseForbidden('Invalid signature')

    payload = json.loads(raw_body)
    print('Verified webhook topic:', request.headers.get('X-Loop-Webhook-Topic'))

    return HttpResponse(status=200)

# urls.py
from django.urls import path
from . import views

urlpatterns = [
    path('webhooks/loop/', views.loop_webhook),
]

package main

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"io"
	"net/http"
	"os"
	"strings"
)

// secretKeys holds all active secret keys.
// Add more keys here during key rotation.
var secretKeys = []string{os.Getenv("LOOP_WEBHOOK_SECRET")}

func verifyLoopWebhook(rawBody []byte, signatureHeader string) bool {
	if signatureHeader == "" {
		return false
	}

	var receivedSignatures []string
	for _, s := range strings.Split(signatureHeader, ",") {
		s = strings.TrimSpace(s)
		if strings.HasPrefix(s, "v1=") {
			receivedSignatures = append(receivedSignatures, s)
		}
	}

	for _, secret := range secretKeys {
		mac := hmac.New(sha256.New, []byte(secret))
		mac.Write(rawBody)
		expected := "v1=" + hex.EncodeToString(mac.Sum(nil))

		for _, received := range receivedSignatures {
			// Use hmac.Equal for constant-time comparison
			if hmac.Equal([]byte(expected), []byte(received)) {
				return true
			}
		}
	}

	return false
}

func loopWebhookHandler(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
		return
	}

	rawBody, err := io.ReadAll(r.Body)
	if err != nil {
		http.Error(w, "Failed to read body", http.StatusInternalServerError)
		return
	}
	defer r.Body.Close()

	signatureHeader := r.Header.Get("X-Loop-Webhook-Signature")
	if !verifyLoopWebhook(rawBody, signatureHeader) {
		http.Error(w, "Invalid signature", http.StatusUnauthorized)
		return
	}

	topic := r.Header.Get("X-Loop-Webhook-Topic")
	fmt.Printf("Verified webhook topic: %s\n", topic)

	w.WriteHeader(http.StatusOK)
}

func main() {
	http.HandleFunc("/webhooks/loop", loopWebhookHandler)
	fmt.Println("Listening on :3000")
	http.ListenAndServe(":3000", nil)
}

require 'sinatra'
require 'openssl'

# Add more keys during key rotation
SECRET_KEYS = [ENV.fetch('LOOP_WEBHOOK_SECRET')]

def verify_loop_webhook(raw_body, signature_header)
  return false if signature_header.nil? || signature_header.empty?

  received_signatures = signature_header
    .split(',')
    .map(&:strip)
    .select { |s| s.start_with?('v1=') }

  SECRET_KEYS.each do |secret|
    digest = OpenSSL::HMAC.hexdigest('SHA256', secret, raw_body)
    expected = "v1=#{digest}"

    received_signatures.each do |received|
      # ActiveSupport::SecurityUtils.secure_compare or manual constant-time check
      return true if Rack::Utils.secure_compare(expected, received)
    end
  end

  false
end

post '/webhooks/loop' do
  request.body.rewind
  raw_body = request.body.read

  signature_header = request.env['HTTP_X_LOOP_WEBHOOK_SIGNATURE']

  unless verify_loop_webhook(raw_body, signature_header)
    halt 401, { error: 'Invalid signature' }.to_json
  end

  payload = JSON.parse(raw_body)
  puts "Verified webhook topic: #{request.env['HTTP_X_LOOP_WEBHOOK_TOPIC']}"

  status 200
end

# app/controllers/webhooks_controller.rb
class WebhooksController < ActionController::API
  # Skip Rails' default CSRF protection; we do our own verification
  skip_before_action :verify_authenticity_token

  # Add more keys during key rotation
  SECRET_KEYS = [ENV.fetch('LOOP_WEBHOOK_SECRET')].freeze

  def loop
    raw_body = request.body.read
    signature_header = request.headers['X-Loop-Webhook-Signature']

    unless verify_signature(raw_body, signature_header)
      render json: { error: 'Invalid signature' }, status: :unauthorized
      return
    end

    payload = JSON.parse(raw_body)
    Rails.logger.info "Verified webhook topic: #{request.headers['X-Loop-Webhook-Topic']}"

    head :ok
  end

  private

  def verify_signature(raw_body, signature_header)
    return false if signature_header.blank?

    received_signatures = signature_header
      .split(',')
      .map(&:strip)
      .select { |s| s.start_with?('v1=') }

    SECRET_KEYS.each do |secret|
      digest = OpenSSL::HMAC.hexdigest('SHA256', secret, raw_body)
      expected = "v1=#{digest}"

      received_signatures.each do |received|
        return true if ActiveSupport::SecurityUtils.secure_compare(expected, received)
      end
    end

    false
  end
end

# config/routes.rb
Rails.application.routes.draw do
  post '/webhooks/loop', to: 'webhooks#loop'
end

{
  "payload": { },
  "metaData": {
    "myshopifyDomain": "your-store.myshopify.com",
    "source": "loop",
    "entity": "subscription"
  },
  "previousPayload": { }
}