feat(auth): per-user OAuth on downstream MCP connections + no-DCR provider support

apps/api/migrations/171-per-user-oauth.ts

@@ -0,0 +1,119 @@
+/**
+ * Migration 171: Per-user OAuth on downstream MCP connections
+ *
+ * Adds `auth_mode` to `connections` and re-introduces `userId` on
+ * `downstream_tokens` as a nullable column. Migration 017
+ * (017-downstream-token-remove-userid) had removed `userId` because all
+ * tokens were treated as connection-scoped.
+ *
+ * This change reintroduces user-scoped tokens as an opt-in per connection:
+ *
+ *   - `connections.auth_mode = 'shared'`  → one token per connection
+ *     (legacy behaviour; `downstream_tokens.userId IS NULL`).
+ *   - `connections.auth_mode = 'per_user'` → one token per
+ *     (connection, user) pair; each member authorizes with their own
+ *     identity, and audit logs at the downstream provider show the real
+ *     person acting.
+ *
+ * Two partial unique indexes enforce the rule:
+ *   - `downstream_tokens_shared_unique` on (connectionId) WHERE userId IS NULL
+ *   - `downstream_tokens_per_user_unique` on (connectionId, userId)
+ *       WHERE userId IS NOT NULL
+ *
+ * Existing rows are untouched: their `userId` becomes NULL (= shared) and
+ * the parent connection keeps `auth_mode = 'shared'` (column default).
+ */
+
+import { sql, type Kysely } from "kysely";
+
+export async function up(db: Kysely<unknown>): Promise<void> {
+  // 1. connections.auth_mode (default 'shared' for backward compatibility)
+  await db.schema
+    .alterTable("connections")
+    .addColumn("auth_mode", "text", (col) => col.notNull().defaultTo("shared"))
+    .execute();
+
+  await sql`
+    ALTER TABLE connections
+    ADD CONSTRAINT connections_auth_mode_check
+    CHECK (auth_mode IN ('shared', 'per_user'))
+  `.execute(db);
+
+  // 2. downstream_tokens.userId (nullable, FK to user.id, cascade on delete)
+  await db.schema
+    .alterTable("downstream_tokens")
+    .addColumn("userId", "text", (col) =>
+      col.references("user.id").onDelete("cascade"),
+    )
+    .execute();
+
+  // 3. Drop legacy unique-on-connectionId index created by migration 017.
+  // Kysely's createIndex concatenated the column name onto the base, so the
+  // actual index in Postgres is `idx_downstream_tokens_connectionId` (quoted,
+  // with the camelCase column). DROP IF EXISTS keeps both spellings safe.
+  await sql`DROP INDEX IF EXISTS "idx_downstream_tokens_connectionId"`.execute(
+    db,
+  );
+  await sql`DROP INDEX IF EXISTS idx_downstream_tokens_connection`.execute(db);
+
+  // 4. New partial unique indexes — one shared row, many per-user rows.
+  await sql`
+    CREATE UNIQUE INDEX downstream_tokens_shared_unique
+    ON downstream_tokens ("connectionId")
+    WHERE "userId" IS NULL
+  `.execute(db);
+
+  await sql`
+    CREATE UNIQUE INDEX downstream_tokens_per_user_unique
+    ON downstream_tokens ("connectionId", "userId")
+    WHERE "userId" IS NOT NULL
+  `.execute(db);
+
+  // 5. Lookup index for "my connections" views.
+  await db.schema
+    .createIndex("idx_downstream_tokens_user")
+    .on("downstream_tokens")
+    .column("userId")
+    .execute();
+}
+
+export async function down(db: Kysely<unknown>): Promise<void> {
+  // Drop indexes/constraints in reverse order, then columns.
+  await db.schema.dropIndex("idx_downstream_tokens_user").execute();
+  await db.schema.dropIndex("downstream_tokens_per_user_unique").execute();
+  await db.schema.dropIndex("downstream_tokens_shared_unique").execute();
+
+  // Restore the legacy single-token-per-connection unique index. Any
+  // per-user rows must be deduplicated first; keep the most recently
+  // updated row (mirrors the strategy used in migration 017).
+  await sql`
+    DELETE FROM downstream_tokens
+    WHERE id NOT IN (
+      SELECT id FROM (
+        SELECT id, "connectionId",
+          ROW_NUMBER() OVER (
+            PARTITION BY "connectionId" ORDER BY "updatedAt" DESC
+          ) AS rn
+        FROM downstream_tokens
+      ) ranked
+      WHERE rn = 1
+    )
+  `.execute(db);
+
+  await sql`
+    CREATE UNIQUE INDEX "idx_downstream_tokens_connectionId"
+    ON downstream_tokens ("connectionId")
+  `.execute(db);
+
+  await db.schema
+    .alterTable("downstream_tokens")
+    .dropColumn("userId")
+    .execute();
+
+  await sql`
+    ALTER TABLE connections
+    DROP CONSTRAINT connections_auth_mode_check
+  `.execute(db);
+
+  await db.schema.alterTable("connections").dropColumn("auth_mode").execute();
+}

