fix(orb): authenticate fleet telemetry ingest (#1285)

* fix(orb): authenticate fleet telemetry ingest

* fix(orb): constant-time compare the ingest token (timingSafeEqual)

Author: JSONbored

src/api/routes.ts

@@ -21,6 +21,7 @@ import {
   extractCookieValue,
   isAuthorizedGitHubSessionLogin,
   revokeSession,
+  timingSafeEqual,
   type AuthIdentity,
 } from "../auth/security";
 import { normalizeGittBountySnapshot } from "../bounties/ingest";
@@ -2914,12 +2915,13 @@ export function createApp() {
     return c.json(result);
   });
 
-  // 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).
+  // Gittensory Orb (#1255) — central fleet-calibration collector. Receives anonymized, reversal-aware outcome
+  // batches from self-hosted instances. Sender-side HMAC anonymization is for privacy, not authentication.
+  // OPTIONAL shared-token gate (#1285): unset ⇒ OPEN ingress (the live fleet keeps working, as before); set
+  // ⇒ the collector REQUIRES it, so an operator can lock the write path down after distributing the matching
+  // ORB_COLLECTOR_TOKEN to exporters. Bounded by a hard body ceiling, and dedup'd via UNIQUE(instance_id, repo_hash, pr_hash).
   app.post("/v1/orb/ingest", async (c) => {
-    // Open ingress (no shared secret — the fleet topology has no per-instance key the collector could
-    // verify), bounded by a hard body ceiling so it can't be used to make us buffer unbounded input.
+    if (!(await isAuthorizedOrbIngest(c.env, extractBearerToken(c.req.header("authorization"))))) return c.json({ error: "unauthorized" }, 401);
     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);
@@ -4945,6 +4947,18 @@ function toIsoQueryDate(value: string): string | undefined {
   return Number.isFinite(timestamp) ? new Date(timestamp).toISOString() : undefined;
 }
 
+
+// Optional Orb-ingest auth (#1285). FAIL-OPEN by default: with no ORB_INGEST_TOKEN configured the ingress stays
+// OPEN (matching today's live fleet — deploying this is non-breaking). Once the operator sets the token, the
+// collector REQUIRES an exact bearer match, so the write path can be locked down after the matching
+// ORB_COLLECTOR_TOKEN is rolled out to exporters.
+async function isAuthorizedOrbIngest(env: Env, token: string | undefined): Promise<boolean> {
+  if (!env.ORB_INGEST_TOKEN) return true;
+  // Constant-time compare (mirrors every other secret check in auth/security) — a `===` here is timing-attack
+  // vulnerable for a shared secret.
+  return timingSafeEqual(token, env.ORB_INGEST_TOKEN);
+}
+
 function requiresApiToken(path: string): boolean {
   if (path === "/health") return false;
   if (path === "/v1/mcp/compatibility") return false;

src/env.d.ts

@@ -103,6 +103,8 @@ declare global {
     GITTENSORY_API_TOKEN: string;
     GITTENSORY_MCP_TOKEN: string;
     INTERNAL_JOB_TOKEN: string;
+    /** Shared bearer secret required by the hosted Orb ingest collector. */
+    ORB_INGEST_TOKEN?: string;
     /** AES-256-GCM master secret for maintainer BYOK provider keys (encrypt/decrypt at rest). A Worker
      *  secret (`wrangler secret put`), never a public var. When absent, BYOK is unavailable and the AI
      *  review silently falls back to free Workers AI. */

src/orb/ingest.ts

@@ -4,6 +4,9 @@
 // aggregate calibration metadata (verdict, outcome, reversal, bucketed reason, cycle time).
 
 const MAX_BATCH = 500;
+const MAX_INSTANCE_ID_CHARS = 64;
+const MAX_HASH_CHARS = 128;
+const MAX_BUCKET_CHARS = 64;
 const VALID_OUTCOMES = new Set(["merged", "closed"]);
 const VALID_REVERSALS = new Set(["none", "reopened", "reverted"]);
 const MIN_CYCLE_MS = 1_000; // <1s is implausible
@@ -87,7 +90,7 @@ export async function handleOrbIngest(body: string, db: D1Database): Promise<Orb
   }
 
   const { instance_id, events } = payload as OrbIngestPayload;
-  if (!instance_id || events.length === 0) {
+  if (!instance_id || instance_id.length > MAX_INSTANCE_ID_CHARS || events.length === 0) {
     return { error: "invalid_payload" };
   }
 
@@ -109,8 +112,8 @@ export async function handleOrbIngest(body: string, db: D1Database): Promise<Orb
 
   for (const event of batch) {
     if (
-      typeof event.repo_hash !== "string" || !event.repo_hash ||
-      typeof event.pr_hash !== "string" || !event.pr_hash ||
+      typeof event.repo_hash !== "string" || !event.repo_hash || event.repo_hash.length > MAX_HASH_CHARS ||
+      typeof event.pr_hash !== "string" || !event.pr_hash || event.pr_hash.length > MAX_HASH_CHARS ||
       !VALID_OUTCOMES.has(event.outcome)
     ) {
       continue;
@@ -136,7 +139,7 @@ export async function handleOrbIngest(body: string, db: D1Database): Promise<Orb
           typeof event.gate_verdict === "string" ? event.gate_verdict : null,
           event.outcome,
           reversal,
-          typeof event.gate_reasoncode_bucket === "string" ? event.gate_reasoncode_bucket : null,
+          typeof event.gate_reasoncode_bucket === "string" && event.gate_reasoncode_bucket.length <= MAX_BUCKET_CHARS ? event.gate_reasoncode_bucket : null,
           clampCycleMs(event.time_to_close_ms),
           typeof event.decision_timestamp === "string" ? event.decision_timestamp : null,
           typeof event.outcome_timestamp === "string" ? event.outcome_timestamp : null,

src/selfhost/orb-collector.ts

@@ -10,6 +10,7 @@
 //   ORB_COLLECTOR_URL=<url>   — endpoint (default: gittensory's hosted collector)
 //   ORB_AIR_GAP=true          — air-gapped/offline deployments only: compute locally, never send
 //   ORB_ANONYMIZE=true        — HMAC-hash repo/PR before export (default: true)
+//   ORB_COLLECTOR_TOKEN=<secret> — bearer credential for the hosted collector
 //
 // No diffs, no code, no comments, no logins, no commit SHAs — only verdict + outcome + reversal + a bucketed
 // reason category + cycle time, with repo/PR identifiers HMAC'd by a key the collector never holds (so it
@@ -195,11 +196,17 @@ export async function exportOrbBatch(db: D1Database, batchSize = 200, fetchFn: t
 
   const body = JSON.stringify(payload);
   const signature = createHmac("sha256", secret).update(body).digest("hex");
+  const collectorToken = process.env.ORB_COLLECTOR_TOKEN;
 
   try {
     const res = await fetchFn(collectorUrl, {
       method: "POST",
-      headers: { "content-type": "application/json", "x-orb-signature": `sha256=${signature}`, "x-orb-instance": instance },
+      headers: {
+        "content-type": "application/json",
+        "x-orb-signature": `sha256=${signature}`,
+        "x-orb-instance": instance,
+        ...(collectorToken ? { authorization: `Bearer ${collectorToken}` } : {}),
+      },
       body,
     });
     if (!res.ok) {

test/integration/orb-ingest.test.ts

@@ -20,19 +20,22 @@ describe("handleOrbIngest()", () => {
     expect(await handleOrbIngest("{not json}", makeDb())).toEqual({ error: "invalid_json" });
   });
 
-  it("returns invalid_payload: instance_id not a string / events not an array / empty instance / empty events", async () => {
+  it("returns invalid_payload: instance_id not a string / events not an array / empty/oversized instance / empty events", async () => {
     const db = makeDb();
     expect(await handleOrbIngest(JSON.stringify({ instance_id: 123, events: [] }), db)).toEqual({ error: "invalid_payload" });
     expect(await handleOrbIngest(JSON.stringify({ instance_id: "abc", events: "bad" }), db)).toEqual({ error: "invalid_payload" });
     expect(await handleOrbIngest(JSON.stringify({ instance_id: "", events: [ev()] }), db)).toEqual({ error: "invalid_payload" });
     expect(await handleOrbIngest(JSON.stringify({ instance_id: "abc", events: [] }), db)).toEqual({ error: "invalid_payload" });
+    expect(await handleOrbIngest(JSON.stringify({ instance_id: "i".repeat(65), events: [ev()] }), db)).toEqual({ error: "invalid_payload" });
   });
 
   it("skips events with bad repo_hash / pr_hash / outcome", async () => {
     expect(await ingest(makeDb(), [ev({ repo_hash: 99 })])).toEqual({ accepted: 0 });
     expect(await ingest(makeDb(), [ev({ repo_hash: "" })])).toEqual({ accepted: 0 });
+    expect(await ingest(makeDb(), [ev({ repo_hash: "r".repeat(129) })])).toEqual({ accepted: 0 });
     expect(await ingest(makeDb(), [ev({ pr_hash: null })])).toEqual({ accepted: 0 });
     expect(await ingest(makeDb(), [ev({ pr_hash: "" })])).toEqual({ accepted: 0 });
+    expect(await ingest(makeDb(), [ev({ pr_hash: "p".repeat(129) })])).toEqual({ accepted: 0 });
     expect(await ingest(makeDb(), [ev({ outcome: "opened" })])).toEqual({ accepted: 0 });
   });
 
@@ -57,9 +60,10 @@ describe("handleOrbIngest()", () => {
 
   it("stores gate_reasoncode_bucket string vs null", async () => {
     const db = makeDb();
-    await ingest(db, [ev({ pr_hash: "b1", gate_reasoncode_bucket: "duplicate_risk" }), ev({ pr_hash: "b2" })]);
+    await ingest(db, [ev({ pr_hash: "b1", gate_reasoncode_bucket: "duplicate_risk" }), ev({ pr_hash: "b2" }), ev({ pr_hash: "b3", gate_reasoncode_bucket: "b".repeat(65) })]);
     expect(await col(db, "b1", "gate_reasoncode_bucket")).toBe("duplicate_risk");
     expect(await col(db, "b2", "gate_reasoncode_bucket")).toBeNull();
+    expect(await col(db, "b3", "gate_reasoncode_bucket")).toBeNull();
   });
 
   it("clamps time_to_close_ms: valid kept; absent / <1s / >1y → null", async () => {
@@ -212,6 +216,21 @@ describe("POST /v1/orb/ingest route", () => {
     expect(res.status).toBe(413);
     expect(((await res.json()) as { error: string }).error).toBe("payload_too_large");
   });
+
+  it("optional collector token (#1285): open when unset; enforced once ORB_INGEST_TOKEN is set", async () => {
+    const body = JSON.stringify({ instance_id: "abc0", events: [{ repo_hash: "rhash", pr_hash: "phash", outcome: "merged" }] });
+    const post = (env: Env, authorization?: string) =>
+      app.request("/v1/orb/ingest", { method: "POST", headers: { "content-type": "application/json", ...(authorization ? { authorization } : {}) }, body }, env);
+
+    // Token UNSET → open ingress (the live fleet keeps working with no auth header).
+    expect((await post(createTestEnv())).status).toBe(200);
+    // Token SET → a missing or wrong bearer is rejected before the body is parsed.
+    const env = createTestEnv({ ORB_INGEST_TOKEN: "fleet-secret" });
+    expect((await post(env)).status).toBe(401);
+    expect((await post(env, "Bearer wrong")).status).toBe(401);
+    // Token SET + the matching bearer → accepted.
+    expect((await post(env, "Bearer fleet-secret")).status).toBe(200);
+  });
 });
 
 describe("Orb instance registry routes (/v1/internal/orb/instances)", () => {

test/unit/selfhost-orb-collector.test.ts

@@ -74,9 +74,10 @@ describe("exportOrbBatch() — always-on; reads review_audit, ships anonymized r
     process.env.ORB_ANONYMIZE = "true";
     delete process.env.ORB_AIR_GAP;
     delete process.env.ORB_COLLECTOR_URL;
+    delete process.env.ORB_COLLECTOR_TOKEN;
   });
   afterEach(() => {
-    for (const k of ["GITHUB_APP_PRIVATE_KEY", "ORB_APP_ID", "ORB_ANONYMIZE", "ORB_AIR_GAP", "ORB_COLLECTOR_URL", "GITHUB_APP_ID"]) delete (process.env as NodeJS.Dict<string>)[k];
+    for (const k of ["GITHUB_APP_PRIVATE_KEY", "ORB_APP_ID", "ORB_ANONYMIZE", "ORB_AIR_GAP", "ORB_COLLECTOR_URL", "ORB_COLLECTOR_TOKEN", "GITHUB_APP_ID"]) delete (process.env as NodeJS.Dict<string>)[k];
   });
 
   it("returns 0 when the App private key is not configured (App not set up → nothing to export)", async () => {
@@ -173,10 +174,12 @@ describe("exportOrbBatch() — always-on; reads review_audit, ships anonymized r
       await audit(db, "o/r", i, "gate_decision", "merge", `2026-03-0${i}T00:00:00Z`);
       await audit(db, "o/r", i, "pr_outcome", "merged", `2026-03-0${i}T01:00:00Z`);
     }
-    let sig: string | undefined;
-    const n = await exportOrbBatch(db, 3, async (_u, init) => { sig = (init!.headers as Record<string, string>)["x-orb-signature"]; return new Response(null, { status: 200 }); });
+    process.env.ORB_COLLECTOR_TOKEN = "collector-secret";
+    let headers: Record<string, string> | undefined;
+    const n = await exportOrbBatch(db, 3, async (_u, init) => { headers = init!.headers as Record<string, string>; return new Response(null, { status: 200 }); });
     expect(n).toBe(3); // batch cap
-    expect(sig).toMatch(/^sha256=[a-f0-9]{64}$/);
+    expect(headers?.["x-orb-signature"]).toMatch(/^sha256=[a-f0-9]{64}$/);
+    expect(headers?.authorization).toBe("Bearer collector-secret");
   });
 
   it("falls back to GITHUB_APP_ID for the instance id and applies the anonymize default when ORB_* are unset", async () => {