Call of Duty webhooks

// server.js  (npm install express)
import crypto from "node:crypto";
import express from "express";

const app = express();
const seen = new Set(); // use your database in production

// express.raw keeps the exact bytes. express.json() here would break the signature.
app.post("/webhooks/cito", express.raw({ type: "*/*" }), (req, res) => {
  const expected = crypto
    .createHmac("sha256", process.env.CITO_WEBHOOK_SECRET)
    .update(req.body)
    .digest("hex");
  const got = req.get("X-Cito-Signature") || "";
  const valid =
    got.length === expected.length &&
    crypto.timingSafeEqual(Buffer.from(got), Buffer.from(expected));
  if (!valid) return res.sendStatus(401);

  res.sendStatus(200); // answer within 10 seconds, then work

  const { id, event, data } = JSON.parse(req.body.toString("utf8"));
  // The id is the same on every retry and replay. A replay you asked for gets through, so make the work safe to repeat.
  // test is true on the envelope of test deliveries: keep those out of your real data.
  if (seen.has(id) && !data.replayed) return;
  seen.add(id);

  switch (event) {
    case "cod.transfer.new":
      // handle cod.transfer.new
      break;
    case "cod.match.completed":
      // handle cod.match.completed
      break;
    case "cod.standings.updated":
      // handle cod.standings.updated
      break;
  }
});

app.listen(3000);

// app/api/webhooks/cito/route.ts
import crypto from "node:crypto";

const seen = new Set<string>(); // use your database in production

export async function POST(req: Request) {
  const raw = await req.text(); // the exact body, before any JSON parsing
  const expected = crypto
    .createHmac("sha256", process.env.CITO_WEBHOOK_SECRET!)
    .update(raw, "utf8")
    .digest("hex");
  const got = req.headers.get("x-cito-signature") ?? "";
  const valid =
    got.length === expected.length &&
    crypto.timingSafeEqual(Buffer.from(got), Buffer.from(expected));
  if (!valid) return new Response("invalid signature", { status: 401 });

  const payload = JSON.parse(raw);
  // The id is the same on every retry and replay. A replay you asked for gets through, so make handle() safe to repeat.
  // payload.test is true on test deliveries: keep those out of your real data.
  if (!seen.has(payload.id) || payload.data?.replayed) {
    seen.add(payload.id);
    // Do slow work in a queue. Answer within 10 seconds.
    handle(payload).catch(console.error);
  }
  return new Response("ok");
}

async function handle({ event, data }: { event: string; data: any }) {
  switch (event) {
    case "cod.transfer.new":
      // handle cod.transfer.new
      break;
    case "cod.match.completed":
      // handle cod.match.completed
      break;
    case "cod.standings.updated":
      // handle cod.standings.updated
      break;
  }
}

# main.py  (pip install fastapi uvicorn)
import hashlib, hmac, json, os
from fastapi import BackgroundTasks, FastAPI, HTTPException, Request

app = FastAPI()
SECRET = os.environ["CITO_WEBHOOK_SECRET"].encode()
seen = set()  # use your database in production

def handle(event, data):
    if event == "cod.transfer.new":
        pass  # handle cod.transfer.new
    elif event == "cod.match.completed":
        pass  # handle cod.match.completed
    elif event == "cod.standings.updated":
        pass  # handle cod.standings.updated

@app.post("/webhooks/cito")
async def cito_webhook(request: Request, tasks: BackgroundTasks):
    raw = await request.body()  # the exact bytes, before any JSON parsing
    expected = hmac.new(SECRET, raw, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected.encode(), request.headers.get("x-cito-signature", "").encode()):
        raise HTTPException(status_code=401, detail="invalid signature")

    payload = json.loads(raw)
    # The id is the same on every retry and replay. A replay you asked for gets through, so make handle() safe to repeat.
    # payload["test"] is True on test deliveries: keep those out of your real data.
    if payload["id"] not in seen or payload["data"].get("replayed"):
        seen.add(payload["id"])
        tasks.add_task(handle, payload["event"], payload["data"])  # runs after we answer
    return {"ok": True}

# app.py  (pip install flask)
import hashlib, hmac, json, os
from flask import Flask, abort, request

app = Flask(__name__)
SECRET = os.environ["CITO_WEBHOOK_SECRET"].encode()
seen = set()  # use your database in production

def handle(event, data):
    if event == "cod.transfer.new":
        pass  # handle cod.transfer.new
    elif event == "cod.match.completed":
        pass  # handle cod.match.completed
    elif event == "cod.standings.updated":
        pass  # handle cod.standings.updated

@app.post("/webhooks/cito")
def cito_webhook():
    raw = request.get_data()  # the exact bytes, before any JSON parsing
    expected = hmac.new(SECRET, raw, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected.encode(), request.headers.get("X-Cito-Signature", "").encode()):
        abort(401)

    payload = json.loads(raw)
    # The id is the same on every retry and replay. A replay you asked for gets through, so make handle() safe to repeat.
    # payload["test"] is True on test deliveries: keep those out of your real data.
    if payload["id"] not in seen or payload["data"].get("replayed"):
        seen.add(payload["id"])
        handle(payload["event"], payload["data"])  # move slow work to a queue: answer within 10 seconds
    return "", 200

// main.go
package main

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"io"
	"log"
	"net/http"
	"os"
	"sync"
)

var (
	secret = []byte(os.Getenv("CITO_WEBHOOK_SECRET"))
	seen   sync.Map // use your database in production
)

type envelope struct {
	Event string          `json:"event"`
	ID    string          `json:"id"`
	Data  json.RawMessage `json:"data"`
}

func handle(event string, data json.RawMessage) {
	switch event {
	case "cod.transfer.new":
		// handle cod.transfer.new
	case "cod.match.completed":
		// handle cod.match.completed
	case "cod.standings.updated":
		// handle cod.standings.updated
	}
}

func citoWebhook(w http.ResponseWriter, r *http.Request) {
	raw, err := io.ReadAll(r.Body) // the exact bytes, before any JSON parsing
	if err != nil {
		http.Error(w, "bad body", http.StatusBadRequest)
		return
	}
	mac := hmac.New(sha256.New, secret)
	mac.Write(raw)
	expected := hex.EncodeToString(mac.Sum(nil))
	if !hmac.Equal([]byte(expected), []byte(r.Header.Get("X-Cito-Signature"))) {
		http.Error(w, "invalid signature", http.StatusUnauthorized)
		return
	}

	var p envelope
	if err := json.Unmarshal(raw, &p); err != nil {
		http.Error(w, "bad json", http.StatusBadRequest)
		return
	}
	// The id is the same on every retry and replay. A replay you asked for gets through, so make handle safe to repeat.
	// "test" is true on test deliveries: keep those out of your real data.
	var replay struct {
		Replayed bool `json:"replayed"`
	}
	_ = json.Unmarshal(p.Data, &replay)
	if _, dup := seen.LoadOrStore(p.ID, true); !dup || replay.Replayed {
		go handle(p.Event, p.Data) // answer first, work after
	}
	w.WriteHeader(http.StatusOK)
}

func main() {
	http.HandleFunc("/webhooks/cito", citoWebhook)
	log.Fatal(http.ListenAndServe(":3000", nil))
}