Webhooks

import crypto from 'crypto';

interface WebhookPayload {
  requestId: string;
  jobId?: string;
  type: string;
  status: 'completed' | 'failed';
  data?: unknown;
  error?: { code: string; message: string };
  timestamp: string;
  signature: string;
}

function verifyWebhookSignature(
  payload: WebhookPayload,
  secret: string
): boolean {
  // Extract signature from payload
  const { signature, ...payloadWithoutSignature } = payload;
  
  // Compute expected signature
  const hmac = crypto.createHmac('sha256', secret);
  hmac.update(JSON.stringify(payloadWithoutSignature));
  const expectedSignature = hmac.digest('hex');
  
  // Compare signatures (timing-safe)
  return crypto.timingSafeEqual(
    Buffer.from(signature),
    Buffer.from(expectedSignature)
  );
}

// Express.js example
app.post('/webhook', (req, res) => {
  const payload = req.body as WebhookPayload;
  const secret = process.env.PREDIKT_WEBHOOK_SECRET!;
  
  if (!verifyWebhookSignature(payload, secret)) {
    console.error('Invalid webhook signature');
    return res.status(401).json({ error: 'Invalid signature' });
  }
  
  // Process the verified webhook
  console.log('Verified webhook:', payload.type, payload.status);
  
  // Always respond quickly
  res.status(200).json({ received: true });
});

import hmac
import hashlib
import json
from flask import Flask, request, jsonify

app = Flask(__name__)
WEBHOOK_SECRET = os.environ.get('PREDIKT_WEBHOOK_SECRET')

def verify_webhook_signature(payload: dict, secret: str) -> bool:
    """Verify the HMAC-SHA256 signature of a webhook payload."""
    signature = payload.pop('signature', None)
    if not signature:
        return False
    
    # Compute expected signature
    payload_bytes = json.dumps(payload, separators=(',', ':')).encode()
    expected = hmac.new(
        secret.encode(),
        payload_bytes,
        hashlib.sha256
    ).hexdigest()
    
    # Timing-safe comparison
    return hmac.compare_digest(signature, expected)

@app.route('/webhook', methods=['POST'])
def handle_webhook():
    payload = request.get_json()
    
    if not verify_webhook_signature(payload.copy(), WEBHOOK_SECRET):
        return jsonify({'error': 'Invalid signature'}), 401
    
    # Process the verified webhook
    print(f"Verified webhook: {payload['type']} - {payload['status']}")
    
    return jsonify({'received': True}), 200

package main

import (
    "crypto/hmac"
    "crypto/sha256"
    "encoding/hex"
    "encoding/json"
)

func verifyWebhookSignature(payload map[string]interface{}, secret string) bool {
    // Extract and remove signature
    signature, ok := payload["signature"].(string)
    if !ok {
        return false
    }
    delete(payload, "signature")
    
    // Compute expected signature
    payloadBytes, _ := json.Marshal(payload)
    mac := hmac.New(sha256.New, []byte(secret))
    mac.Write(payloadBytes)
    expected := hex.EncodeToString(mac.Sum(nil))
    
    // Timing-safe comparison
    return hmac.Equal([]byte(signature), []byte(expected))
}