From bafb49acb8d7aa5749e7d80d5a1eb7f5cc31bd4b Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Sat, 27 Jun 2026 18:21:48 -0700 Subject: [PATCH] feat(orb): pull-mode relay delivery for NAT/tailnet self-hosts A self-host engine behind NAT/Tailscale has no public inbound URL, so the Orb's push relay delivery is unreachable (observed: 257 dropped events, no fresh-PR reviews). The server side of pull mode already existed; this adds the engine side: - registerOrbRelayTarget honors ORB_RELAY_MODE=pull (registers no relay URL, no PUBLIC_API_ORIGIN required); push (default) unchanged. mode sent to register. - drainOrbRelay: POST /v1/orb/relay/pull with the secret + prior-batch ack ids, returns the validated event batch; best-effort ([] on non-broker/non-ok/throw). - enqueueWebhookByEnv: env-based core extracted from enqueueVerifiedWebhook (behavior identical) so the Context-less drain loop can enqueue. - server.ts: a 15s drain loop (when ORB_RELAY_MODE=pull) pulls events outbound + enqueues into the WEBHOOKS lane + acks handled ids. No inbound endpoint exposed. Closes #1638. --- src/github/webhook.ts | 60 ++++++++++++++++------------- src/orb/broker-client.ts | 44 +++++++++++++++++++-- src/server.ts | 32 ++++++++++++++- test/unit/orb-broker-client.test.ts | 44 ++++++++++++++++++++- 4 files changed, 145 insertions(+), 35 deletions(-) diff --git a/src/github/webhook.ts b/src/github/webhook.ts index 817e402edc..380c4c93f0 100644 --- a/src/github/webhook.ts +++ b/src/github/webhook.ts @@ -35,60 +35,66 @@ export async function handleGitHubWebhook(c: Context<{ Bindings: Env }>): Promis * webhook receiver above AND the Orb relay receiver below (they verify the body differently — GitHub's HMAC vs the * Orb relay HMAC — then share everything after). */ export async function enqueueVerifiedWebhook(c: Context<{ Bindings: Env }>, deliveryId: string, eventName: string, rawBody: string): Promise { + const result = await enqueueWebhookByEnv(c.env, deliveryId, eventName, rawBody); + switch (result) { + case "invalid_json": + return c.json({ error: "invalid_json" }, 400); + case "duplicate": + return c.json({ ok: true, deliveryId, eventName, status: "duplicate" }, 202); + case "enqueue_failed": + return c.json({ error: "enqueue_failed", deliveryId }, 500); + default: + return c.json({ ok: true, deliveryId, eventName, status: "queued" }, 202); + } +} + +export type EnqueueWebhookResult = "queued" | "duplicate" | "invalid_json" | "enqueue_failed"; + +/** Env-based core of the webhook enqueue (parse → dedup → record → WEBHOOKS lane), with NO Hono Context. Shared by + * the request-context receiver above AND the pull-mode relay drain loop (server.ts), which has no Context. Returns + * a status the caller maps to a response / an ack decision. */ +export async function enqueueWebhookByEnv(env: Env, deliveryId: string, eventName: string, rawBody: string): Promise { let payload: GitHubWebhookPayload; try { payload = JSON.parse(rawBody) as GitHubWebhookPayload; } catch { - return c.json({ error: "invalid_json" }, 400); + return "invalid_json"; } const payloadHash = await sha256Hex(rawBody); - const existingEvent = await getWebhookEvent(c.env, deliveryId); + const existingEvent = await getWebhookEvent(env, deliveryId); // Suppress redelivery of an already-processed event (on success its payloadHash is overwritten to a // "processed" sentinel, so a hash match alone misses it and the event re-runs its side effects) or one // still in flight with the same payload. "error" rows are never suppressed so a failed enqueue/processing // can still be retried (#789). if (existingEvent && existingEvent.status !== "error" && (existingEvent.status === "processed" || existingEvent.payloadHash === payloadHash)) { - return c.json({ ok: true, deliveryId, eventName, status: "duplicate" }, 202); + return "duplicate"; } - await recordWebhookEvent(c.env, { + const eventRow = { deliveryId, eventName, action: payload.action, installationId: payload.installation?.id, repositoryFullName: payload.repository?.full_name, payloadHash, - status: "queued", - }); - - const message: JobMessage = { - type: "github-webhook", - deliveryId, - eventName, - payload, }; + await recordWebhookEvent(env, { ...eventRow, status: "queued" }); + + const message: JobMessage = { type: "github-webhook", deliveryId, eventName, payload }; try { // Send to the dedicated WEBHOOKS lane (not the shared JOBS queue) so a maintenance burst on JOBS can never // starve real GitHub events into the DLQ. (#audit-webhook-queue) - await c.env.WEBHOOKS.send(message); + await env.WEBHOOKS.send(message); } catch { - // Enqueue failed: flip the event to "error" so the dedup guard above lets GitHub redeliver, - // and return 500 so GitHub retries instead of treating the webhook as handled (#786). This also covers the - // deploy-ordering case where the WEBHOOKS queue is not yet provisioned — no event is lost. - await recordWebhookEvent(c.env, { - deliveryId, - eventName, - action: payload.action, - installationId: payload.installation?.id, - repositoryFullName: payload.repository?.full_name, - payloadHash, - status: "error", - }); - return c.json({ error: "enqueue_failed", deliveryId }, 500); + // Enqueue failed: flip the event to "error" so the dedup guard above lets GitHub redeliver / the next pull + // re-deliver, instead of treating the webhook as handled (#786). Also covers the deploy-ordering case where + // the WEBHOOKS queue is not yet provisioned — no event is lost. + await recordWebhookEvent(env, { ...eventRow, status: "error" }); + return "enqueue_failed"; } - return c.json({ ok: true, deliveryId, eventName, status: "queued" }, 202); + return "queued"; } /** The brokered self-host's relay RECEIVER. The central Orb forwards an event here, HMAC-signed (x-orb-signature- diff --git a/src/orb/broker-client.ts b/src/orb/broker-client.ts index 7f250a768e..d313a2542e 100644 --- a/src/orb/broker-client.ts +++ b/src/orb/broker-client.ts @@ -81,17 +81,22 @@ export async function fetchBrokeredInstallationToken( * relay until the next boot — it never blocks startup or throws. The relay URL is the container's public origin + * /v1/orb/relay (the receiver); the Orb SSRF-validates it, so PUBLIC_API_ORIGIN must be a real public https host. */ export async function registerOrbRelayTarget( - env: { ORB_ENROLLMENT_SECRET?: string | undefined; ORB_BROKER_URL?: string | undefined; PUBLIC_API_ORIGIN?: string | undefined }, + env: { ORB_ENROLLMENT_SECRET?: string | undefined; ORB_BROKER_URL?: string | undefined; PUBLIC_API_ORIGIN?: string | undefined; ORB_RELAY_MODE?: string | undefined }, fetchImpl: typeof fetch = fetch, ): Promise<"registered" | "skipped" | "failed"> { - if (!isOrbBrokerMode(env) || !env.PUBLIC_API_ORIGIN) return "skipped"; - const relayUrl = `${env.PUBLIC_API_ORIGIN.replace(/\/+$/, "")}/v1/orb/relay`; + if (!isOrbBrokerMode(env)) return "skipped"; + // Pull mode (#secure-relay): the engine DRAINS events outbound from the Orb, so NO inbound endpoint is exposed — + // the right fit for a NAT/tailnet self-host (a public push URL would otherwise be unreachable). Push mode needs a + // public relay URL the Orb can reach. + const mode = env.ORB_RELAY_MODE === "pull" ? "pull" : "push"; + if (mode === "push" && !env.PUBLIC_API_ORIGIN) return "skipped"; + const relayUrl = mode === "push" ? `${env.PUBLIC_API_ORIGIN!.replace(/\/+$/, "")}/v1/orb/relay` : ""; try { const base = orbBrokerBaseUrl(env); const res = await fetchImpl(`${base}/v1/orb/relay/register`, { method: "POST", headers: { authorization: `Bearer ${env.ORB_ENROLLMENT_SECRET}`, "content-type": "application/json" }, // present — isOrbBrokerMode required it - body: JSON.stringify({ relayUrl }), + body: JSON.stringify({ relayUrl, mode }), signal: AbortSignal.timeout(10_000), }); return res.ok ? "registered" : "failed"; @@ -99,3 +104,34 @@ export async function registerOrbRelayTarget( return "failed"; } } + +/** Pull-mode drain (#secure-relay): fetch this install's queued events from the Orb, acking the previous batch's + * delivery ids so the Orb deletes them. Lets a NAT/tailnet engine receive events WITHOUT exposing an inbound + * endpoint. BEST-EFFORT — returns [] on a non-broker / unsafe-URL / non-ok / thrown case; the next tick retries. */ +export async function drainOrbRelay( + env: { ORB_ENROLLMENT_SECRET?: string | undefined; ORB_BROKER_URL?: string | undefined }, + ack: string[] = [], + fetchImpl: typeof fetch = fetch, +): Promise<{ deliveryId: string; eventName: string; rawBody: string }[]> { + if (!isOrbBrokerMode(env)) return []; + try { + const base = orbBrokerBaseUrl(env); + const res = await fetchImpl(`${base}/v1/orb/relay/pull`, { + method: "POST", + headers: { authorization: `Bearer ${env.ORB_ENROLLMENT_SECRET}`, "content-type": "application/json" }, + body: JSON.stringify({ ack }), + signal: AbortSignal.timeout(15_000), + }); + if (!res.ok) return []; + const body = (await res.json()) as { events?: Array<{ deliveryId?: unknown; eventName?: unknown; rawBody?: unknown }> }; + const out: { deliveryId: string; eventName: string; rawBody: string }[] = []; + for (const e of body.events ?? []) { + if (typeof e.deliveryId === "string" && typeof e.eventName === "string" && typeof e.rawBody === "string") { + out.push({ deliveryId: e.deliveryId, eventName: e.eventName, rawBody: e.rawBody }); + } + } + return out; + } catch { + return []; + } +} diff --git a/src/server.ts b/src/server.ts index eed679f2ce..2faaf12f7a 100644 --- a/src/server.ts +++ b/src/server.ts @@ -735,12 +735,14 @@ async function main(): Promise { void runOrbExport(); // flush any pending events at startup setInterval(runOrbExport, 3_600_000); // then hourly - // Brokered self-host: register our public relay URL with the central Orb so it forwards this install's events - // here (best-effort, fire-and-forget — a no-op unless ORB_ENROLLMENT_SECRET + PUBLIC_API_ORIGIN are set). + // Brokered self-host: register our relay target with the central Orb (best-effort, fire-and-forget). PUSH mode + // (default) registers a public relay URL the Orb POSTs to; PULL mode (ORB_RELAY_MODE=pull) registers no URL and + // the drain loop below pulls events outbound — the right fit behind NAT/tailnet (no inbound endpoint exposed). void registerOrbRelayTarget({ ORB_ENROLLMENT_SECRET: process.env.ORB_ENROLLMENT_SECRET, ORB_BROKER_URL: process.env.ORB_BROKER_URL, PUBLIC_API_ORIGIN: process.env.PUBLIC_API_ORIGIN, + ORB_RELAY_MODE: process.env.ORB_RELAY_MODE, }) .then((r) => { if (r === "registered") { @@ -753,6 +755,32 @@ async function main(): Promise { }) .catch((error) => captureError(error, { kind: "orb_relay_register" })); + // Pull-mode relay drain (#secure-relay): when ORB_RELAY_MODE=pull, the engine DRAINS its events from the Orb on a + // timer instead of exposing an inbound endpoint — the right fit behind NAT/tailnet. Acks the previous batch so the + // Orb deletes delivered events; best-effort (a failed tick retries next interval). Each event enqueues into the + // same WEBHOOKS lane the push receiver uses. + if (process.env.ORB_RELAY_MODE === "pull" && process.env.ORB_ENROLLMENT_SECRET) { + const { drainOrbRelay } = await import("./orb/broker-client"); + const { enqueueWebhookByEnv } = await import("./github/webhook"); + let pendingAck: string[] = []; + const drainRelay = async (): Promise => { + const events = await drainOrbRelay( + { ORB_ENROLLMENT_SECRET: process.env.ORB_ENROLLMENT_SECRET, ORB_BROKER_URL: process.env.ORB_BROKER_URL }, + pendingAck, + ); + pendingAck = []; + for (const ev of events) { + const result = await enqueueWebhookByEnv(env, ev.deliveryId, ev.eventName, ev.rawBody); + // Ack everything durably handled (queued / duplicate / invalid_json) so the Orb deletes it; retry only a + // real enqueue failure on the next pull (don't ack → the Orb keeps it). + if (result !== "enqueue_failed") pendingAck.push(ev.deliveryId); + } + if (events.length > 0) console.log(JSON.stringify({ event: "orb_relay_drained", count: events.length })); + }; + void drainRelay(); + setInterval(() => void drainRelay().catch((error) => captureError(error, { kind: "orb_relay_drain" })), 15_000); + } + // Graceful shutdown: stop accepting HTTP, let the queue finish, close the backend. let shuttingDown = false; const shutdown = async (signal: string): Promise => { diff --git a/test/unit/orb-broker-client.test.ts b/test/unit/orb-broker-client.test.ts index 681437b337..b1cc4941b3 100644 --- a/test/unit/orb-broker-client.test.ts +++ b/test/unit/orb-broker-client.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest"; -import { fetchBrokeredInstallationToken, isOrbBrokerMode, registerOrbRelayTarget } from "../../src/orb/broker-client"; +import { drainOrbRelay, fetchBrokeredInstallationToken, isOrbBrokerMode, registerOrbRelayTarget } from "../../src/orb/broker-client"; /** A fetch stub that records the URL + init and returns a fixed response. */ function captureFetch(resp: Response): { fetchImpl: typeof fetch; calls: { url: string; init?: RequestInit | undefined }[] } { @@ -106,7 +106,7 @@ describe("registerOrbRelayTarget", () => { expect(await registerOrbRelayTarget({ ORB_ENROLLMENT_SECRET: "orbsec_x", PUBLIC_API_ORIGIN: "https://me.example/", ORB_BROKER_URL: "https://broker.example/" }, fetchImpl)).toBe("registered"); expect(calls[0]?.url).toBe("https://broker.example/v1/orb/relay/register"); // ORB_BROKER_URL trailing slash stripped expect((calls[0]?.init?.headers as Record).authorization).toBe("Bearer orbsec_x"); - expect(JSON.parse(String(calls[0]?.init?.body))).toEqual({ relayUrl: "https://me.example/v1/orb/relay" }); // PUBLIC_API_ORIGIN trailing slash stripped + expect(JSON.parse(String(calls[0]?.init?.body))).toEqual({ relayUrl: "https://me.example/v1/orb/relay", mode: "push" }); // PUBLIC_API_ORIGIN trailing slash stripped }); it("uses the default broker base when ORB_BROKER_URL is unset", async () => { @@ -128,4 +128,44 @@ describe("registerOrbRelayTarget", () => { expect(await registerOrbRelayTarget(cfg, (async () => new Response("no", { status: 403 })) as typeof fetch)).toBe("failed"); expect(await registerOrbRelayTarget(cfg, (async () => { throw new Error("down"); }) as typeof fetch)).toBe("failed"); }); + + it("pull mode (ORB_RELAY_MODE=pull) registers with NO relay URL and works without a public origin (NAT/tailnet)", async () => { + const { fetchImpl, calls } = captureFetch(new Response("ok")); + // No PUBLIC_API_ORIGIN — push would skip, but pull doesn't need an inbound URL. + expect(await registerOrbRelayTarget({ ORB_ENROLLMENT_SECRET: "s", ORB_RELAY_MODE: "pull" }, fetchImpl)).toBe("registered"); + expect(JSON.parse(String(calls[0]?.init?.body))).toEqual({ relayUrl: "", mode: "pull" }); + }); +}); + +describe("drainOrbRelay (pull-mode drain)", () => { + it("returns [] when not in broker mode (no enrollment secret)", async () => { + expect(await drainOrbRelay({})).toEqual([]); + }); + + it("POSTs the ack list, parses returned events, and filters malformed ones", async () => { + const { fetchImpl, calls } = captureFetch( + Response.json({ + events: [ + { deliveryId: "d1", eventName: "pull_request", rawBody: "{\"a\":1}" }, + { deliveryId: "d2", eventName: "check_suite", rawBody: "{}" }, + { deliveryId: "bad", eventName: "x" }, // no rawBody → filtered out + ], + }), + ); + const out = await drainOrbRelay({ ORB_ENROLLMENT_SECRET: "s" }, ["prev-1"], fetchImpl); + expect(out).toEqual([ + { deliveryId: "d1", eventName: "pull_request", rawBody: "{\"a\":1}" }, + { deliveryId: "d2", eventName: "check_suite", rawBody: "{}" }, + ]); + expect(calls[0]?.url).toBe("https://gittensory-api.aethereal.dev/v1/orb/relay/pull"); + expect((calls[0]?.init?.headers as Record).authorization).toBe("Bearer s"); + expect(JSON.parse(String(calls[0]?.init?.body))).toEqual({ ack: ["prev-1"] }); + }); + + it("tolerates a missing events array (?? [] arm) and returns [] on non-ok / thrown / unsafe-URL", async () => { + expect(await drainOrbRelay({ ORB_ENROLLMENT_SECRET: "s" }, [], (async () => Response.json({})) as typeof fetch)).toEqual([]); + expect(await drainOrbRelay({ ORB_ENROLLMENT_SECRET: "s" }, [], (async () => new Response("no", { status: 403 })) as typeof fetch)).toEqual([]); + expect(await drainOrbRelay({ ORB_ENROLLMENT_SECRET: "s" }, [], (async () => { throw new Error("down"); }) as typeof fetch)).toEqual([]); + expect(await drainOrbRelay({ ORB_ENROLLMENT_SECRET: "s", ORB_BROKER_URL: "http://broker.example" }, [], (async () => { throw new Error("unsafe should not fetch"); }) as typeof fetch)).toEqual([]); + }); });