Verify signatures

import express from "express"
import crypto from "crypto"

const app = express()

// IMPORTANT: you need the raw body. Use express.raw for this route.
app.post(
  "/webhook/ryvo",
  express.raw({ type: "application/json" }),
  (req, res) => {
    const signatureHeader = req.headers["x-ryvo-signature"]
    if (!signatureHeader) return res.status(401).send("missing signature")

    const parts = signatureHeader.split(",").reduce((acc, p) => {
      const [k, v] = p.split("=")
      acc[k] = v
      return acc
    }, {})

    const ts = parts.t
    const sig = parts.v1
    if (!ts || !sig) return res.status(401).send("invalid signature format")

    // 5 minute tolerance
    if (Math.abs(Date.now() / 1000 - Number(ts)) > 300) {
      return res.status(401).send("signature too old")
    }

    const expected = crypto
      .createHmac("sha256", process.env.RYVO_WEBHOOK_SECRET)
      .update(`${ts}.${req.body.toString()}`)
      .digest("hex")

    const a = Buffer.from(sig, "hex")
    const b = Buffer.from(expected, "hex")
    if (a.length !== b.length || !crypto.timingSafeEqual(a, b)) {
      return res.status(401).send("invalid signature")
    }

    const event = JSON.parse(req.body.toString())
    console.log("Verified event:", event.type, event.id)

    res.sendStatus(200)
  }
)

import hmac, hashlib, time
from flask import Flask, request, abort

app = Flask(__name__)
SECRET = os.environ["RYVO_WEBHOOK_SECRET"]

@app.post("/webhook/ryvo")
def webhook():
    sig_header = request.headers.get("X-Ryvo-Signature")
    if not sig_header:
        abort(401)

    parts = dict(p.split("=") for p in sig_header.split(","))
    ts = parts.get("t")
    sig = parts.get("v1")
    if not ts or not sig:
        abort(401)

    # 5 minute tolerance
    if abs(time.time() - int(ts)) > 300:
        abort(401)

    expected = hmac.new(
        SECRET.encode(),
        f"{ts}.{request.data.decode()}".encode(),
        hashlib.sha256,
    ).hexdigest()

    if not hmac.compare_digest(sig, expected):
        abort(401)

    event = request.get_json()
    print("Verified event:", event["type"], event["id"])
    return "", 200

require "sinatra"
require "openssl"

SECRET = ENV.fetch("RYVO_WEBHOOK_SECRET")

post "/webhook/ryvo" do
  sig_header = request.env["HTTP_X_RYVO_SIGNATURE"]
  halt 401 unless sig_header

  parts = sig_header.split(",").map { |p| p.split("=") }.to_h
  ts, sig = parts["t"], parts["v1"]
  halt 401 unless ts && sig

  halt 401 if (Time.now.to_i - ts.to_i).abs > 300

  body = request.body.read
  expected = OpenSSL::HMAC.hexdigest("SHA256", SECRET, "#{ts}.#{body}")

  halt 401 unless Rack::Utils.secure_compare(sig, expected)

  event = JSON.parse(body)
  puts "Verified event: #{event["type"]} #{event["id"]}"
  status 200
end

package main

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

var secret = []byte(os.Getenv("RYVO_WEBHOOK_SECRET"))

func handleRyvoWebhook(w http.ResponseWriter, r *http.Request) {
  sigHeader := r.Header.Get("X-Ryvo-Signature")
  if sigHeader == "" {
    http.Error(w, "missing signature", 401); return
  }

  var ts, sig string
  for _, part := range strings.Split(sigHeader, ",") {
    kv := strings.SplitN(part, "=", 2)
    if len(kv) != 2 { continue }
    if kv[0] == "t" { ts = kv[1] }
    if kv[0] == "v1" { sig = kv[1] }
  }
  if ts == "" || sig == "" {
    http.Error(w, "bad signature format", 401); return
  }

  tsInt, _ := strconv.ParseInt(ts, 10, 64)
  if time.Now().Unix()-tsInt > 300 || tsInt-time.Now().Unix() > 300 {
    http.Error(w, "signature too old", 401); return
  }

  body, _ := io.ReadAll(r.Body)
  mac := hmac.New(sha256.New, secret)
  mac.Write([]byte(ts + "." + string(body)))
  expected := hex.EncodeToString(mac.Sum(nil))

  sigBytes, _ := hex.DecodeString(sig)
  expBytes, _ := hex.DecodeString(expected)
  if !hmac.Equal(sigBytes, expBytes) {
    http.Error(w, "invalid signature", 401); return
  }

  w.WriteHeader(200)
}