API Webhooks

const crypto = require('crypto');
 
function verifyWebhookSignature(rawBody, signature, timestamp, secret) {
  // Rejeter si le timestamp est trop ancien (protection replay)
  const now = Math.floor(Date.now() / 1000);
  if (Math.abs(now - parseInt(timestamp)) > 300) {
    return false;
  }
 
  const signedPayload = `${timestamp}.${rawBody}`;
  const expectedSignature = crypto
    .createHmac('sha256', secret)
    .update(signedPayload)
    .digest('hex');
 
  return crypto.timingSafeEqual(
    Buffer.from(signature),
    Buffer.from(expectedSignature)
  );
}
 
// Express.js
app.post('/webhooks/auris', express.raw({ type: 'application/json' }), (req, res) => {
  const signature = req.headers['x-webhook-signature'];
  const timestamp = req.headers['x-webhook-timestamp'];
  const rawBody = req.body.toString();
 
  if (!verifyWebhookSignature(rawBody, signature, timestamp, process.env.WEBHOOK_SECRET)) {
    return res.status(401).json({ error: 'Signature invalide' });
  }
 
  const payload = JSON.parse(rawBody);
  console.log('Événement webhook reçu :', payload.event);
  res.status(200).json({ received: true });
});

import hmac
import hashlib
import time
import json
from flask import Flask, request, jsonify
 
app = Flask(__name__)
 
def verify_webhook_signature(raw_body: bytes, signature: str, timestamp: str, secret: str) -> bool:
    # Rejeter si le timestamp est trop ancien (protection replay)
    now = int(time.time())
    if abs(now - int(timestamp)) > 300:
        return False
 
    signed_payload = f"{timestamp}.{raw_body.decode('utf-8')}"
    expected_signature = hmac.new(
        secret.encode('utf-8'),
        signed_payload.encode('utf-8'),
        hashlib.sha256
    ).hexdigest()
 
    return hmac.compare_digest(signature, expected_signature)
 
@app.route('/webhooks/auris', methods=['POST'])
def webhook():
    signature = request.headers.get('X-Webhook-Signature')
    timestamp = request.headers.get('X-Webhook-Timestamp')
    raw_body = request.get_data()
 
    if not verify_webhook_signature(raw_body, signature, timestamp, WEBHOOK_SECRET):
        return jsonify({'error': 'Signature invalide'}), 401
 
    payload = json.loads(raw_body)
    print(f"Événement webhook reçu : {payload['event']}")
    return jsonify({'received': True}), 200