diff --git a/migrations/0063_orb_webhook_events.sql b/migrations/0063_orb_webhook_events.sql new file mode 100644 index 0000000000..fbd389a50b --- /dev/null +++ b/migrations/0063_orb_webhook_events.sql @@ -0,0 +1,20 @@ +-- Gittensory Orb central GitHub App (#1255) — webhook delivery dedup + audit for POST /v1/orb/webhook. +-- The central Orb App is a SEPARATE GitHub App from the review app, with its OWN webhook secret and its OWN +-- delivery IDs, so it gets its OWN dedup table (not webhook_events) — a GitHub delivery_id is only unique per +-- App, so sharing one table across two Apps could collide. This receiver just verifies + records (PR1); +-- install-registry + PR-outcome processing land in later PRs. +CREATE TABLE IF NOT EXISTS orb_webhook_events ( + delivery_id TEXT PRIMARY KEY NOT NULL, + event_name TEXT NOT NULL, + action TEXT, + installation_id INTEGER, + repository_full_name TEXT, + payload_hash TEXT NOT NULL, + -- 'received' (recorded, not yet processed) | 'processed' | 'error'. Processing is added in a later PR. + status TEXT NOT NULL DEFAULT 'received', + received_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + processed_at TEXT +); + +CREATE INDEX IF NOT EXISTS orb_webhook_events_status_idx ON orb_webhook_events(status); +CREATE INDEX IF NOT EXISTS orb_webhook_events_installation_idx ON orb_webhook_events(installation_id); diff --git a/src/api/routes.ts b/src/api/routes.ts index f20d366a59..c0177844de 100644 --- a/src/api/routes.ts +++ b/src/api/routes.ts @@ -123,6 +123,7 @@ import { } from "../github/commands"; import { handleGitHubWebhook } from "../github/webhook"; import { handleOrbIngest, readOrbIngestBody } from "../orb/ingest"; +import { handleOrbWebhook } from "../orb/webhook"; import { computeFleetAnalytics } from "../orb/analytics"; import { handleMcpRequest } from "../mcp/server"; import { buildOpenApiSpec } from "../openapi/spec"; @@ -2864,6 +2865,11 @@ export function createApp() { app.post("/v1/github/webhook", handleGitHubWebhook); + // Gittensory Orb central GitHub App (#1255) — inbound webhook for the ONE shared Orb App maintainers install. + // Verifies the Orb App's OWN webhook secret, dedups, and records install + PR/review events (the homepage + // fleet-metrics data spine). Separate App + secret from the review-app /v1/github/webhook above. + app.post("/v1/orb/webhook", handleOrbWebhook); + // Gittensory Orb (#1255) — central fleet-calibration collector. Receives anonymized, reversal-aware // outcome batches from self-hosted instances. No auth required: all data is HMAC-anonymized by the sender; // dedup is enforced via UNIQUE(instance_id, repo_hash, pr_hash) in orb_signals. Rate-limited (strict, #1254). @@ -4866,6 +4872,7 @@ function requiresApiToken(path: string): boolean { if (path === "/v1/drafts" || path.startsWith("/v1/drafts/")) return false; if (path.startsWith("/v1/auth/")) return false; if (path === "/v1/github/webhook") return false; + if (path === "/v1/orb/webhook") return false; if (path === "/v1/orb/ingest") return false; if (path.startsWith("/v1/internal/")) return false; return path.startsWith("/v1/"); diff --git a/src/auth/rate-limit.ts b/src/auth/rate-limit.ts index 79dd046a29..49949ecc77 100644 --- a/src/auth/rate-limit.ts +++ b/src/auth/rate-limit.ts @@ -96,6 +96,9 @@ export async function enforceRateLimit(c: Context<{ Bindings: Env }>, routeClass export function routeClassForPath(path: string): RateLimitClass { if (path === "/v1/github/webhook") return "strict"; + // Orb central-App inbound webhook — same class as the review-app webhook above (GitHub delivers from a + // narrow IP range; the per-IP strict cap is proven for /v1/github/webhook and #1292 reserves headroom). + if (path === "/v1/orb/webhook") return "strict"; // Orb telemetry ingest: unauthenticated + write, accepting anonymized batches from untrusted // self-host instances. Strict (10/min per IP) caps abuse — legitimate instances export hourly. if (path === "/v1/orb/ingest") return "strict"; diff --git a/src/env.d.ts b/src/env.d.ts index 7d898869ef..64358ae912 100644 --- a/src/env.d.ts +++ b/src/env.d.ts @@ -38,6 +38,9 @@ declare global { ADMIN_GITHUB_LOGINS?: string; GITHUB_WEBHOOK_SECRET: string; GITHUB_WEBHOOK_MAX_BODY_BYTES?: string; + /** Webhook secret for the central Gittensory Orb GitHub App (#1255) — distinct from the review app's + * GITHUB_WEBHOOK_SECRET. Verifies inbound POST /v1/orb/webhook deliveries. Inject as a wrangler secret. */ + ORB_GITHUB_WEBHOOK_SECRET?: string; GITHUB_APP_PRIVATE_KEY: string; GITHUB_APP_ID: string; GITHUB_APP_SLUG: string; diff --git a/src/orb/webhook.ts b/src/orb/webhook.ts new file mode 100644 index 0000000000..b41866e057 --- /dev/null +++ b/src/orb/webhook.ts @@ -0,0 +1,114 @@ +// Gittensory Orb central GitHub App (#1255) — inbound webhook receiver (POST /v1/orb/webhook). +// +// The central Orb App is a SEPARATE GitHub App that maintainers INSTALL (one shared app, like +// das-github-mirror's). GitHub delivers its install + PR/review events here, to gittensory-api. This is the +// data spine for the homepage fleet metrics (reviews initiated / merged / closed / reversals). +// +// PR1 scope: receive + verify (the Orb App's OWN webhook secret) + dedup + record. NO processing yet — the +// install registry and PR-outcome aggregation land in later PRs, reading from orb_webhook_events. This mirrors +// the proven src/github/webhook.ts handler verbatim; only the secret + dedup table differ. +import type { Context } from "hono"; +import type { GitHubWebhookPayload } from "../types"; +import { sha256Hex, verifyGitHubSignature } from "../utils/crypto"; + +const DEFAULT_MAX_ORB_WEBHOOK_BODY_BYTES = 1024 * 1024; + +export async function handleOrbWebhook(c: Context<{ Bindings: Env }>): Promise { + const deliveryId = c.req.header("x-github-delivery") ?? null; + const eventName = c.req.header("x-github-event") ?? null; + const signature = c.req.header("x-hub-signature-256") ?? null; + if (!deliveryId || !eventName) { + return c.json({ error: "missing_github_headers" }, 400); + } + + const maxBodyBytes = parsePositiveInt(c.env.GITHUB_WEBHOOK_MAX_BODY_BYTES) ?? DEFAULT_MAX_ORB_WEBHOOK_BODY_BYTES; + const contentLength = parsePositiveInt(c.req.header("content-length")); + if (contentLength !== null && contentLength > maxBodyBytes) { + return c.json({ error: "payload_too_large", maxBytes: maxBodyBytes }, 413); + } + + const rawBody = await readBodyWithLimit(c.req.raw, maxBodyBytes); + if (rawBody === null) { + return c.json({ error: "payload_too_large", maxBytes: maxBodyBytes }, 413); + } + // The Orb App's OWN webhook secret — distinct from the review app's GITHUB_WEBHOOK_SECRET. Absent secret → + // verifyGitHubSignature returns false → 401 (fail-closed), so this route is inert until the secret is injected. + const verified = await verifyGitHubSignature(rawBody, signature, c.env.ORB_GITHUB_WEBHOOK_SECRET ?? ""); + if (!verified) { + return c.json({ error: "invalid_signature" }, 401); + } + + let payload: GitHubWebhookPayload; + try { + payload = JSON.parse(rawBody) as GitHubWebhookPayload; + } catch { + return c.json({ error: "invalid_json" }, 400); + } + + const payloadHash = await sha256Hex(rawBody); + const existing = await getOrbWebhookEvent(c.env, deliveryId); + // Suppress redelivery of an already-recorded delivery (same payload) or a processed one; "error" rows are + // never suppressed so a failed record can be retried — same semantics as the review-app handler (#789). + if (existing && existing.status !== "error" && (existing.status === "processed" || existing.payloadHash === payloadHash)) { + return c.json({ ok: true, deliveryId, eventName, status: "duplicate" }, 202); + } + + await recordOrbWebhookEvent(c.env, { + deliveryId, + eventName, + action: payload.action ?? null, + installationId: payload.installation?.id ?? null, + repositoryFullName: payload.repository?.full_name ?? null, + payloadHash, + }); + + return c.json({ ok: true, deliveryId, eventName, status: "received" }, 202); +} + +async function getOrbWebhookEvent(env: Env, deliveryId: string): Promise<{ payloadHash: string; status: string } | null> { + const row = await env.DB.prepare("SELECT payload_hash AS payloadHash, status FROM orb_webhook_events WHERE delivery_id = ?") + .bind(deliveryId) + .first<{ payloadHash: string; status: string }>(); + return row ?? null; +} + +async function recordOrbWebhookEvent( + env: Env, + e: { deliveryId: string; eventName: string; action: string | null; installationId: number | null; repositoryFullName: string | null; payloadHash: string }, +): Promise { + await env.DB.prepare( + `INSERT INTO orb_webhook_events (delivery_id, event_name, action, installation_id, repository_full_name, payload_hash, status) + VALUES (?, ?, ?, ?, ?, ?, 'received') + ON CONFLICT(delivery_id) DO UPDATE SET + status = 'received', payload_hash = excluded.payload_hash, action = excluded.action, + installation_id = excluded.installation_id, repository_full_name = excluded.repository_full_name`, + ) + .bind(e.deliveryId, e.eventName, e.action, e.installationId, e.repositoryFullName, e.payloadHash) + .run(); +} + +function parsePositiveInt(value: string | null | undefined): number | null { + if (!value) return null; + const parsed = Number.parseInt(value, 10); + if (!Number.isFinite(parsed) || parsed <= 0) return null; + return parsed; +} + +async function readBodyWithLimit(request: Request, maxBytes: number): Promise { + const stream = request.body; + if (!stream) return ""; + const reader = stream.getReader(); + const decoder = new TextDecoder(); + const chunks: string[] = []; + let total = 0; + while (true) { + const { done, value } = await reader.read(); + if (done) break; + if (!value) continue; + total += value.byteLength; + if (total > maxBytes) return null; + chunks.push(decoder.decode(value, { stream: true })); + } + chunks.push(decoder.decode()); + return chunks.join(""); +} diff --git a/test/integration/orb-webhook.test.ts b/test/integration/orb-webhook.test.ts new file mode 100644 index 0000000000..f1091852c3 --- /dev/null +++ b/test/integration/orb-webhook.test.ts @@ -0,0 +1,161 @@ +import type { Context } from "hono"; +import { describe, expect, it } from "vitest"; +import { createApp } from "../../src/api/routes"; +import { handleOrbWebhook } from "../../src/orb/webhook"; +import { createTestEnv, type TestD1Database } from "../helpers/d1"; + +const SECRET = "orb-test-secret"; +const env = (over: Record = {}): Env => createTestEnv({ ORB_GITHUB_WEBHOOK_SECRET: SECRET, ...over }); + +async function sign(body: string, secret: string): Promise { + const key = await crypto.subtle.importKey("raw", new TextEncoder().encode(secret), { name: "HMAC", hash: "SHA-256" }, false, ["sign"]); + const signed = await crypto.subtle.sign("HMAC", key, new TextEncoder().encode(body)); + return `sha256=${[...new Uint8Array(signed)].map((b) => b.toString(16).padStart(2, "0")).join("")}`; +} + +function ctx(e: Env, headers: Record, request: Request): Context<{ Bindings: Env }> { + return { + req: { raw: request, header: (n: string) => headers[n.toLowerCase()] ?? null }, + env: e, + json: (payload: unknown, status?: number) => Response.json(payload, status === undefined ? undefined : { status }), + } as unknown as Context<{ Bindings: Env }>; +} + +async function post( + e: Env, + body: string, + opts: { delivery?: string | null; event?: string | null; sig?: string; headers?: Record; request?: Request } = {}, +): Promise { + const headers: Record = { + "x-github-delivery": opts.delivery === undefined ? "d1" : opts.delivery, + "x-github-event": opts.event === undefined ? "installation" : opts.event, + "x-hub-signature-256": opts.sig ?? (await sign(body, SECRET)), + ...opts.headers, + }; + const request = opts.request ?? new Request("https://collector/v1/orb/webhook", { method: "POST", body }); + return handleOrbWebhook(ctx(e, headers, request)); +} + +const INSTALL = JSON.stringify({ action: "created", installation: { id: 42 }, repository: { full_name: "JSONbored/gittensory" } }); +const row = (e: Env, delivery: string) => + (e.DB as unknown as TestD1Database).prepare("SELECT event_name, action, installation_id, repository_full_name, status FROM orb_webhook_events WHERE delivery_id=?").bind(delivery).first<{ event_name: string; action: string; installation_id: number; repository_full_name: string; status: string }>(); + +describe("handleOrbWebhook (POST /v1/orb/webhook)", () => { + it("400 when the GitHub delivery or event header is missing", async () => { + expect((await post(env(), INSTALL, { delivery: null as unknown as string })).status).toBe(400); + expect((await post(env(), INSTALL, { event: null as unknown as string })).status).toBe(400); + }); + + it("401 on an invalid signature (and 401 when the Orb secret is absent — fail-closed)", async () => { + expect((await post(env(), INSTALL, { sig: "sha256=deadbeef" })).status).toBe(401); + const noSecret = createTestEnv(); // ORB_GITHUB_WEBHOOK_SECRET unset + expect((await post(noSecret, INSTALL)).status).toBe(401); + }); + + it("401 when the signature header is absent entirely", async () => { + expect((await post(env(), INSTALL, { headers: { "x-hub-signature-256": null } })).status).toBe(401); + }); + + it("ignores a non-numeric content-length and processes normally", async () => { + const res = await post(env(), INSTALL, { delivery: "cl-abc", headers: { "content-length": "abc" } }); + expect(res.status).toBe(202); + }); + + it("413 when content-length exceeds the cap", async () => { + const res = await post(env(), INSTALL, { headers: { "content-length": "99999999" } }); + expect(res.status).toBe(413); + }); + + it("413 when the streamed body exceeds the cap (no content-length declared)", async () => { + const res = await post(env({ GITHUB_WEBHOOK_MAX_BODY_BYTES: "16" }), "x".repeat(40)); + expect(res.status).toBe(413); + }); + + it("400 on a signed-but-non-JSON body", async () => { + expect((await post(env(), "not json")).status).toBe(400); + }); + + it("202 + records the install event (action/installation/repo extracted)", async () => { + const e = env(); + const res = await post(e, INSTALL, { delivery: "ok-1" }); + expect(res.status).toBe(202); + await expect(res.json()).resolves.toMatchObject({ status: "received", eventName: "installation" }); + expect(await row(e, "ok-1")).toMatchObject({ action: "created", installation_id: 42, repository_full_name: "JSONbored/gittensory", status: "received" }); + }); + + it("stores null fields for a payload with no action/installation/repository (e.g. ping)", async () => { + const e = env(); + await post(e, JSON.stringify({ zen: "keep it logically awesome" }), { delivery: "ping-1", event: "ping" }); + expect(await row(e, "ping-1")).toMatchObject({ action: null, installation_id: null, repository_full_name: null }); + }); + + it("dedups a redelivery of the same (delivery, payload) as a duplicate", async () => { + const e = env(); + expect((await post(e, INSTALL, { delivery: "dup-1" })).status).toBe(202); + const second = await post(e, INSTALL, { delivery: "dup-1" }); + expect(second.status).toBe(202); + await expect(second.json()).resolves.toMatchObject({ status: "duplicate" }); + }); + + it("re-records a redelivery whose payload CHANGED (same delivery id, different hash)", async () => { + const e = env(); + await post(e, INSTALL, { delivery: "chg-1" }); + const changed = JSON.stringify({ action: "deleted", installation: { id: 42 } }); + const res = await post(e, changed, { delivery: "chg-1" }); + await expect(res.json()).resolves.toMatchObject({ status: "received" }); // not a duplicate + expect((await row(e, "chg-1"))?.action).toBe("deleted"); + }); + + it("treats an already-processed delivery as a duplicate regardless of payload", async () => { + const e = env(); + await (e.DB as unknown as TestD1Database) + .prepare("INSERT INTO orb_webhook_events (delivery_id, event_name, payload_hash, status) VALUES ('proc-1','installation','oldhash','processed')") + .run(); + const res = await post(e, INSTALL, { delivery: "proc-1" }); + await expect(res.json()).resolves.toMatchObject({ status: "duplicate" }); + }); + + it("does NOT suppress an 'error' row — the delivery is retried", async () => { + const e = env(); + await (e.DB as unknown as TestD1Database) + .prepare("INSERT INTO orb_webhook_events (delivery_id, event_name, payload_hash, status) VALUES ('err-1','installation','oldhash','error')") + .run(); + const res = await post(e, INSTALL, { delivery: "err-1" }); + await expect(res.json()).resolves.toMatchObject({ status: "received" }); // re-recorded, not suppressed + expect((await row(e, "err-1"))?.status).toBe("received"); + }); + + it("treats a missing request body as empty (→ 400 invalid JSON after a valid empty-body signature)", async () => { + const e = env(); + const sig = await sign("", SECRET); + const res = await post(e, "", { request: new Request("https://collector/v1/orb/webhook", { method: "POST" }), sig }); + expect(res.status).toBe(400); // empty body verifies, then fails JSON.parse + }); + + it("skips undefined stream chunks while reading the body", async () => { + const e = env(); + const body = new ReadableStream({ + start(controller) { + controller.enqueue(undefined as unknown as Uint8Array); + controller.close(); + }, + }); + const request = { body } as unknown as Request; + const res = await post(e, "", { request, sig: "sha256=bad" }); + expect(res.status).toBe(401); // empty (undefined-skipped) body + bad sig + }); +}); + +describe("POST /v1/orb/webhook route (through the app middleware)", () => { + const app = createApp(); + + it("is token-exempt + rate-classified, routing to the handler (401 on a bad signature)", async () => { + // Exercises requiresApiToken (exempt) + routeClassForPath (strict) for the new path, then the handler. + const res = await app.request( + "/v1/orb/webhook", + { method: "POST", headers: { "x-github-delivery": "rt-1", "x-github-event": "ping", "x-hub-signature-256": "sha256=bad" }, body: "{}" }, + createTestEnv({ ORB_GITHUB_WEBHOOK_SECRET: "s" }), + ); + expect(res.status).toBe(401); + }); +});