Documentation

{
  "id": "jd7abc123def456",
  "type": "youtube.video.public_published",
  "occurredAt": "2026-05-10T14:30:00.000Z",
  "data": {
    "video": {
      "id": "dQw4w9WgXcQ",
      "title": "New video title",
      "url": "https://www.youtube.com/watch?v=dQw4w9WgXcQ",
      "thumbnailUrl": "https://i.ytimg.com/vi/dQw4w9WgXcQ/hqdefault.jpg",
      "publishedAt": "2026-05-10T14:28:00.000Z",
      "updatedAt": "2026-05-10T14:30:00.000Z"
    },
    "channel": {
      "id": "UCBcRF18a7Qf58cCRy5xuWwQ",
      "title": "Channel Name",
      "handle": "@channelhandle",
      "url": "https://www.youtube.com/@channelhandle",
      "avatarUrl": "https://yt3.ggpht.com/..."
    }
  }
}

import { createHmac, timingSafeEqual } from "node:crypto";

const SIGNATURE_TOLERANCE_SECONDS = 5 * 60;
const SIGNATURE_PATTERN = /^t=(\d+),v2=([0-9a-f]{64})$/;

function verifyTubeHookSignature(
  rawBody: string,
  headers: Record<string, string | string[] | undefined>,
  secret: string,
): boolean {
  const signature = headers["tubehook-signature-v2"];
  const deliveryId = headers["tubehook-delivery-id"];
  const eventId = headers["tubehook-event-id"];
  const eventType = headers["tubehook-event-type"];
  if (
    typeof signature !== "string" ||
    typeof deliveryId !== "string" ||
    typeof eventId !== "string" ||
    typeof eventType !== "string"
  ) return false;

  const match = SIGNATURE_PATTERN.exec(signature);
  if (!match) return false;
  const [, timestamp, providedDigest] = match;
  const signedContent = [
    "tubehook-webhook-v2",
    timestamp,
    deliveryId,
    eventId,
    eventType,
    rawBody,
  ].join("\n");
  const expectedDigest = createHmac("sha256", secret)
    .update(signedContent)
    .digest("hex");
  const expected = Buffer.from(expectedDigest);
  const provided = Buffer.from(providedDigest);
  if (
    expected.length !== provided.length ||
    !timingSafeEqual(expected, provided)
  ) return false;

  const age = Math.abs(Math.floor(Date.now() / 1000) - Number(timestamp));
  return age <= SIGNATURE_TOLERANCE_SECONDS;
}

// Express example: keep the body raw until after verification.
app.post("/webhook", express.text({ type: "application/json" }), (req, res) => {
  if (!verifyTubeHookSignature(req.body, req.headers, process.env.TUBEHOOK_SECRET!)) {
    return res.status(401).send("Invalid signature");
  }
  const event = JSON.parse(req.body);
  // handle event.type ...
  res.sendStatus(200);
});

import hashlib
import hmac
import re
import time

SIGNATURE_TOLERANCE_SECONDS = 5 * 60
SIGNATURE_PATTERN = re.compile(r"^t=(\d+),v2=([0-9a-f]{64})$")

def verify_tubehook_signature(raw_body: bytes, headers, secret: str) -> bool:
    signature = headers.get("tubehook-signature-v2", "")
    delivery_id = headers.get("tubehook-delivery-id", "")
    event_id = headers.get("tubehook-event-id", "")
    event_type = headers.get("tubehook-event-type", "")
    match = SIGNATURE_PATTERN.fullmatch(signature)
    if not match or not delivery_id or not event_id or not event_type:
        return False

    timestamp, provided_digest = match.groups()
    signed_content = b"\n".join([
        b"tubehook-webhook-v2",
        timestamp.encode(),
        delivery_id.encode(),
        event_id.encode(),
        event_type.encode(),
        raw_body,
    ])
    expected_digest = hmac.new(
        secret.encode(),
        signed_content,
        hashlib.sha256,
    ).hexdigest()
    if not hmac.compare_digest(expected_digest, provided_digest):
        return False

    return abs(int(time.time()) - int(timestamp)) <= SIGNATURE_TOLERANCE_SECONDS

# Flask example: keep the body raw until after verification.
@app.route("/webhook", methods=["POST"])
def webhook():
    raw_body = request.get_data()
    if not verify_tubehook_signature(raw_body, request.headers, TUBEHOOK_SECRET):
        return "Invalid signature", 401
    event = request.get_json(force=True)
    # handle event["type"] ...
    return "", 200