Webhook Signatures

import crypto from 'crypto';

const TIMESTAMP_TOLERANCE = 300; // 5 minutes in seconds

function verifyWebhookSignature(rawBody, signatureHeader, timestampHeader, secret) {
  const timestamp = parseInt(timestampHeader, 10);

  // 1. Reject stale timestamps
  const now = Math.floor(Date.now() / 1000);
  if (Math.abs(now - timestamp) > TIMESTAMP_TOLERANCE) {
    throw new Error('Webhook timestamp too old');
  }

  // 2. Compute expected signature
  const signedContent = `${timestamp}.${rawBody}`;
  const expected = crypto
    .createHmac('sha256', secret)
    .update(signedContent)
    .digest('hex');

  // 3. Constant-time comparison
  const received = signatureHeader.replace('sha256=', '');
  const isValid = crypto.timingSafeEqual(
    Buffer.from(received),
    Buffer.from(expected)
  );

  if (!isValid) {
    throw new Error('Invalid webhook signature');
  }

  return true;
}

// Express example
app.post('/webhooks/klara', express.raw({ type: 'application/json' }), (req, res) => {
  try {
    verifyWebhookSignature(
      req.body.toString(),
      req.headers['x-klara-signature'],
      req.headers['x-klara-timestamp'],
      process.env.KLARA_WEBHOOK_SECRET
    );

    const event = JSON.parse(req.body);
    // Process event...

    res.status(200).send('OK');
  } catch (err) {
    res.status(401).send('Invalid signature');
  }
});

import hmac
import hashlib
import time

TIMESTAMP_TOLERANCE = 300  # 5 minutes

def verify_webhook_signature(raw_body: bytes, signature_header: str, timestamp_header: str, secret: str) -> bool:
    timestamp = int(timestamp_header)

    # 1. Reject stale timestamps
    now = int(time.time())
    if abs(now - timestamp) > TIMESTAMP_TOLERANCE:
        raise ValueError("Webhook timestamp too old")

    # 2. Compute expected signature
    signed_content = f"{timestamp}.{raw_body.decode()}"
    expected = hmac.new(
        secret.encode(),
        signed_content.encode(),
        hashlib.sha256
    ).hexdigest()

    # 3. Constant-time comparison
    received = signature_header.replace("sha256=", "")
    if not hmac.compare_digest(received, expected):
        raise ValueError("Invalid webhook signature")

    return True

# Flask example
@app.route("/webhooks/klara", methods=["POST"])
def handle_webhook():
    try:
        verify_webhook_signature(
            request.get_data(),
            request.headers["X-Klara-Signature"],
            request.headers["X-Klara-Timestamp"],
            os.environ["KLARA_WEBHOOK_SECRET"],
        )
    except ValueError:
        return "Invalid signature", 401

    event = request.get_json()
    # Process event...
    return "OK", 200

require 'openssl'

TIMESTAMP_TOLERANCE = 300 # 5 minutes

def verify_webhook_signature(raw_body, signature_header, timestamp_header, secret)
  timestamp = timestamp_header.to_i

  # 1. Reject stale timestamps
  now = Time.now.to_i
  raise "Webhook timestamp too old" if (now - timestamp).abs > TIMESTAMP_TOLERANCE

  # 2. Compute expected signature
  signed_content = "#{timestamp}.#{raw_body}"
  expected = OpenSSL::HMAC.hexdigest("SHA256", secret, signed_content)

  # 3. Constant-time comparison
  received = signature_header.sub("sha256=", "")
  raise "Invalid webhook signature" unless Rack::Utils.secure_compare(received, expected)

  true
end