A webhook you have not verified is a public endpoint that anyone can call.

import { Notix, WebhookVerificationError } from "notix-js";

const notix = new Notix(process.env.NOTIX_API_KEY!);
const webhooks = notix.webhooks(process.env.NOTIX_WEBHOOK_SECRET!);

export async function POST(request: Request) {
  // request.text(), never request.json(): the signature covers the exact bytes.
  const rawBody = await request.text();

  let event;
  try {
    event = webhooks.constructEvent(rawBody, { headers: request.headers });
  } catch (error) {
    if (error instanceof WebhookVerificationError) {
      return new Response(error.code, { status: 401 });
    }
    throw error;
  }

  // Narrowing on type narrows data with it.
  if (event.type === "email.bounced") {
    await markInvalid(event.data.to, event.id);
  }

  return new Response("ok", { status: 200 });
}

import { createHmac, timingSafeEqual } from "node:crypto";
import express from "express";

const TOLERANCE_MS = 5 * 60 * 1000;

function verify(secret: string, rawBody: Buffer, headers: express.Request["headers"]) {
  const signature = String(headers["x-notix-signature"] ?? "");
  const timestamp = String(headers["x-notix-timestamp"] ?? "");
  if (!signature.startsWith("v1=") || !/^\d+$/.test(timestamp)) return false;
  if (Math.abs(Date.now() - Number(timestamp)) > TOLERANCE_MS) return false;

  const expected =
    "v1=" + createHmac("sha256", secret).update(`${timestamp}.`).update(rawBody).digest("hex");
  const a = Buffer.from(expected);
  const b = Buffer.from(signature);
  return a.length === b.length && timingSafeEqual(a, b);
}

const app = express();
// Raw bytes for this route only; express.json() elsewhere would re-serialise them.
app.post("/webhooks/notix", express.raw({ type: "application/json" }), (req, res) => {
  if (!verify(process.env.NOTIX_WEBHOOK_SECRET!, req.body, req.headers)) {
    return res.status(401).send("invalid signature");
  }
  const event = JSON.parse(req.body.toString("utf8"));
  res.status(200).send("ok");
  queue.add(event); // process after answering
});

from flask import Flask, request
from notix import Notix
from notix.webhooks import WebhookVerificationError

app = Flask(__name__)
notix = Notix(os.environ["NOTIX_API_KEY"])
webhooks = notix.webhooks(os.environ["NOTIX_WEBHOOK_SECRET"])


@app.post("/webhooks/notix")
def notix_webhook():
    try:
        # request.data is the raw bytes; request.get_json() would not be.
        event = webhooks.construct_event(request.data, headers=request.headers)
    except WebhookVerificationError as error:
        return error.code, 401

    if event["type"] == "email.bounced":
        mark_invalid(event["data"]["to"], event["id"])

    return "ok", 200


# The same check without the package, for Django or anything else:
import hashlib, hmac, time

def verify(secret: str, raw_body: bytes, signature: str, timestamp: str) -> bool:
    if not signature.startswith("v1=") or not timestamp.isdigit():
        return False
    if abs(time.time() * 1000 - int(timestamp)) > 5 * 60 * 1000:
        return False
    digest = hmac.new(secret.encode(), timestamp.encode() + b"." + raw_body, hashlib.sha256)
    return hmac.compare_digest("v1=" + digest.hexdigest(), signature)

package main

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

const toleranceMs = 5 * 60 * 1000

func verify(secret string, rawBody []byte, signature, timestamp string) bool {
    if !strings.HasPrefix(signature, "v1=") {
        return false
    }
    ts, err := strconv.ParseInt(timestamp, 10, 64)
    if err != nil || math.Abs(float64(time.Now().UnixMilli()-ts)) > toleranceMs {
        return false
    }
    mac := hmac.New(sha256.New, []byte(secret))
    mac.Write([]byte(timestamp + "."))
    mac.Write(rawBody)
    expected := "v1=" + hex.EncodeToString(mac.Sum(nil))
    return hmac.Equal([]byte(expected), []byte(signature))
}

func handler(w http.ResponseWriter, r *http.Request) {
    rawBody, err := io.ReadAll(r.Body)
    if err != nil {
        http.Error(w, "unreadable body", http.StatusBadRequest)
        return
    }
    if !verify(os.Getenv("NOTIX_WEBHOOK_SECRET"), rawBody,
        r.Header.Get("X-Notix-Signature"), r.Header.Get("X-Notix-Timestamp")) {
        http.Error(w, "invalid signature", http.StatusUnauthorized)
        return
    }
    w.WriteHeader(http.StatusOK) // answer first, then parse rawBody and do the work
}

use Illuminate\Http\Request;
use Notix\Exception\SignatureVerificationException;
use Notix\Webhooks;

Route::post('/webhooks/notix', function (Request $request) {
    $webhooks = new Webhooks(config('services.notix.webhook_secret'));

    try {
        // getContent() is the raw body; $request->all() is not.
        $event = $webhooks->constructEvent($request->getContent(), $request->headers->all());
    } catch (SignatureVerificationException $e) {
        return response($e->getMessage(), 401);
    }

    if ($event['type'] === 'email.bounced') {
        MarkInvalid::dispatch($event['data']['to'], $event['id']);
    }

    return response('ok', 200);
})->withoutMiddleware([\App\Http\Middleware\VerifyCsrfToken::class]);

// The same check without the package:
function verifyNotix(string $secret, string $rawBody, string $signature, string $timestamp): bool
{
    if (!str_starts_with($signature, 'v1=') || !ctype_digit($timestamp)) {
        return false;
    }
    if (abs((int) round(microtime(true) * 1000) - (int) $timestamp) > 5 * 60 * 1000) {
        return false;
    }
    $expected = 'v1=' . hash_hmac('sha256', $timestamp . '.' . $rawBody, $secret);
    return hash_equals($expected, $signature);
}