Webhooks

import crypto from 'crypto';

function verifySignature(
  payload: string,
  signatureHeader: string,
  deliveryId: string,
  timestampHeader: string,
  secret: string,
): boolean {
  const segments = signatureHeader.split(',').map((part) => part.trim());
  const hasV1Marker = segments.includes('v1');
  const parts = Object.fromEntries(segments.filter((part) => part.includes('=')).map((part) => {
    const [key, value] = part.trim().split('=');
    return [key, value ?? ''];
  }));
  if (!deliveryId || !hasV1Marker || parts.t !== timestampHeader) return false;

  const timestamp = Number(parts.t);
  if (!Number.isFinite(timestamp)) return false;
  if (Math.abs(Math.floor(Date.now() / 1000) - timestamp) > 300) return false;
  if (!/^[0-9a-f]{64}$/i.test(parts.h ?? '')) return false;

  const expected = crypto
    .createHmac('sha256', secret)
    .update(`${deliveryId}.${parts.t}.${payload}`)
    .digest('hex');

  if (parts.h.length !== expected.length) return false;

  return crypto.timingSafeEqual(
    Buffer.from(parts.h, 'hex'),
    Buffer.from(expected, 'hex'),
  );
}

app.post('/webhooks/shoppex', (req, res) => {
  const signature = req.headers['x-shoppex-signature-v2'] as string;
  const deliveryId = req.headers['x-shoppex-delivery'] as string;
  const timestamp = req.headers['x-shoppex-timestamp'] as string;

  if (!verifySignature(req.rawBody, signature, deliveryId, timestamp, WEBHOOK_SECRET)) {
    return res.status(401).send('Invalid signature');
  }

  const { event, data } = req.body;

  switch (event) {
    case 'order:paid':
      handleOrderPaid(data);
      break;
    case 'order:cancelled':
      handleOrderCancelled(data);
      break;
    case 'subscription:created':
      handleSubscriptionCreated(data);
      break;
  }

  res.status(200).send('OK');
});

import hmac
import hashlib
import time
import re

def verify_signature(payload: bytes, signature_header: str, delivery_id: str, timestamp_header: str, secret: str) -> bool:
    segments = [part.strip() for part in signature_header.split(",")]
    parts = dict(part.split("=", 1) for part in segments if "=" in part)
    if not delivery_id or "v1" not in segments or parts.get("t") != timestamp_header:
        return False
    if not re.fullmatch(r"[0-9a-fA-F]{64}", parts.get("h", "")):
        return False
    if abs(int(time.time()) - int(parts["t"])) > 300:
        return False
    expected = hmac.new(
        secret.encode(),
        f"{delivery_id}.{parts['t']}.".encode() + payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(parts["h"], expected)