Webhooks

const crypto = require("crypto");

const SIGNATURE_TOLERANCE_SECONDS = 5 * 60;

function verifyWebhookSignature(rawBody, header, secret) {
  if (!header) return false;

  const parts = {};
  for (const part of header.split(",")) {
    const [k, v] = part.split("=");
    if (k && v) parts[k.trim()] = v.trim();
  }

  const timestamp = parts.t;
  const provided = parts.v1;
  if (!timestamp || !provided || !/^[a-f0-9]{64}$/i.test(provided)) {
    return false;
  }

  const age = Math.abs(Math.floor(Date.now() / 1000) - Number(timestamp));
  if (!Number.isFinite(age) || age > SIGNATURE_TOLERANCE_SECONDS) {
    return false;
  }

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

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

import hmac
import hashlib
import re
import time

SIGNATURE_TOLERANCE_SECONDS = 5 * 60
_HEX64 = re.compile(r"^[a-f0-9]{64}$", re.IGNORECASE)

def verify_webhook_signature(raw_body: bytes, header: str, secret: str) -> bool:
    if not header:
        return False

    parts = {}
    for part in header.split(","):
        key, _, value = part.partition("=")
        if key and value:
            parts[key.strip()] = value.strip()

    timestamp = parts.get("t")
    provided = parts.get("v1")
    if not timestamp or not provided or not _HEX64.match(provided):
        return False

    try:
        age = abs(int(time.time()) - int(timestamp))
    except ValueError:
        return False
    if age > SIGNATURE_TOLERANCE_SECONDS:
        return False

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

    return hmac.compare_digest(expected, provided)

const express = require("express");
const app = express();

const SECRET = process.env.MONITORING_WEBHOOK_SECRET;

// Capture the raw body so signature verification matches exactly what was signed
app.post(
  "/webhooks/monitoring",
  express.raw({ type: "application/json" }),
  (req, res) => {
    const signature = req.header("X-Webhook-Signature") ?? "";
    if (!verifyWebhookSignature(req.body, signature, SECRET)) {
      return res.status(401).send("invalid signature");
    }

    const envelope = JSON.parse(req.body.toString("utf8"));

    // Ack immediately, then process in the background
    res.status(200).send("ok");
    setImmediate(() => forwardToLinear(envelope).catch(console.error));
  }
);

app.listen(3000);

import os
import threading
from flask import Flask, request

app = Flask(__name__)
SECRET = os.environ["MONITORING_WEBHOOK_SECRET"]

@app.post("/webhooks/monitoring")
def handle():
    raw_body = request.get_data()  # raw bytes, not request.json
    signature = request.headers.get("X-Webhook-Signature", "")
    if not verify_webhook_signature(raw_body, signature, SECRET):
        return "invalid signature", 401

    envelope = request.get_json()

    # Ack immediately, then process in the background
    threading.Thread(
        target=forward_to_linear, args=(envelope,), daemon=True
    ).start()
    return "ok", 200

const LINEAR_API_KEY = process.env.LINEAR_API_KEY;
const LINEAR_TEAM_ID = process.env.LINEAR_TEAM_ID;

// Linear priority enum: 0 none, 1 urgent, 2 high, 3 medium, 4 low
const PRIORITY = { high: 2, medium: 3, low: 4 };

async function forwardToLinear(envelope) {
  if (envelope.type !== "monitoring_event") return;

  const { view, latest_development: dev, assets = [] } = envelope.data;
  const assetList = assets.map((a) => a.name).join(", ") || "none";

  const input = {
    teamId: LINEAR_TEAM_ID,
    title: `[${view.name}] ${dev.title}`,
    description: [
      `**View:** ${view.name}`,
      `**Assets:** ${assetList}`,
      `**Detected:** ${dev.created_at}`,
      "",
      dev.summary,
    ].join("\n"),
    priority: PRIORITY[dev.importance] ?? 0,
  };

  const res = await fetch("https://api.linear.app/graphql", {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      Authorization: LINEAR_API_KEY,
    },
    body: JSON.stringify({
      query: `mutation($input: IssueCreateInput!) {
        issueCreate(input: $input) { success issue { identifier url } }
      }`,
      variables: { input },
    }),
  });

  if (!res.ok) throw new Error(`Linear ${res.status}: ${await res.text()}`);
}

import os
import requests

LINEAR_API_KEY = os.environ["LINEAR_API_KEY"]
LINEAR_TEAM_ID = os.environ["LINEAR_TEAM_ID"]

# Linear priority enum: 0 none, 1 urgent, 2 high, 3 medium, 4 low
PRIORITY = {"high": 2, "medium": 3, "low": 4}

def forward_to_linear(envelope):
    if envelope.get("type") != "monitoring_event":
        return

    data = envelope["data"]
    view = data["view"]
    dev = data["latest_development"]
    assets = data.get("assets", [])

    asset_list = ", ".join(a["name"] for a in assets) or "none"
    description = "\n".join([
        f"**View:** {view['name']}",
        f"**Assets:** {asset_list}",
        f"**Detected:** {dev['created_at']}",
        "",
        dev["summary"],
    ])

    input_ = {
        "teamId": LINEAR_TEAM_ID,
        "title": f"[{view['name']}] {dev['title']}",
        "description": description,
        "priority": PRIORITY.get(dev.get("importance"), 0),
    }

    res = requests.post(
        "https://api.linear.app/graphql",
        headers={
            "Content-Type": "application/json",
            "Authorization": LINEAR_API_KEY,
        },
        json={
            "query": """
              mutation($input: IssueCreateInput!) {
                issueCreate(input: $input) { success issue { identifier url } }
              }
            """,
            "variables": {"input": input_},
        },
        timeout=10,
    )
    res.raise_for_status()