From 6b4d25c1f72defa5944b5db53a7bf544094dec42 Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Wed, 24 Jun 2026 15:38:18 -0700 Subject: [PATCH] feat(orb): registration-gate the fleet + cap ingest body (das-github-mirror model) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Answers the "how should /v1/orb/ingest be protected?" question by following das-github-mirror's ingress model rather than a shared ingest secret (which would break #1257's hardwired-on, no-shared-key telemetry — every self-hoster would 401): - Open ingest, bounded. Read the body behind a 1 MiB ceiling (streaming, with a content-length fast-path) → 413 on oversize. Mirrors the body limit das-github-mirror puts in front of its open webhook ingress. Dedup already exists via UNIQUE(instance_id, repo_hash, pr_hash). - Registration gate (the trust anchor). New orb_instances table — every instance that ingests is recorded (registered=0 by default, like das-github-mirror's registered flag); signals are still stored for everyone (so registration is retroactive), but computeFleetAnalytics counts ONLY registered instances toward the fleet median. A stranger — or a ring of them — cannot move calibration until an operator opts them in. - Operator endpoints (internal-token gated): GET /v1/internal/orb/instances lists pending + registered instances with their stored-signal counts; POST /v1/internal/orb/instances/register opts one in (or out). Behavioral note: the fleet surfaces shipped in #1268 (operator dashboard + MCP tool) now read empty until the operator registers at least one instance — the intended trust posture. migrations/0061 + portable ON CONFLICT upserts (no pg-dialect change). Supersedes #1248. --- migrations/0061_orb_instances.sql | 17 ++++ src/api/routes.ts | 41 ++++++++- src/orb/analytics.ts | 10 ++- src/orb/ingest.ts | 49 +++++++++++ test/integration/orb-ingest.test.ts | 117 +++++++++++++++++++++++++- test/unit/mcp-fleet-analytics.test.ts | 2 + test/unit/operator-dashboard.test.ts | 3 + test/unit/orb-analytics.test.ts | 27 ++++++ 8 files changed, 261 insertions(+), 5 deletions(-) create mode 100644 migrations/0061_orb_instances.sql diff --git a/migrations/0061_orb_instances.sql b/migrations/0061_orb_instances.sql new file mode 100644 index 0000000000..96fe71ee0b --- /dev/null +++ b/migrations/0061_orb_instances.sql @@ -0,0 +1,17 @@ +-- Gittensory Orb (#1255) — instance registration gate, modeled on das-github-mirror's `registered=false` +-- default. Every self-host instance that POSTs anonymized batches to /v1/orb/ingest is recorded here on +-- first contact, but its signals only count toward fleet calibration once an operator REGISTERS it +-- (registered=1). This is the fleet's trust anchor: ingest stays open + frictionless (no shared secret — +-- the topology has no per-instance key the collector could verify), but a stranger — or a ring of them — +-- cannot move the fleet median until a human opts them in. Signals are still stored for everyone (so a +-- later registration is retroactive); computeFleetAnalytics is what filters to registered instances. +CREATE TABLE IF NOT EXISTS orb_instances ( + instance_id TEXT PRIMARY KEY NOT NULL, + -- 0 until an operator opts the instance into fleet calibration; computeFleetAnalytics counts only registered. + registered INTEGER NOT NULL DEFAULT 0, + first_seen_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + last_seen_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, + registered_at TEXT +); + +CREATE INDEX IF NOT EXISTS orb_instances_registered_idx ON orb_instances(registered); diff --git a/src/api/routes.ts b/src/api/routes.ts index 227ea95d46..f20d366a59 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 } from "../orb/ingest"; import { computeFleetAnalytics } from "../orb/analytics"; import { handleMcpRequest } from "../mcp/server"; import { buildOpenApiSpec } from "../openapi/spec"; @@ -2868,7 +2868,10 @@ export function createApp() { // 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). app.post("/v1/orb/ingest", async (c) => { - const body = await c.req.text().catch(() => null); + // 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. + 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 result = await handleOrbIngest(body, c.env.DB); if ("error" in result) return c.json(result, 400); @@ -2883,6 +2886,40 @@ export function createApp() { return c.json(await computeFleetAnalytics(c.env, { windowDays: days })); }); + // Orb instance registry — the fleet trust gate. Every self-host instance that ingests is recorded here, + // but only REGISTERED ones count toward fleet calibration (computeFleetAnalytics). Bearer-gated by the + // `/v1/internal/*` middleware (INTERNAL_JOB_TOKEN). List shows pending + registered instances with their + // stored-signal counts so an operator knows what they're opting in before they register it. + app.get("/v1/internal/orb/instances", async (c) => { + const rows = await c.env.DB + .prepare( + `SELECT i.instance_id AS instanceId, i.registered AS registered, i.first_seen_at AS firstSeenAt, + i.last_seen_at AS lastSeenAt, i.registered_at AS registeredAt, + (SELECT COUNT(*) FROM orb_signals s WHERE s.instance_id = i.instance_id) AS signalCount + FROM orb_instances i ORDER BY i.last_seen_at DESC`, + ) + .all<{ instanceId: string; registered: number; firstSeenAt: string; lastSeenAt: string; registeredAt: string | null; signalCount: number }>(); + return c.json({ instances: (rows.results ?? []).map((r) => ({ ...r, registered: r.registered === 1 })) }); + }); + + // Opt an instance into (or out of) fleet calibration. Body: { instanceId, registered? } (registered + // defaults true). Upserts so an operator can register an instance that has ingested but isn't recorded yet. + app.post("/v1/internal/orb/instances/register", async (c) => { + const payload = (await c.req.json().catch(() => null)) as { instanceId?: unknown; registered?: unknown } | null; + const instanceId = typeof payload?.instanceId === "string" ? payload.instanceId : ""; + if (!instanceId) return c.json({ error: "instanceId required" }, 400); + const registered = payload?.registered === false ? 0 : 1; + await c.env.DB + .prepare( + `INSERT INTO orb_instances (instance_id, registered, registered_at) VALUES (?, ?, CURRENT_TIMESTAMP) + ON CONFLICT(instance_id) DO UPDATE SET registered = excluded.registered, + registered_at = CASE WHEN excluded.registered = 1 THEN CURRENT_TIMESTAMP ELSE NULL END`, + ) + .bind(instanceId, registered) + .run(); + return c.json({ instanceId, registered: registered === 1 }); + }); + // Convergence (ops / observability, flag GITTENSORY_REVIEW_OPS). Cross-repo review-OUTCOME aggregate (gate-block // ledger + recommendation/slop calibration) for an operator dashboard. Bearer-gated by the `/v1/internal/*` // middleware above (INTERNAL_JOB_TOKEN). Flag-OFF (default) → 404, so the endpoint does not exist and the diff --git a/src/orb/analytics.ts b/src/orb/analytics.ts index 0d773996fa..cc9d4f5292 100644 --- a/src/orb/analytics.ts +++ b/src/orb/analytics.ts @@ -91,6 +91,7 @@ export async function computeFleetAnalytics(env: Env, opts: { windowDays?: numbe let cells: Cell[] = []; let cycle: number[] = []; + let registered = new Set(); try { const matrix = await env.DB .prepare( @@ -106,6 +107,10 @@ export async function computeFleetAnalytics(env: Env, opts: { windowDays?: numbe .bind(cutoff) .all<{ ms: number }>(); cycle = (cy.results ?? []).map((r) => r.ms); + // The fleet trust gate: only operator-registered instances count toward the median (open ingest stores + // everyone's signals, but a stranger can't move calibration until a human opts them in — #1255). + const reg = await env.DB.prepare(`SELECT instance_id FROM orb_instances WHERE registered = 1`).all<{ instance_id: string }>(); + registered = new Set((reg.results ?? []).map((r) => r.instance_id)); } catch { return { windowDays, instanceCount: 0, fleet: { mergePrecision: null, closePrecision: null, fpRate: null, reversalRate: null, cycleP50Ms: null, cycleP95Ms: null }, instances: [], outliers: [] }; } @@ -119,8 +124,9 @@ export async function computeFleetAnalytics(env: Env, opts: { windowDays?: numbe } const instances = [...byInstance.entries()].map(([id, cs]) => foldInstance(id, cs)).sort((a, b) => a.instanceId.localeCompare(b.instanceId)); - // Fleet = median across instances with enough volume (robust to a single bad contributor). - const eligible = instances.filter((i) => i.decided >= MIN_DECIDED); + // Fleet = median across REGISTERED instances with enough volume (robust to a single bad contributor and + // to unregistered/untrusted senders — registration is the fleet's trust anchor). + const eligible = instances.filter((i) => i.decided >= MIN_DECIDED && registered.has(i.instanceId)); const nums = (sel: (i: InstanceMetrics) => number | null): number[] => eligible.map(sel).filter((v): v is number => v !== null); const fleetMergeP = median(nums((i) => i.mergePrecision)); const fleetCloseP = median(nums((i) => i.closePrecision)); diff --git a/src/orb/ingest.ts b/src/orb/ingest.ts index 05f6d946b9..10ce40c6a6 100644 --- a/src/orb/ingest.ts +++ b/src/orb/ingest.ts @@ -9,6 +9,42 @@ const VALID_REVERSALS = new Set(["none", "reopened", "reverted"]); const MIN_CYCLE_MS = 1_000; // <1s is implausible const MAX_CYCLE_MS = 31_536_000_000; // >1y is implausible +// 1 MiB comfortably holds a full MAX_BATCH (500) of small anonymized events (~hashes + numbers) with +// headroom, while bounding how much a hostile sender can make the collector buffer. Mirrors the +// body limit das-github-mirror puts in front of its open webhook ingress. +export const MAX_ORB_INGEST_BODY_BYTES = 1_048_576; + +function parseContentLength(header: string | null | undefined): number | null { + if (typeof header !== "string") return null; + const n = Number(header); + return Number.isInteger(n) && n >= 0 ? n : null; +} + +/** Read the request body with a hard byte ceiling so a hostile sender can't make us buffer unbounded + * input. Returns null when the body exceeds MAX_ORB_INGEST_BODY_BYTES (the caller answers 413). */ +export async function readOrbIngestBody(request: Request, contentLengthHeader: string | null | undefined): Promise { + const declared = parseContentLength(contentLengthHeader); + if (declared !== null && declared > MAX_ORB_INGEST_BODY_BYTES) return null; + + const stream = request.body; + if (!stream) return ""; + const reader = stream.getReader(); + const decoder = new TextDecoder(); + let total = 0; + let out = ""; + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + total += value.byteLength; + if (total > MAX_ORB_INGEST_BODY_BYTES) { + await reader.cancel(); + return null; + } + out += decoder.decode(value, { stream: true }); + } + return out + decoder.decode(); +} + interface OrbIngestEvent { repo_hash: string; pr_hash: string; @@ -55,6 +91,19 @@ export async function handleOrbIngest(body: string, db: D1Database): Promise { @@ -127,6 +127,61 @@ describe("handleOrbIngest()", () => { const db = { prepare: () => ({ bind: () => ({ run: () => Promise.resolve({ meta: { changes: 0 } }) }) }) } as unknown as D1Database; expect(await ingest(db, [ev()])).toEqual({ accepted: 0 }); }); + + it("records the instance on first contact (registered=0) and bumps last_seen on re-ingest", async () => { + const db = makeDb(); + await ingest(db, [ev({ pr_hash: "i1" })], "instX"); + const row = await (db as unknown as TestD1Database) + .prepare("SELECT registered, first_seen_at, last_seen_at FROM orb_instances WHERE instance_id=?") + .bind("instX") + .first<{ registered: number; first_seen_at: string; last_seen_at: string }>(); + expect(row?.registered).toBe(0); // not trusted until an operator registers it + await ingest(db, [ev({ pr_hash: "i2" })], "instX"); // same instance again → still one row + const cnt = await (db as unknown as TestD1Database).prepare("SELECT COUNT(*) AS n FROM orb_instances WHERE instance_id=?").bind("instX").first<{ n: number }>(); + expect(cnt?.n).toBe(1); + }); + + it("does not fail ingest if the instance bookkeeping upsert throws", async () => { + // First prepare() (orb_instances upsert) rejects; ingest must still process the batch best-effort. + let call = 0; + const db = { + prepare: (sql: string) => { + call++; + if (sql.includes("orb_instances")) return { bind: () => ({ run: () => Promise.reject(new Error("boom")) }) }; + return new TestD1Database().prepare(sql); + }, + } as unknown as D1Database; + expect(await ingest(db, [ev()])).toBeTruthy(); + expect(call).toBeGreaterThan(0); + }); +}); + +describe("readOrbIngestBody()", () => { + const reqWithBody = (body: BodyInit, headers?: Record) => + new Request("http://collector/v1/orb/ingest", { method: "POST", body, ...(headers ? { headers } : {}) }); + + it("reads a normal body", async () => { + expect(await readOrbIngestBody(reqWithBody("hello"), "5")).toBe("hello"); + }); + + it("returns '' when there is no request body", async () => { + expect(await readOrbIngestBody(new Request("http://collector", { method: "POST" }), null)).toBe(""); + }); + + it("rejects (null) when the declared content-length exceeds the cap — without reading", async () => { + expect(await readOrbIngestBody(reqWithBody("tiny"), String(MAX_ORB_INGEST_BODY_BYTES + 1))).toBeNull(); + }); + + it("ignores a non-numeric content-length and reads normally", async () => { + expect(await readOrbIngestBody(reqWithBody("ok"), "not-a-number")).toBe("ok"); + }); + + it("rejects (null) when the streamed body exceeds the cap with no declared length", async () => { + const big = new Uint8Array(MAX_ORB_INGEST_BODY_BYTES + 8); + const stream = new ReadableStream({ start(ctrl) { ctrl.enqueue(big); ctrl.close(); } }); + const req = new Request("http://collector", { method: "POST", body: stream, ...({ duplex: "half" } as object) }); + expect(await readOrbIngestBody(req, null)).toBeNull(); + }); }); describe("POST /v1/orb/ingest route", () => { @@ -150,6 +205,66 @@ describe("POST /v1/orb/ingest route", () => { const res = await app.request("/v1/orb/ingest", { method: "POST", body: "" }, createTestEnv()); expect(res.status).toBe(400); }); + + it("returns 413 when the body exceeds the ingest byte ceiling", async () => { + const huge = "x".repeat(MAX_ORB_INGEST_BODY_BYTES + 16); + const res = await app.request("/v1/orb/ingest", { method: "POST", body: huge }, createTestEnv()); + expect(res.status).toBe(413); + expect(((await res.json()) as { error: string }).error).toBe("payload_too_large"); + }); +}); + +describe("Orb instance registry routes (/v1/internal/orb/instances)", () => { + const app = createApp(); + const auth = { authorization: "Bearer dev-internal-token" }; + const ingestOne = (env: Env, instance: string) => + app.request("/v1/orb/ingest", { method: "POST", body: JSON.stringify({ instance_id: instance, events: [{ repo_hash: "r", pr_hash: `${instance}-p`, outcome: "merged" }] }) }, env); + + it("lists ingested instances as unregistered with their stored-signal count", async () => { + const env = createTestEnv(); + await ingestOne(env, "inst-a"); + const res = await app.request("/v1/internal/orb/instances", { headers: auth }, env); + expect(res.status).toBe(200); + const { instances } = (await res.json()) as { instances: Array<{ instanceId: string; registered: boolean; signalCount: number }> }; + expect(instances).toEqual([expect.objectContaining({ instanceId: "inst-a", registered: false, signalCount: 1 })]); + }); + + it("401 without the internal token", async () => { + expect((await app.request("/v1/internal/orb/instances", {}, createTestEnv())).status).toBe(401); + }); + + it("registers an instance (and can unregister it)", async () => { + const env = createTestEnv(); + await ingestOne(env, "inst-b"); + const reg = await app.request("/v1/internal/orb/instances/register", { method: "POST", headers: auth, body: JSON.stringify({ instanceId: "inst-b" }) }, env); + expect(((await reg.json()) as { registered: boolean }).registered).toBe(true); + const off = await app.request("/v1/internal/orb/instances/register", { method: "POST", headers: auth, body: JSON.stringify({ instanceId: "inst-b", registered: false }) }, env); + expect(((await off.json()) as { registered: boolean }).registered).toBe(false); + }); + + it("registers an instance that has not ingested yet (upsert)", async () => { + const env = createTestEnv(); + const reg = await app.request("/v1/internal/orb/instances/register", { method: "POST", headers: auth, body: JSON.stringify({ instanceId: "never-seen" }) }, env); + expect(reg.status).toBe(200); + const list = (await (await app.request("/v1/internal/orb/instances", { headers: auth }, env)).json()) as { instances: Array<{ instanceId: string; registered: boolean }> }; + expect(list.instances).toEqual([expect.objectContaining({ instanceId: "never-seen", registered: true })]); + }); + + it("400 when instanceId is missing", async () => { + const res = await app.request("/v1/internal/orb/instances/register", { method: "POST", headers: auth, body: JSON.stringify({}) }, createTestEnv()); + expect(res.status).toBe(400); + }); + + it("400 on a non-JSON register body (json().catch → null)", async () => { + const res = await app.request("/v1/internal/orb/instances/register", { method: "POST", headers: auth, body: "{bad" }, createTestEnv()); + expect(res.status).toBe(400); + }); + + it("tolerates a list query that omits results (rows.results ?? [])", async () => { + const env = { ...createTestEnv(), DB: { prepare: () => ({ all: () => Promise.resolve({}) }) } } as unknown as Env; + const res = await app.request("/v1/internal/orb/instances", { headers: auth }, env); + expect(((await res.json()) as { instances: unknown[] }).instances).toEqual([]); + }); }); describe("GET /v1/internal/fleet/analytics route", () => { diff --git a/test/unit/mcp-fleet-analytics.test.ts b/test/unit/mcp-fleet-analytics.test.ts index 2657a427cc..3ac63d3796 100644 --- a/test/unit/mcp-fleet-analytics.test.ts +++ b/test/unit/mcp-fleet-analytics.test.ts @@ -22,6 +22,8 @@ async function seedMergeSignals(env: Env, instance: string, n: number): Promise< .bind(instance, `repo${seq}`, `pr${seq++}`) .run(); } + // Register the instance so it counts toward the fleet (only registered instances are aggregated). + await env.DB.prepare(`INSERT INTO orb_instances (instance_id, registered) VALUES (?, 1) ON CONFLICT(instance_id) DO UPDATE SET registered=1`).bind(instance).run(); } describe("gittensory_get_fleet_analytics MCP tool", () => { diff --git a/test/unit/operator-dashboard.test.ts b/test/unit/operator-dashboard.test.ts index 70c40066e4..08170716ef 100644 --- a/test/unit/operator-dashboard.test.ts +++ b/test/unit/operator-dashboard.test.ts @@ -47,6 +47,9 @@ describe("operator dashboard payload", () => { await seed("good1", 5, "merged"); // precision 1.0 await seed("good2", 5, "merged"); // precision 1.0 await seed("bad", 5, "closed"); // precision 0.0 → outlier vs the median (1.0) + for (const id of ["good1", "good2", "bad"]) { + await env.DB.prepare(`INSERT INTO orb_instances (instance_id, registered) VALUES (?, 1)`).bind(id).run(); // only registered instances count + } const payload = await buildOperatorDashboardPayload(env); expect(payload.fleetMetrics.instanceCount).toBe(3); expect(payload.fleetMetrics.outliers.map((o) => o.instanceId)).toContain("bad"); diff --git a/test/unit/orb-analytics.test.ts b/test/unit/orb-analytics.test.ts index 65a78b4b5a..55c44823f0 100644 --- a/test/unit/orb-analytics.test.ts +++ b/test/unit/orb-analytics.test.ts @@ -21,6 +21,13 @@ async function signals( } } +/** Opt instances into fleet calibration — only registered instances count toward instanceCount/fleet. */ +async function register(env: Env, ...ids: string[]): Promise { + for (const id of ids) { + await env.DB.prepare(`INSERT INTO orb_instances (instance_id, registered) VALUES (?, 1) ON CONFLICT(instance_id) DO UPDATE SET registered=1`).bind(id).run(); + } +} + describe("computeFleetAnalytics()", () => { it("empty store → zeroed report (and a custom/clamped window)", async () => { const env = createTestEnv(); @@ -48,6 +55,13 @@ describe("computeFleetAnalytics()", () => { expect(a.instances).toEqual([]); }); + it("tolerates a registered-instances query that omits results (registered ?? [])", async () => { + // matrix/cycle use .bind().all(); the registered-set query uses .all() directly and returns no `results`. + const env = { DB: { prepare: () => ({ bind: () => ({ all: () => Promise.resolve({ results: [] }) }), all: () => Promise.resolve({}) }) } } as unknown as Env; + const a = await computeFleetAnalytics(env); + expect(a.instanceCount).toBe(0); + }); + it("computes per-instance precision incl. reversals (reverted merge = false positive)", async () => { const env = createTestEnv(); await signals(env, "inst1", 3, { verdict: "merge", outcome: "merged", reversal: "none" }); // confirmed @@ -88,6 +102,7 @@ describe("computeFleetAnalytics()", () => { await signals(env, "good2", 5, { verdict: "merge", outcome: "merged", ms: 2000 }); // precision 1.0 await signals(env, "bad", 5, { verdict: "merge", outcome: "closed", ms: 9000 }); // precision 0.0 → outlier await signals(env, "tiny", 2, { verdict: "merge", outcome: "closed" }); // below MIN_DECIDED → excluded from fleet + await register(env, "good1", "good2", "bad", "tiny"); // all trusted; only MIN_DECIDED gates the fleet here const a = await computeFleetAnalytics(env); expect(a.instanceCount).toBe(3); // good1, good2, bad (tiny excluded) expect(a.fleet.mergePrecision).toBe(1); // median of [1,1,0] @@ -101,7 +116,19 @@ describe("computeFleetAnalytics()", () => { const env = createTestEnv(); await signals(env, "a", 5, { verdict: "merge", outcome: "merged" }); // 1.0 await signals(env, "b", 5, { verdict: "merge", outcome: "closed" }); // 0.0 + await register(env, "a", "b"); const a = await computeFleetAnalytics(env); expect(a.fleet.mergePrecision).toBeCloseTo(0.5); // (1+0)/2 }); + + it("excludes unregistered instances from the fleet even with enough volume (registration is the trust gate)", async () => { + const env = createTestEnv(); + await signals(env, "trusted", 5, { verdict: "merge", outcome: "merged" }); + await signals(env, "stranger", 5, { verdict: "merge", outcome: "closed" }); // enough volume, but NOT registered + await register(env, "trusted"); + const a = await computeFleetAnalytics(env); + expect(a.instanceCount).toBe(1); // only the registered instance counts + expect(a.fleet.mergePrecision).toBe(1); // the stranger's 0.0 does not drag the median + expect(a.instances.map((i) => i.instanceId)).toContain("stranger"); // still visible per-instance for the operator + }); });