Webhooks

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

// The current secret, and during a rollover the old one too.
const secrets = [
  process.env.SEVALLA_WEBHOOK_SECRET,
  process.env.SEVALLA_WEBHOOK_SECRET_OLD,
].filter(Boolean);

function isFromSevalla(header) {
  if (typeof header !== "string") return false;
  const received = Buffer.from(header);
  return secrets.some((secret) => {
    const expected = Buffer.from(secret);
    return expected.length === received.length && timingSafeEqual(expected, received);
  });
}

const app = express();

app.post("/sevalla", express.json(), (req, res) => {
  if (!isFromSevalla(req.get("svl-signature"))) {
    return res.sendStatus(401);
  }
  res.sendStatus(200); // Answer first, then do the work.

  const { event_id, type, data } = req.body;
  if (type === "APP_DEPLOY" && data.status === "failed") {
    console.log(`Deployment ${data.deployment_id} of ${data.app_display_name} failed (event ${event_id})`);
  }
});

app.listen(process.env.PORT || 8080);

import hmac
import os

from flask import Flask, abort, request

# The current secret, and during a rollover the old one too.
SECRETS = [
    s
    for s in (
        os.environ.get("SEVALLA_WEBHOOK_SECRET"),
        os.environ.get("SEVALLA_WEBHOOK_SECRET_OLD"),
    )
    if s
]

app = Flask(__name__)


def is_from_sevalla(header):
    if header is None:
        return False
    return any(hmac.compare_digest(header.encode(), s.encode()) for s in SECRETS)


@app.post("/sevalla")
def sevalla_webhook():
    if not is_from_sevalla(request.headers.get("svl-signature")):
        abort(401)

    event = request.get_json()
    if event["type"] == "APP_DEPLOY" and event["data"]["status"] == "failed":
        print(f"Deployment {event['data']['deployment_id']} failed (event {event['event_id']})")
    return "", 200