diff --git a/apps/gittensory-ui/public/openapi.json b/apps/gittensory-ui/public/openapi.json index 709c2fcd5a..e54daadfbc 100644 --- a/apps/gittensory-ui/public/openapi.json +++ b/apps/gittensory-ui/public/openapi.json @@ -15225,6 +15225,12 @@ }, "400": { "description": "Malformed JSON or invalid payload shape" + }, + "401": { + "description": "Invalid Orb HMAC signature" + }, + "413": { + "description": "Payload exceeds the Orb ingest byte limit" } }, "security": [ @@ -15234,6 +15240,18 @@ { "GittensorySessionCookie": [] } + ], + "parameters": [ + { + "schema": { + "type": "string", + "description": "sha256=" + }, + "required": true, + "description": "sha256=", + "name": "x-orb-signature", + "in": "header" + } ] } } diff --git a/src/api/routes.ts b/src/api/routes.ts index 64d54426ab..3f7d7c017d 100644 --- a/src/api/routes.ts +++ b/src/api/routes.ts @@ -122,7 +122,7 @@ import { type GittensoryMentionCommandName, } from "../github/commands"; import { handleGitHubWebhook } from "../github/webhook"; -import { handleOrbIngest } from "../orb/ingest"; +import { handleOrbIngest, readOrbIngestBody, verifyOrbIngestSignature } from "../orb/ingest"; import { handleMcpRequest } from "../mcp/server"; import { buildOpenApiSpec } from "../openapi/spec"; import { generateSignalSnapshots } from "../queue/processors"; @@ -2864,11 +2864,18 @@ export function createApp() { app.post("/v1/github/webhook", handleGitHubWebhook); // Gittensory Orb (#1219) — central collector. Receives anonymized outcome signal batches - // from self-hosted instances. No auth required: all data is HMAC-anonymized by the sender; + // from self-hosted instances. Verifies the exporter HMAC before parsing, then // dedup is enforced via UNIQUE(instance_id, pr_hash) in orb_signals. app.post("/v1/orb/ingest", async (c) => { - const body = await c.req.text().catch(() => null); + const body = await readOrbIngestBody(c.req.raw, c.req.header("content-length")); + if (body === null) return c.json({ error: "payload_too_large" }, 413); if (!body) return c.json({ error: "invalid_request" }, 400); + const verified = await verifyOrbIngestSignature( + body, + c.req.header("x-orb-signature") ?? null, + c.env.ORB_INGEST_SECRET, + ); + if (!verified) return c.json({ error: "invalid_signature" }, 401); const result = await handleOrbIngest(body, c.env.DB); if ("error" in result) return c.json(result, 400); return c.json(result, 200); diff --git a/src/env.d.ts b/src/env.d.ts index 7d898869ef..a2f1ba0524 100644 --- a/src/env.d.ts +++ b/src/env.d.ts @@ -38,6 +38,8 @@ declare global { ADMIN_GITHUB_LOGINS?: string; GITHUB_WEBHOOK_SECRET: string; GITHUB_WEBHOOK_MAX_BODY_BYTES?: string; + /** Shared HMAC secret for self-hosted Orb exporters posting to /v1/orb/ingest. */ + ORB_INGEST_SECRET?: string; GITHUB_APP_PRIVATE_KEY: string; GITHUB_APP_ID: string; GITHUB_APP_SLUG: string; diff --git a/src/openapi/spec.ts b/src/openapi/spec.ts index 451147d957..30f7788093 100644 --- a/src/openapi/spec.ts +++ b/src/openapi/spec.ts @@ -667,9 +667,14 @@ export function buildOpenApiSpec() { registry.registerPath({ method: "post", path: "/v1/orb/ingest", + request: { + headers: z.object({ "x-orb-signature": z.string().describe("sha256=") }), + }, responses: { 200: { description: "Batch accepted; returns { accepted: number }" }, 400: { description: "Malformed JSON or invalid payload shape" }, + 401: { description: "Invalid Orb HMAC signature" }, + 413: { description: "Payload exceeds the Orb ingest byte limit" }, }, }); registry.registerPath({ diff --git a/src/orb/ingest.ts b/src/orb/ingest.ts index 6e91ddbcfb..fb22a69bfa 100644 --- a/src/orb/ingest.ts +++ b/src/orb/ingest.ts @@ -3,7 +3,14 @@ // No raw repo names, owner identifiers, or PR content is accepted or stored — only HMAC-anonymized // hashes + aggregate outcome metadata (verdict, timing). +import { verifyGitHubSignature } from "../utils/crypto"; + const MAX_BATCH = 500; +export const MAX_ORB_INGEST_BODY_BYTES = 128 * 1024; +const MAX_INSTANCE_ID_CHARS = 64; +const MAX_HASH_CHARS = 128; +const MAX_GATE_VERDICT_CHARS = 64; +const MAX_CREATED_AT_CHARS = 64; const VALID_OUTCOMES = new Set(["merged", "closed"]); interface OrbIngestEvent { @@ -22,6 +29,38 @@ interface OrbIngestPayload { export type OrbIngestResult = { accepted: number } | { error: string }; +export async function verifyOrbIngestSignature( + body: string, + signatureHeader: string | null, + secret: string | undefined, +): Promise { + return verifyGitHubSignature(body, signatureHeader, secret ?? ""); +} + +export async function readOrbIngestBody(request: Request, contentLengthHeader: string | null | undefined): Promise { + const contentLength = parsePositiveInt(contentLengthHeader); + if (contentLength !== null && contentLength > MAX_ORB_INGEST_BODY_BYTES) return null; + + 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 > MAX_ORB_INGEST_BODY_BYTES) return null; + chunks.push(decoder.decode(value, { stream: true })); + } + + chunks.push(decoder.decode()); + return chunks.join(""); +} + export async function handleOrbIngest(body: string, db: D1Database): Promise { let payload: unknown; try { @@ -38,7 +77,7 @@ export async function handleOrbIngest(body: string, db: D1Database): Promise 0) accepted++; @@ -79,3 +122,21 @@ export async function handleOrbIngest(body: string, db: D1Database): Promise 0 && value.length <= maxChars; +} + +function normalizeOptionalString(value: unknown, maxChars: number): string | null | undefined { + if (value === null || value === undefined) return null; + if (typeof value !== "string") return null; + if (value.length === 0 || value.length > maxChars) return undefined; + return value; +} + +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; +} diff --git a/test/integration/orb-ingest.test.ts b/test/integration/orb-ingest.test.ts index 1a25c33a42..26057fa766 100644 --- a/test/integration/orb-ingest.test.ts +++ b/test/integration/orb-ingest.test.ts @@ -1,8 +1,23 @@ import { describe, expect, it } from "vitest"; import { createApp } from "../../src/api/routes"; -import { handleOrbIngest } from "../../src/orb/ingest"; +import { handleOrbIngest, readOrbIngestBody, verifyOrbIngestSignature } from "../../src/orb/ingest"; import { createTestEnv, TestD1Database } from "../helpers/d1"; +const ORB_INGEST_SECRET = "orb-ingest-test-secret"; + +async function signOrbBody(body: string, secret = ORB_INGEST_SECRET): Promise { + const key = await crypto.subtle.importKey( + "raw", + new TextEncoder().encode(secret), + { name: "HMAC", hash: "SHA-256" }, + false, + ["sign"], + ); + const signature = await crypto.subtle.sign("HMAC", key, new TextEncoder().encode(body)); + const hex = [...new Uint8Array(signature)].map((byte) => byte.toString(16).padStart(2, "0")).join(""); + return `sha256=${hex}`; +} + // ── handleOrbIngest unit-style tests ────────────────────────────────────────── describe("handleOrbIngest()", () => { @@ -197,6 +212,32 @@ describe("handleOrbIngest()", () => { expect(result).toEqual({ accepted: 500 }); }); + + it("rejects an overlong instance_id before inserting rows", async () => { + const db = makeDb(); + const result = await handleOrbIngest( + JSON.stringify({ instance_id: "i".repeat(65), events: [{ repo_hash: "rh", pr_hash: "ph", outcome: "merged" }] }), + db, + ); + expect(result).toEqual({ error: "invalid_payload" }); + }); + + it("skips events with empty or overlong optional strings", async () => { + const db = makeDb(); + const result = await handleOrbIngest( + JSON.stringify({ + instance_id: "inst1", + events: [ + { repo_hash: "rh-overlong-gate", pr_hash: "ph-overlong-gate", outcome: "merged", gate_verdict: "v".repeat(65) }, + { repo_hash: "rh-overlong-date", pr_hash: "ph-overlong-date", outcome: "closed", created_at: "2".repeat(65) }, + { repo_hash: "rh-empty-gate", pr_hash: "ph-empty-gate", outcome: "merged", gate_verdict: "" }, + ], + }), + db, + ); + expect(result).toEqual({ accepted: 0 }); + }); + it("does not throw when the DB throws on insert (covers inner catch branch)", async () => { const brokenDb = { prepare: () => ({ bind: () => ({ run: () => Promise.reject(new Error("disk full")) }) }), @@ -209,33 +250,113 @@ describe("handleOrbIngest()", () => { }); }); + +// ── Orb ingest transport guards ─────────────────────────────────────────────── + +describe("Orb ingest transport guards", () => { + it("verifies valid signatures and rejects missing secrets", async () => { + const body = JSON.stringify({ ok: true }); + expect(await verifyOrbIngestSignature(body, await signOrbBody(body), ORB_INGEST_SECRET)).toBe(true); + expect(await verifyOrbIngestSignature(body, await signOrbBody(body), undefined)).toBe(false); + }); + + it("reads a body under the byte limit", async () => { + const request = new Request("https://example.test/v1/orb/ingest", { method: "POST", body: "hello" }); + await expect(readOrbIngestBody(request, "5")).resolves.toBe("hello"); + }); + + it("returns an empty string when the request has no body stream", async () => { + const request = new Request("https://example.test/v1/orb/ingest"); + await expect(readOrbIngestBody(request, null)).resolves.toBe(""); + }); + + it("ignores invalid content-length values", async () => { + const request = new Request("https://example.test/v1/orb/ingest", { method: "POST", body: "hello" }); + await expect(readOrbIngestBody(request, "not-a-number")).resolves.toBe("hello"); + }); + + it("rejects content-length values over the byte limit", async () => { + const request = new Request("https://example.test/v1/orb/ingest", { method: "POST", body: "hello" }); + await expect(readOrbIngestBody(request, "131073")).resolves.toBeNull(); + }); + + it("rejects streamed bodies that cross the byte limit without content-length", async () => { + const request = new Request("https://example.test/v1/orb/ingest", { method: "POST", body: "x".repeat(131073) }); + await expect(readOrbIngestBody(request, null)).resolves.toBeNull(); + }); +}); + // ── Route integration tests (covers routes.ts new lines) ────────────────────── describe("POST /v1/orb/ingest route", () => { const app = createApp(); it("returns 200 with accepted count for a valid batch", async () => { - const env = createTestEnv(); const body = JSON.stringify({ instance_id: "abc123def456abc0", events: [{ repo_hash: "rhash1234567890123456", pr_hash: "phash1234567890123456", outcome: "merged" }], }); - const res = await app.request("/v1/orb/ingest", { method: "POST", headers: { "content-type": "application/json" }, body }, env); + const res = await app.request( + "/v1/orb/ingest", + { + method: "POST", + headers: { "content-type": "application/json", "x-orb-signature": await signOrbBody(body) }, + body, + }, + createTestEnv({ ORB_INGEST_SECRET }), + ); expect(res.status).toBe(200); const json = await res.json() as { accepted: number }; expect(json.accepted).toBe(1); }); it("returns 400 for invalid JSON (covers error-in-result branch)", async () => { - const env = createTestEnv(); - const res = await app.request("/v1/orb/ingest", { method: "POST", headers: { "content-type": "application/json" }, body: "{bad" }, env); + const env = createTestEnv({ ORB_INGEST_SECRET }); + const body = "{bad"; + const res = await app.request( + "/v1/orb/ingest", + { method: "POST", headers: { "content-type": "application/json", "x-orb-signature": await signOrbBody(body) }, body }, + env, + ); expect(res.status).toBe(400); const json = await res.json() as { error: string }; expect(json.error).toBe("invalid_json"); }); + + it("rejects unsigned batches before parsing (regression for unauthenticated signal poisoning)", async () => { + const env = createTestEnv({ ORB_INGEST_SECRET }); + const body = JSON.stringify({ + instance_id: "abc123def456abc0", + events: [{ repo_hash: "rhash1234567890123456", pr_hash: "phash1234567890123456", outcome: "merged" }], + }); + const res = await app.request("/v1/orb/ingest", { method: "POST", headers: { "content-type": "application/json" }, body }, env); + expect(res.status).toBe(401); + await expect(res.json()).resolves.toMatchObject({ error: "invalid_signature" }); + }); + + it("rejects oversized batches before reading the full body", async () => { + const env = createTestEnv({ ORB_INGEST_SECRET }); + const body = JSON.stringify({ instance_id: "abc123def456abc0", events: [] }); + const res = await app.request( + "/v1/orb/ingest", + { + method: "POST", + headers: { + "content-type": "application/json", + "content-length": "131073", + "x-orb-signature": await signOrbBody(body), + }, + body, + }, + env, + ); + expect(res.status).toBe(413); + await expect(res.json()).resolves.toMatchObject({ error: "payload_too_large" }); + }); + it("returns 400 for an empty body (covers !body branch in route)", async () => { - const env = createTestEnv(); + const env = createTestEnv({ ORB_INGEST_SECRET }); const res = await app.request("/v1/orb/ingest", { method: "POST", body: "" }, env); expect(res.status).toBe(400); });