Receive events by webhook

curl -X POST https://api.sealtrust.io/v1/partner/webhooks \
  -H "Authorization: Bearer st_test_0000000000000000000000000000000000000000000000" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://exemple-sas.example/webhooks/sealtrust",
    "events": ["product.scanned", "transfer.accepted"],
    "secret": "secret-de-demonstration-a-remplacer"
  }'

import { SealTrustClient } from "@sealtrust-io/sdk";

const sealtrust = new SealTrustClient({
  apiKey: "st_test_0000000000000000000000000000000000000000000000",
});

const abonnement = await sealtrust.webhooks.create({
  url: "https://exemple-sas.example/webhooks/sealtrust",
  events: ["product.scanned", "transfer.accepted"],
  secret: "secret-de-demonstration-a-remplacer",
});

console.log(abonnement.id, abonnement.health);

import requests

reponse = requests.post(
    "https://api.sealtrust.io/v1/partner/webhooks",
    headers={
        "Authorization": "Bearer st_test_0000000000000000000000000000000000000000000000",
    },
    json={
        "url": "https://exemple-sas.example/webhooks/sealtrust",
        "events": ["product.scanned", "transfer.accepted"],
        "secret": "secret-de-demonstration-a-remplacer",
    },
    timeout=30,
)
reponse.raise_for_status()
print(reponse.json())

import crypto from "node:crypto";
import express from "express";

const SECRET = process.env.SEALTRUST_WEBHOOK_SECRET ?? "";
const TOLERANCE_SECONDES = 300;

const app = express();
const dejaVus = new Set<string>();

// express.raw leaves the body in the form of bytes. That is what is signed.
app.post(
  "/webhooks/sealtrust",
  express.raw({ type: "application/json" }),
  (req, res) => {
    const entete = req.get("X-Webhook-Signature") ?? "";
    const parties = new Map<string, string>();
    for (const morceau of entete.split(",")) {
      const separateur = morceau.indexOf("=");
      if (separateur > 0) {
        parties.set(
          morceau.slice(0, separateur).trim(),
          morceau.slice(separateur + 1).trim(),
        );
      }
    }

    const horodatage = parties.get("t");
    const recue = parties.get("v1");
    if (!horodatage || !recue) {
      return res.status(400).send("signature absente");
    }

    const age = Math.abs(Math.floor(Date.now() / 1000) - Number(horodatage));
    if (!Number.isFinite(age) || age > TOLERANCE_SECONDES) {
      return res.status(400).send("horodatage hors fenêtre");
    }

    const attendue = crypto
      .createHmac("sha256", SECRET)
      .update(`${horodatage}.`)
      .update(req.body as Buffer)
      .digest("hex");

    const a = Buffer.from(attendue, "hex");
    const b = Buffer.from(recue, "hex");
    if (a.length !== b.length || !crypto.timingSafeEqual(a, b)) {
      return res.status(400).send("signature invalide");
    }

    // Answer before handling: the delivery expires after 10 seconds.
    res.status(200).send("ok");

    // Deduplication on the pair event name + identifier.
    const evenement = req.get("X-Webhook-Event") ?? "";
    const identifiant = req.get("X-Webhook-Id") ?? "";
    const cle = `${evenement}:${identifiant}`;
    if (dejaVus.has(cle)) {
      return;
    }
    dejaVus.add(cle);

    const corps = JSON.parse((req.body as Buffer).toString("utf8"));
    console.log("événement reçu", evenement, corps);
  },
);

app.listen(8080);

import hashlib
import hmac
import os
import time

from flask import Flask, request

SECRET = os.environ["SEALTRUST_WEBHOOK_SECRET"].encode()
TOLERANCE_SECONDES = 300

app = Flask(__name__)
deja_vus: set[tuple[str, str]] = set()


