From f77620b9a4b76cc05508612ee894a20b751651bb Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 31 Oct 2026 14:50:21 -0500 Subject: [PATCH] Refresh GitHub event connector after live enrollment Resolve the current identity only on admitted work, retain unchanged connector negotiation, and replace removed or rotated identity configuration. Prove real signed enrollment and encrypted event delivery without recreating the registered worker or adding idle polling. Co-Authored-By: Paperclip --- .../2026-10-30-github-connection-sleep.md | 25 ++++ .../github-connection-enrollment-work.test.ts | 139 ++++++++++++++++++ .../src/services/github-connection-events.ts | 19 ++- 3 files changed, 181 insertions(+), 2 deletions(-) create mode 100644 server/src/__tests__/github-connection-enrollment-work.test.ts diff --git a/doc/plans/2026-10-30-github-connection-sleep.md b/doc/plans/2026-10-30-github-connection-sleep.md index 3e6de1148bb..287b479b07c 100644 --- a/doc/plans/2026-10-30-github-connection-sleep.md +++ b/doc/plans/2026-10-30-github-connection-sleep.md @@ -114,3 +114,28 @@ concurrent replay, signed callbacks, outer commit/rollback, drain/resume, startu wiring in both scheduler branches, credential backoff and changed deadlines. Cloud acceptance uses its actual broker/store/notifier boundaries and a controlled HTTP tenant. These tests do not establish a deployed fleet rollout or cost savings. + +## Enrollment after worker startup + +The event owner resolves the current connector only when a delivery or setup pass +is admitted. A service created before self-host enrollment can therefore use a +later completed enrollment without being recreated. The full normalized identity +and target configuration keys the connector cache, preserving negotiated event +wake support while it is unchanged. An absent identity clears the cache; a +replacement identity or changed key creates a new connector. Explicitly injected +connectors keep their original lifetime. + +Enrollment does not add a timer, database scan or network probe. An ordinary +credential/configuration producer, authenticated control wake or active setup +request admits the next pass. Unconfigured passes return before owner SQL or +HTTP. A pass already in flight keeps its connector for that lease and its matching +acknowledgment; identity discovery never changes authority halfway through a pass. + +The real PostgreSQL/loopback regression starts the registered owner unenrolled, +completes the canonical signed enrollment flow, then applies and acknowledges an +encrypted installation event. Removing the identity parks the same owner, and a +second enrollment proves the replacement signing/sealing identity is used. Each +applied installation update has one bounded empty follow-up from its credential +hint; the test joins that saved deadline once before checking 25-minute SQL/HTTP +quiet windows. This is process-local enrollment recovery, not a deployed Cloud +rollout or a new cancellation policy. diff --git a/server/src/__tests__/github-connection-enrollment-work.test.ts b/server/src/__tests__/github-connection-enrollment-work.test.ts new file mode 100644 index 00000000000..c888565b3d9 --- /dev/null +++ b/server/src/__tests__/github-connection-enrollment-work.test.ts @@ -0,0 +1,139 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { createServer, type Server } from "node:http"; +import { once } from "node:events"; +import { createCipheriv, createPublicKey, diffieHellman, generateKeyPairSync, hkdfSync, randomBytes, randomUUID, verify } from "node:crypto"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { companies, connectionEventDeliveries, connectionGrants, createDb, toolApplications, toolConnections } from "@paperclipai/db"; +import { eq } from "drizzle-orm"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { githubConnectionEventService } from "../services/github-connection-events.js"; +import { registerGitHubConnectionEventWork } from "../services/github-connection-work.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { wakeConnectionEvents } from "../services/connection-event-wake.js"; +import { completePaperclipCloudConnectorEnrollment, loadPaperclipCloudConnectorIdentity, startPaperclipCloudConnectorEnrollment } from "../services/paperclip-cloud-connector-enrollment.js"; +import { paperclipCloudConnectorIdentityPath } from "../services/paperclip-cloud-connector-enrollment.js"; + +const support = await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("GitHub enrollment on a live event worker (real PostgreSQL and signed loopback Cloud)", () => { + let temporary: Awaited>; + let db: ReturnType; + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("github-live-enrollment-"); + db = createDb(temporary.connectionString, { maxConnections: 4, idleTimeoutSeconds: 0, connectTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + }, 90_000); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + + it("enrolls without recreation, drains an authenticated event, forgets removed identity and re-enrolls from quiet", async () => { + const root = await mkdtemp(path.join(os.tmpdir(), "github-live-enrollment-")); + const previous = { home: process.env.PAPERCLIP_HOME, instance: process.env.PAPERCLIP_INSTANCE_ID }; + process.env.PAPERCLIP_HOME = root; process.env.PAPERCLIP_INSTANCE_ID = "enrollment-worker"; + let server: Server | undefined; + let worker: ReturnType | undefined; + let coordinator: ReturnType | undefined; + const errors: unknown[] = [], httpErrors: unknown[] = []; + let requests = 0, acknowledged = 0, leases = 0, enrollment: Record | undefined; + let origin = "", eventId = randomUUID(), grantId = "", companyId = ""; + const identities: string[] = []; + const delivered = new Set(); + try { + server = createServer(async (req, res) => { + try { + requests++; + const chunks: Buffer[] = []; for await (const chunk of req) chunks.push(Buffer.from(chunk)); + const body = JSON.parse(Buffer.concat(chunks).toString()); + if (req.url === "/v1/connector/enrollments") { + enrollment = body; + res.setHeader("content-type", "application/json"); + res.end(JSON.stringify({ enrollmentId: "enrollment", verificationUrl: `${origin}/connections/enroll?id=enrollment`, expiresAt: new Date(Date.now() + 60_000).toISOString() })); return; + } + expect(enrollment).toBeDefined(); + const [header, payload, signature] = body.request.split("."); + expect(verify(null, Buffer.from(`${header}.${payload}`), createPublicKey({ key: { kty: "OKP", crv: "Ed25519", x: enrollment!.signPublicKey }, format: "jwk" }), Buffer.from(signature, "base64url"))).toBe(true); + const claims = JSON.parse(Buffer.from(payload, "base64url").toString()); + expect(claims.iss).toBe(enrollment!.instanceId); expect(claims.env).toBe("test"); expect(claims.aud).toBe(`${origin}${req.url}`); + res.setHeader("content-type", "application/json"); + if (req.url === "/v1/connector/enrollment-claims") { + res.end(JSON.stringify({ id: enrollment!.instanceId, environment: "test", origins: [enrollment!.origin] })); return; + } + identities.push(claims.iss); + if (req.url === "/v1/connector/events/ack") { acknowledged++; delivered.add(eventId); res.end(JSON.stringify({ acknowledged: 1 })); return; } + expect(req.url).toBe("/v1/connector/events/lease"); leases++; + if (delivered.has(eventId)) { res.end(JSON.stringify({ events: [], eventWakeVersion: 1 })); return; } + const ephemeral = generateKeyPairSync("x25519"), epk = ephemeral.publicKey.export({ format: "jwk" }).x!; + const recipient = createPublicKey({ key: { kty: "OKP", crv: "X25519", x: enrollment!.sealPublicKey }, format: "jwk" }); + const aad = Buffer.from([1, "X25519-HKDF-SHA256-A256GCM", "events", claims.iss, "test", "github", "github.code", ""].join("\n")); + const key = Buffer.from(hkdfSync("sha256", diffieHellman({ privateKey: ephemeral.privateKey, publicKey: recipient }), Buffer.concat([Buffer.from(epk, "base64url"), Buffer.from(enrollment!.sealPublicKey, "base64url")]), aad, 32)); + const iv = randomBytes(12), cipher = createCipheriv("aes-256-gcm", key, iv); cipher.setAAD(aad); + const packet = { v: 1, instanceId: claims.iss, environment: "test", leaseId: "lease", events: [{ id: eventId, provider: "github", event: "installation", action: "unsuspend", installationId: "101", repositoryId: null, createdAt: new Date().toISOString(), bindingIds: [`${grantId}_101`], payload: { installationId: "101" } }] }; + const ct = Buffer.concat([cipher.update(JSON.stringify(packet)), cipher.final(), cipher.getAuthTag()]); + res.end(JSON.stringify({ leaseId: "lease", eventWakeVersion: 1, sealed: { v: 1, alg: "X25519-HKDF-SHA256-A256GCM", purpose: "events", provider: "github", profile: "github.code", epk, iv: iv.toString("base64url"), ct: ct.toString("base64url") } })); + } catch (error) { httpErrors.push(error); res.writeHead(500).end(); } + }); + const listening = once(server, "listening"); server.listen(0, "127.0.0.1"); await listening; + const address = server.address(); if (!address || typeof address === "string") throw new Error("No loopback listener"); + origin = `http://127.0.0.1:${address.port}`; + companyId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Live enrollment", issuePrefix: "GLE" }); + const [app] = await db.insert(toolApplications).values({ companyId, applicationKey: randomUUID(), name: "GitHub", type: "mcp_http", status: "active" }).returning(); + const [connection] = await db.insert(toolConnections).values({ companyId, applicationId: app!.id, uid: randomUUID(), name: "GitHub", status: "active", enabled: true, transport: "mcp_remote", authKind: "oauth", credentialPolicy: "per_user", config: { sourceTemplateKey: "github", oauth: { connectorProfile: "github.code" } }, transportConfig: {} }).returning(); + const [grant] = await db.insert(connectionGrants).values({ companyId, connectionId: connection!.id, kind: "user", subjectUserId: "owner", status: "active", providerTenant: { github: { userId: "42", login: "octocat", installationIds: ["101"], installationCount: 1, repositoryCount: 0, repositorySelection: "all", installationOwnerLogins: [], webhookHealth: "unhealthy" } } }).returning(); + grantId = grant!.id; + const env = { PAPERCLIP_CLOUD_CONNECTOR_BASE_URL: origin, PAPERCLIP_CLOUD_CONNECTOR_ENVIRONMENT: "test" }; + const service = githubConnectionEventService(db, { env }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: e => errors.push(e) }); + const selects = vi.spyOn(db, "select"), transactions = vi.spyOn(db, "transaction"), executes = vi.spyOn(db, "execute"); + worker = registerGitHubConnectionEventWork(db, coordinator, service, () => true); + await worker.ready; + async function quiet(stage: string) { + await vi.waitFor(() => { expect(idleWorkSnapshot().active, stage).toBe(0); expect(coordinator!.nextWakeAt(), stage).toBeNull(); }); + selects.mockClear(); transactions.mockClear(); executes.mockClear(); const before = requests; + await vi.advanceTimersByTimeAsync(25 * 60_000); + expect(selects).not.toHaveBeenCalled(); expect(transactions).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled(); expect(requests).toBe(before); + } + await quiet("unenrolled startup"); expect(requests).toBe(0); + async function enroll() { + await startPaperclipCloudConnectorEnrollment({ origin: "https://self-hosted.example.test", env, companyId }); + const pending = loadPaperclipCloudConnectorIdentity()!; + await completePaperclipCloudConnectorEnrollment({ enrollmentId: pending.pending!.enrollmentId, state: pending.pending!.returnState, approvalCode: "approved", env }); + return loadPaperclipCloudConnectorIdentity()!.instanceId; + } + const firstIdentity = await enroll(); + expect(wakeConnectionEvents(db)).toBe(true); + await vi.advanceTimersByTimeAsync(6_000); + await vi.waitFor(() => expect(acknowledged).toBe(1)); + expect(await db.select().from(connectionEventDeliveries).where(eq(connectionEventDeliveries.providerDeliveryId, eventId))).toMatchObject([{ status: "processed" }]); + expect((await db.select().from(connectionGrants).where(eq(connectionGrants.id, grantId)))[0]!.providerTenant!.github!.webhookHealth).toBe("healthy"); + // The applied installation update publishes one credential-change hint. + // Join that existing minimum-spacing pass once; the broker now has no debt. + const followup = coordinator.nextWakeAt(); expect(followup).not.toBeNull(); + expect(followup! - Date.now()).toBeLessThanOrEqual(5_000); + await vi.advanceTimersByTimeAsync(Math.max(0, followup! - Date.now()) + 1); + await vi.waitFor(() => expect(leases).toBe(2)); + await quiet("first event plus its one empty follow-up"); + await rm(paperclipCloudConnectorIdentityPath()); + const oldLeases = leases; expect(wakeConnectionEvents(db)).toBe(true); await vi.advanceTimersByTimeAsync(6_000); + await quiet("removed identity"); expect(leases).toBe(oldLeases); + eventId = randomUUID(); const secondIdentity = await enroll(); expect(secondIdentity).not.toBe(firstIdentity); + expect(wakeConnectionEvents(db)).toBe(true); await vi.advanceTimersByTimeAsync(6_000); + await vi.waitFor(() => expect(acknowledged).toBe(2)); + expect(identities).toContain(firstIdentity); expect(identities).toContain(secondIdentity); + expect(identities.slice(identities.indexOf(secondIdentity))).toEqual(expect.arrayContaining([secondIdentity, secondIdentity])); + expect(identities.slice(identities.indexOf(secondIdentity)).every(id => id === secondIdentity)).toBe(true); + const replacementFollowup = coordinator.nextWakeAt(); expect(replacementFollowup).not.toBeNull(); + expect(replacementFollowup! - Date.now()).toBeLessThanOrEqual(5_000); + await vi.advanceTimersByTimeAsync(Math.max(0, replacementFollowup! - Date.now()) + 1); + await vi.waitFor(() => expect(leases).toBe(oldLeases + 2)); + await quiet("replacement event plus its one empty follow-up"); expect(errors).toEqual([]); expect(httpErrors).toEqual([]); + } finally { + await worker?.stop(); await coordinator?.stop(); vi.restoreAllMocks(); vi.useRealTimers(); + if (server) { const closed = once(server, "close"); server.close(); server.closeAllConnections(); await closed; } + if (previous.home === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = previous.home; + if (previous.instance === undefined) delete process.env.PAPERCLIP_INSTANCE_ID; else process.env.PAPERCLIP_INSTANCE_ID = previous.instance; + await rm(root, { recursive: true, force: true }); + } + }, 30_000); +}); diff --git a/server/src/services/github-connection-events.ts b/server/src/services/github-connection-events.ts index a70213dc44b..af936353bdd 100644 --- a/server/src/services/github-connection-events.ts +++ b/server/src/services/github-connection-events.ts @@ -4,6 +4,7 @@ import { notifyToolConnectionHealth } from "./tool-connection-health-notificatio import { dispatchGitHubBotCloudEvent } from "./chat-github-cloud-ingress.js"; import { chatEndpoints } from "@paperclipai/db"; import { randomUUID } from "node:crypto"; +import { isDeepStrictEqual } from "node:util"; import { connectionEventDeliveries, connectionGrants, @@ -223,8 +224,21 @@ export function githubConnectionEventService( ) { const now = options.now ?? (() => new Date()); let nextPollAt = 0; - const config = options.connector ? null : paperclipCloudConnectorConfigFromEnv(options.env); - const connector = options.connector ?? (config ? createPaperclipCloudConnector({ config }) : null); + let configuredConnector: { + config: NonNullable>; + connector: PaperclipCloudConnector; + } | undefined; + function currentConnector() { + if (options.connector) return options.connector; + // Enrollment and identity replacement can happen after this owner starts. + // Read only on an admitted pass; preserve negotiated delivery while unchanged. + const config = paperclipCloudConnectorConfigFromEnv(options.env); + if (!config) { configuredConnector = undefined; return null; } + if (!configuredConnector || !isDeepStrictEqual(configuredConnector.config, config)) { + configuredConnector = { config, connector: createPaperclipCloudConnector({ config }) }; + } + return configuredConnector.connector; + } const forcePolling = (options.env ?? process.env).PAPERCLIP_GITHUB_EVENT_POLLING === "1"; let emptyPolls = 0; let nextSetupPollAt = 0; @@ -484,6 +498,7 @@ export function githubConnectionEventService( } async function runPoll(): Promise { + const connector = currentConnector(); if (!connector) { nextPollAt = Number.POSITIVE_INFINITY; return { leased: 0, processed: 0, duplicate: 0, ignored: 0, failed: 0 };