Webhooks

1import crypto from 'crypto';
2 
3function verifyWebhookSignature(payload, signature, secret) {
4  const expectedSignature = crypto
5    .createHmac('sha256', secret)
6    .update(payload)
7    .digest('hex');
8 
9  const trusted = Buffer.from(`sha256=${expectedSignature}`, 'ascii');
10  const untrusted = Buffer.from(signature, 'ascii');
11 
12  return crypto.timingSafeEqual(trusted, untrusted);
13}
14 
15// Express.js Handler
16app.post('/webhook', express.raw({ type: 'application/json' }), (req, res) => {
17  const signature = req.headers['x-xhub-signature'];
18  const payload = req.body;
19 
20  if (!verifyWebhookSignature(payload, signature, process.env.WEBHOOK_SECRET)) {
21    return res.status(401).send('Invalid signature');
22  }
23 
24  const event = JSON.parse(payload);
25 
26  switch (event.type) {
27    case 'invoice.created':
28      console.log('Invoice created:', event.data.object.id);
29      break;
30    case 'invoice.validation_failed':
31      console.log('Validation failed:', event.data.object.errors);
32      break;
33  }
34 
35  res.status(200).send('OK');
36});

1import hmac
2import hashlib
3from flask import Flask, request, abort
4 
5app = Flask(__name__)
6WEBHOOK_SECRET = 'your_webhook_secret'
7 
8def verify_signature(payload, signature, secret):
9    expected = hmac.new(
10        secret.encode(),
11        payload,
12        hashlib.sha256
13    ).hexdigest()
14    return hmac.compare_digest(f'sha256={expected}', signature)
15 
16@app.route('/webhook', methods=['POST'])
17def webhook():
18    signature = request.headers.get('X-Xhub-Signature')
19    payload = request.data
20 
21    if not verify_signature(payload, signature, WEBHOOK_SECRET):
22        abort(401)
23 
24    event = request.json
25 
26    if event['type'] == 'invoice.created':
27        print(f"Invoice created: {event['data']['object']['id']}")
28    elif event['type'] == 'invoice.validation_failed':
29        print(f"Validation failed: {event['data']['object']['errors']}")
30 
31    return 'OK', 200