Webhooks و Slack

import crypto from 'node:crypto';
import express from 'express';

const secret = process.env.LERIX_WEBHOOK_SECRET!; // whsec_...

function verifyLerixSignature(rawBody: string, header: string, toleranceSeconds = 300): boolean {
  const parts = Object.fromEntries(
    header.split(',').map((part) => {
      const i = part.indexOf('=');
      return [part.slice(0, i).trim(), part.slice(i + 1).trim()];
    }),
  );
  const t = Number(parts.t);
  if (!Number.isInteger(t) || Math.abs(Date.now() / 1000 - t) > toleranceSeconds) return false;

  const expected = crypto.createHmac('sha256', secret).update(`${t}.${rawBody}`).digest('hex');
  const a = Buffer.from(parts.v1 ?? '', 'hex');
  const b = Buffer.from(expected, 'hex');
  return a.length === b.length && crypto.timingSafeEqual(a, b);
}

const app = express();

app.post('/lerix/webhook', express.raw({ type: 'application/json' }), (req, res) => {
  const rawBody = req.body.toString('utf8');
  if (!verifyLerixSignature(rawBody, req.header('Lerix-Signature') ?? '')) {
    return res.status(400).send('invalid signature');
  }
  const event = JSON.parse(rawBody);
  // Use req.header('Lerix-Delivery') to skip deliveries you've already handled.
  console.log(event.type, event.data);
  res.sendStatus(200);
});

import hashlib
import hmac
import json
import os
import time

from flask import Flask, abort, request

SECRET = os.environ["LERIX_WEBHOOK_SECRET"]  # whsec_...


def verify_lerix_signature(raw_body: bytes, header: str, tolerance_seconds: int = 300) -> bool:
    parts = dict(part.strip().split("=", 1) for part in header.split(",") if "=" in part)
    try:
        t = int(parts.get("t", ""))
    except ValueError:
        return False
    if abs(time.time() - t) > tolerance_seconds:
        return False
    signed = f"{t}.".encode() + raw_body
    expected = hmac.new(SECRET.encode(), signed, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, parts.get("v1", ""))


app = Flask(__name__)


@app.post("/lerix/webhook")
def lerix_webhook():
    raw_body = request.get_data()  # the raw bytes, before any JSON parsing
    if not verify_lerix_signature(raw_body, request.headers.get("Lerix-Signature", "")):
        abort(400)
    event = json.loads(raw_body)
    # Use request.headers["Lerix-Delivery"] to skip deliveries you've already handled.
    print(event["type"], event["data"])
    return "", 200