Verify webhook signatures

const crypto = require('crypto');

function verifyBlooioSignature(rawBody, signatureHeader, secret) {
  // Parse the signature header
  const parts = signatureHeader.split(',');
  const timestamp = parts.find(p => p.startsWith('t=')).split('=')[1];
  const signature = parts.find(p => p.startsWith('v1=')).split('=')[1];

  // Reject webhooks older than 5 minutes (replay protection)
  const age = Math.floor(Date.now() / 1000) - parseInt(timestamp, 10);
  if (age > 300) {
    throw new Error('Webhook timestamp too old');
  }

  // Compute expected signature
  const expectedSignature = crypto
    .createHmac('sha256', secret)
    .update(`${timestamp}.${rawBody}`)
    .digest('hex');

  // Constant-time comparison to prevent timing attacks
  return crypto.timingSafeEqual(
    Buffer.from(signature, 'hex'),
    Buffer.from(expectedSignature, 'hex')
  );
}

// Express middleware
app.post('/webhook', express.raw({ type: 'application/json' }), (req, res) => {
  const signature = req.headers['x-blooio-signature'];

  if (!signature) {
    return res.status(401).send('Missing signature');
  }

  try {
    const isValid = verifyBlooioSignature(
      req.body.toString(),
      signature,
      process.env.BLOOIO_WEBHOOK_SECRET
    );

    if (!isValid) {
      return res.status(401).send('Invalid signature');
    }
  } catch (err) {
    return res.status(401).send(err.message);
  }

  const event = JSON.parse(req.body);
  // Process the webhook...

  res.sendStatus(200);
});

import hmac
import hashlib
import time

def verify_blooio_signature(raw_body: bytes, signature_header: str, secret: str) -> bool:
    # Parse the signature header
    parts = dict(p.split('=') for p in signature_header.split(','))
    timestamp = parts.get('t')
    signature = parts.get('v1')

    if not timestamp or not signature:
        raise ValueError('Invalid signature format')

    # Reject webhooks older than 5 minutes
    age = int(time.time()) - int(timestamp)
    if age > 300:
        raise ValueError('Webhook timestamp too old')

    # Compute expected signature
    signed_payload = f"{timestamp}.{raw_body.decode('utf-8')}"
    expected = hmac.new(
        secret.encode('utf-8'),
        signed_payload.encode('utf-8'),
        hashlib.sha256
    ).hexdigest()

    # Constant-time comparison
    return hmac.compare_digest(signature, expected)


# Flask example
from flask import Flask, request

app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def webhook():
    signature = request.headers.get('X-Blooio-Signature')

    if not signature:
        return 'Missing signature', 401

    try:
        is_valid = verify_blooio_signature(
            request.get_data(),
            signature,
            os.environ['BLOOIO_WEBHOOK_SECRET']
        )

        if not is_valid:
            return 'Invalid signature', 401
    except ValueError as e:
        return str(e), 401

    event = request.get_json()
    # Process the webhook...

    return '', 200

package main

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"io"
	"net/http"
	"strconv"
	"strings"
	"time"
)

func verifyBlooioSignature(body []byte, signatureHeader, secret string) error {
	// Parse signature header
	parts := strings.Split(signatureHeader, ",")
	var timestamp, signature string
	for _, part := range parts {
		if strings.HasPrefix(part, "t=") {
			timestamp = strings.TrimPrefix(part, "t=")
		}
		if strings.HasPrefix(part, "v1=") {
			signature = strings.TrimPrefix(part, "v1=")
		}
	}

	if timestamp == "" || signature == "" {
		return errors.New("invalid signature format")
	}

	// Check timestamp age
	ts, _ := strconv.ParseInt(timestamp, 10, 64)
	age := time.Now().Unix() - ts
	if age > 300 {
		return errors.New("webhook timestamp too old")
	}

	// Compute expected signature
	signedPayload := timestamp + "." + string(body)
	mac := hmac.New(sha256.New, []byte(secret))
	mac.Write([]byte(signedPayload))
	expected := hex.EncodeToString(mac.Sum(nil))

	// Constant-time comparison
	if !hmac.Equal([]byte(signature), []byte(expected)) {
		return errors.New("signature mismatch")
	}

	return nil
}

func webhookHandler(w http.ResponseWriter, r *http.Request) {
	body, _ := io.ReadAll(r.Body)
	signature := r.Header.Get("X-Blooio-Signature")

	err := verifyBlooioSignature(body, signature, os.Getenv("BLOOIO_WEBHOOK_SECRET"))
	if err != nil {
		http.Error(w, err.Error(), http.StatusUnauthorized)
		return
	}

	// Process webhook...
	w.WriteHeader(http.StatusOK)
}