fix(orb): harden the federated peer-intelligence trust boundary (#9147, #9148, #9149, #9150, #9166)

migrations/0194_orb_relay_pending_enroll_id.sql

@@ -0,0 +1,11 @@
+-- Pull-mode drain isolation (#9150): orb_relay_pending was scoped only by installation_id, so ANY enrollment
+-- secret valid for an installation could drain and destructively-ack another live consumer's queued webhooks
+-- (a stale re-enrolled container silently stealing a fresh one's events). This column lets an enqueue tag
+-- which enrollment a GitHub-webhook row was queued for (the same "winning enrollment" forwardOrbEvent already
+-- picks for the push path, #1783); NULL for every existing row and for config_push notices (which remain
+-- intentionally installation-wide, not consumer-specific -- see enqueueConfigPushRelay). NULL also means "not
+-- yet tagged", so pullRelayPending's WHERE clause matches untagged rows regardless of which enrollment asks --
+-- a safe default during rollout, not a bypass (an enrollment can only ever add matches, never lose ones scoped
+-- to a DIFFERENT enroll_id).
+ALTER TABLE orb_relay_pending ADD COLUMN enroll_id TEXT;
+CREATE INDEX IF NOT EXISTS idx_orb_relay_pending_enroll ON orb_relay_pending (installation_id, enroll_id);

packages/loopover-engine/src/telemetry/anonymize.ts

@@ -14,6 +14,12 @@ import { createHmac, randomBytes } from "node:crypto";
  * once per instance in its own store and reuses it on every export, so the same raw value always hashes the
  * same way. Never derived from, or shared with, any other credential (App private keys, webhook secrets) --
  * key separation means a leaked anonymization secret can't be used to forge or decrypt anything else.
+ *
+ * #9147: this now also holds for Orb's opt-in federated peer-benchmark export (src/orb/federated-bundle.ts).
+ * That feature signs its bundles with its OWN dedicated secret (getOrCreateFederatedSigningSecret) rather
+ * than this one, and derives its instance identity from a third, independent secret
+ * (getOrCreateInstanceIdentitySecret) -- so enrolling a federated peer, or rotating any one of the three
+ * secrets, can never affect this anonymization secret or the always-on export that uses it.
  */
 export function generateAnonSecret(): string {
   return randomBytes(32).toString("hex");

src/api/routes.ts

@@ -166,6 +166,7 @@ import {
   isOrbBrokerEnabled,
   issueOrbEnrollment,
   issueOrbStoredSecret,
+  ORB_SECRET_TYPE_GITHUB_TOKEN,
   ORB_SECRET_TYPE_TENANT_DB_CREDENTIAL,
   revokeOrbEnrollment,
 } from "../orb/broker";
@@ -1761,14 +1762,15 @@ export function createApp() {
     });
     // Federated benchmark (#6481): "your gate precision vs peer median". Reads the opt-in from the loopover
     // self-repo's manifest (mirrors prReconciliation/publicStats/etc.'s fleet-wide override lookup) rather
-    // than any of the maintainer's own repos — federatedIntelligence is operator-level, not per-repo. Bounded
-    // to a single, short-timeout attempt so an unreachable or slow collector degrades the panel to its
-    // existing empty state instead of holding up the whole dashboard load.
+    // than any of the maintainer's own repos — federatedIntelligence is operator-level, not per-repo.
+    // #9148: buildFederatedBenchmark no longer pulls a peer collector on this request path at all — the
+    // local half is a single fast DB query, and the peer half is read from a cache the "federated-peer-sync"
+    // background queue job refreshes on its own cadence. This used to be a live, rate-limited network call
+    // on every dashboard load (N maintainers hitting refresh was N requests/second at the peer collector);
+    // now a slow/unreachable collector can only make the CACHE stale, never hold up this request.
     const federatedIntelligenceManifest = await loadRepoFocusManifest(c.env, resolveLoopOverSelfRepoFullName(c.env));
     const federatedBenchmark = await buildFederatedBenchmark(federatedIntelligenceManifest, c.env.DB, {
       now: Date.parse(generatedAt),
-      timeoutMs: 5_000,
-      maxAttempts: 1,
     });
     const qualityDashboard = {
       ...buildMaintainerQualityDashboard({
@@ -4441,16 +4443,19 @@ export function createApp() {
     } catch {
       ack = undefined; // tolerate an empty/invalid body — just no ack this round
     }
-    const events = await pullRelayPending(c.env, enrollment.installationId, { ack }).catch(dbBrokerError);
+    // #9150: scope the drain to THIS enrollment (or untagged/legacy rows) — not merely the installation — so a
+    // second live enrollment for the same install can never steal and destructively-ack this one's events.
+    const events = await pullRelayPending(c.env, enrollment.installationId, { ack, enrollId: enrollment.enrollId }).catch(dbBrokerError);
     if (events === null) return c.json({ error: "broker_error" }, 503);
     return c.json({ events }, 200);
   });
 
   // LoopOver 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).
+  // #9046/#9166: FAILS CLOSED when ORB_INGEST_TOKEN is unset — isAuthorizedIngest (below, shared with
+  // /v1/ams/ingest) requires an exact bearer match against the configured token; an unconfigured collector
+  // rejects rather than accepting anonymous writes. Bounded by a hard body ceiling, and dedup'd via
+  // UNIQUE(instance_id, repo_hash, pr_hash).
   app.post("/v1/orb/ingest", async (c) => {
     if (!(await isAuthorizedIngest(c.env.ORB_INGEST_TOKEN, extractBearerToken(c.req.header("authorization"))))) return c.json({ error: "unauthorized" }, 401);
     const body = await readOrbIngestBody(c.req.raw, c.req.header("content-length"));
@@ -4575,15 +4580,22 @@ export function createApp() {
   // records lands at registered=0; only REGISTERED ones count toward the global public counter (getOrbGlobalStats)
   // and are eligible for token brokering. Bearer-gated by the `/v1/internal/*` middleware (INTERNAL_JOB_TOKEN). The
   // list shows pending + registered installs so an operator knows what they're opting in.
+  //
+  // liveEnrollmentCount (#9149): how many enrollment secrets are currently live ('enrolled', not revoked) for
+  // this install — an operator previously had no way to see that a re-enrollment (issueOrbEnrollment was a bare
+  // INSERT, never revoking) had accumulated a second standing credential. A correlated subquery rather than a
+  // JOIN: each installation has at most a handful of enrollment rows, so this stays cheap without reshaping the
+  // one-row-per-installation result set a GROUP BY would require.
   app.get("/v1/internal/orb/installations", async (c) => {
     const rows = await c.env.DB
       .prepare(
         `SELECT installation_id AS installationId, account_login AS accountLogin, account_type AS accountType,
                 repository_selection AS repositorySelection, registered, suspended_at AS suspendedAt,
-                removed_at AS removedAt, first_seen_at AS firstSeenAt, last_event_at AS lastEventAt
+                removed_at AS removedAt, first_seen_at AS firstSeenAt, last_event_at AS lastEventAt,
+                (SELECT COUNT(*) FROM orb_enrollments oe WHERE oe.installation_id = orb_github_installations.installation_id AND oe.state = 'enrolled' AND oe.revoked_at IS NULL) AS liveEnrollmentCount
          FROM orb_github_installations ORDER BY last_event_at DESC`,
       )
-      .all<{ installationId: number; accountLogin: string | null; accountType: string | null; repositorySelection: string | null; registered: number; suspendedAt: string | null; removedAt: string | null; firstSeenAt: string; lastEventAt: string }>();
+      .all<{ installationId: number; accountLogin: string | null; accountType: string | null; repositorySelection: string | null; registered: number; suspendedAt: string | null; removedAt: string | null; firstSeenAt: string; lastEventAt: string; liveEnrollmentCount: number }>();
     return c.json({ installations: (rows.results ?? []).map((r) => ({ ...r, registered: r.registered === 1 })) });
   });
 
@@ -4619,9 +4631,15 @@ export function createApp() {
   // secret issuance path control-plane's hosted provisioning core (#7180/#8066) calls instead, for a credential
   // that already exists (a tenant's Postgres connection string) rather than a GitHub installation to bind.
   // `installationId` is irrelevant to that path (see issueOrbStoredSecret's own header comment for why).
+  //
+  // Also accepts an optional `{ rotate: true }` (#9149): when set, every PRIOR live enrollment for this
+  // installation is revoked before the new one is minted, mirroring the OAuth landing page's `state=<id>:rotate`
+  // flow (oauth.ts) for the operator-issued path. Defaults to false (append, not replace) -- unchanged behavior
+  // for every existing caller, since a blue/green container swap legitimately relies on two live enrollments
+  // briefly overlapping (see #9150's sibling fix, which makes that overlap safe rather than assuming away).
   app.post("/v1/internal/orb/enrollments", async (c) => {
     if (!isOrbBrokerEnabled(c.env)) return c.json({ error: "not_found" }, 404);
-    const payload = (await c.req.json().catch(() => null)) as { installationId?: unknown; secretType?: unknown; secretValue?: unknown } | null;
+    const payload = (await c.req.json().catch(() => null)) as { installationId?: unknown; secretType?: unknown; secretValue?: unknown; rotate?: unknown } | null;
     if (payload?.secretType === ORB_SECRET_TYPE_TENANT_DB_CREDENTIAL) {
       const secretValue = typeof payload.secretValue === "string" ? payload.secretValue : "";
       const result = await issueOrbStoredSecret(c.env, ORB_SECRET_TYPE_TENANT_DB_CREDENTIAL, secretValue);
@@ -4630,7 +4648,7 @@ export function createApp() {
     }
     const installationId = Number(payload?.installationId);
     if (!Number.isInteger(installationId) || installationId <= 0) return c.json({ error: "installationId required" }, 400);
-    const result = await issueOrbEnrollment(c.env, installationId);
+    const result = await issueOrbEnrollment(c.env, installationId, undefined, ORB_SECRET_TYPE_GITHUB_TOKEN, { rotate: payload?.rotate === true });
     if ("error" in result) return c.json(result, result.error === "installation_not_found" ? 404 : 409);
     return c.json(result); // { enrollId, secret } — secret shown exactly once
   });
@@ -6690,10 +6708,6 @@ function toIsoQueryDate(value: string): string | 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.
 /**
  * #9046: the SINGLE authorization rule for every telemetry-collector ingest endpoint (Orb and AMS), so the two
  * products cannot drift apart again. They previously had separate, near-identical copies — and AMS ended up

src/db/schema.ts

@@ -977,6 +977,11 @@ export const orbRelayPending = sqliteTable(
     // enqueueConfigPushRelay). The dispatch side (#7523) branches on this before treating raw_body as a
     // GitHubWebhookPayload.
     kind: text("kind").notNull().default("github_webhook"),
+    // Pull-mode drain isolation (#9150): which enrollment a GitHub-webhook row was queued for (the same
+    // "winning enrollment" forwardOrbEvent picks for the push path). NULL for config_push rows (installation-
+    // wide, not consumer-specific) and for any pre-#9150 row -- pullRelayPending's WHERE matches NULL
+    // regardless of the asking enrollment, so an untagged row is never silently orphaned.
+    enrollId: text("enroll_id"),
   },
   (table) => ({
     installation: index("idx_orb_relay_pending_install").on(table.installationId, table.createdAt),
@@ -986,6 +991,8 @@ export const orbRelayPending = sqliteTable(
     // pruneRelayPending (src/orb/relay.ts) filters/deletes by created_at alone (fleet-wide TTL sweep, not
     // scoped to one installation) -- neither index above leads with created_at, so that scan was unindexed.
     createdAt: index("idx_orb_relay_pending_created_at").on(table.createdAt),
+    // #9150: pullRelayPending's SELECT/DELETE now filter on (installation_id, enroll_id) together.
+    enroll: index("idx_orb_relay_pending_enroll").on(table.installationId, table.enrollId),
   }),
 );
 

src/index.ts

@@ -14,6 +14,7 @@ import { isLoopEscalationSweepEnabled } from "./review/loop-escalation-wire";
 import { isAprRepoTransferPollEnabled } from "./orb/apr-repo-transfer";
 import { isPrReconciliationEnabled, resolvePrReconciliationManifestOverride } from "./review/pr-reconciliation";
 import { isActiveReviewReconciliationEnabled, resolveActiveReviewReconciliationManifestOverride } from "./review/active-review-reconciliation";
+import { resolveFederatedIntelligenceManifestOverride } from "./orb/federated-benchmark";
 import { isRagEnabled } from "./review/rag-wire";
 import { isDecisionAuditEnabled } from "./review/decision-audit";
 import { isRiskControlEnabled } from "./review/risk-control-wire";
@@ -226,6 +227,16 @@ async function enqueueScheduledJobs(env: Env, controller: ScheduledController):
       jobs.push({ type: "reconcile-active-review-tracking", requestedBy: "schedule" });
     }
   }
+  // Federated peer-sync tick (#9148/#9166). Same 10-minute cadence as the reconciliation jobs above — a
+  // best-effort background sync has no business running any more often than that, and the collector's own
+  // module-level rate limiter (federated-collector.ts, 6/min) further bounds any single tick's calls. Opt-in
+  // via the loopover self-repo's `.loopover.yml federatedIntelligence:` block only — there is no env-var
+  // equivalent for this one (unlike prReconciliation/activeReviewReconciliation above), since federated intel
+  // has never had one; flag-OFF (default, `present: false`) this job is never created.
+  if (isReconciliationWindow) {
+    const federatedIntelligenceManifestOverride = await resolveFederatedIntelligenceManifestOverride(env);
+    if (federatedIntelligenceManifestOverride.enabled) jobs.push({ type: "federated-peer-sync", requestedBy: "schedule" });
+  }
   if (isHourly) {
     // Isolation (#experimental-gittensor-plugin): on self-host, refresh-registry both FETCHES from and
     // PERSISTS the whole upstream gittensor-subnet registry (entrius/gittensor has no server-side filtering,

src/orb/broker.ts

@@ -53,16 +53,26 @@ export type IssueResult = { enrollId: string; secret: string } | { error: "insta
  *  hashed). Issued by the operator (internal endpoint) OR by a maintainer who proved install-admin via OAuth —
  *  in the latter case the maintainer's GitHub identity is recorded for audit. installation_id is bound here and
  *  read back (never from the request) at token-exchange time, so a secret can never mint a token for another
- *  install. `secretType` defaults to the only mintable type today; every existing caller is unaffected. */
+ *  install. `secretType` defaults to the only mintable type today; every existing caller is unaffected.
+ *
+ *  `opts.rotate` (#9149): this was previously a bare INSERT, so re-running the enrollment flow (e.g. after a
+ *  secret leak) minted a SECOND simultaneously-valid secret and revoked nothing — the leaked one kept working
+ *  forever. Deliberately NOT the default: a blue/green container swap or a deliberate multi-container setup
+ *  legitimately relies on two live enrollments overlapping for a short window (see #9150's sibling fix, which
+ *  makes that overlap safe rather than assuming it can't happen) — auto-revoking on every issuance would break
+ *  that. `rotate: true` is the caller's explicit "I want this to be the only valid secret now" confirmation
+ *  (the OAuth landing page's `?rotate=1`/`state=<id>:rotate` flow, oauth.ts). */
 export async function issueOrbEnrollment(
   env: Env,
   installationId: number,
   maintainer?: { login: string; githubId?: number | null | undefined },
   secretType: string = ORB_SECRET_TYPE_GITHUB_TOKEN,
+  opts: { rotate?: boolean } = {},
 ): Promise<IssueResult> {
   const install = await env.DB.prepare("SELECT registered FROM orb_github_installations WHERE installation_id = ?").bind(installationId).first<{ registered: number }>();
   if (!install) return { error: "installation_not_found" };
   if (install.registered !== 1) return { error: "installation_not_registered" };
+  if (opts.rotate === true) await revokeAllLiveEnrollmentsForInstallation(env, installationId);
   const enrollId = createOpaqueToken("orbenr");
   const secret = createOpaqueToken("orbsec");
   await env.DB.prepare(
@@ -74,6 +84,28 @@ export async function issueOrbEnrollment(
   return { enrollId, secret };
 }
 
+/** Revokes every currently-live ('enrolled', not yet revoked) enrollment for an installation, regardless of
+ *  secret_type — the self-hoster-reachable counterpart to the single-enrollment {@link revokeOrbEnrollment}
+ *  (#9149). Returns the count actually revoked (0 is a valid, non-error result: an install with no live
+ *  enrollment is not a failure, it's just already in the state the caller wanted). Deliberately does NOT gate
+ *  on `registered`/`suspended_at`/`removed_at` the way issuance does — revoking access should work EVEN ON an
+ *  installation the operator has since disabled or suspended, since a maintainer revoking a leaked secret has
+ *  no reason to be blocked by an unrelated administrative state. */
+export async function revokeAllLiveEnrollmentsForInstallation(env: Env, installationId: number): Promise<number> {
+  const result = await env.DB.prepare("UPDATE orb_enrollments SET revoked_at = CURRENT_TIMESTAMP, state = 'revoked' WHERE installation_id = ? AND state = 'enrolled' AND revoked_at IS NULL")
+    .bind(installationId)
+    .run();
+  return result.meta.changes;
+}
+
+/** How many enrollments are currently live ('enrolled', not revoked) for an installation (#9149) — the count
+ *  surfaced to an operator (the internal installations list) and to a maintainer (the OAuth secret page) so
+ *  neither has to guess whether a re-enrollment minted an extra standing credential. */
+export async function countLiveEnrollmentsForInstallation(env: Env, installationId: number): Promise<number> {
+  const row = await env.DB.prepare("SELECT COUNT(*) AS c FROM orb_enrollments WHERE installation_id = ? AND state = 'enrolled' AND revoked_at IS NULL").bind(installationId).first<{ c: number }>();
+  return row?.c ?? 0;
+}
+
 export type IssueStoredSecretResult = IssueResult | { error: "secret_value_required" | "encryption_unavailable" };
 
 /** Issues a one-time enrollment secret for a STORED (not minted) credential (#8064) -- e.g. control-plane's