Push subscriptions

import { createServer } from "node:http"
import { createHmac, timingSafeEqual } from "node:crypto"

// During a rotation, list the new secret and the previous one.
const secrets = [process.env.EZGH_QUEUES_SIGNING_SECRET]

function verifySignature(header, body, secrets, now = Date.now()) {
  let t = -1
  const signatures = []
  for (const part of (header ?? "").split(",")) {
    const [key, value = ""] = part.trim().split("=")
    if (key === "t" && /^\d+$/.test(value)) t = Number(value)
    if (key === "v1" && /^[0-9a-f]{64}$/.test(value)) signatures.push(Buffer.from(value, "hex"))
  }
  if (t < 0 || signatures.length === 0) return false
  if (Math.abs(now / 1000 - t) > 300) return false // more than 5 minutes old or ahead
  return secrets.some((secret) => {
    const key = Buffer.from(secret.slice("whsec_".length), "base64url")
    const expected = createHmac("sha256", key).update(`${t}.`).update(body).digest()
    return signatures.some((signature) => timingSafeEqual(expected, signature))
  })
}

createServer(async (req, res) => {
  const chunks = []
  for await (const chunk of req) chunks.push(chunk)
  const body = Buffer.concat(chunks)
  if (!verifySignature(req.headers["ezgh-signature"], body, secrets)) {
    res.writeHead(401).end()
    return
  }
  const delivery = JSON.parse(body.toString("utf8"))
  const failed = []
  for (const message of delivery.messages) {
    try {
      await handle(message) // your code
    } catch {
      failed.push(message.messageId)
    }
  }
  res.writeHead(200, { "Content-Type": "application/json" }).end(JSON.stringify({ failed }))
}).listen(8080)

package main

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"io"
	"net/http"
	"strconv"
	"strings"
	"time"
)

func verifySignature(header string, body []byte, secrets []string, now time.Time) bool {
	t := int64(-1)
	var signatures [][]byte
	for _, part := range strings.Split(header, ",") {
		key, value, _ := strings.Cut(strings.TrimSpace(part), "=")
		switch key {
		case "t":
			if n, err := strconv.ParseUint(value, 10, 63); err == nil {
				t = int64(n)
			}
		case "v1":
			if b, err := hex.DecodeString(value); err == nil && len(b) == sha256.Size {
				signatures = append(signatures, b)
			}
		}
	}
	if t < 0 || len(signatures) == 0 {
		return false
	}
	if age := now.Unix() - t; age > 300 || age < -300 { // more than 5 minutes old or ahead
		return false
	}
	for _, secret := range secrets {
		key, err := base64.RawURLEncoding.DecodeString(strings.TrimPrefix(secret, "whsec_"))
		if err != nil {
			continue
		}
		mac := hmac.New(sha256.New, key)
		mac.Write([]byte(strconv.FormatInt(t, 10) + "."))
		mac.Write(body)
		expected := mac.Sum(nil)
		for _, signature := range signatures {
			if hmac.Equal(expected, signature) {
				return true
			}
		}
	}
	return false
}

type message struct {
	MessageID  string            `json:"messageId"`
	Body       string            `json:"body"`
	Attributes map[string]string `json:"attributes"`
}

// deliveries serves a subscription's endpoint. During a rotation, pass the new secret and the
// previous one.
func deliveries(secrets []string, handle func(*http.Request, message) error) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, 16<<20))
		if err != nil {
			http.Error(w, "unreadable body", http.StatusBadRequest)
			return
		}
		if !verifySignature(r.Header.Get("EZGH-Signature"), body, secrets, time.Now()) {
			http.Error(w, "bad signature", http.StatusUnauthorized)
			return
		}
		var delivery struct {
			Messages []message `json:"messages"`
		}
		if err := json.Unmarshal(body, &delivery); err != nil {
			http.Error(w, "bad JSON", http.StatusBadRequest)
			return
		}
		failed := []string{}
		for _, m := range delivery.Messages {
			if err := handle(r, m); err != nil {
				failed = append(failed, m.MessageID)
			}
		}
		w.Header().Set("Content-Type", "application/json")
		json.NewEncoder(w).Encode(map[string][]string{"failed": failed})
	}
}

import base64
import hashlib
import hmac
import json
import os
import time

from fastapi import FastAPI, Request, Response


def verify_signature(header, body, secrets, now=None):
    t, signatures = None, []
    for part in (header or "").split(","):
        key, _, value = part.strip().partition("=")
        if key == "t" and value.isascii() and value.isdigit():
            t = int(value)
        elif key == "v1" and len(value) == 64 and value.isascii():
            signatures.append(value)
    if t is None or not signatures:
        return False
    if abs((time.time() if now is None else now) - t) > 300:  # more than 5 minutes old or ahead
        return False
    for secret in secrets:
        encoded = secret.removeprefix("whsec_")
        key = base64.urlsafe_b64decode(encoded + "=" * (-len(encoded) % 4))
        expected = hmac.new(key, f"{t}.".encode() + body, hashlib.sha256).hexdigest()
        if any(hmac.compare_digest(expected, s) for s in signatures):
            return True
    return False


app = FastAPI()
# During a rotation, list the new secret and the previous one.
SECRETS = [os.environ["EZGH_QUEUES_SIGNING_SECRET"]]


@app.post("/queues/orders")
async def deliveries(request: Request):
    body = await request.body()
    if not verify_signature(request.headers.get("EZGH-Signature"), body, SECRETS):
        return Response(status_code=401)
    failed = []
    for message in json.loads(body)["messages"]:
        try:
            await handle(message)  # your code
        except Exception:
            failed.append(message["messageId"])
    return {"failed": failed}