diff --git a/.loopover.yml.example b/.loopover.yml.example index 06446800af..43c0f050bd 100644 --- a/.loopover.yml.example +++ b/.loopover.yml.example @@ -1277,3 +1277,9 @@ settings: # peer-to-peer and assumes no central service at all. # federatedIntelligence: # enabled: true # Bool. Default: false (nothing is exported). +# # The collector YOU run or trust (#6479). There is deliberately NO default and no auto-discovery -- with +# # this unset, `enabled: true` still sends nothing, because there is nowhere to send it. Must be a public +# # HTTPS URL: it is validated at config-read time against the same SSRF guard every other URL field here +# # uses, so an http:// or localhost/private-range endpoint is dropped with a warning. +# collectorUrl: https://collector.example.org/v1/federated +# collectorMode: both # push | pull | both. Default: both. diff --git a/config/examples/loopover.full.yml b/config/examples/loopover.full.yml index 5a4fefb295..95c88f04ac 100644 --- a/config/examples/loopover.full.yml +++ b/config/examples/loopover.full.yml @@ -1291,3 +1291,9 @@ settings: # peer-to-peer and assumes no central service at all. # federatedIntelligence: # enabled: true # Bool. Default: false (nothing is exported). +# # The collector YOU run or trust (#6479). There is deliberately NO default and no auto-discovery -- with +# # this unset, `enabled: true` still sends nothing, because there is nowhere to send it. Must be a public +# # HTTPS URL: it is validated at config-read time against the same SSRF guard every other URL field here +# # uses, so an http:// or localhost/private-range endpoint is dropped with a warning. +# collectorUrl: https://collector.example.org/v1/federated +# collectorMode: both # push | pull | both. Default: both. diff --git a/packages/loopover-engine/src/focus-manifest.ts b/packages/loopover-engine/src/focus-manifest.ts index 1572a92fda..02f2dd433a 100644 --- a/packages/loopover-engine/src/focus-manifest.ts +++ b/packages/loopover-engine/src/focus-manifest.ts @@ -435,10 +435,29 @@ export type FocusManifestUpstreamDriftIssuesConfig = { * federated env var; not present ⇒ disabled ⇒ nothing is bundled and no network call is made, byte-identical * to before this override existed. (ORB_AIR_GAP gates the separate, always-on #1255 orb telemetry path in * src/selfhost/orb-collector.ts and is unrelated to this opt-in.) + * + * `collectorUrl`/`collectorMode` (#6479) additionally arm the transport client + * (src/orb/federated-collector.ts), which pushes this instance's bundle to — and/or pulls peer bundles from — + * an endpoint the OPERATOR configures. `enabled: true` alone still exports nothing over the wire: without a + * `collectorUrl` there is nowhere to send it, and there is deliberately no default collector to fall back to. */ +export const FEDERATED_COLLECTOR_MODES = ["push", "pull", "both"] as const; +export type FederatedCollectorMode = (typeof FEDERATED_COLLECTOR_MODES)[number]; + export type FocusManifestFederatedIntelligenceConfig = { present: boolean; enabled: boolean; + /** + * The operator-configured collector this instance pushes its own bundle to and/or pulls peer bundles from + * (#6479). Null unless the operator sets one: there is deliberately NO hardcoded or auto-discovered default, + * because this codebase's self-host posture assumes no central/managed collector exists. Validated at + * config-read time against the same `isSafeHttpUrl` SSRF guard every other URL-valued manifest field uses, + * so it must be a public HTTPS host — a collector an operator runs is reachable at one, and a loopback/ + * private-range target would be both unreachable from a Worker and an SSRF footgun. + */ + collectorUrl: string | null; + /** Which directions the client may use against `collectorUrl`. Null ⇒ `both`. */ + collectorMode: FederatedCollectorMode | null; }; /** @@ -1191,6 +1210,8 @@ const EMPTY_UPSTREAM_DRIFT_ISSUES_CONFIG: FocusManifestUpstreamDriftIssuesConfig const EMPTY_FEDERATED_INTELLIGENCE_CONFIG: FocusManifestFederatedIntelligenceConfig = { present: false, enabled: false, + collectorUrl: null, + collectorMode: null, }; const EMPTY_MANIFEST: FocusManifest = { @@ -2076,7 +2097,23 @@ function parseFederatedIntelligenceConfig(value: JsonValue | undefined, warnings } const record = value as Record; const enabled = normalizeOptionalBoolean(record.enabled, "federatedIntelligence.enabled", warnings) ?? false; - return { present: true, enabled }; + const collectorUrl = parseFederatedCollectorUrl(record.collectorUrl, warnings); + const collectorMode = normalizeOptionalEnum(record.collectorMode, "federatedIntelligence.collectorMode", FEDERATED_COLLECTOR_MODES, warnings); + return { present: true, enabled, collectorUrl, collectorMode }; +} + +/** Parse `federatedIntelligence.collectorUrl` (#6479) — validated at CONFIG-READ time against the same + * `isSafeHttpUrl` SSRF guard every other URL-valued manifest field uses (mirrors + * {@link parseVisualProductionUrl}). A non-HTTPS or private/loopback host is dropped with a warning rather + * than accepted, so an unsafe endpoint can never reach the transport client at all. */ +function parseFederatedCollectorUrl(value: JsonValue | undefined, warnings: string[]): string | null { + const url = parsePublicSafeText(value, "federatedIntelligence.collectorUrl", warnings); + if (url === null) return null; + if (!isSafeHttpUrl(url)) { + warnings.push(`Manifest "federatedIntelligence.collectorUrl" must be a valid HTTPS URL targeting a public host; ignoring it.`); + return null; + } + return url; } /** Serialize a federatedIntelligence config back into the parse-compatible shape so a cached snapshot @@ -2084,7 +2121,7 @@ function parseFederatedIntelligenceConfig(value: JsonValue | undefined, warnings * configured. */ export function federatedIntelligenceConfigToJson(config: FocusManifestFederatedIntelligenceConfig): JsonValue { if (!config.present) return null; - return { enabled: config.enabled }; + return { enabled: config.enabled, collectorUrl: config.collectorUrl, collectorMode: config.collectorMode }; } function normalizeOptionalEnum(value: JsonValue | undefined, field: string, allowed: readonly T[], warnings: string[]): T | null { diff --git a/packages/loopover-engine/src/index.ts b/packages/loopover-engine/src/index.ts index 6d551d6ddd..72be673ce8 100644 --- a/packages/loopover-engine/src/index.ts +++ b/packages/loopover-engine/src/index.ts @@ -758,6 +758,7 @@ export { draftFlowConfigToJson, upstreamDriftIssuesConfigToJson, federatedIntelligenceConfigToJson, + FEDERATED_COLLECTOR_MODES, settingsOverrideToJson, MAX_FOCUS_MANIFEST_BYTES, CONVERGED_FEATURE_KEYS, @@ -797,6 +798,7 @@ export { type FocusManifestDraftFlowConfig, type FocusManifestUpstreamDriftIssuesConfig, type FocusManifestFederatedIntelligenceConfig, + type FederatedCollectorMode, type FocusManifestSettings, type FocusManifestSource, type LinkedIssueSatisfactionMode, diff --git a/src/orb/federated-collector.ts b/src/orb/federated-collector.ts new file mode 100644 index 0000000000..0e6149c08c --- /dev/null +++ b/src/orb/federated-collector.ts @@ -0,0 +1,190 @@ +// LoopOver federated fleet intelligence (#1970) — OPT-IN collector TRANSPORT client (#6479). +// +// Moves the anonymized bundles built by ./federated-bundle.ts (#6478) between self-hosted instances. Two +// directions, both best-effort and both off by default: +// push — POST this instance's own bundle to the operator's configured collector. +// pull — GET peer bundles from it. +// +// SCOPE — deliberately NOT the import side. A pulled bundle is fetched, shape-checked and RETURNED; it is +// never signature-verified, never trust-gated, and never persisted. That is #6480's job, and #6480 is blocked +// on #6477 (the key-trust/anti-poisoning design). Verifying here would not merely be out of scope, it would be +// WRONG: there is no trust anchor to verify against yet, and inventing one is exactly what #6477 exists to +// prevent (see the TODO(#6477) note on signFederatedBundle in ./federated-bundle.ts). +// +// NO DEFAULT COLLECTOR, BY DESIGN. The client only ever talks to an endpoint the operator configured in +// `.loopover.yml`. There is no hardcoded fallback and no auto-discovery — this codebase's self-host posture +// assumes no central/managed collector exists. (Contrast the #1255 orb path, which does POST to a hosted +// default at src/selfhost/orb-collector.ts:168; that is a different feature with a different contract.) +// +// FAIL-SAFE, ALWAYS. Every entry point resolves the opt-in BEFORE touching the database or the network, wraps +// its whole body in a catch, and degrades to a falsy result. Nothing here throws, so the review/gate path can +// never be slowed or broken by a collector that is unreachable, slow, rate-limited or returning garbage. The +// gate never awaits this; it is background, best-effort sync. +import { + evaluateLocalRateLimit, + jitteredBackoffMs, + type LocalRateBucket, +} from "@loopover/engine"; +import { isSafeHttpUrl } from "../review/content-lane/safe-url"; +import { buildFederatedBundle, FEDERATED_BUNDLE_SCHEMA_VERSION, type FederatedSignalBundle } from "./federated-bundle"; +import type { FocusManifest } from "../signals/focus-manifest"; + +/** Matches every other outbound call in this subsystem (orb-collector.ts:215's 30s export tick). */ +const DEFAULT_TIMEOUT_MS = 30_000; +/** Total attempts per direction, including the first. Mirrors the miner's fetchWithRetry contract. */ +const DEFAULT_MAX_ATTEMPTS = 3; +/** Exponential base for the jittered backoff between retries. */ +const RETRY_BASE_MS = 500; +/** A best-effort background sync has no business hammering a peer's collector. */ +const RATE_LIMIT: { limit: number; windowMs: number } = { limit: 6, windowMs: 60_000 }; + +type ManifestSlice = Pick | null | undefined; + +export type CollectorOpts = { + /** Injected so tests never touch the real network (orb-collector.ts:155's fetchFn idiom). */ + fetchFn?: typeof fetch; + timeoutMs?: number; + maxAttempts?: number; + /** Injected so a retry costs no wall-clock in tests. */ + sleepFn?: (ms: number) => Promise; + /** Injected random for the jitter — jitteredBackoffMs never reads Math.random itself. */ + randomFn?: () => number; + /** Caller-owned rolling-window bucket. Omitted ⇒ no local rate limiting is applied. */ + bucket?: LocalRateBucket; + now?: number; +}; + +/** + * The collector endpoint armed for `direction`, or null when this instance must not talk to anyone: not opted + * in, no collector configured, the configured URL failed the SSRF guard at parse time, or the operator scoped + * `collectorMode` to the other direction. Callers MUST consult this before touching the network or the DB. + */ +export function resolveCollectorEndpoint(manifest: ManifestSlice, direction: "push" | "pull"): string | null { + const config = manifest?.federatedIntelligence; + if (config?.enabled !== true) return null; + const url = config.collectorUrl; + if (url === null || url === undefined) return null; + // Defense in depth: the URL was already guarded at config-read time, but re-check at call time exactly as + // src/orb/relay.ts:230 does — a snapshot round-tripped through KV must not be trusted to have been parsed + // by the current guard. + if (!isSafeHttpUrl(url)) return null; + const mode = config.collectorMode ?? "both"; + if (mode !== "both" && mode !== direction) return null; + return url; +} + +/** True when the caller's bucket still permits an attempt. No bucket ⇒ unlimited. */ +function rateLimitAllows(opts: CollectorOpts, now: number): boolean { + if (!opts.bucket) return true; + return evaluateLocalRateLimit(opts.bucket, RATE_LIMIT, now).allowed; +} + +/** A 4xx is the operator's own misconfiguration and will fail identically on a retry; only 5xx/network is + * worth another attempt. Mirrors packages/loopover-miner/lib/http-retry.js's 5xx-only contract. */ +function isRetryableStatus(status: number): boolean { + return status >= 500; +} + +/** + * One fetch with a bounded timeout, retried on 5xx/network with jittered exponential backoff. Returns the + * Response on a 2xx, or null once attempts are exhausted / a non-retryable status arrives. Never throws. + */ +async function fetchWithRetry(url: string, init: RequestInit, opts: CollectorOpts): Promise { + const doFetch = opts.fetchFn ?? globalThis.fetch; + const timeoutMs = Number.isFinite(opts.timeoutMs) ? (opts.timeoutMs as number) : DEFAULT_TIMEOUT_MS; + const maxAttempts = Number.isFinite(opts.maxAttempts) ? Math.max(1, opts.maxAttempts as number) : DEFAULT_MAX_ATTEMPTS; + const sleep = opts.sleepFn ?? ((ms: number) => new Promise((resolve) => setTimeout(resolve, ms))); + const random = opts.randomFn ?? Math.random; + + for (let attempt = 0; attempt < maxAttempts; attempt++) { + try { + const response = await doFetch(url, { ...init, signal: AbortSignal.timeout(timeoutMs) }); + if (response.ok) return response; + // A 4xx is the operator's own misconfiguration and will fail identically next time. + if (!isRetryableStatus(response.status)) return null; + } catch { + // A timeout, DNS failure or connection reset — indistinguishable to us, and all worth one more try. + } + if (attempt === maxAttempts - 1) return null; + await sleep(jitteredBackoffMs(RETRY_BASE_MS, attempt, random)); + } + /* v8 ignore next -- unreachable: maxAttempts is clamped to >= 1, so the final iteration always returns above */ + return null; +} + +/** Is this parsed value a bundle we understand? Shape only — NOT a signature/trust check (#6477/#6480). */ +function isBundleShaped(value: unknown): value is FederatedSignalBundle { + if (typeof value !== "object" || value === null) return false; + const b = value as Record; + return ( + b.schemaVersion === FEDERATED_BUNDLE_SCHEMA_VERSION && + typeof b.instanceId === "string" && + typeof b.generatedAt === "string" && + typeof b.windowDays === "number" && + typeof b.decided === "number" && + typeof b.signature === "string" + ); +} + +/** + * Push this instance's own bundle to the operator's configured collector. + * + * Returns false — having touched neither the database nor the network — unless the operator opted in AND + * configured a push-armed collector. Returns false rather than throwing on any failure. The body is exactly + * the anonymized bundle from #6478: no code, no diffs, no logins, no repo names. + */ +export async function pushFederatedBundle(manifest: ManifestSlice, db: D1Database, opts: CollectorOpts = {}): Promise { + const endpoint = resolveCollectorEndpoint(manifest, "push"); + if (endpoint === null) return false; + + try { + const now = Number.isFinite(opts.now) ? (opts.now as number) : Date.now(); + if (!rateLimitAllows(opts, now)) return false; + + const bundle = await buildFederatedBundle(manifest, db, opts.now === undefined ? {} : { now: opts.now }); + // The builder already fails safe to null; nothing to send is not a failure worth retrying. + if (bundle === null) return false; + + const response = await fetchWithRetry( + endpoint, + { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(bundle) }, + opts, + ); + return response !== null; + } catch (error) { + console.error( + JSON.stringify({ level: "error", event: "federated_push_failed", message: String(error).slice(0, 200) }), + ); + return false; + } +} + +/** + * Pull peer bundles from the operator's configured collector. + * + * Returns [] — having touched nothing — unless the operator opted in AND configured a pull-armed collector. + * Bundles are shape-checked and returned; unrecognized entries are dropped. They are deliberately NOT + * signature-verified or trust-gated — that is #6480, blocked on #6477. Returns [] rather than throwing on any + * failure, so an unreachable or hostile collector is indistinguishable from "no peers yet" to every caller. + */ +export async function pullPeerBundles(manifest: ManifestSlice, opts: CollectorOpts = {}): Promise { + const endpoint = resolveCollectorEndpoint(manifest, "pull"); + if (endpoint === null) return []; + + try { + const now = Number.isFinite(opts.now) ? (opts.now as number) : Date.now(); + if (!rateLimitAllows(opts, now)) return []; + + const response = await fetchWithRetry(endpoint, { method: "GET", headers: { accept: "application/json" } }, opts); + if (response === null) return []; + + const payload: unknown = await response.json(); + if (!Array.isArray(payload)) return []; + return payload.filter(isBundleShaped); + } catch (error) { + console.error( + JSON.stringify({ level: "error", event: "federated_pull_failed", message: String(error).slice(0, 200) }), + ); + return []; + } +} diff --git a/src/signals/focus-manifest.ts b/src/signals/focus-manifest.ts index 69127423a5..8617ff86ee 100644 --- a/src/signals/focus-manifest.ts +++ b/src/signals/focus-manifest.ts @@ -40,6 +40,7 @@ export { draftFlowConfigToJson, upstreamDriftIssuesConfigToJson, federatedIntelligenceConfigToJson, + FEDERATED_COLLECTOR_MODES, settingsOverrideToJson, type AiReviewCadence, type AutoReviewConfig, @@ -71,6 +72,7 @@ export { type FocusManifestDraftFlowConfig, type FocusManifestUpstreamDriftIssuesConfig, type FocusManifestFederatedIntelligenceConfig, + type FederatedCollectorMode, type FocusManifestSettings, type FocusManifestSource, type LinkedIssueSatisfactionMode, diff --git a/test/unit/federated-bundle.test.ts b/test/unit/federated-bundle.test.ts index 5ccb4521e4..00b3b4fd10 100644 --- a/test/unit/federated-bundle.test.ts +++ b/test/unit/federated-bundle.test.ts @@ -69,7 +69,14 @@ async function resolved( /** A manifest carrying only the block the builder reads. */ function manifest(enabled: boolean | undefined): Pick { - return { federatedIntelligence: { present: enabled !== undefined, enabled: enabled ?? false } }; + return { + federatedIntelligence: { + present: enabled !== undefined, + enabled: enabled ?? false, + collectorUrl: null, + collectorMode: null, + }, + }; } /** A db that fails the test if it is touched at all — proves the opted-out path reads nothing. */ diff --git a/test/unit/federated-collector.test.ts b/test/unit/federated-collector.test.ts new file mode 100644 index 0000000000..83fd3d27dd --- /dev/null +++ b/test/unit/federated-collector.test.ts @@ -0,0 +1,374 @@ +import { DatabaseSync } from "node:sqlite"; +import { describe, expect, it, vi } from "vitest"; +import { createD1Adapter, nodeSqliteDriver } from "../../src/selfhost/d1-adapter"; +import { FEDERATED_BUNDLE_SCHEMA_VERSION, type FederatedSignalBundle } from "../../src/orb/federated-bundle"; +import { + pullPeerBundles, + pushFederatedBundle, + resolveCollectorEndpoint, + type CollectorOpts, +} from "../../src/orb/federated-collector"; +import type { FederatedCollectorMode, FocusManifest } from "../../src/signals/focus-manifest"; + +const URL_OK = "https://collector.example.org/v1/federated"; + +/** In-memory DB with the tables the bundle builder reads (mirrors federated-bundle.test.ts's makeDb). */ +function makeDb(): D1Database { + const driver = nodeSqliteDriver(new DatabaseSync(":memory:") as never); + driver.exec(` + CREATE TABLE review_audit ( + id TEXT PRIMARY KEY NOT NULL, project TEXT NOT NULL, target_id TEXT NOT NULL, + event_type TEXT NOT NULL DEFAULT 'gate_decision', decision TEXT, + source TEXT NOT NULL DEFAULT 'gittensory-native', head_sha TEXT, summary TEXT, + created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ','now')) + ); + CREATE TABLE system_flags ( + key TEXT PRIMARY KEY, value TEXT, + updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ','now')) + ); + `); + return createD1Adapter(driver); +} + +let seq = 0; +/** One fully-resolved PR so the builder has something to bundle. */ +async function resolved(db: D1Database, pr: number): Promise { + for (const [type, decision, at] of [ + ["gate_decision", "merge", "2026-07-10T10:00:00Z"], + ["pr_outcome", "merged", "2026-07-10T12:00:00Z"], + ] as const) { + await db + .prepare( + `INSERT INTO review_audit (id, project, target_id, event_type, decision, source, summary, created_at) VALUES (?, ?, ?, ?, ?, 'gittensory-native', NULL, ?)`, + ) + .bind(`c${seq++}`, "owner/repo", `owner/repo#${pr}`, type, decision, at) + .run(); + } +} + +function manifest( + o: { enabled?: boolean; collectorUrl?: string | null; collectorMode?: FederatedCollectorMode | null; present?: boolean } = {}, +): Pick { + return { + federatedIntelligence: { + present: o.present ?? true, + enabled: o.enabled ?? true, + collectorUrl: o.collectorUrl === undefined ? URL_OK : o.collectorUrl, + collectorMode: o.collectorMode ?? null, + }, + }; +} + +/** Fails the test if the network is touched at all — the literal proof of "opted out ⇒ zero network calls". */ +function untouchableFetch(): typeof fetch { + return new Proxy((() => {}) as unknown as typeof fetch, { + apply() { + throw new Error("opted-out client must not make a network call"); + }, + }); +} + +/** Same, for the database. */ +function untouchableDb(): D1Database { + return new Proxy({} as D1Database, { + get() { + throw new Error("opted-out client must not touch the database"); + }, + }); +} + +const NOW = Date.parse("2026-07-16T00:00:00Z"); +/** Deterministic retry: no wall-clock, no Math.random. */ +const DETERMINISTIC: CollectorOpts = { sleepFn: async () => undefined, randomFn: () => 0.5, now: NOW }; + +function bundle(over: Partial = {}): Record { + return { + schemaVersion: FEDERATED_BUNDLE_SCHEMA_VERSION, + instanceId: "abc123", + generatedAt: "2026-07-16T00:00:00.000Z", + windowDays: 90, + decided: 7, + mergePrecision: 0.9, + closePrecision: 1, + fpRate: 0.1, + fnRate: 0, + reversalRate: 0.1, + cycleP50Ms: 1000, + cycleP95Ms: 2000, + slopRate: 0, + copycatRate: 0, + signature: "f".repeat(64), + ...over, + }; +} + +function jsonResponse(body: unknown, status = 200): Response { + return { ok: status >= 200 && status < 300, status, json: async () => body } as unknown as Response; +} + +describe("resolveCollectorEndpoint()", () => { + it("is null unless opted in AND a collector is configured", () => { + expect(resolveCollectorEndpoint(null, "push")).toBeNull(); + expect(resolveCollectorEndpoint(undefined, "push")).toBeNull(); + expect(resolveCollectorEndpoint({} as Pick, "push")).toBeNull(); + expect(resolveCollectorEndpoint(manifest({ enabled: false }), "push")).toBeNull(); + expect(resolveCollectorEndpoint(manifest({ collectorUrl: null }), "push")).toBeNull(); + }); + + it("re-checks the SSRF guard at call time, so a round-tripped snapshot cannot smuggle an unsafe URL", () => { + expect(resolveCollectorEndpoint(manifest({ collectorUrl: "http://collector.example.org" }), "push")).toBeNull(); + expect(resolveCollectorEndpoint(manifest({ collectorUrl: "https://127.0.0.1/v1" }), "pull")).toBeNull(); + }); + + it("honors collectorMode in both directions, defaulting to both", () => { + expect(resolveCollectorEndpoint(manifest({ collectorMode: null }), "push")).toBe(URL_OK); + expect(resolveCollectorEndpoint(manifest({ collectorMode: null }), "pull")).toBe(URL_OK); + expect(resolveCollectorEndpoint(manifest({ collectorMode: "both" }), "push")).toBe(URL_OK); + expect(resolveCollectorEndpoint(manifest({ collectorMode: "push" }), "push")).toBe(URL_OK); + expect(resolveCollectorEndpoint(manifest({ collectorMode: "push" }), "pull")).toBeNull(); + expect(resolveCollectorEndpoint(manifest({ collectorMode: "pull" }), "pull")).toBe(URL_OK); + expect(resolveCollectorEndpoint(manifest({ collectorMode: "pull" }), "push")).toBeNull(); + }); +}); + +describe("pushFederatedBundle() — opted out", () => { + it("returns false touching neither the database nor the network", async () => { + for (const m of [null, manifest({ enabled: false }), manifest({ collectorUrl: null }), manifest({ collectorMode: "pull" })]) { + expect(await pushFederatedBundle(m, untouchableDb(), { fetchFn: untouchableFetch() })).toBe(false); + } + }); +}); + +describe("pushFederatedBundle() — opted in", () => { + it("POSTs exactly the anonymized bundle and leaks no identifier", async () => { + const db = makeDb(); + await resolved(db, 1); + const calls: Array<{ url: string; init: RequestInit }> = []; + const fetchFn = (async (url: string, init: RequestInit) => { + calls.push({ url, init }); + return jsonResponse({ ok: true }); + }) as unknown as typeof fetch; + + expect(await pushFederatedBundle(manifest(), db, { ...DETERMINISTIC, fetchFn })).toBe(true); + expect(calls).toHaveLength(1); + expect(calls[0]!.url).toBe(URL_OK); + expect(calls[0]!.init.method).toBe("POST"); + expect((calls[0]!.init.headers as Record)["content-type"]).toBe("application/json"); + + const sent = JSON.parse(calls[0]!.init.body as string); + expect(sent.schemaVersion).toBe(FEDERATED_BUNDLE_SCHEMA_VERSION); + expect(sent.signature).toMatch(/^[0-9a-f]{64}$/); + // The privacy regression test: the wire body carries no identifier of any kind. + const wire = calls[0]!.init.body as string; + expect(wire).not.toMatch(/owner\/repo/); + expect(wire).not.toMatch(/target_id|project|repo_hash|pr_hash/); + }); + + it("returns false when the builder has nothing to send", async () => { + // enabled:true but the builder returns null only if it throws; an empty DB still yields a bundle, so + // drive the null path through a DB whose read fails (the builder's own fail-safe). + const brokenDb = { + prepare() { + throw new Error("d1 down"); + }, + } as unknown as D1Database; + const err = vi.spyOn(console, "error").mockImplementation(() => {}); + const fetchFn = untouchableFetch(); // nothing to send ⇒ no request + expect(await pushFederatedBundle(manifest(), brokenDb, { ...DETERMINISTIC, fetchFn })).toBe(false); + err.mockRestore(); + }); +}); + +describe("push/pull — failure handling never reaches the gate", () => { + it("retries a 5xx with jittered backoff and gives up after maxAttempts", async () => { + const db = makeDb(); + await resolved(db, 1); + let calls = 0; + const sleeps: number[] = []; + const fetchFn = (async () => { + calls += 1; + return jsonResponse({ err: "boom" }, 503); + }) as unknown as typeof fetch; + + const ok = await pushFederatedBundle(manifest(), db, { + ...DETERMINISTIC, + fetchFn, + maxAttempts: 3, + sleepFn: async (ms: number) => { + sleeps.push(ms); + }, + }); + expect(ok).toBe(false); + expect(calls).toBe(3); // all attempts used + expect(sleeps).toHaveLength(2); // no sleep after the final attempt + expect(sleeps.every((ms) => ms > 0)).toBe(true); // backoff was actually consulted + }); + + it("does NOT retry a 4xx — an operator misconfiguration fails identically next time", async () => { + const db = makeDb(); + await resolved(db, 1); + let calls = 0; + const fetchFn = (async () => { + calls += 1; + return jsonResponse({ err: "bad request" }, 400); + }) as unknown as typeof fetch; + expect(await pushFederatedBundle(manifest(), db, { ...DETERMINISTIC, fetchFn })).toBe(false); + expect(calls).toBe(1); + }); + + it("swallows a timeout / network throw on both directions", async () => { + const db = makeDb(); + await resolved(db, 1); + const fetchFn = (async () => { + throw new DOMException("The operation was aborted due to timeout", "TimeoutError"); + }) as unknown as typeof fetch; + await expect(pushFederatedBundle(manifest(), db, { ...DETERMINISTIC, fetchFn })).resolves.toBe(false); + await expect(pullPeerBundles(manifest(), { ...DETERMINISTIC, fetchFn })).resolves.toEqual([]); + }); + + it("returns [] when the response body is not JSON at all", async () => { + const err = vi.spyOn(console, "error").mockImplementation(() => {}); + const fetchFn = (async () => + ({ + ok: true, + status: 200, + json: async () => { + throw new SyntaxError("Unexpected token < in JSON"); + }, + }) as unknown as Response) as unknown as typeof fetch; + await expect(pullPeerBundles(manifest(), { ...DETERMINISTIC, fetchFn })).resolves.toEqual([]); + expect(err).toHaveBeenCalled(); + err.mockRestore(); + }); +}); + +describe("pullPeerBundles()", () => { + it("returns [] touching nothing when opted out or scoped to push only", async () => { + for (const m of [null, manifest({ enabled: false }), manifest({ collectorUrl: null }), manifest({ collectorMode: "push" })]) { + expect(await pullPeerBundles(m, { fetchFn: untouchableFetch() })).toEqual([]); + } + }); + + it("fetches, shape-checks and returns peer bundles", async () => { + const fetchFn = (async (url: string, init: RequestInit) => { + expect(url).toBe(URL_OK); + expect(init.method).toBe("GET"); + return jsonResponse([bundle(), bundle({ instanceId: "def456" })]); + }) as unknown as typeof fetch; + const got = await pullPeerBundles(manifest(), { ...DETERMINISTIC, fetchFn }); + expect(got).toHaveLength(2); + expect(got[1]!.instanceId).toBe("def456"); + }); + + it("drops entries it does not understand and never verifies signatures (that is #6480/#6477)", async () => { + const fetchFn = (async () => + jsonResponse([ + bundle(), + bundle({ schemaVersion: 99 }), // a future schema + { ...bundle(), signature: undefined }, // no signature + { nope: true }, + null, + "string", + ])) as unknown as typeof fetch; + const got = await pullPeerBundles(manifest(), { ...DETERMINISTIC, fetchFn }); + expect(got).toHaveLength(1); + // A well-shaped bundle is returned AS-IS, with its signature untouched and unverified. + expect(got[0]!.signature).toBe("f".repeat(64)); + }); + + it("returns [] for a non-array payload", async () => { + const fetchFn = (async () => jsonResponse({ bundles: [bundle()] })) as unknown as typeof fetch; + expect(await pullPeerBundles(manifest(), { ...DETERMINISTIC, fetchFn })).toEqual([]); + }); +}); + +describe("rate limiting — a best-effort sync must not hammer a peer", () => { + it("skips the request entirely when the caller's bucket is exhausted", async () => { + const db = makeDb(); + await resolved(db, 1); + const exhausted = { count: 999, windowStartMs: NOW }; + const fetchFn = untouchableFetch(); + expect(await pushFederatedBundle(manifest(), db, { ...DETERMINISTIC, fetchFn, bucket: exhausted })).toBe(false); + expect(await pullPeerBundles(manifest(), { ...DETERMINISTIC, fetchFn, bucket: exhausted })).toEqual([]); + }); + + it("allows the request when the bucket has room, and when no bucket is supplied at all", async () => { + const db = makeDb(); + await resolved(db, 1); + const fetchFn = (async () => jsonResponse([])) as unknown as typeof fetch; + const fresh = { count: 0, windowStartMs: NOW }; + expect(await pushFederatedBundle(manifest(), db, { ...DETERMINISTIC, fetchFn, bucket: fresh })).toBe(true); + expect(await pullPeerBundles(manifest(), { ...DETERMINISTIC, fetchFn })).toEqual([]); + }); +}); + +describe("defaults", () => { + it("uses the platform clock and fetch when neither is injected", async () => { + const original = globalThis.fetch; + let called = 0; + globalThis.fetch = (async () => { + called += 1; + return jsonResponse([]); + }) as unknown as typeof fetch; + try { + // No `now`, no fetchFn, no sleepFn, no randomFn — exercises every default arm. + expect(await pullPeerBundles(manifest(), {})).toEqual([]); + expect(called).toBe(1); + } finally { + globalThis.fetch = original; + } + }); + + it("clamps a nonsensical maxAttempts to at least one attempt", async () => { + let calls = 0; + const fetchFn = (async () => { + calls += 1; + return jsonResponse(null, 500); + }) as unknown as typeof fetch; + expect(await pullPeerBundles(manifest(), { ...DETERMINISTIC, fetchFn, maxAttempts: 0 })).toEqual([]); + expect(calls).toBe(1); + }); + + it("honors an explicit timeoutMs and uses the built-in sleep when none is injected", async () => { + let calls = 0; + const fetchFn = (async (_url: string, init: RequestInit) => { + calls += 1; + expect(init.signal).toBeInstanceOf(AbortSignal); // the timeout is armed on every attempt + return jsonResponse(null, 500); + }) as unknown as typeof fetch; + // No sleepFn => the real setTimeout-backed sleep runs; randomFn 0 keeps the single backoff short. + const started = Date.now(); + expect(await pullPeerBundles(manifest(), { now: NOW, randomFn: () => 0, fetchFn, timeoutMs: 1_000, maxAttempts: 2 })).toEqual([]); + expect(calls).toBe(2); + expect(Date.now() - started).toBeGreaterThanOrEqual(200); // it genuinely waited + }); + + it("pushes with the platform clock when no now is injected", async () => { + const db = makeDb(); + await resolved(db, 1); + const fetchFn = (async () => jsonResponse({ ok: true })) as unknown as typeof fetch; + expect(await pushFederatedBundle(manifest(), db, { fetchFn })).toBe(true); + }); +}); + +describe("total fail-safety", () => { + // The gate must survive even a misbehaving injected dependency: nothing in this module may throw. + it("degrades to false rather than throwing when an injected dependency itself throws", async () => { + const db = makeDb(); + await resolved(db, 1); + const err = vi.spyOn(console, "error").mockImplementation(() => {}); + const fetchFn = (async () => jsonResponse(null, 500)) as unknown as typeof fetch; + await expect( + pushFederatedBundle(manifest(), db, { + now: NOW, + fetchFn, + randomFn: () => 0.5, + sleepFn: async () => { + throw new Error("scheduler exploded"); + }, + }), + ).resolves.toBe(false); + expect(err).toHaveBeenCalled(); + err.mockRestore(); + }); +}); diff --git a/test/unit/focus-manifest.test.ts b/test/unit/focus-manifest.test.ts index 81648f6123..028e111dd5 100644 --- a/test/unit/focus-manifest.test.ts +++ b/test/unit/focus-manifest.test.ts @@ -924,7 +924,7 @@ describe("compileFocusManifestPolicy", () => { publicStats: { present: false, enabled: false }, draftFlow: { present: false, enabled: false }, upstreamDriftIssues: { present: false, enabled: false }, - federatedIntelligence: { present: false, enabled: false }, + federatedIntelligence: { present: false, enabled: false, collectorUrl: null, collectorMode: null }, warnings: [], }); expect(policy.publicSafe.entryGuidance).toContain("Keep PRs focused."); @@ -2123,12 +2123,12 @@ describe("parseFocusManifest gate config", () => { describe("federatedIntelligence: (#1970, opt-in federated fleet intelligence export config-as-code toggle)", () => { it("defaults to fully disabled/absent when the key is omitted, and does not make the manifest present on its own", () => { const m = parseFocusManifest({}); - expect(m.federatedIntelligence).toEqual({ present: false, enabled: false }); + expect(m.federatedIntelligence).toEqual({ present: false, enabled: false, collectorUrl: null, collectorMode: null }); expect(m.present).toBe(false); }); it("treats an explicit null the same as an omitted key", () => { - expect(parseFocusManifest({ federatedIntelligence: null }).federatedIntelligence).toEqual({ present: false, enabled: false }); + expect(parseFocusManifest({ federatedIntelligence: null }).federatedIntelligence).toEqual({ present: false, enabled: false, collectorUrl: null, collectorMode: null }); }); it("warns and falls back to the default when the value is a non-mapping type (string or array)", () => { @@ -2142,13 +2142,13 @@ describe("parseFocusManifest gate config", () => { it("parses enabled: true, making the manifest present", () => { const m = parseFocusManifest({ federatedIntelligence: { enabled: true } }); - expect(m.federatedIntelligence).toEqual({ present: true, enabled: true }); + expect(m.federatedIntelligence).toEqual({ present: true, enabled: true, collectorUrl: null, collectorMode: null }); expect(m.present).toBe(true); }); it("parses enabled: false explicitly, still making the manifest present", () => { const m = parseFocusManifest({ federatedIntelligence: { enabled: false } }); - expect(m.federatedIntelligence).toEqual({ present: true, enabled: false }); + expect(m.federatedIntelligence).toEqual({ present: true, enabled: false, collectorUrl: null, collectorMode: null }); expect(m.present).toBe(true); }); @@ -2166,6 +2166,55 @@ describe("parseFocusManifest gate config", () => { it("federatedIntelligenceConfigToJson returns null for an absent config", () => { expect(federatedIntelligenceConfigToJson(parseFocusManifest(null).federatedIntelligence)).toBeNull(); }); + + it("parses a public HTTPS collectorUrl and a collectorMode (#6479)", () => { + const m = parseFocusManifest({ + federatedIntelligence: { enabled: true, collectorUrl: "https://collector.example.org/v1/federated", collectorMode: "push" }, + }); + expect(m.federatedIntelligence).toEqual({ + present: true, + enabled: true, + collectorUrl: "https://collector.example.org/v1/federated", + collectorMode: "push", + }); + }); + + // The SSRF guard is what makes an operator-configured collector safe to arm from config-as-code: an + // unsafe endpoint is dropped at parse time so it can never reach the transport client at all. + it("drops a collectorUrl that fails the HTTPS/public-host guard, with a warning, rather than accepting it", () => { + for (const bad of [ + "http://collector.example.org", + "https://localhost/v1", + "https://127.0.0.1/v1", + "https://10.0.0.1/v1", + "https://192.168.1.10/v1", + "https://collector.internal/v1", + "not-a-url", + ]) { + const m = parseFocusManifest({ federatedIntelligence: { enabled: true, collectorUrl: bad } }); + expect(m.federatedIntelligence.collectorUrl, `should have rejected ${bad}`).toBeNull(); + expect(m.warnings.some((w) => /federatedIntelligence\.collectorUrl/.test(w))).toBe(true); + } + }); + + it("warns and drops a non-string collectorUrl and an unknown collectorMode", () => { + const m = parseFocusManifest({ + federatedIntelligence: { enabled: true, collectorUrl: 42 as unknown as string, collectorMode: "sideways" as never }, + }); + expect(m.federatedIntelligence.collectorUrl).toBeNull(); + expect(m.federatedIntelligence.collectorMode).toBeNull(); + expect(m.warnings.some((w) => /federatedIntelligence\.collectorMode/.test(w))).toBe(true); + }); + + it("round-trips collectorUrl + collectorMode through federatedIntelligenceConfigToJson unchanged", () => { + const m = parseFocusManifest({ + federatedIntelligence: { enabled: true, collectorUrl: "https://collector.example.org/v1", collectorMode: "both" }, + }); + expect( + parseFocusManifest({ federatedIntelligence: federatedIntelligenceConfigToJson(m.federatedIntelligence) }) + .federatedIntelligence, + ).toEqual(m.federatedIntelligence); + }); }); it("parses aiReviewAllAuthors from the settings: block (generic override)", () => {