feat(connections): add brokered OAuth instance seam

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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"),

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 : "";

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<string>();
+  const claims = new Map([["claim-code-at-least-22-chars", { owner: INSTANCE_ID, attempts: 0, consumed: false }]]);
+  const saved: Parameters<ConnectionBrokerStore["saveGrant"]>[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"));
+  });
+});

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<typeof createSignedConnectBrokerClient>;
+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<SecretRef> };
+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<void>;
+};
+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;
+    },
+  };
+}

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<typeof connectionBrokerService>;
+  brokeredCustodyMode?: "A" | "B2";
 };
 
 type DbTransaction = Parameters<Parameters<Db["transaction"]>[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<ToolOAuthStartResult> {
     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,