Webhooks

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

const app = express();
const SECRET = process.env.DAYDREAM_WEBHOOK_SECRET!;
const TOLERANCE_SECONDS = 300;

function verify(rawBody: Buffer, header: string): boolean {
  const parts = Object.fromEntries(
    header.split(",").map(kv => kv.split("=") as [string, string]),
  );
  const timestamp = Number(parts.t);
  if (!timestamp || Math.abs(Date.now() / 1000 - timestamp) > TOLERANCE_SECONDS) {
    return false;
  }

  const expected = crypto
    .createHmac("sha256", SECRET)
    .update(`${timestamp}.${rawBody.toString("utf8")}`)
    .digest("hex");

  const a = Buffer.from(expected, "hex");
  const b = Buffer.from(parts.v1 ?? "", "hex");
  return a.length === b.length && crypto.timingSafeEqual(a, b);
}

// express.raw, not express.json — the signature covers the raw bytes.
app.post(
  "/webhooks/daydream",
  express.raw({ type: "application/json" }),
  (req, res) => {
    if (!verify(req.body, req.header("Daydream-Webhook-Signature") ?? "")) {
      return res.status(401).send("bad signature");
    }

    const event = JSON.parse(req.body.toString("utf8"));
    if (event.type === "stream.ended") {
      console.log(event.data.stream_id, "stopped at", event.data.ended_at);
    }

    // Acknowledge first, work later.
    res.sendStatus(200);
  },
);

import hashlib
import hmac
import os
import time

from flask import Flask, request

app = Flask(__name__)
SECRET = os.environ["DAYDREAM_WEBHOOK_SECRET"]
TOLERANCE_SECONDS = 300


def verify(raw_body: bytes, header: str) -> bool:
    parts = dict(kv.split("=", 1) for kv in header.split(","))
    try:
        timestamp = int(parts["t"])
    except (KeyError, ValueError):
        return False

    if abs(time.time() - timestamp) > TOLERANCE_SECONDS:
        return False

    expected = hmac.new(
        SECRET.encode(),
        f"{timestamp}.".encode() + raw_body,
        hashlib.sha256,
    ).hexdigest()

    return hmac.compare_digest(expected, parts.get("v1", ""))


@app.post("/webhooks/daydream")
def handle():
    # request.data is the raw body; do not use request.json here.
    if not verify(request.data, request.headers.get("Daydream-Webhook-Signature", "")):
        return "bad signature", 401

    event = request.get_json()
    if event["type"] == "stream.ended":
        print(event["data"]["stream_id"], "stopped at", event["data"]["ended_at"])

    return "", 200