Webhooks

import express from 'express';
import { constructEvent } from '@kalixo/checkout-node';

app.post('/webhooks/kalixo-checkout', express.raw({ type: 'application/json' }), (req, res) => {
  let event;
  try {
    event = constructEvent(req.body, req.header('Kalixo-Signature'), process.env.KALIXO_CHECKOUT_WEBHOOK_SECRET);
  } catch {
    return res.status(400).end();
  }
  res.status(200).end();
  queue.add(event); // process asynchronously
});

import hmac, hashlib, time

def verify(raw_body: bytes, header: str, secret: str) -> bool:
    parts = dict(p.split("=", 1) for p in header.split(",") if "=" in p)
    t = int(parts.get("t", 0))
    if abs(time.time() - t) > 300:
        return False
    expected = hmac.new(secret.encode(), f"{t}.".encode() + raw_body, hashlib.sha256).hexdigest()
    candidates = [v for k, v in (p.split("=", 1) for p in header.split(",")) if k.strip() == "v1"]
    return any(hmac.compare_digest(expected, c) for c in candidates)