Webhooks

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

const app = express();
const SECRET = process.env.BLNK_SECRET; // must match server.secret_key

app.use(express.json({ verify: (req, _, buf) => { req.rawBody = buf; } }));

app.post("/webhook", (req, res) => {
  const sig = req.header("x-blnk-signature");
  const ts = req.header("x-blnk-timestamp");
  if (!sig || !ts) return res.sendStatus(400);

  const expected = crypto.createHmac("sha256", SECRET)
    .update(`${ts}.${req.rawBody}`)
    .digest("hex");

  if (!crypto.timingSafeEqual(Buffer.from(sig), Buffer.from(expected)))
    return res.sendStatus(401);

  res.sendStatus(200);
});

app.listen(3000);

package main

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

var secret = []byte(os.Getenv("BLNK_SECRET")) // must match server.secret_key

func webhook(w http.ResponseWriter, r *http.Request) {
  sig := r.Header.Get("X-Blnk-Signature")
  ts := r.Header.Get("X-Blnk-Timestamp")
  if sig == "" || ts == "" {
    http.Error(w, "missing headers", http.StatusBadRequest)
    return
  }

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

  if !hmac.Equal([]byte(sig), []byte(expected)) {
    http.Error(w, "invalid signature", http.StatusUnauthorized)
    return
  }
  w.WriteHeader(http.StatusOK)
}

func main() {
  http.HandleFunc("/webhook", webhook)
  http.ListenAndServe(":3000", nil)
}

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

app = Flask(__name__)
SECRET = os.environ.get("BLNK_SECRET").encode()  # must match server.secret_key

@app.route("/webhook", methods=["POST"])
def webhook():
    sig = request.headers.get("X-Blnk-Signature")
    ts = request.headers.get("X-Blnk-Timestamp")
    if not sig or not ts:
        abort(400)

    signed = f"{ts}.".encode() + request.get_data()
    expected = hmac.new(SECRET, signed, hashlib.sha256).hexdigest()

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

if __name__ == "__main__":
    app.run(port=3000)