Overview

import crypto from "node:crypto";
import express, { Request, Response } from "express";
import dotenv from "dotenv";

dotenv.config();

const WEBHOOK_SECRET = process.env.HYPERBROWSER_WEBHOOK_SECRET;
const app = express();

function verifySignature(req: Request): boolean {
  const timestamp = req.header("X-Webhook-Timestamp");
  const signature = req.header("X-Webhook-Signature");
  if (!timestamp || !signature || !req.body) return false;

  const signedPayload = `${timestamp}.${req.body}`;
  const expected = crypto
    .createHmac("sha256", WEBHOOK_SECRET)
    .update(signedPayload)
    .digest("hex");

  return crypto.timingSafeEqual(
    Buffer.from(signature, "hex"),
    Buffer.from(expected, "hex")
  );
}

app.post(
  "/webhooks/hyperbrowser",
  express.raw({ type: "application/json" }),
  (request: Request, response: Response) => {
    if (WEBHOOK_SECRET) {
      try {
        if (!verifySignature(request)) {
          console.log(`⚠️ Webhook signature verification failed.`);
          response.sendStatus(400);
          return;
        }
      } catch (err) {
        const message = err instanceof Error ? err.message : String(err);
        console.log(`⚠️ Webhook signature verification failed.`, message);
        response.sendStatus(400);
        return;
      }
    }

    let event;
    try {
      event = JSON.parse(request.body.toString("utf8"));
    } catch (err) {
      console.log(`⚠️ Webhook parsing failed:`, err);
      response.sendStatus(400);
      return;
    }

    const eventId = request.header("X-Webhook-Event-Id");

    console.log(`[${eventId}] ${event.type}`);

    switch (event.type) {
      case "session.downloads.updated":
        // TODO: Handle the session downloads updated event
        console.log("Session Downloads Updated:", event.data);
        break;
      case "session.tab.opened":
        // TODO: Handle the tab opened event
        console.log("Session tab opened:", event.data);
        break;
      case "session.captcha.detected":
        // TODO: Handle the captcha detected event
        console.log("Captcha detected:", event.data);
        break;
      case "session.captcha.solved":
        // TODO: Handle the captcha solved event
        console.log("Captcha solved:", event.data);
        break;
      case "session.closed":
        // TODO: Handle the session closed event
        console.log("Session closed:", event.data);
        break;
      default:
        console.warn("Unhandled event type:", event.type);
    }

    response.json({ received: true });
  }
);

app.listen(3000, () => {
  console.log("Listening for Hyperbrowser webhooks on port 3000");
});

import os
from dotenv import load_dotenv
import hmac
import hashlib
from fastapi import FastAPI, Header, HTTPException, Request, BackgroundTasks
import uvicorn

load_dotenv()
WEBHOOK_SECRET = os.getenv("HYPERBROWSER_WEBHOOK_SECRET")
app = FastAPI()


def verify_signature(raw_body: bytes, timestamp: str, signature: str) -> bool:
    signed_payload = f"{timestamp}.{raw_body.decode('utf-8')}"
    expected = hmac.new(
        WEBHOOK_SECRET.encode("utf-8"),
        signed_payload.encode("utf-8"),
        hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(expected, signature)


def handle_event(event: dict):
    event_type = event["type"]
    if event_type == "session.downloads.updated":
        print("Session Downloads Updated:", event["data"])
    elif event_type == "session.tab.opened":
        print("Session tab opened:", event["data"])
    elif event_type == "session.captcha.detected":
        print("Captcha detected:", event["data"])
    elif event_type == "session.captcha.solved":
        print("Captcha solved:", event["data"])
    elif event_type == "session.closed":
        print("Session closed:", event["data"])
    else:
        print("Unknown event:", event_type)


@app.post("/webhooks/hyperbrowser")
async def hyperbrowser_webhook(
    request: Request,
    background_tasks: BackgroundTasks,
    x_webhook_timestamp: str = Header(alias="X-Webhook-Timestamp"),
    x_webhook_signature: str = Header(alias="X-Webhook-Signature"),
    x_webhook_event_id: str = Header(alias="X-Webhook-Event-Id"),
):
    raw_body = await request.body()
    if not verify_signature(raw_body, x_webhook_timestamp, x_webhook_signature):
        raise HTTPException(status_code=400, detail="Invalid signature")

    event = await request.json()
    print(f"[{x_webhook_event_id}] {event['type']}")

    background_tasks.add_task(handle_event, event)
    return {"received": True}


if __name__ == "__main__":
    print("Starting Hyperbrowser webhook receiver server...")
    print(f"Webhook endpoint: http://localhost:3000/webhooks/hyperbrowser")
    uvicorn.run(app, host="0.0.0.0", port=3000)