Receiving events

curl https://api.quic.chat/platform/v1/webhooks \
  -H "Authorization: Bearer $QUIC_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://example.com/quic/webhook",
    "verifyToken": "a-long-random-string-you-choose",
    "events": ["message.received", "message.status", "consent.updated", "app.status_changed"]
  }'

GET https://example.com/quic/webhook?quic.mode=subscribe&quic.verify_token=<your token>&quic.challenge=<random>

import crypto from 'node:crypto';

/**
 * Verify a QuiC webhook. rawBody must be the exact bytes you received
 * (a Buffer or string) — not re-serialised JSON.
 */
export function verifyQuicSignature(rawBody, header, secret, toleranceSec = 300, now = Math.floor(Date.now() / 1000)) {
  if (typeof header !== 'string' || header.length === 0) return false;
  let timestamp = null;
  const signatures = [];
  for (const part of header.split(',')) {
    const eq = part.indexOf('=');
    if (eq === -1) continue;
    const key = part.slice(0, eq).trim();
    const value = part.slice(eq + 1).trim();
    if (key === 't') timestamp = Number(value);
    else if (key === 'v1') signatures.push(value);
  }
  if (!Number.isInteger(timestamp) || Math.abs(now - timestamp) > toleranceSec) return false;
  const expected = crypto
    .createHmac('sha256', secret)
    .update(`${timestamp}.`)
    .update(rawBody)
    .digest();
  // During a secret rotation QuiC sends two v1 values for 24 hours.
  return signatures.some((sig) => {
    const given = Buffer.from(sig, 'hex');
    return given.length === expected.length && crypto.timingSafeEqual(given, expected);
  });
}

import hashlib
import hmac
import time


def verify_quic_signature(raw_body: bytes, header: str, secret: str,
                          tolerance: int = 300, now: int | None = None) -> bool:
    """Verify a QuiC webhook. raw_body must be the exact bytes you received."""
    if not header:
        return False
    timestamp, signatures = None, []
    for part in header.split(","):
        key, _, value = part.strip().partition("=")
        if key == "t" and value.isdigit():
            timestamp = int(value)
        elif key == "v1":
            signatures.append(value)
    now = int(time.time()) if now is None else now
    if timestamp is None or abs(now - timestamp) > tolerance:
        return False
    expected = hmac.new(secret.encode("utf-8"),
                        f"{timestamp}.".encode("utf-8") + raw_body,
                        hashlib.sha256).hexdigest()
    # During a secret rotation QuiC sends two v1 values for 24 hours.
    return any(hmac.compare_digest(expected, sig) for sig in signatures)

import express from 'express';
import { verifyQuicSignature } from './verifyQuicSignature.js';

const app = express();
const seen = new Set(); // use a database or Redis in production

// 1. Verification handshake: echo the challenge if the token matches.
app.get('/quic/webhook', (req, res) => {
  if (req.query['quic.mode'] === 'subscribe' && req.query['quic.verify_token'] === process.env.QUIC_VERIFY_TOKEN) {
    return res.status(200).type('text/plain').send(req.query['quic.challenge']);
  }
  return res.sendStatus(403);
});

// 2. Events: verify the signature on the RAW body, dedupe on event id, answer fast.
app.post('/quic/webhook', express.raw({ type: 'application/json' }), (req, res) => {
  if (!verifyQuicSignature(req.body, req.get('QuiC-Signature'), process.env.QUIC_WEBHOOK_SECRET)) {
    return res.sendStatus(401);
  }
  const event = JSON.parse(req.body.toString('utf8'));
  if (seen.has(event.id)) return res.sendStatus(200);
  seen.add(event.id);
  res.sendStatus(200); // acknowledge within 10 seconds, then do the work
  handleEvent(event);
});

function handleEvent(event) {
  switch (event.type) {
    case 'message.received':
      console.log(`${event.data.from} says: ${event.data.text}`);
      break;
    case 'consent.updated':
      // opted_out or blocked: stop messaging this appUserId immediately.
      break;
  }
}

app.listen(3000);

import json
import os

from flask import Flask, abort, request

from verify import verify_quic_signature

app = Flask(__name__)


@app.get("/quic/webhook")
def subscribe():
    if (request.args.get("quic.mode") == "subscribe"
            and request.args.get("quic.verify_token") == os.environ["QUIC_VERIFY_TOKEN"]):
        return request.args.get("quic.challenge", ""), 200, {"Content-Type": "text/plain"}
    abort(403)


@app.post("/quic/webhook")
def events():
    raw = request.get_data()  # the exact bytes, before any JSON parsing
    if not verify_quic_signature(raw, request.headers.get("QuiC-Signature", ""),
                                 os.environ["QUIC_WEBHOOK_SECRET"]):
        abort(401)
    event = json.loads(raw)
    # Dedupe on event["id"], then handle event["type"].
    return "", 200