Receive webhook events from Meow

import hmac, hashlib, base64, time
from fastapi import HTTPException

SECRET = "whsec_..."  # rotate via PATCH /webhooks/subscriptions/{id}

# Strip the `whsec_` prefix and base64-decode the rest to get the HMAC key.
_HMAC_KEY = base64.b64decode(SECRET.removeprefix("whsec_"))


def verify(request_body: bytes, msg_id: str, ts: str, sig_header: str) -> None:
    if abs(int(ts) - int(time.time())) > 300:
        raise HTTPException(400, "stale timestamp")

    signed = f"{msg_id}.{ts}.{request_body.decode()}".encode()
    expected = base64.b64encode(
        hmac.new(_HMAC_KEY, signed, hashlib.sha256).digest()
    ).decode()

    # `webhook-signature` may carry multiple `v1,<sig>` during rotation,
    # match any of them.
    for token in sig_header.split():
        version, _, candidate = token.partition(",")
        if version == "v1" and hmac.compare_digest(candidate, expected):
            return
    raise HTTPException(400, "bad signature")

import crypto from "node:crypto";

const SECRET = process.env.WEBHOOK_SECRET;
// Strip `whsec_` and base64-decode to get the HMAC key.
const HMAC_KEY = Buffer.from(SECRET.replace(/^whsec_/, ""), "base64");

export function verify(body, msgId, ts, sigHeader) {
  if (Math.abs(Date.now() / 1000 - Number(ts)) > 300) {
    throw new Error("stale timestamp");
  }
  const expected = Buffer.from(
    crypto
      .createHmac("sha256", HMAC_KEY)
      .update(`${msgId}.${ts}.${body}`)
      .digest("base64"),
  );

  // `webhook-signature` may carry multiple `v1,<sig>` during rotation,
  // guard buffer length first because `timingSafeEqual` throws on mismatched
  // lengths, which would 500 on a malformed token before checking the next.
  const ok = sigHeader
    .split(/\s+/)
    .filter(Boolean)
    .some((token) => {
      const [version, sig] = token.split(",");
      if (version !== "v1" || !sig) return false;
      const candidate = Buffer.from(sig);
      return (
        candidate.length === expected.length &&
        crypto.timingSafeEqual(candidate, expected)
      );
    });

  if (!ok) throw new Error("bad signature");
}

package webhooks

import (
    "crypto/hmac"
    "crypto/sha256"
    "encoding/base64"
    "errors"
    "fmt"
    "strconv"
    "strings"
    "time"
)

// Strip the `whsec_` prefix and base64-decode the rest to get the HMAC key.
func decodeSecret(secret string) ([]byte, error) {
    return base64.StdEncoding.DecodeString(strings.TrimPrefix(secret, "whsec_"))
}

func Verify(body []byte, msgID, ts, sigHeader, secret string) error {
    key, err := decodeSecret(secret)
    if err != nil {
        return fmt.Errorf("bad secret: %w", err)
    }
    epoch, err := strconv.ParseInt(ts, 10, 64)
    if err != nil {
        return fmt.Errorf("bad timestamp: %w", err)
    }
    if d := time.Now().Unix() - epoch; d > 300 || d < -300 {
        return errors.New("stale timestamp")
    }
    mac := hmac.New(sha256.New, key)
    fmt.Fprintf(mac, "%s.%s.%s", msgID, ts, body)
    expected := base64.StdEncoding.EncodeToString(mac.Sum(nil))

    for _, token := range strings.Fields(sigHeader) {
        if v, sig, ok := strings.Cut(token, ","); ok && v == "v1" {
            if hmac.Equal([]byte(sig), []byte(expected)) {
                return nil
            }
        }
    }
    return errors.New("bad signature")
}