Webhook setup

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

const SECRET = process.env.SHIELDLABS_WEBHOOK_SECRET; // whsec_… for this endpoint
const app = express();

// Keep the raw body so the bytes you hash match the bytes that were signed.
app.use(express.json({ verify: (req, _res, buf) => { req.rawBody = buf; } }));

app.post("/webhooks/shieldlabs", (req, res) => {
  const received = req.get("X-Shield-Signature") ?? "";
  const expected =
    "sha256=" +
    crypto.createHmac("sha256", SECRET).update(req.rawBody).digest("hex");

  const a = Buffer.from(expected);
  const b = Buffer.from(received);
  if (a.length !== b.length || !crypto.timingSafeEqual(a, b)) {
    return res.sendStatus(401);
  }

  res.sendStatus(200);          // acknowledge fast
  if (req.body.event_type === "webhook.ping") return;
  handleScore(req.body.data);   // idempotent on request_id
});

package main

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

var secret = []byte(os.Getenv("SHIELDLABS_WEBHOOK_SECRET")) // whsec_… for this endpoint

func webhook(w http.ResponseWriter, r *http.Request) {
	body, err := io.ReadAll(r.Body)
	if err != nil {
		w.WriteHeader(http.StatusBadRequest)
		return
	}

	mac := hmac.New(sha256.New, secret)
	mac.Write(body) // HMAC over the raw request body
	expected := "sha256=" + hex.EncodeToString(mac.Sum(nil))

	if !hmac.Equal([]byte(expected), []byte(r.Header.Get("X-Shield-Signature"))) {
		w.WriteHeader(http.StatusUnauthorized)
		return
	}

	w.WriteHeader(http.StatusOK) // acknowledge fast

	var envelope struct {
		EventType string          `json:"event_type"`
		Data      json.RawMessage `json:"data"`
	}
	if err := json.Unmarshal(body, &envelope); err != nil || envelope.EventType == "webhook.ping" {
		return
	}
	handleScore(envelope.Data) // idempotent on request_id
}

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

SECRET = os.environ["SHIELDLABS_WEBHOOK_SECRET"].encode()  # whsec_… for this endpoint
app = Flask(__name__)

@app.post("/webhooks/shieldlabs")
def webhook():
    # Read the raw request body bytes exactly as received.
    raw = request.get_data()
    received = request.headers.get("X-Shield-Signature", "")
    expected = "sha256=" + hmac.new(SECRET, raw, hashlib.sha256).hexdigest()

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

    payload = request.get_json()
    if payload.get("event_type") == "webhook.ping":
        return "", 200

    handle_score(payload.get("data"))  # idempotent on request_id
    return "", 200