@app.post("/webhooks/sealtrust")
def recevoir():
    entete = request.headers.get("X-Webhook-Signature", "")
    parties = {}
    for morceau in entete.split(","):
        if "=" in morceau:
            nom, valeur = morceau.split("=", 1)
            parties[nom.strip()] = valeur.strip()

    horodatage = parties.get("t")
    recue = parties.get("v1")
    if not horodatage or not recue:
        return "signature absente", 400

    try:
        age = abs(int(time.time()) - int(horodatage))
    except ValueError:
        return "horodatage illisible", 400
    if age > TOLERANCE_SECONDES:
        return "horodatage hors fenêtre", 400

    # get_data() returns the raw bytes. That is what is signed.
    corps = request.get_data()
    attendue = hmac.new(
        SECRET,
        horodatage.encode() + b"." + corps,
        hashlib.sha256,
    ).hexdigest()
    if not hmac.compare_digest(attendue, recue):
        return "signature invalide", 400

    # Deduplication on the pair event name + identifier.
    evenement = request.headers.get("X-Webhook-Event", "")
    identifiant = request.headers.get("X-Webhook-Id", "")
    cle = (evenement, identifiant)
    if cle in deja_vus:
        return "", 200
    deja_vus.add(cle)

    charge = request.get_json(force=True)
    print("événement reçu", evenement, charge)
    return "", 200


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

curl https://api.sealtrust.io/v1/partner/webhooks/1 \
  -H "Authorization: Bearer st_test_0000000000000000000000000000000000000000000000"

import { SealTrustClient } from "@sealtrust-io/sdk";

const sealtrust = new SealTrustClient({
  apiKey: "st_test_0000000000000000000000000000000000000000000000",
});

const abonnement = await sealtrust.webhooks.get(1);
console.log(abonnement.health, abonnement.is_active);

import requests

reponse = requests.get(
    "https://api.sealtrust.io/v1/partner/webhooks/1",
    headers={
        "Authorization": "Bearer st_test_0000000000000000000000000000000000000000000000",
    },
    timeout=30,
)
reponse.raise_for_status()
print(reponse.json()["health"], reponse.json()["is_active"])

curl -X PUT https://api.sealtrust.io/v1/partner/webhooks/1 \
  -H "Authorization: Bearer st_test_0000000000000000000000000000000000000000000000" \
  -H "Content-Type: application/json" \
  -d '{
    "events": ["product.scanned", "transfer.accepted", "clone.alert"],
    "secret": "nouveau-secret-de-demonstration"
  }'

import { SealTrustClient } from "@sealtrust-io/sdk";

const sealtrust = new SealTrustClient({
  apiKey: "st_test_0000000000000000000000000000000000000000000000",
});

const abonnement = await sealtrust.webhooks.update(1, {
  events: ["product.scanned", "transfer.accepted", "clone.alert"],
  secret: "nouveau-secret-de-demonstration",
});

console.log(abonnement.events);

import requests

reponse = requests.put(
    "https://api.sealtrust.io/v1/partner/webhooks/1",
    headers={
        "Authorization": "Bearer st_test_0000000000000000000000000000000000000000000000",
    },
    json={
        "events": ["product.scanned", "transfer.accepted", "clone.alert"],
        "secret": "nouveau-secret-de-demonstration",
    },
    timeout=30,
)
reponse.raise_for_status()
print(reponse.json()["events"])

curl -X DELETE "https://api.sealtrust.io/v1/partner/webhooks/1?confirm=https%3A%2F%2Fexemple-sas.example%2Fwebhooks%2Fsealtrust" \
  -H "Authorization: Bearer st_test_0000000000000000000000000000000000000000000000"

import { SealTrustClient } from "@sealtrust-io/sdk";

const sealtrust = new SealTrustClient({
  apiKey: "st_test_0000000000000000000000000000000000000000000000",
});

const abonnement = await sealtrust.webhooks.get(1);
await sealtrust.webhooks.delete(abonnement.id, abonnement.url);

import requests

entetes = {
    "Authorization": "Bearer st_test_0000000000000000000000000000000000000000000000",
}

lecture = requests.get(
    "https://api.sealtrust.io/v1/partner/webhooks/1",
    headers=entetes,
    timeout=30,
)
lecture.raise_for_status()

suppression = requests.delete(
    "https://api.sealtrust.io/v1/partner/webhooks/1",
    headers=entetes,
    params={"confirm": lecture.json()["url"]},
    timeout=30,
)
suppression.raise_for_status()