Die Social-Media-API zum Posten, Planen und für Webhooks

class RobinreachWebhooksController < ActionController::API
  def create
    body = request.raw_post
    return head :unauthorized unless valid_signature?(body, request.headers["X-RobinReach-Signature"])

    event = JSON.parse(body)
    # Skip event["id"] if you have seen it, then queue the work.
    head :ok
  end

  private

  def valid_signature?(body, header, tolerance: 300)
    parts = header.to_s.split(",").map { |p| p.split("=", 2) }.select { |kv| kv.size == 2 }.to_h
    t = Integer(parts["t"].to_s, exception: false)
    return false unless t && parts["v1"] && (Time.now.to_i - t).abs <= tolerance

    expected = OpenSSL::HMAC.hexdigest("SHA256", ENV.fetch("ROBINREACH_WEBHOOK_SECRET"), "#{t}.#{body}")
    ActiveSupport::SecurityUtils.secure_compare(expected, parts["v1"])
  end
end

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

const app = express();

app.post("/webhooks/robinreach", express.raw({ type: "application/json" }), (req, res) => {
  const raw = req.body.toString("utf8");
  if (!verify(raw, req.get("X-RobinReach-Signature"), process.env.ROBINREACH_WEBHOOK_SECRET)) {
    return res.sendStatus(401);
  }
  const event = JSON.parse(raw);
  // Skip event.id if you have seen it, then queue the work.
  res.sendStatus(200);
});

function verify(rawBody, header, secret, toleranceSec = 300) {
  const parts = Object.fromEntries((header || "").split(",").map((p) => p.split("=", 2)));
  const t = Number(parts.t);
  if (!t || !parts.v1 || Math.abs(Date.now() / 1000 - t) > toleranceSec) return false;
  const expected = crypto.createHmac("sha256", secret).update(`${t}.${rawBody}`).digest("hex");
  const a = Buffer.from(expected), b = Buffer.from(parts.v1);
  return a.length === b.length && crypto.timingSafeEqual(a, b);
}

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

app = Flask(__name__)

@app.post("/webhooks/robinreach")
def robinreach_webhook():
    raw = request.get_data()
    if not verify(raw, request.headers.get("X-RobinReach-Signature", ""), os.environ["ROBINREACH_WEBHOOK_SECRET"]):
        abort(401)
    event = request.get_json()
    # Skip event["id"] if you have seen it, then queue the work.
    return "", 200

def verify(raw_body, header, secret, tolerance=300):
    parts = dict(p.split("=", 1) for p in header.split(",") if "=" in p)
    try:
        t = int(parts["t"])
    except (KeyError, ValueError):
        return False
    if "v1" not in parts or abs(time.time() - t) > tolerance:
        return False
    expected = hmac.new(secret.encode(), f"{t}.".encode() + raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, parts["v1"])