Webhook Signature Validation Guide

Copy

• Copy for LLM

  Copy page as Markdown for LLMs
• View as Markdown

  Open this page as Markdown
• Open in ChatGPT

  Get insights from ChatGPT
• Open in Claude

  Get insights from Claude

Synqly-Signature: sha256=a1b2c3d4e5f6...

import hashlib
import hmac

def validate_webhook_signature(payload_body, signature_header, integrator_key):
    """
    Validate a webhook signature from Synqly
    
    Args:
        payload_body (bytes): Raw request body
        signature_header (str): Value from Synqly-Signature header
        integrator_key (str): Your webhook secret (Integrator Key)
    
    Returns:
        bool: True if signature is valid, False otherwise
    """
    # Remove 'sha256=' prefix if present
    received_signature = signature_header.replace('sha256=', '')
    
    # Generate expected signature
    expected_signature = hmac.new(
        integrator_key.encode('utf-8'),
        payload_body,
        hashlib.sha256
    ).hexdigest()
    
    # Use timing-safe comparison
    return hmac.compare_digest(received_signature, expected_signature)

# Usage example
def handle_webhook(request):
    payload_body = request.body  # Raw bytes
    signature_header = request.headers.get('Synqly-Signature')
    integrator_key = 'your-integrator-key-here'
    
    if not validate_webhook_signature(payload_body, signature_header, integrator_key):
        return {'error': 'Invalid signature'}, 401
    
    # Process webhook payload
    return {'status': 'success'}, 200

package main

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

// ValidateWebhookSignature validates a webhook signature from Synqly
func ValidateWebhookSignature(payloadBody []byte, signatureHeader, integratorKey string) bool {
    // Remove 'sha256=' prefix if present
    receivedSignature := strings.TrimPrefix(signatureHeader, "sha256=")
    
    // Generate expected signature
    mac := hmac.New(sha256.New, []byte(integratorKey))
    mac.Write(payloadBody)
    expectedSignature := hex.EncodeToString(mac.Sum(nil))
    
    // Use timing-safe comparison
    return hmac.Equal([]byte(receivedSignature), []byte(expectedSignature))
}

// Usage example
func webhookHandler(w http.ResponseWriter, r *http.Request) {
    payloadBody, _ := io.ReadAll(r.Body)
    signatureHeader := r.Header.Get("Synqly-Signature")
    integratorKey := "your-integrator-key-here"
    
    if !ValidateWebhookSignature(payloadBody, signatureHeader, integratorKey) {
        http.Error(w, "Invalid signature", http.StatusUnauthorized)
        return
    }
    
    // Process webhook payload
    w.WriteHeader(http.StatusOK)
}

def validate_webhook_signature(payload_body, signature_header, integrator_key):
    if not signature_header:
        raise ValueError("Missing signature header")
    
    if not signature_header.startswith('sha256='):
        raise ValueError("Invalid signature format")
    
    # Continue with validation...

from flask import Flask, request, jsonify
import os

app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def handle_webhook():
    payload_body = request.get_data()  # Raw bytes
    signature_header = request.headers.get('Synqly-Signature')
    integrator_key = os.environ.get('INTEGRATOR_KEY')
    
    if not validate_webhook_signature(payload_body, signature_header, integrator_key):
        return jsonify({'error': 'Invalid signature'}), 401
    
    # Parse JSON after validation
    payload = request.get_json()
    
    # Process webhook
    print(f'Webhook received: {payload}')
    return jsonify({'status': 'success'})

package main

import (
    "encoding/json"
    "fmt"
    "io"
    "log"
    "net/http"
    "os"
)

func main() {
    http.HandleFunc("/webhook", webhookHandler)
    log.Println("Starting webhook server on :8080")
    log.Fatal(http.ListenAndServe(":8080", nil))
}

func webhookHandler(w http.ResponseWriter, r *http.Request) {
    if r.Method != "POST" {
        http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
        return
    }
    
    payloadBody, err := io.ReadAll(r.Body)
    if err != nil {
        http.Error(w, "Failed to read body", http.StatusBadRequest)
        return
    }
    
    signatureHeader := r.Header.Get("Synqly-Signature")
    integrationKey := os.Getenv("INTEGRATOR_KEY")
    
    if !ValidateWebhookSignature(payloadBody, signatureHeader, integrationKey) {
        http.Error(w, "Invalid signature", http.StatusUnauthorized)
        return
    }
    
    // Parse JSON after validation
    var payload map[string]interface{}
    if err := json.Unmarshal(payloadBody, &payload); err != nil {
        http.Error(w, "Invalid JSON", http.StatusBadRequest)
        return
    }
    
    // Process webhook
    fmt.Printf("Webhook received: %+v\n", payload)
    
    w.Header().Set("Content-Type", "application/json")
    w.WriteHeader(http.StatusOK)
    json.NewEncoder(w).Encode(map[string]string{"status": "success"})
}

Payload: {"test": "data"}
Secret: "test-secret"
Expected Signature: sha256=4b04c13cf8b8fa3b993c8a7e6c9dc6e0eddb0b2cee7b468cf3ed6b4b6fdda1a5

def debug_signature_validation(payload_body, signature_header, integrator_key):
    print(f"Payload length: {len(payload_body)}")
    print(f"Payload (first 100 chars): {payload_body[:100]}")
    print(f"Received signature: {signature_header}")
    print(f"Secret length: {len(integrator_key)}")
    
    # Generate expected signature
    expected_signature = hmac.new(
        integrator_key.encode('utf-8'),
        payload_body,
        hashlib.sha256
    ).hexdigest()
    
    print(f"Expected signature: sha256={expected_signature}")
    
    return validate_webhook_signature(payload_body, signature_header, integrator_key)