Webhooks for Pay with Warp

import crypto from 'node:crypto';

export function verifyWarpSignature(rawBody, header, secret, toleranceSeconds = 300) {
  if (!header || !secret) return false;
  const parts = Object.fromEntries(
    header.split(',').map((part) => {
      const i = part.indexOf('=');
      return [part.slice(0, i).trim(), part.slice(i + 1).trim()];
    }),
  );
  const { t, v1 } = parts;
  if (!t || !v1 || !/^\d+$/.test(t)) return false;
  if (Math.abs(Date.now() / 1000 - Number(t)) > toleranceSeconds) return false;

  const expected = crypto
    .createHmac('sha256', secret)
    .update(`${t}.`)
    .update(rawBody) // Buffer from express.raw()
    .digest('hex');

  const a = Buffer.from(expected, 'hex');
  const b = Buffer.from(v1, 'hex');
  return a.length === b.length && crypto.timingSafeEqual(a, b);
}

import hashlib
import hmac
import time


def verify_warp_signature(raw_body: bytes, header: str, secret: str, tolerance: int = 300) -> bool:
    try:
        parts = dict(p.strip().split("=", 1) for p in header.split(","))
        t, v1 = parts["t"], parts["v1"]
        timestamp = int(t)
    except (AttributeError, KeyError, ValueError):
        return False
    if abs(time.time() - timestamp) > tolerance:
        return False
    expected = hmac.new(secret.encode(), f"{t}.".encode() + raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, v1)

import express from 'express';
import { verifyWarpSignature } from './warp-webhooks.js';

const app = express();

// Use the raw body: the signature is computed over the exact bytes Warp sent.
app.post('/webhooks/warp', express.raw({ type: 'application/json' }), async (req, res) => {
  if (!verifyWarpSignature(req.body, req.get('Warp-Signature'), process.env.WARP_WEBHOOK_SECRET)) {
    return res.status(400).send('Invalid signature');
  }

  const event = JSON.parse(req.body.toString('utf8'));

  // Deliveries can repeat. Process each event ID once.
  if (await alreadyProcessed(event.id)) return res.sendStatus(200);

  const session = event.data.object;
  switch (event.type) {
    case 'payment.completed':
      await markOrderPaid(session.metadata.order_id, session.payment.tx_hash);
      break;
    case 'payment.failed':
    case 'payment.expired':
      await releaseOrder(session.metadata.order_id);
      break;
  }

  await recordProcessed(event.id);
  res.sendStatus(200);
});

import json
import os

from flask import Flask, request

from warp_webhooks import verify_warp_signature

app = Flask(__name__)


@app.post("/webhooks/warp")
def warp_webhook():
    raw = request.get_data()  # raw bytes, before JSON parsing
    if not verify_warp_signature(raw, request.headers.get("Warp-Signature", ""), os.environ["WARP_WEBHOOK_SECRET"]):
        return "Invalid signature", 400

    event = json.loads(raw)
    if already_processed(event["id"]):
        return "", 200

    session = event["data"]["object"]
    if event["type"] == "payment.completed":
        mark_order_paid(session["metadata"]["order_id"], session["payment"]["tx_hash"])
    elif event["type"] in ("payment.failed", "payment.expired"):
        release_order(session["metadata"]["order_id"])

    record_processed(event["id"])
    return "", 200