Verifying Webhooks from WaveSpeedAI

const (
  WebhookSignatureVersion = "v3"
  WebhookSignatureSeparator = ","
  WebhookContentSeparator = "."
  WebhookMaxAgeSeconds = 300 // 5 minutes
)

import hmac
import hashlib
import os
import time
 
def verify_wavespeed_signature(payload: bytes, headers: dict, secret: str, max_age: int = 300) -> bool:
    webhook_id = headers.get("webhook-id")
    timestamp = headers.get("webhook-timestamp")
    signature_header = headers.get("webhook-signature")
 
    if not (webhook_id and timestamp and signature_header):
        raise Exception("Missing required headers")
 
    parts = signature_header.split(',')
    if len(parts) != 2 or parts[0] != 'v3':
        raise Exception("Invalid signature header format")
 
    received_signature = parts[1]
    # payload must be raw request body bytes
    signed_content = f"{webhook_id}.{timestamp}.{payload.decode()}".encode()
 
    # Remove whsec_ prefix; do not base64 decode
    key_without_prefix = secret[6:] if secret.startswith('whsec_') else secret
    expected_signature = hmac.new(key_without_prefix.encode(), signed_content, hashlib.sha256).hexdigest()
 
    # Validate timestamp
    if abs(time.time() - int(timestamp)) > max_age:
        raise Exception("Signature timestamp too old")
 
    if not hmac.compare_digest(expected_signature, received_signature):
        raise Exception("Invalid signature")
 
    return True
 
# Local self-test. In your handler, pass the raw request body and received headers.
webhook_id = "pred_abc123"
payload = b'{"id":"pred_abc123","model":"wavespeed-ai/z-image/turbo","outputs":["https://example.com/output/example.jpeg"],"urls":{"get":"https://api.wavespeed.ai/api/v3/predictions/pred_abc123/result"},"status":"completed"}'
timestamp = str(int(time.time()))
secret = os.environ["WAVESPEED_WEBHOOK_SECRET"]
key = secret[6:] if secret.startswith("whsec_") else secret
signature = hmac.new(key.encode(), f"{webhook_id}.{timestamp}.".encode() + payload, hashlib.sha256).hexdigest()
headers = {
    "webhook-id": webhook_id,
    "webhook-timestamp": timestamp,
    "webhook-signature": f"v3,{signature}",
}
 
print(verify_wavespeed_signature(payload, headers, secret))

const crypto = require('crypto');
 
function verifyWaveSpeedAISignature(payload, headers, secret, maxAgeSeconds = 300) {
  const id = headers['webhook-id'];
  const timestamp = headers['webhook-timestamp'];
  const signatureHeader = headers['webhook-signature'];
 
  if (!id || !timestamp || !signatureHeader) {
    throw new Error('Missing required headers');
  }
 
  const parts = signatureHeader.split(',');
  if (parts.length !== 2 || parts[0] !== 'v3' || !/^[a-f0-9]{64}$/i.test(parts[1])) {
    throw new Error('Invalid signature header format');
  }
  const receivedSignature = parts[1];
 
  // Use raw body string (do not re-serialize)
  const signedContent = `${id}.${timestamp}.${payload}`;
 
  // Remove whsec_ prefix; do not base64 decode
  const keyWithoutPrefix = secret.startsWith('whsec_') ? secret.slice(6) : secret;
  const expectedSignature = crypto
    .createHmac('sha256', keyWithoutPrefix)
    .update(signedContent)
    .digest('hex');
 
  const ageSeconds = Math.abs(Date.now() / 1000 - parseInt(timestamp, 10));
  if (ageSeconds > maxAgeSeconds) {
    throw new Error('Signature timestamp too old');
  }
 
  if (!crypto.timingSafeEqual(Buffer.from(expectedSignature, 'hex'), Buffer.from(receivedSignature, 'hex'))) {
    throw new Error('Invalid signature');
  }
 
  return true;
}
 
// Local self-test. In your handler, pass the raw request body and received headers.
const id = 'pred_abc123';
const payload = '{"id":"pred_abc123","model":"wavespeed-ai/z-image/turbo","outputs":["https://example.com/output/example.jpeg"],"urls":{"get":"https://api.wavespeed.ai/api/v3/predictions/pred_abc123/result"},"status":"completed"}';
const timestamp = Math.floor(Date.now() / 1000).toString();
const secret = process.env.WAVESPEED_WEBHOOK_SECRET;
if (!secret) throw new Error('Set WAVESPEED_WEBHOOK_SECRET');
const key = secret.startsWith('whsec_') ? secret.slice(6) : secret;
const signature = crypto.createHmac('sha256', key).update(`${id}.${timestamp}.${payload}`).digest('hex');
const headers = {
  'webhook-id': id,
  'webhook-timestamp': timestamp,
  'webhook-signature': `v3,${signature}`,
};
 
