Webhooks

{
  "id": "evt_abc123xyz",
  "type": "fraud.detected",
  "api_version": "2025-01-15",
  "created_at": "2025-01-15T14:30:00Z",
  "data": {
    "object": "fraud_alert",
    "transaction_id": "txn_abc123",
    "user_id": "usr_123456",
    "fraud_score": 0.92,
    "fraud_level": "critical",
    "amount": 5234.00,
    "currency": "EUR",
    "risk_factors": [
      "Unusual transaction velocity",
      "New device fingerprint",
      "High-risk merchant category"
    ],
    "recommendation": "deny",
    "detected_at": "2025-01-15T14:29:58Z"
  },
  "metadata": {
    "webhook_id": "wh_abc123xyz",
    "delivery_attempt": 1
  }
}

{
  "id": "evt_def456xyz",
  "type": "refund.completed",
  "api_version": "2025-01-15",
  "created_at": "2025-01-15T14:31:00Z",
  "data": {
    "object": "refund",
    "refund_id": "ref_abc123xyz",
    "transaction_id": "txn_abc123",
    "amount": 5234.00,
    "currency": "EUR",
    "status": "completed",
    "beneficiary": {
      "name": "Juan García López",
      "account_masked": "JO94****0302"
    },
    "completed_at": "2025-01-15T14:30:58Z",
    "processing_time_seconds": 18,
    "reference": "CLIQ-20250115-789012"
  },
  "metadata": {
    "webhook_id": "wh_abc123xyz",
    "delivery_attempt": 1
  }
}

import crypto from 'crypto';

interface WebhookHeaders {
  'x-dguard-signature': string;
  'x-dguard-timestamp': string;
  'x-dguard-event-id': string;
}

function verifyWebhookSignature(
  payload: string,
  headers: WebhookHeaders,
  secret: string,
  toleranceSeconds: number = 300 // 5 minutes
): boolean {
  const signature = headers['x-dguard-signature'];
  const timestamp = headers['x-dguard-timestamp'];
  
  // 1. Check timestamp to prevent replay attacks
  const currentTime = Math.floor(Date.now() / 1000);
  const webhookTime = parseInt(timestamp, 10);
  
  if (Math.abs(currentTime - webhookTime) > toleranceSeconds) {
    console.error('Webhook timestamp outside tolerance window');
    return false;
  }
  
  // 2. Compute expected signature
  // Signature is computed over: timestamp + '.' + payload
  const signedPayload = `${timestamp}.${payload}`;
  const expectedSignature = crypto
    .createHmac('sha256', secret)
    .update(signedPayload)
    .digest('hex');
  
  // 3. Compare signatures using timing-safe comparison
  const signatureBuffer = Buffer.from(signature, 'hex');
  const expectedBuffer = Buffer.from(expectedSignature, 'hex');
  
  if (signatureBuffer.length !== expectedBuffer.length) {
    return false;
  }
  
  return crypto.timingSafeEqual(signatureBuffer, expectedBuffer);
}

// Usage in Express.js
app.post('/webhooks/dguard', express.raw({ type: 'application/json' }), (req, res) => {
  const payload = req.body.toString();
  const headers = req.headers as unknown as WebhookHeaders;
  
  if (!verifyWebhookSignature(payload, headers, process.env.DGUARD_WEBHOOK_SECRET!)) {
    console.error('Invalid webhook signature');
    return res.status(401).send('Invalid signature');
  }
  
  // Process the verified webhook
  const event = JSON.parse(payload);
  console.log('Received verified event:', event.type);
  
  // Always respond quickly with 200
  res.status(200).send('OK');
  
  // Process event asynchronously
  processWebhookEvent(event);
});

import hmac
import hashlib
import time

def verify_webhook_signature(
    payload: bytes,
    signature: str,
    timestamp: str,
    secret: str,
    tolerance_seconds: int = 300
) -> bool:
    # 1. Check timestamp to prevent replay attacks
    current_time = int(time.time())
    webhook_time = int(timestamp)
    
    if abs(current_time - webhook_time) > tolerance_seconds:
        print("Webhook timestamp outside tolerance window")
        return False
    
    # 2. Compute expected signature
    signed_payload = f"{timestamp}.{payload.decode('utf-8')}"
    expected_signature = hmac.new(
        secret.encode('utf-8'),
        signed_payload.encode('utf-8'),
        hashlib.sha256
    ).hexdigest()
    
    # 3. Compare signatures using timing-safe comparison
    return hmac.compare_digest(signature, expected_signature)

# Usage in Flask
@app.route('/webhooks/dguard', methods=['POST'])
def handle_webhook():
    payload = request.get_data()
    signature = request.headers.get('X-DGuard-Signature')
    timestamp = request.headers.get('X-DGuard-Timestamp')
    
    if not verify_webhook_signature(payload, signature, timestamp, WEBHOOK_SECRET):
        return 'Invalid signature', 401
    
    event = json.loads(payload)
    print(f"Received verified event: {event['type']}")
    
    # Process event asynchronously (use Celery, etc.)
    process_webhook_event.delay(event)
    
    return 'OK', 200

{
  "event_type": "fraud.detected"
}