apps/api/migrations/index.ts

@@ -169,6 +169,7 @@ import * as migration167taskboardrunretry from "./167-task-board-run-retry.ts";
 import * as migration168orgreposync from "./168-org-repo-sync.ts";
 import * as migration169taskboardmergefailedactivity from "./169-task-board-merge-failed-activity.ts";
 import * as migration170taskboarditemrepo from "./170-task-board-item-repo.ts";
+import * as migration171peruseroauth from "./171-per-user-oauth.ts";
 
 /**
  * Core migrations for the Studio application.
@@ -367,6 +368,7 @@ const migrations: Record<string, Migration> = {
   "169-task-board-merge-failed-activity":
     migration169taskboardmergefailedactivity,
   "170-task-board-item-repo": migration170taskboarditemrepo,
+  "171-per-user-oauth": migration171peruseroauth,
 };
 
 export default migrations;

apps/api/src/api/app.ts

@@ -465,12 +465,20 @@ const oauthProxyHandler: MiddlewareHandler<Env> = async (c) => {
       : undefined;
   const resourceIndicator = resourceOverride ?? connection.connection_url;
 
+  // Scopes advertised by the downstream resource (RFC 9728). Used as the
+  // authorize-leg scope fallback for pre-registered-client providers (e.g.
+  // Google) that reject authorization requests without an explicit scope.
+  let resourceScopes: string[] | undefined;
   if (resourceRes.ok) {
     // Origin has Protected Resource Metadata - use authorization_servers from it
     const resourceData = (await resourceRes.json()) as {
       authorization_servers?: string[];
+      scopes_supported?: string[];
     };
     originAuthServer = resourceData.authorization_servers?.[0];
+    if (Array.isArray(resourceData.scopes_supported)) {
+      resourceScopes = resourceData.scopes_supported;
+    }
   }
 
   // Fall back to origin root if:
@@ -493,6 +501,17 @@ const oauthProxyHandler: MiddlewareHandler<Env> = async (c) => {
     registration_endpoint?: string;
   };
 
+  // Pre-registered OAuth client for providers WITHOUT Dynamic Client
+  // Registration (e.g. accounts.google.com). Operators store the client in
+  // `connection.oauth_config`; the proxy then (a) synthesizes the DCR
+  // response, (b) pins the authorize client_id, and (c) injects the client
+  // secret on the token leg — the secret never reaches the browser.
+  const preRegisteredClient =
+    connection.oauth_config?.clientId &&
+    connection.oauth_config.grantType !== "client_credentials"
+      ? connection.oauth_config
+      : undefined;
+
   // Map endpoint name to URL
   let originEndpointUrl: string | undefined;
   if (endpoint === "authorize") {
@@ -501,6 +520,31 @@ const oauthProxyHandler: MiddlewareHandler<Env> = async (c) => {
     originEndpointUrl = endpoints.token_endpoint;
   } else if (endpoint === "register") {
     originEndpointUrl = endpoints.registration_endpoint;
+    // No downstream DCR but an operator-supplied client exists: answer the
+    // registration ourselves so the standard MCP SDK flow proceeds. The
+    // response advertises `token_endpoint_auth_method: "none"` — the SDK
+    // must NOT try to authenticate the client itself; the proxy's token leg
+    // injects the real secret server-side.
+    if (!originEndpointUrl && preRegisteredClient) {
+      let redirectUris: string[] = [];
+      try {
+        const body = (await c.req.json()) as { redirect_uris?: string[] };
+        if (Array.isArray(body.redirect_uris))
+          redirectUris = body.redirect_uris;
+      } catch {
+        // No/invalid body — fine, redirect_uris stays empty.
+      }
+      return c.json(
+        {
+          client_id: preRegisteredClient.clientId,
+          redirect_uris: redirectUris,
+          token_endpoint_auth_method: "none",
+          grant_types: ["authorization_code", "refresh_token"],
+          response_types: ["code"],
+        },
+        201,
+      );
+    }
   }
 
   if (!originEndpointUrl) {
@@ -554,6 +598,32 @@ const oauthProxyHandler: MiddlewareHandler<Env> = async (c) => {
       targetUrl.searchParams.set("resource", resourceIndicator);
     }
 
+    // Pre-registered client (no-DCR providers): pin the client_id and make
+    // sure a scope is present — Google rejects authorization requests with
+    // no scope. Prefer the operator's `oauth_config.scopes`, falling back to
+    // the resource's advertised `scopes_supported`.
+    if (preRegisteredClient) {
+      targetUrl.searchParams.set("client_id", preRegisteredClient.clientId);
+      if (!targetUrl.searchParams.get("scope")) {
+        const scopes = preRegisteredClient.scopes?.length
+          ? preRegisteredClient.scopes
+          : (resourceScopes ?? []);
+        if (scopes.length) {
+          targetUrl.searchParams.set("scope", scopes.join(" "));
+        }
+      }
+      // Google only issues a refresh_token with explicit offline consent.
+      // Harmless no-ops for other providers, so gate on the Google AS.
+      if (targetUrl.hostname.endsWith("google.com")) {
+        if (!targetUrl.searchParams.has("access_type")) {
+          targetUrl.searchParams.set("access_type", "offline");
+        }
+        if (!targetUrl.searchParams.has("prompt")) {
+          targetUrl.searchParams.set("prompt", "consent");
+        }
+      }
+    }
+
     // Add smart OAuth params for deco-hosted MCPs to skip org/project selection
     // Wrapped in try-catch to ensure OAuth redirect proceeds even if smart params fail
     if (isDecoHostedMcp(connection.connection_url)) {
@@ -626,6 +696,22 @@ const oauthProxyHandler: MiddlewareHandler<Env> = async (c) => {
       if (formData.has("resource")) {
         formData.set("resource", resourceIndicator);
       }
+      // Pre-registered client (no-DCR providers): the browser flow runs with
+      // `token_endpoint_auth_method: "none"`, so the exchange arrives without
+      // credentials. Inject them here — server-side only — so providers that
+      // require a confidential client (Google) accept the exchange.
+      if (preRegisteredClient) {
+        if (!formData.get("client_id")) {
+          formData.set("client_id", preRegisteredClient.clientId);
+        }
+        if (
+          preRegisteredClient.clientSecret &&
+          !formData.get("client_secret") &&
+          formData.get("client_id") === preRegisteredClient.clientId
+        ) {
+          formData.set("client_secret", preRegisteredClient.clientSecret);
+        }
+      }
       const cidRaw = formData.get("client_id");
       const csRaw = formData.get("client_secret");
       if (typeof cidRaw === "string" && cidRaw) capturedClientId = cidRaw;

apps/api/src/api/routes/dev-connection.ts

@@ -159,6 +159,7 @@ export async function resolveDevConnection(
     connection_token: null,
     connection_headers: null,
     oauth_config: null,
+    auth_mode: "shared",
     configuration_state: null,
     configuration_scopes: null,
     metadata: {

apps/api/src/api/routes/downstream-token.ts

@@ -115,9 +115,17 @@ export const createDownstreamTokenRoutes = () => {
     // Create storage instance
     const tokenStorage = new DownstreamTokenStorage(ctx.db, ctx.vault);
 
+    // Pick the storage scope based on auth_mode:
+    //   - per_user: this token belongs to the authenticated caller.
+    //   - shared:   this token is org-wide. The endpoint is admin-facing
+    //               in this mode (any member with org-admin can authorise
+    //               the bot account once).
+    const tokenUserId = connection.auth_mode === "per_user" ? userId : null;
+
     // Save token
     const tokenData: DownstreamTokenData = {
       connectionId,
+      userId: tokenUserId,
       accessToken: body.accessToken,
       refreshToken: body.refreshToken ?? null,
       scope: body.scope ?? null,
@@ -164,7 +172,10 @@ export const createDownstreamTokenRoutes = () => {
     }
 
     const tokenStorage = new DownstreamTokenStorage(ctx.db, ctx.vault);
-    await tokenStorage.delete(connectionId);
+    // Same scoping rule as the upsert: per-user callers only erase their
+    // own token; admins of a shared connection erase the org-wide one.
+    const tokenUserId = connection.auth_mode === "per_user" ? userId : null;
+    await tokenStorage.delete(connectionId, tokenUserId);
 
     return c.json({ success: true });
   });
@@ -198,7 +209,11 @@ export const createDownstreamTokenRoutes = () => {
     }
 
     const tokenStorage = new DownstreamTokenStorage(ctx.db, ctx.vault);
-    const token = await tokenStorage.get(connectionId);
+    // Status reflects what THIS user can do. For per-user connections we
+    // look up their own token; for shared connections we look at the
+    // org-wide one.
+    const tokenUserId = connection.auth_mode === "per_user" ? userId : null;
+    const token = await tokenStorage.get(connectionId, tokenUserId);
 
     if (!token) {
       return c.json({

apps/api/src/api/routes/oauth-proxy.ts

@@ -801,6 +801,21 @@ const authServerMetadataHandler = async (c: {
     // Rewrite OAuth endpoint URLs to go through our proxy
     const rewrittenData = rewriteAuthServerMetadata(data, proxyBase);
 
+    // Providers without Dynamic Client Registration (e.g. accounts.google.com)
+    // publish no registration_endpoint, and the MCP SDK aborts with
+    // "Incompatible auth server" before ever attempting the flow. When the
+    // operator stored a pre-registered client on the connection
+    // (`oauth_config.clientId`), advertise the proxy's own register endpoint —
+    // the proxy answers it by synthesizing the DCR response from that config
+    // (see the register branch in the oauth-proxy dispatcher).
+    if (
+      !rewrittenData.registration_endpoint &&
+      connection.oauth_config?.clientId &&
+      connection.oauth_config.grantType !== "client_credentials"
+    ) {
+      rewrittenData.registration_endpoint = `${proxyBase}/register`;
+    }
+
     return new Response(JSON.stringify(rewrittenData), {
       status: 200,
       headers: { "Content-Type": "application/json" },

apps/api/src/api/routes/proxy.ts

@@ -30,6 +30,10 @@ import {
   recordFailure,
   recordSuccess,
 } from "@/mcp-clients/circuit-breaker";
+import {
+  isPerUserAuthorizationRequiredError,
+  renderPerUserAuthorizationRequired,
+} from "@/mcp-clients/outbound/errors";
 import { getConnectionCircuitStore } from "@/mcp-clients/connection-circuit-store";
 import { CONNECTION_ERROR_REPROBE_COOLDOWN_MS } from "@/core/constants";
 import { peekRpcMethod, probeDecision } from "./proxy-handshake";
@@ -413,6 +417,15 @@ export const createProxyRoutes = () => {
         if (error instanceof CircuitOpenError) {
           throw error;
         }
+        // Per-user authorization required is an EXPECTED state for
+        // `auth_mode: "per_user"` connections when the caller has no token yet
+        // — not a downstream outage. Return the 401 challenge (with authorize
+        // URL) WITHOUT tripping the breaker or auto-disabling the connection.
+        // Handled before recordFailure below, which would otherwise open the
+        // circuit and 503 the connection (hiding the OAuth the error asks for).
+        if (isPerUserAuthorizationRequiredError(error)) {
+          return renderPerUserAuthorizationRequired(error);
+        }
         // Check if this is an auth error - if so, return appropriate 401
         // Note: This only applies to HTTP connections
         const connection = await ctx.storage.connections.findById(

apps/api/src/api/routes/virtual-mcp.ts

@@ -21,6 +21,10 @@ import { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/
 import { Hono } from "hono";
 import { getUserId, type StudioContext } from "../../core/studio-context";
 import { MCP_TOOL_CALL_TIMEOUT_MS } from "@/core/constants";
+import {
+  isPerUserAuthorizationRequiredError,
+  renderPerUserAuthorizationRequired,
+} from "../../mcp-clients/outbound/errors";
 import { createVirtualClientFrom } from "../../mcp-clients/virtual-mcp";
 import { resolveDevConnection } from "./dev-connection";
 import { readSandboxMap } from "../../tools/sandbox/sandbox-map";
@@ -237,6 +241,11 @@ export async function handleVirtualMcpRequest(
     );
   } catch (error) {
     const err = error as Error;
+    // Per-user OAuth: a tool in the aggregated set requires the caller to
+    // authorise their own account first. Bubble the actionable 401 up.
+    if (isPerUserAuthorizationRequiredError(err)) {
+      return renderPerUserAuthorizationRequired(err);
+    }
     console.error("[virtual-mcp] Error handling virtual MCP request:", err);
     return c.json(
       { error: "Internal server error", message: err.message },