Connections: the platform owns OAuth sign-ins, the flow, and the audit page

backend/druks/api/app.py

@@ -35,6 +35,7 @@
 from druks.notifications.routes import router as notifications_router
 from druks.redis import close_client
 from druks.services.exceptions import ServiceNotConnectedError
+from druks.services.routes import oauth_router
 from druks.services.routes import router as service_identities_router
 from druks.settings import Settings, ensure_data_dirs, load_settings, setup_logging
 from druks.skills.routes import router as skills_router
@@ -258,6 +259,7 @@ async def _unhandled_exception_handler(
 app.include_router(settings_router, dependencies=_identity_gate)
 app.include_router(extensions_router, dependencies=_identity_gate)
 app.include_router(service_identities_router, dependencies=_identity_gate)
+app.include_router(oauth_router, dependencies=_identity_gate)
 app.include_router(skills_router, dependencies=_identity_gate)
 app.include_router(mcp_router, dependencies=_identity_gate)
 app.include_router(notifications_router, dependencies=_identity_gate)

backend/druks/core/templates/service_oauth_callback.html

@@ -0,0 +1,9 @@
+{% extends "page.html" %}
+{% block content %}
+<p>Connected <b>{{ name }}</b>.
+  You can close this tab and return to druks.</p>
+<script>
+  new BroadcastChannel('druks-service-connect').postMessage({{ name | tojson }});
+  window.close();
+</script>
+{% endblock %}

backend/druks/mcp/constants.py

@@ -17,16 +17,6 @@
 REGISTRY_SEARCH_CACHE_PREFIX = "mcp:registry:search:"
 REGISTRY_CACHE_TTL_SECONDS = 300
 
-# OAuth connect + mint plumbing rides the shared engine (druks.services'
-# OauthClient) under this provider namespace. The namespace keys the engine's
-# connect-state, token-cache, and refresh-lock Redis entries, so it is pinned:
-# a rolling deploy's old and new processes must elect one refresher per grant.
-# The prefixes spell out the derived keys; the token and lock keys append
-# {name}:{account_id}. The callback path is public API surface — the
-# authorization server redirects the operator's browser to
-# {urls.endpoint}{OAUTH_CALLBACK_PATH} after consent.
-OAUTH_PROVIDER = "mcp:oauth"
+# Every dynamically-registered client pins this path as a redirect_uri, so
+# renaming it orphans existing registrations.
 OAUTH_CALLBACK_PATH = "/api/mcp-servers/oauth/callback"
-OAUTH_CONNECT_STATE_PREFIX = f"{OAUTH_PROVIDER}:connect:"
-OAUTH_ACCESS_TOKEN_PREFIX = f"{OAUTH_PROVIDER}:access_token:"
-OAUTH_REFRESH_LOCK_PREFIX = f"{OAUTH_PROVIDER}:refresh_lock:"

backend/druks/mcp/helpers.py

@@ -1,5 +1,26 @@
+from druks.accounts.constants import SYSTEM_ACCOUNT_ID
 from druks.mcp.constants import TOKEN_ENV_PREFIX, TOKEN_ENV_SUFFIX
+from druks.mcp.enums import IdentityMode
+from druks.mcp.exceptions import UnresolvedGrantAccountError
 
 
 def get_bearer_token_env_var(name: str) -> str:
     return f"{TOKEN_ENV_PREFIX}{name.upper()}{TOKEN_ENV_SUFFIX}"
+
+
+def grant_provider(name: str) -> str:
+    # Namespaced so a server name can never collide with a Service name in
+    # the shared grant table and Redis keys.
+    return f"mcp:{name}"
+
+
+def get_grant_account(identity_mode: str | None, run_account_id: str | None) -> str:
+    # Whose grant serves this caller: a shared server's grant lives under
+    # the system account whoever asks; a per-user server's under the asker.
+    if identity_mode == IdentityMode.PER_USER and run_account_id:
+        return run_account_id
+    if identity_mode == IdentityMode.PER_USER:
+        raise UnresolvedGrantAccountError(identity_mode, run_account_id)
+    if identity_mode == IdentityMode.SHARED:
+        return SYSTEM_ACCOUNT_ID
+    raise UnresolvedGrantAccountError(identity_mode, run_account_id)

backend/druks/mcp/models.py

@@ -2,24 +2,22 @@
 from datetime import datetime
 from typing import Any
 
-from sqlalchemy import Boolean, ForeignKey, String, UniqueConstraint, select, update
+from sqlalchemy import Boolean, ForeignKey, String, UniqueConstraint, select
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.orm import Mapped, mapped_column
 
 from druks.accounts.constants import SYSTEM_ACCOUNT_ID
 from druks.core.models import Uuid7Pk
-from druks.database import db_session, get_session
+from druks.database import db_session
 from druks.extensions.registry import mcp_servers
 from druks.mcp.constants import NAME_PATTERN
-from druks.mcp.enums import IdentityMode, TokenSource
-from druks.mcp.exceptions import (
-    InvalidServerNameError,
-    MissingGrantError,
-    UnresolvedGrantAccountError,
-)
+from druks.mcp.enums import TokenSource
+from druks.mcp.exceptions import InvalidServerNameError
+from druks.mcp.helpers import get_grant_account, grant_provider
 from druks.models import Base
 from druks.secrets.fields import EncryptedJsonField, EncryptedTextField, Secret
+from druks.services.models import OauthConnection
 
 
 class McpServer(Base, Uuid7Pk):
@@ -109,11 +107,11 @@ def get_resolved(cls, account_id: str | None) -> dict[str, dict]:
             elif source == TokenSource.OAUTH:
                 server["has_token"] = False
                 if server["identity_mode"]:
-                    grant_account = McpOauthGrant.get_grant_account(
-                        server["identity_mode"], account_id
-                    )
+                    grant_account = get_grant_account(server["identity_mode"], account_id)
                     server["has_token"] = bool(
-                        McpOauthGrant.get_for_account(server["name"], grant_account)
+                        OauthConnection.list_for_account(
+                            grant_provider(server["name"]), grant_account
+                        )
                     )
             else:
                 server["has_token"] = bool(server["token"])
@@ -172,130 +170,63 @@ def delete(self) -> None:
         session.flush()
 
 
-class McpOauthGrant(Base, Uuid7Pk):
-    __tablename__ = "mcp_oauth_grants"
-    __table_args__ = (UniqueConstraint("server_name", "account_id"),)
+class McpClientRegistration(Base, Uuid7Pk):
+    __tablename__ = "mcp_client_registrations"
+    __table_args__ = (UniqueConstraint("server_id", "account_id"),)
 
-    # One grant per (server, account): the durable outcome of an OAuth
-    # connect flow — exactly what mint needs to refresh an access token.
-    # Connect-time material (authorization endpoint, PKCE verifier, state) is
-    # transient and lives in Redis, never here. The refresh token never leaves
-    # the backend; the API exposes only that a grant exists.
-    server_name: Mapped[str] = mapped_column(String)
+    # One RFC 7591 registration per grant: druks registers a fresh client on
+    # every connect, so each account's grant refreshes as the client it
+    # consented through. The refresh token lives on the platform's OauthConnection.
+    server_id: Mapped[str] = mapped_column(ForeignKey("mcp_servers.id", ondelete="CASCADE"))
     account_id: Mapped[str] = mapped_column(
         ForeignKey("accounts.id", ondelete="RESTRICT"), default=SYSTEM_ACCOUNT_ID
     )
-    # Ciphertext at rest; decrypted only into the refresh request body.
-    refresh_token = EncryptedTextField()
     token_endpoint: Mapped[str] = mapped_column(String)
-    # The MCP server url the grant is bound to (RFC 8707): an audience-binding
-    # authorization server rejects a refresh that doesn't carry the same
-    # ``resource`` the code exchange did.
-    resource: Mapped[str] = mapped_column(String)
     client_id: Mapped[str] = mapped_column(String)
     # "" for public clients (PKCE-only); some authorization servers issue one
     # even for token_endpoint_auth_method "none" and then expect it on refresh.
     client_secret = EncryptedTextField(default="")
-    # When the operator last completed consent. Stamped on every store — the
-    # row is upserted on re-connect, so row-creation time would lie.
-    connected_at: Mapped[datetime] = mapped_column(default=Base.utc_now)
-
-    @staticmethod
-    def get_grant_account(identity_mode: str | None, run_account_id: str | None) -> str:
-        # Whose grant serves this caller: a shared server's grant lives under
-        # the system account whoever asks; a per-user server's under the asker.
-        if identity_mode == IdentityMode.PER_USER and run_account_id:
-            return run_account_id
-        if identity_mode == IdentityMode.PER_USER:
-            raise UnresolvedGrantAccountError(identity_mode, run_account_id)
-        if identity_mode == IdentityMode.SHARED:
-            return SYSTEM_ACCOUNT_ID
-        raise UnresolvedGrantAccountError(identity_mode, run_account_id)
 
     @classmethod
-    def get_for_account(cls, server_name: str, account_id: str) -> "McpOauthGrant | None":
+    def get_for_account(cls, server_name: str, account_id: str) -> "McpClientRegistration | None":
         return (
             db_session()
             .execute(
-                select(cls).where(cls.server_name == server_name, cls.account_id == account_id)
+                select(cls)
+                .join(McpServer, McpServer.id == cls.server_id)
+                .where(McpServer.name == server_name, cls.account_id == account_id)
             )
             .scalar_one_or_none()
         )
 
-    @classmethod
-    def list_for_server(cls, server_name: str) -> list["McpOauthGrant"]:
-        return list(db_session().scalars(select(cls).where(cls.server_name == server_name)))
-
     @classmethod
     def store(
         cls,
         *,
-        server_name: str,
+        server_id: str,
         account_id: str,
-        refresh_token: str,
         token_endpoint: str,
-        resource: str,
         client_id: str,
         client_secret: str = "",
-    ) -> "McpOauthGrant":
+    ) -> "McpClientRegistration":
         session = db_session()
         statement = pg_insert(cls).values(
-            server_name=server_name,
+            server_id=server_id,
             account_id=account_id,
-            refresh_token=refresh_token,
             token_endpoint=token_endpoint,
-            resource=resource,
             client_id=client_id,
             client_secret=client_secret,
-            connected_at=cls.utc_now(),
         )
         statement = statement.on_conflict_do_update(
-            index_elements=["server_name", "account_id"],
+            index_elements=["server_id", "account_id"],
             set_={
-                "refresh_token": statement.excluded.refresh_token,
                 "token_endpoint": statement.excluded.token_endpoint,
-                "resource": statement.excluded.resource,
                 "client_id": statement.excluded.client_id,
                 "client_secret": statement.excluded.client_secret,
-                "connected_at": statement.excluded.connected_at,
             },
         ).returning(cls)
         return session.scalars(statement, execution_options={"populate_existing": True}).one()
 