console.log(verifyWaveSpeedAISignature(payload, headers, secret));

 
package main
 
import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"os"
	"strconv"
	"strings"
	"time"
)
 
// verifyWaveSpeedSignature verifies a WaveSpeedAI webhook signature
// Returns true if valid, error if invalid
func verifyWaveSpeedSignature(payload []byte, headers map[string]string, secret string) (bool, error) {
	webhookID := headers["webhook-id"]
	timestamp := headers["webhook-timestamp"]
	signatureHeader := headers["webhook-signature"]
 
	if webhookID == "" || timestamp == "" || signatureHeader == "" {
		return false, fmt.Errorf("missing required headers")
	}
 
	parts := strings.Split(signatureHeader, ",")
	if len(parts) != 2 || parts[0] != "v3" {
		return false, fmt.Errorf("invalid signature header format")
	}
 
	receivedSignature := parts[1]
	signedContent := fmt.Sprintf("%s.%s.%s", webhookID, timestamp, string(payload))
 
	// Remove whsec_ prefix
	keyWithoutPrefix := secret
	if strings.HasPrefix(secret, "whsec_") {
		keyWithoutPrefix = secret[6:]
	}
 
	expectedSignature := hmac.New(sha256.New, []byte(keyWithoutPrefix))
	expectedSignature.Write([]byte(signedContent))
	expectedSigHex := hex.EncodeToString(expectedSignature.Sum(nil))
 
	// Timestamp validation
	webhookTime, err := strconv.ParseInt(timestamp, 10, 64)
	if err != nil {
		return false, fmt.Errorf("invalid timestamp: %v", err)
	}
 
	age := time.Now().Unix() - webhookTime
	if age < 0 {
		age = -age
	}
	if age > 300 {
		return false, fmt.Errorf("signature timestamp too old")
	}
 
	// Constant-time comparison
	if !hmac.Equal([]byte(expectedSigHex), []byte(receivedSignature)) {
		return false, fmt.Errorf("invalid signature")
	}
 
	return true, nil
}
 
func main() {
	// Local self-test. In your handler, pass the raw request body and received headers.
	webhookID := "pred_abc123"
	payload := []byte(`{"id":"pred_abc123","model":"wavespeed-ai/z-image/turbo","outputs":["https://example.com/output/example.jpeg"],"urls":{"get":"https://api.wavespeed.ai/api/v3/predictions/pred_abc123/result"},"status":"completed"}`)
	timestamp := strconv.FormatInt(time.Now().Unix(), 10)
	secret := os.Getenv("WAVESPEED_WEBHOOK_SECRET")
	if secret == "" {
		panic("set WAVESPEED_WEBHOOK_SECRET")
	}
	key := strings.TrimPrefix(secret, "whsec_")
	mac := hmac.New(sha256.New, []byte(key))
	mac.Write([]byte(fmt.Sprintf("%s.%s.%s", webhookID, timestamp, payload)))
	signature := hex.EncodeToString(mac.Sum(nil))
 
	headers := map[string]string{
		"webhook-id":        webhookID,
		"webhook-timestamp": timestamp,
		"webhook-signature": "v3," + signature,
	}
 
	result, err := verifyWaveSpeedSignature(payload, headers, secret)
	if err != nil {
		fmt.Printf("Verification failed: %v\n", err)
		return
	}
	
	fmt.Printf("Verification result: %t\n", result)
}
 

ID="pred_abc123"
TIMESTAMP=$(date +%s)
BODY='{"id":"pred_abc123","model":"wavespeed-ai/z-image/turbo","outputs":["https://example.com/output/example.jpeg"],"urls":{"get":"https://api.wavespeed.ai/api/v3/predictions/pred_abc123/result"},"status":"completed"}'
MESSAGE="$ID.$TIMESTAMP.$BODY"
: "${WAVESPEED_WEBHOOK_SECRET:?Set WAVESPEED_WEBHOOK_SECRET}"
SECRET="$WAVESPEED_WEBHOOK_SECRET"
 
# Remove whsec_ prefix (do not base64 decode)
KEY_WITHOUT_PREFIX="${SECRET#whsec_}"
 
# Compute hex signature
SIGNATURE=$(printf "%s" "$MESSAGE" | openssl dgst -sha256 -hmac "$KEY_WITHOUT_PREFIX" | sed 's/^.* //')
 
echo  "\"webhook-id\"": "\"$ID\"",
echo "\"webhook-timestamp\"": \""$TIMESTAMP\"",
echo "\"webhook-signature\"": \""v3,$SIGNATURE\""
 
curl --fail-with-body --connect-timeout 10 --max-time 30 \
  -X POST https://your-webhook-url.com \
  -H "webhook-id: $ID" \
  -H "webhook-timestamp: $TIMESTAMP" \
  -H "webhook-signature: v3,$SIGNATURE" \
  -H "Content-Type: application/json" \
  -d "$BODY"