From 10e7a6f7e99766afeb5f7ec5aba9d04e217da558 Mon Sep 17 00:00:00 2001 From: Dotta Date: Tue, 21 Jul 2026 17:16:48 +0000 Subject: [PATCH] feat(connections): add brokered OAuth instance seam Co-Authored-By: Paperclip --- packages/connect-protocol/src/index.ts | 5 + server/src/routes/tool-access.ts | 19 ++++ server/src/services/connection-broker.test.ts | 96 +++++++++++++++++++ server/src/services/connection-broker.ts | 81 ++++++++++++++++ server/src/services/tool-access.ts | 42 +++++++- 5 files changed, 242 insertions(+), 1 deletion(-) create mode 100644 server/src/services/connection-broker.test.ts create mode 100644 server/src/services/connection-broker.ts diff --git a/packages/connect-protocol/src/index.ts b/packages/connect-protocol/src/index.ts index f4f85cfa105..839e0645cf0 100644 --- a/packages/connect-protocol/src/index.ts +++ b/packages/connect-protocol/src/index.ts @@ -64,6 +64,11 @@ export const claimResponseSchema = z.union([ z.object({ custodyMode: z.enum(["B", "B2"]), grantSealed: z.string().min(1), tokenMeta: tokenMetadataSchema }), ]); +export const grantTokenRequestSchema = z.object({ grantWrapKey: z.string().min(1) }); +export const grantTokenResponseSchema = z.object({ grantSealed: z.string().min(1), tokenMeta: tokenMetadataSchema }); +export const relayRegistrationRequestSchema = z.object({ connectionRef: z.string().uuid(), providerSlug: z.string().min(1), intakeUrls: z.array(z.string().url()).min(1) }); +export const relayRegistrationResponseSchema = z.object({ connectionPublicRef: opaqueId("cn"), intakeUrls: z.array(z.string().url()).min(1), relaySecret: z.string().min(1) }); + export const relayEnvelopeSchema = z.object({ v: z.literal(1), deliveryId: opaqueId("dl"), diff --git a/server/src/routes/tool-access.ts b/server/src/routes/tool-access.ts index 3c02c0d3c42..34e728f71af 100644 --- a/server/src/routes/tool-access.ts +++ b/server/src/routes/tool-access.ts @@ -313,10 +313,29 @@ export function toolAccessRoutes( const result = await svc.startOAuth(existing.companyId, existing.id, { redirectUri: oauthRedirectUri(), actor: getActorInfo(req), + brokered: req.body?.brokered === true, }); res.json(result); }); + router.post("/tools/oauth/:connectionId/claim", async (req, res) => { + const existing = await svc.getConnection(req.params.connectionId as string); + assertToolAppMutationAccess(req, existing.companyId); + const claimCode = typeof req.body?.claimCode === "string" ? req.body.claimCode : ""; + const kind = req.body?.kind === "user" ? "user" : "workspace"; + if (claimCode.length < 22) throw badRequest("Invalid claim code"); + res.json(await svc.claimBrokeredOAuth(existing.companyId, existing.id, { claimCode, kind, subjectUserId: kind === "user" ? req.body?.subjectUserId ?? req.actor.userId ?? null : null })); + }); + + router.post("/tool-connections/:connectionId/relay/register", async (req, res) => { + const existing = await svc.getConnection(req.params.connectionId as string); + assertToolAppMutationAccess(req, existing.companyId); + const providerSlug = typeof req.body?.providerSlug === "string" ? req.body.providerSlug : ""; + const intakeUrls = Array.isArray(req.body?.intakeUrls) ? req.body.intakeUrls.filter((value: unknown): value is string => typeof value === "string") : []; + if (!providerSlug || intakeUrls.length === 0) throw badRequest("Provider and intake URLs are required"); + res.status(201).json(await svc.registerConnectionRelay(existing.companyId, existing.id, { providerSlug, intakeUrls })); + }); + router.get("/tools/oauth/callback", async (req, res) => { assertBoard(req); const state = typeof req.query.state === "string" ? req.query.state : ""; diff --git a/server/src/services/connection-broker.test.ts b/server/src/services/connection-broker.test.ts new file mode 100644 index 00000000000..414c1318997 --- /dev/null +++ b/server/src/services/connection-broker.test.ts @@ -0,0 +1,96 @@ +import { generateKeyPairSync } from "node:crypto"; +import { describe, expect, it } from "vitest"; +import { verifyRequestEnvelope, type ReplayStore } from "@paperclip/connect-protocol"; +import { + ConnectClaimInterceptionError, + ConnectServiceResponseError, + connectionBrokerService, + createSignedConnectBrokerClient, + type ConnectionBrokerStore, +} from "./connection-broker.js"; + +const INSTANCE_ID = "in_testinstance"; +const CONNECTION_ID = "44444444-4444-4444-8444-444444444444"; +function fixture(mode: "A" | "B2" = "A") { + const { privateKey, publicKey } = generateKeyPairSync("ed25519"); + const replay = new Set(); + const claims = new Map([["claim-code-at-least-22-chars", { owner: INSTANCE_ID, attempts: 0, consumed: false }]]); + const saved: Parameters[0][] = []; + const secrets: Array<{ purpose: string; value: string }> = []; + const replayStore: ReplayStore = { consume: (_instance, jti) => !replay.has(jti) && Boolean(replay.add(jti)) }; + let capturedEnvelope = ""; + const transport = { + async request(path: string, compactJws: string) { + capturedEnvelope = compactJws; + try { + const envelope = await verifyRequestEnvelope({ compactJws, publicKey, expectedInstanceId: INSTANCE_ID, expectedPath: path, replayStore }); + if (path === "/v1/claims") { + const code = (envelope.body as { claimCode: string }).claimCode; + const claim = claims.get(code); + if (!claim) return { status: 404, body: { error: "unknown_claim" } }; + if (claim.attempts >= 5) return { status: 404, body: { error: "unknown_claim" } }; + if (claim.owner !== envelope.iss) { claim.attempts += 1; return { status: 403, body: { error: "claim_instance_mismatch" } }; } + if (claim.consumed) return { status: 409, body: { error: "claim_already_consumed" } }; + claim.consumed = true; + return mode === "A" + ? { status: 200, body: { custodyMode: "A", grantRef: "gr_testgrant", tokenMeta: { scopes: ["read"], expiresAt: null } } } + : { status: 200, body: { custodyMode: "B2", grantSealed: "sealed-refresh", tokenMeta: { scopes: ["read"], expiresAt: null } } }; + } + if (path.endsWith("/token")) return { status: 200, body: { grantSealed: "sealed-access", tokenMeta: { scopes: ["read"], expiresAt: null } } }; + if (path === "/v1/handshakes") return { status: 201, body: { handshakeId: "hs_test", authorizeUrl: "https://connect.example/authorize", expiresAt: new Date(Date.now() + 60_000).toISOString() } }; + if (path === "/v1/connections/register") return { status: 201, body: { connectionPublicRef: "cn_test", intakeUrls: ["https://connect.example/intake/cn_test"], relaySecret: "relay-secret" } }; + return { status: 404, body: { error: "not_found" } }; + } catch (error) { + return { status: 401, body: { error: error instanceof Error ? error.message : "invalid_envelope" } }; + } + }, + }; + const client = createSignedConnectBrokerClient({ instanceId: INSTANCE_ID, keyId: "key-1", privateKey, transport }); + const store: ConnectionBrokerStore = { + saveGrant: async (input) => { saved.push(input); return { id: "grant-local" }; }, + registerRelay: async () => undefined, + }; + const service = connectionBrokerService({ client, store, enabled: true, vault: { put: async (input) => { secrets.push(input); return { secretId: `secret-${input.purpose}`, versionSelector: "latest", configPath: `broker.${input.purpose}`, required: true, label: input.purpose }; } } }); + return { service, saved, secrets, claims, transport, getCapturedEnvelope: () => capturedEnvelope }; +} + +describe("connection broker", () => { + it.each(["A", "B2"] as const)("claims and persists custody mode %s", async (mode) => { + const f = fixture(mode); + const result = await f.service.claimOAuth({ companyId: "company", connectionId: CONNECTION_ID, claimCode: "claim-code-at-least-22-chars", kind: "workspace" }); + expect(result.custodyMode).toBe(mode); + expect(f.saved[0]?.custodyMode).toBe(mode); + expect(f.secrets[0]?.value).toBe(mode === "A" ? "sealed-access" : "sealed-refresh"); + }); + + it("surfaces a 409 redemption as an interception alert", async () => { + const f = fixture(); + await f.service.claimOAuth({ companyId: "company", connectionId: CONNECTION_ID, claimCode: "claim-code-at-least-22-chars", kind: "workspace" }); + await expect(f.service.claimOAuth({ companyId: "company", connectionId: CONNECTION_ID, claimCode: "claim-code-at-least-22-chars", kind: "workspace" })).rejects.toBeInstanceOf(ConnectClaimInterceptionError); + }); + + it("does not consume mismatched claims and locks the service-side double after five attempts", async () => { + const f = fixture(); + const claim = f.claims.get("claim-code-at-least-22-chars")!; + claim.owner = "in_otherinstance"; + for (let attempt = 0; attempt < 5; attempt += 1) { + await expect(f.service.claimOAuth({ companyId: "company", connectionId: CONNECTION_ID, claimCode: "claim-code-at-least-22-chars", kind: "workspace" })).rejects.toMatchObject({ status: 403, code: "claim_instance_mismatch" }); + expect(claim.consumed).toBe(false); + } + expect(claim.attempts).toBe(5); + await expect(f.service.claimOAuth({ companyId: "company", connectionId: CONNECTION_ID, claimCode: "claim-code-at-least-22-chars", kind: "workspace" })).rejects.toMatchObject({ status: 404, code: "unknown_claim" }); + }); + + it("rejects replay of an already signed request", async () => { + const f = fixture(); + await f.service.startOAuth({ providerSlug: "vercel", methodKey: "oauth", custodyMode: "A", clientOwnership: "platform_shared", connectionRef: CONNECTION_ID, returnUrl: "https://instance.example/claim", scopes: [] }); + const replay = await f.transport.request("/v1/handshakes", f.getCapturedEnvelope()); + expect(replay).toEqual({ status: 401, body: { error: "replayed_jti" } }); + }); + + it("defaults brokered mode off", async () => { + const f = fixture(); + const disabled = connectionBrokerService({ client: (f.service as never), store: {} as never, vault: {} as never }); + await expect(disabled.startOAuth({})).rejects.toEqual(new ConnectServiceResponseError(403, "mode_not_allowed")); + }); +}); diff --git a/server/src/services/connection-broker.ts b/server/src/services/connection-broker.ts new file mode 100644 index 00000000000..4563a85b3ed --- /dev/null +++ b/server/src/services/connection-broker.ts @@ -0,0 +1,81 @@ +import { randomBytes, type KeyObject } from "node:crypto"; +import { and, eq } from "drizzle-orm"; +import type { Db } from "@paperclipai/db"; +import { connectionGrants, toolConnections } from "@paperclipai/db"; +import { + claimRequestSchema, claimResponseSchema, grantTokenRequestSchema, grantTokenResponseSchema, + handshakeRequestSchema, handshakeResponseSchema, relayRegistrationRequestSchema, + relayRegistrationResponseSchema, signRequestEnvelope, +} from "@paperclip/connect-protocol"; + +export class ConnectServiceResponseError extends Error { + constructor(readonly status: number, readonly code: string) { super(code); } +} +export class ConnectClaimInterceptionError extends ConnectServiceResponseError { + constructor() { super(409, "claim_already_consumed"); } +} +export type ConnectBrokerTransport = { request(path: string, compactJws: string): Promise<{ status: number; body: unknown }> }; +export function createSignedConnectBrokerClient(input: { instanceId: string; keyId: string; privateKey: KeyObject; transport: ConnectBrokerTransport }) { + async function request(path: string, body: unknown) { + const compactJws = signRequestEnvelope({ body, instanceId: input.instanceId, keyId: input.keyId, privateKey: input.privateKey, path }); + const response = await input.transport.request(path, compactJws); + if (response.status >= 400) { + const code = typeof (response.body as { error?: unknown })?.error === "string" ? (response.body as { error: string }).error : "connect_service_error"; + if (response.status === 409 && code === "claim_already_consumed") throw new ConnectClaimInterceptionError(); + throw new ConnectServiceResponseError(response.status, code); + } + return response.body; + } + return { + createHandshake: async (body: unknown) => handshakeResponseSchema.parse(await request("/v1/handshakes", handshakeRequestSchema.parse(body))), + claim: async (claimCode: string) => claimResponseSchema.parse(await request("/v1/claims", claimRequestSchema.parse({ claimCode }))), + mintGrantToken: async (grantRef: string, grantWrapKey: string) => grantTokenResponseSchema.parse(await request(`/v1/grants/${encodeURIComponent(grantRef)}/token`, grantTokenRequestSchema.parse({ grantWrapKey }))), + registerRelay: async (body: unknown) => relayRegistrationResponseSchema.parse(await request("/v1/connections/register", relayRegistrationRequestSchema.parse(body))), + }; +} +export type ConnectBrokerClient = ReturnType; +type SecretRef = { secretId: string; versionSelector: "latest"; configPath: string; required: boolean; label: string }; +export type ConnectionBrokerVault = { put(input: { companyId: string; connectionId: string; purpose: "grant" | "relay"; value: string }): Promise }; +export type ConnectionBrokerStore = { + saveGrant(input: { companyId: string; connectionId: string; kind: "workspace" | "user"; subjectUserId: string | null; credentialSecretRefs: SecretRef[]; custodyMode: "A" | "B2"; serviceGrantRef: string | null; tokenMeta: { scopes: string[]; expiresAt?: string | null } }): Promise<{ id: string }>; + registerRelay(input: { companyId: string; connectionId: string; connectionPublicRef: string; intakeUrls: string[]; relaySecretRef: { secretId: string; versionSelector: "latest" } }): Promise; +}; +export function connectionBrokerStore(db: Db): ConnectionBrokerStore { + return { + async saveGrant(input) { + const providerTenant = { name: `brokered:${input.custodyMode}`, externalId: input.serviceGrantRef ?? undefined, tokenMeta: input.tokenMeta }; + const condition = input.kind === "user" ? eq(connectionGrants.subjectUserId, input.subjectUserId!) : eq(connectionGrants.isDefault, true); + const existing = await db.select({ id: connectionGrants.id }).from(connectionGrants).where(and(eq(connectionGrants.connectionId, input.connectionId), condition)).limit(1).then((rows) => rows[0] ?? null); + if (existing) return db.update(connectionGrants).set({ credentialSecretRefs: input.credentialSecretRefs, providerTenant, status: "active", updatedAt: new Date() }).where(eq(connectionGrants.id, existing.id)).returning({ id: connectionGrants.id }).then((rows) => rows[0]); + return db.insert(connectionGrants).values({ companyId: input.companyId, connectionId: input.connectionId, kind: input.kind, subjectUserId: input.subjectUserId, credentialSecretRefs: input.credentialSecretRefs, providerTenant, isDefault: input.kind === "workspace" }).returning({ id: connectionGrants.id }).then((rows) => rows[0]); + }, + async registerRelay(input) { + const connection = await db.select({ config: toolConnections.config }).from(toolConnections).where(and(eq(toolConnections.id, input.connectionId), eq(toolConnections.companyId, input.companyId))).limit(1).then((rows) => rows[0]); + if (!connection) throw new Error("Tool connection not found"); + await db.update(toolConnections).set({ config: { ...connection.config, relay: { publicRef: input.connectionPublicRef, intakeUrls: input.intakeUrls, secretId: input.relaySecretRef.secretId, secretVersion: input.relaySecretRef.versionSelector } }, updatedAt: new Date() }).where(eq(toolConnections.id, input.connectionId)); + }, + }; +} +export function connectionBrokerService(input: { client: ConnectBrokerClient; store: ConnectionBrokerStore; vault: ConnectionBrokerVault; enabled?: boolean }) { + const assertEnabled = () => { if (!input.enabled) throw new ConnectServiceResponseError(403, "mode_not_allowed"); }; + return { + startOAuth: async (body: unknown) => { assertEnabled(); return input.client.createHandshake(body); }, + async claimOAuth(params: { companyId: string; connectionId: string; claimCode: string; kind: "workspace" | "user"; subjectUserId?: string | null }) { + assertEnabled(); + const claim = await input.client.claim(params.claimCode); + if (claim.custodyMode !== "A" && claim.custodyMode !== "B2") throw new ConnectServiceResponseError(403, "mode_not_allowed"); + const serviceGrantRef = claim.custodyMode === "A" ? claim.grantRef : null; + const sealedGrant = claim.custodyMode === "A" ? (await input.client.mintGrantToken(claim.grantRef, randomBytes(32).toString("base64url"))).grantSealed : claim.grantSealed; + const secretRef = await input.vault.put({ companyId: params.companyId, connectionId: params.connectionId, purpose: "grant", value: sealedGrant }); + const grant = await input.store.saveGrant({ companyId: params.companyId, connectionId: params.connectionId, kind: params.kind, subjectUserId: params.kind === "user" ? params.subjectUserId ?? null : null, credentialSecretRefs: [secretRef], custodyMode: claim.custodyMode, serviceGrantRef, tokenMeta: claim.tokenMeta }); + return { grantId: grant.id, custodyMode: claim.custodyMode, tokenMeta: claim.tokenMeta }; + }, + async registerRelay(params: { companyId: string; connectionId: string; providerSlug: string; intakeUrls: string[] }) { + assertEnabled(); + const registration = await input.client.registerRelay({ connectionRef: params.connectionId, providerSlug: params.providerSlug, intakeUrls: params.intakeUrls }); + const secretRef = await input.vault.put({ companyId: params.companyId, connectionId: params.connectionId, purpose: "relay", value: registration.relaySecret }); + await input.store.registerRelay({ companyId: params.companyId, connectionId: params.connectionId, connectionPublicRef: registration.connectionPublicRef, intakeUrls: registration.intakeUrls, relaySecretRef: secretRef }); + return registration; + }, + }; +} diff --git a/server/src/services/tool-access.ts b/server/src/services/tool-access.ts index 0abcf380190..9f4e67ad039 100644 --- a/server/src/services/tool-access.ts +++ b/server/src/services/tool-access.ts @@ -118,6 +118,7 @@ import { readSignedToolArgumentsPayload } from "./tool-content-guards.js"; import { narrowestScopeBindings, profileIdsInBindingOrder } from "./tool-profile-binding-precedence.js"; import { recordToolRuntimeAuditWriteFailure, TOOL_RUNTIME_AUDIT_WRITE_FAILURE_METRIC } from "./tool-runtime-metrics.js"; import { createToolRuntimeSupervisor, ToolRuntimeSupervisorError } from "./tool-runtime-supervisor.js"; +import type { connectionBrokerService } from "./connection-broker.js"; type ActorInfo = { actorType?: "agent" | "user" | "system" | "plugin"; @@ -143,6 +144,8 @@ type ToolAccessServiceOptions = { deploymentExposure?: DeploymentExposure; trustedLocalStdioRuntimeHost?: string | null; now?: () => Date; + connectionBroker?: ReturnType; + brokeredCustodyMode?: "A" | "B2"; }; type DbTransaction = Parameters[0]>[0]; @@ -4738,7 +4741,15 @@ export function toolAccessService(db: Db, options: ToolAccessServiceOptions = {} async function startOAuth( companyId: string, connectionId: string, - input: { redirectUri: string; actor: ActorInfo; subjectUserId?: string; scopes?: string[]; returnTo?: string; issueId?: string }, + input: { + redirectUri: string; + actor: ActorInfo; + brokered?: boolean; + subjectUserId?: string; + scopes?: string[]; + returnTo?: string; + issueId?: string; + }, ): Promise { const connection = await getConnectionRow(connectionId, companyId); if (connection.status === "archived") throw conflict("Archived app connections cannot start sign in"); @@ -4749,6 +4760,23 @@ export function toolAccessService(db: Db, options: ToolAccessServiceOptions = {} const client = oauthClientForConnection(connection, endpoints.provider); if (!client.clientId) throw unprocessable(`OAuth client id is not configured for ${endpoints.provider}`); + if (input.brokered) { + if (!options.connectionBroker) throw unprocessable("Brokered OAuth is not configured for this instance"); + const custodyMode = options.brokeredCustodyMode ?? "A"; + const result = await options.connectionBroker.startOAuth({ + providerSlug: endpoints.provider, + methodKey: "oauth", + custodyMode, + clientOwnership: connection.ownership === "customer" ? "customer" : connection.ownership, + connectionRef: connection.id, + grantWrapKey: custodyMode === "A" ? undefined : randomOauthToken(48), + returnUrl: input.redirectUri, + scopes: endpoints.scopes, + byoClient: connection.ownership === "customer" ? { clientId: client.clientId } : undefined, + }); + return { connectionId: connection.id, provider: endpoints.provider, authorizationUrl: result.authorizeUrl, expiresAt: result.expiresAt }; + } + await db.delete(toolOauthStates).where(lt(toolOauthStates.expiresAt, new Date())); const state = randomOauthToken(); @@ -5291,6 +5319,18 @@ export function toolAccessService(db: Db, options: ToolAccessServiceOptions = {} }); }, + claimBrokeredOAuth: async (companyId: string, connectionId: string, input: { claimCode: string; kind: "workspace" | "user"; subjectUserId?: string | null }) => { + await getConnectionRow(connectionId, companyId); + if (!options.connectionBroker) throw unprocessable("Brokered OAuth is not configured for this instance"); + return options.connectionBroker.claimOAuth({ companyId, connectionId, ...input }); + }, + + registerConnectionRelay: async (companyId: string, connectionId: string, input: { providerSlug: string; intakeUrls: string[] }) => { + await getConnectionRow(connectionId, companyId); + if (!options.connectionBroker) throw unprocessable("Connection relay is not configured for this instance"); + return options.connectionBroker.registerRelay({ companyId, connectionId, ...input }); + }, + peekOAuthState, completeOAuthCallback,