-    def load_refresh_token(self) -> str:
-        # Under the refresh lock: another process may have rotated and
-        # committed, and this transaction may already hold the row —
-        # populate_existing re-reads it past the identity map.
-        fresh = (
-            db_session()
-            .scalars(
-                select(McpOauthGrant)
-                .where(McpOauthGrant.id == self.id)
-                .execution_options(populate_existing=True)
-            )
-            .one_or_none()
-        )
-        if not fresh:
-            raise MissingGrantError(self.server_name, self.account_id)
-        # The grant's secret halves are ciphertext at rest; the plaintext
-        # exists only in the refresh request body.
-        return fresh.refresh_token.decrypt()
-
-    def save_refresh_token(self, rotated: str) -> None:
-        # The provider invalidated the old token the moment it rotated, so
-        # the write commits on its own session, never the enclosing
-        # transaction — a step that rolls back later must not brick the grant.
-        with get_session(db_session().get_bind()) as session:
-            session.execute(
-                update(McpOauthGrant)
-                .where(McpOauthGrant.id == self.id)
-                .values(refresh_token=rotated)
-            )
-            session.commit()
-        # Keep the enclosing transaction's copy true as well.
-        self.refresh_token = rotated
-        db_session().flush()
-
     def delete(self) -> None:
         session = db_session()
         session.delete(self)