Connections carry who signed in

backend/druks/mcp/oauth.py

@@ -1,3 +1,4 @@
+import logging
 from urllib.parse import urlparse
 
 import httpx
@@ -14,6 +15,9 @@
 from druks.services.constants import OAUTH_MINT_WAIT_ATTEMPTS, OAUTH_MINT_WAIT_INTERVAL_SECONDS
 from druks.services.models import OauthConnection
 from druks.services.oauth import complete_connect as complete_oauth_exchange
+from druks.services.oauth import fetch_identity
+
+logger = logging.getLogger(__name__)
 
 
 def _http() -> httpx.AsyncClient:
@@ -36,6 +40,18 @@ def _origin(url: str) -> str:
     return f"{parts.scheme}://{parts.netloc}"
 
 
+def _same_origin_userinfo(metadata: dict) -> str:
+    # The token goes here as a bearer at consent, so only trust a userinfo
+    # endpoint on the issuer's own origin — off-issuer would exfiltrate it.
+    endpoint = metadata.get("userinfo_endpoint", "")
+    if not endpoint or _origin(endpoint) == _origin(metadata["issuer"]):
+        return endpoint
+    logger.warning(
+        "ignoring userinfo_endpoint %s off the issuer origin %s", endpoint, metadata["issuer"]
+    )
+    return ""
+
+
 async def _get_json(client: httpx.AsyncClient, url: str) -> dict | None:
     # A discovery probe: any failure — network, non-2xx, non-JSON, non-object —
     # just means this candidate url isn't it.
@@ -193,6 +209,7 @@ async def begin_connect(
             "server_url": server_url,
             "account_id": account_id,
             "identity_mode": identity_mode,
+            "userinfo_endpoint": _same_origin_userinfo(metadata),
         },
         extra_authorize_params={"resource": server_url},
     )
@@ -234,9 +251,12 @@ async def complete_connect(*, state: str, code: str) -> str:
         client_id=pending["client_id"],
         client_secret=pending["client_secret"],
     )
+    identity = {}
+    if pending["userinfo_endpoint"]:
+        identity = await fetch_identity(pending["userinfo_endpoint"], tokens["access_token"])
     connection = get_connection(name, account_id)
     if connection:
-        connection.reconnect(refresh_token=tokens["refresh_token"], scopes=[])
+        connection.reconnect(refresh_token=tokens["refresh_token"], scopes=[], identity=identity)
         # A reconsent's stale cached token must not serve until its TTL runs out.
         await evict_access_token(name, account_id)
     else:
@@ -245,6 +265,7 @@ async def complete_connect(*, state: str, code: str) -> str:
             account_id=account_id,
             refresh_token=tokens["refresh_token"],
             scopes=[],
+            identity=identity,
         )
     return name
 

backend/druks/services/base.py

@@ -9,7 +9,7 @@
 
 from .exceptions import ServiceConnectError, ServiceNotConnectedError
 from .models import OauthConnection, ServiceIdentity
-from .oauth import OauthClient
+from .oauth import OauthClient, fetch_identity
 
 
 class Connection:
@@ -93,12 +93,16 @@ class Service:
     # Set both endpoints when the registered app is an OAuth client;
     # ``get_oauth_client()`` then hands back the connected identity as a
     # configured ``OauthClient``. Scopes are not declared here — the
-    # extensions that use the service declare them (``connection``), and
+    # extensions that use the service declare them (``with_scopes``), and
     # the connect door asks for their union.
     authorization_endpoint: ClassVar[str] = ""
     token_endpoint: ClassVar[str] = ""
     # HTTP Basic on the token endpoint; False sends the secret in the body.
     basic_auth: ClassVar[bool] = False
+    # The endpoint that returns the signed-in account's facts;
+    # identity_scopes join the consent ask.
+    identity_endpoint: ClassVar[str] = ""
+    identity_scopes: ClassVar[tuple[str, ...]] = ()
 
     def __init_subclass__(cls, **kwargs: Any) -> None:
         super().__init_subclass__(**kwargs)
@@ -173,8 +177,17 @@ def declarations(cls) -> "list[ScopedService]":
     def required_scopes(cls) -> tuple[str, ...]:
         """The union of every installed declaration's scopes — the consent ask."""
         scopes = {scope for declaration in cls.declarations() for scope in declaration.scopes}
+        scopes.update(cls.identity_scopes)
         return tuple(sorted(scopes))
 
+    @classmethod
+    async def get_identity(cls, access_token: str) -> dict[str, Any]:
+        """The signed-in account's facts, read from ``identity_endpoint``.
+        Override when the provider needs a different call."""
+        if cls.identity_endpoint:
+            return await fetch_identity(cls.identity_endpoint, access_token)
+        return {}
+
     @classmethod
     def get_oauth_client(cls) -> OauthClient:
         """The connected identity as a configured ``OauthClient``, keyed by

backend/druks/services/models.py

@@ -63,6 +63,9 @@ class OauthConnection(Base, Uuid7Pk):
     # The token response's ``scope`` when the provider echoes one, else the
     # scopes the consent asked for.
     scopes: Mapped[list[str]] = mapped_column(JSONB, default=list)
+    # The provider's facts for the sign-in (email, username, name), set at
+    # consent; {} when the provider gave none.
+    identity: Mapped[dict[str, Any]] = mapped_column(JSONB, default=dict)
     connected_at: Mapped[datetime] = mapped_column(default=Base.utc_now)
 
     @classmethod
@@ -71,10 +74,20 @@ def get(cls, connection_id: str) -> "OauthConnection | None":
 
     @classmethod
     def create(
-        cls, *, provider: str, account_id: str, refresh_token: str, scopes: list[str]
+        cls,
+        *,
+        provider: str,
+        account_id: str,
+        refresh_token: str,
+        scopes: list[str],
+        identity: dict[str, Any] | None = None,
     ) -> "OauthConnection":
         connection = cls(
-            provider=provider, account_id=account_id, refresh_token=refresh_token, scopes=scopes
+            provider=provider,
+            account_id=account_id,
+            refresh_token=refresh_token,
+            scopes=scopes,
+            identity=identity or {},
         )
         db_session().add(connection)
         db_session().flush()
@@ -102,9 +115,13 @@ def list_owned_by(cls, account_id: str | None) -> "list[OauthConnection]":
             )
         )
 
-    def reconnect(self, *, refresh_token: str, scopes: list[str]) -> None:
+    def reconnect(
+        self, *, refresh_token: str, scopes: list[str], identity: dict[str, Any] | None = None
+    ) -> None:
         self.refresh_token = refresh_token
         self.scopes = scopes
+        if identity:
+            self.identity = identity
         self.connected_at = Base.utc_now()
         db_session().flush()
 

backend/druks/services/oauth.py

@@ -330,3 +330,16 @@ async def complete_connect(*, state: str, code: str) -> tuple[dict, dict]:
             context=pending,
         )
     return tokens, pending
+
+
+async def fetch_identity(endpoint: str, access_token: str) -> dict:
+    """The provider's facts for a fresh token. Any failure returns {} —
+    a missing label must not fail the consent."""
+    async with _http() as http:
+        try:
+            response = await http.get(endpoint, headers={"Authorization": f"Bearer {access_token}"})
+            response.raise_for_status()
+            payload = response.json()
+        except (httpx.HTTPError, ValueError):
+            return {}
+    return payload if isinstance(payload, dict) else {}

backend/druks/services/routes.py

@@ -119,18 +119,20 @@ async def oauth_callback(state: str = "", code: str = "", error: str = "") -> Re
     except OauthExchangeError as exchange_error:
         raise HTTPException(status_code=400, detail=str(exchange_error)) from exchange_error
     provider = pending["provider"]
-    if not services.get(provider):
+    service = services.get(provider)
+    if not service:
         # A state begun by another door (an MCP connect) finishes at its own callback.
         raise HTTPException(status_code=400, detail=f"No OAuth service {provider!r}.")
     granted = tokens.get("scope", "").split() or pending["scopes"]
+    identity = await service.get_identity(tokens["access_token"])
     reconsent = bool(pending["connection_id"])
     if reconsent:
         row = OauthConnection.get(pending["connection_id"])
         if not row:
             raise HTTPException(
                 status_code=400, detail="The connection was removed while consent was open."
             )
-        row.reconnect(refresh_token=tokens["refresh_token"], scopes=granted)
+        row.reconnect(refresh_token=tokens["refresh_token"], scopes=granted, identity=identity)
         # A reconsent's narrower cached token must not serve until its TTL runs out.
         await OauthClient(provider=provider).evict_access_token(row.id)
     else:
@@ -139,6 +141,7 @@ async def oauth_callback(state: str = "", code: str = "", error: str = "") -> Re
             account_id=pending["account_id"],
             refresh_token=tokens["refresh_token"],
             scopes=granted,
+            identity=identity,
         )
     await publish(
         "oauth.connected",

backend/druks/services/schemas.py

@@ -26,6 +26,7 @@ class ConnectionResponse(BaseResponse):
     id: str
     provider: str
     scopes: list[str]
+    identity: dict[str, Any]
     connected_at: datetime
 
 

backend/migrations/versions/c4f8a2d97e15_connections_carry_identity.py

@@ -0,0 +1,30 @@
+"""connections carry identity
+
+Revision ID: c4f8a2d97e15
+Revises: a7c2e9f14b38
+Create Date: 2026-08-20 00:00:00.000000
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = "c4f8a2d97e15"
+down_revision: str | Sequence[str] | None = "a7c2e9f14b38"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "oauth_connections",
+        sa.Column("identity", postgresql.JSONB(), server_default="{}", nullable=False),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("oauth_connections", "identity")

backend/tests/test_mcp_oauth.py

@@ -48,6 +48,7 @@ def __init__(self) -> None:
         self.resource = _SERVER_URL
         self.issuer = _AUTH_BASE
         self.code_challenge_methods = ["S256"]
+        self.userinfo_endpoint = f"{_AUTH_BASE}/userinfo"
         self.token_status = 200
         self.token_malformed = False
         self.token_response = {
@@ -75,11 +76,14 @@ def handler(self, request: httpx.Request) -> httpx.Response:
                 "issuer": self.issuer,
                 "authorization_endpoint": f"{_AUTH_BASE}/authorize",
                 "token_endpoint": f"{_AUTH_BASE}/token",
+                "userinfo_endpoint": self.userinfo_endpoint,
                 "code_challenge_methods_supported": self.code_challenge_methods,
             }
             if self.registration_supported:
                 metadata["registration_endpoint"] = f"{_AUTH_BASE}/register"
             return httpx.Response(200, json=metadata)
+        if path == "/userinfo":
+            return httpx.Response(200, json={"email": "op@linear.test"})
         if path == "/register":
             return httpx.Response(201, json={"client_id": "client-123"})
         if path == "/token":
@@ -308,6 +312,7 @@ async def test_complete_connect_exchanges_code_and_stores_the_grant(auth_server,
     assert name == _NAME
     grant = oauth.get_connection(_NAME, SYSTEM_ACCOUNT_ID)
     assert grant.refresh_token.decrypt() == "rt-1"
+    assert grant.identity == {"email": "op@linear.test"}
     registration = McpClientRegistration.get_for_account(_NAME, SYSTEM_ACCOUNT_ID)
     assert registration.client_id == "client-123"
     assert registration.token_endpoint == f"{_AUTH_BASE}/token"
@@ -330,6 +335,24 @@ async def test_complete_connect_exchanges_code_and_stores_the_grant(auth_server,
         await oauth.complete_connect(state=state, code="code-1")
 
 
+async def test_off_issuer_userinfo_is_dropped_and_logged(auth_server, druks_db, caplog):
+    auth_server.userinfo_endpoint = "https://evil.test/collect"
+    url = await oauth.begin_connect(
+        _NAME,
+        _SERVER_URL,
+        _ENDPOINT,
+        account_id=SYSTEM_ACCOUNT_ID,
+        identity_mode=IdentityMode.SHARED,
+    )
+    state = dict(parse_qsl(urlparse(url).query))["state"]
+
+    with caplog.at_level("WARNING"):
+        await oauth.complete_connect(state=state, code="code-1")
+
+    assert oauth.get_connection(_NAME, SYSTEM_ACCOUNT_ID).identity == {}
+    assert "evil.test" in caplog.text
+
+
 async def test_complete_connect_without_refresh_token_stores_nothing(auth_server, druks_db):
     auth_server.token_response = {"access_token": "at-1", "expires_in": 3600}
     url = await oauth.begin_connect(

backend/tests/test_services.py

@@ -494,6 +494,23 @@ class Digest:
     assert not NightWatch.acme.get("missing")
 
 
+async def test_get_identity_without_a_declared_endpoint_is_empty(declared_services):
+    from druks.services import Service
+    from pydantic import BaseModel, SecretStr
+
+    class Quiet(Service):
+        name = "quiet_provider"
+        title = "Quiet"
+        authorization_endpoint = "https://quiet.test/authorize"
+        token_endpoint = "https://quiet.test/token"
+
+        class Settings(BaseModel):
+            client_id: str
+            client_secret: SecretStr
+
+    assert await Quiet.get_identity("at-1") == {}
+
+
 def test_with_scopes_requires_oauth_endpoints(declared_services):
     from druks.services import Service
     from pydantic import BaseModel, SecretStr
@@ -522,6 +539,8 @@ class Acme(Service):
         title = "Acme OAuth app"
         authorization_endpoint = "https://acme.test/authorize"
         token_endpoint = "https://acme.test/token"
+        identity_endpoint = "https://acme.test/whoami"
+        identity_scopes = ("openid",)
 
         class Settings(BaseModel):
             client_id: str
@@ -538,11 +557,15 @@ class NightWatch:
         "expires_in": 3600,
         "scope": "profile.read posts.write",
     }
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        if request.url.path == "/whoami":
+            return httpx.Response(200, json={"email": "op@acme.test"})
+        return httpx.Response(200, json=tokens)
+
     monkeypatch.setattr(
         "druks.services.oauth._http",
-        lambda: httpx.AsyncClient(
-            transport=httpx.MockTransport(lambda request: httpx.Response(200, json=tokens))
-        ),
+        lambda: httpx.AsyncClient(transport=httpx.MockTransport(handler)),
     )
     return Acme
 
@@ -565,7 +588,7 @@ def test_oauth_connect_redirects_to_consent_with_the_scope_union(
     consent = urlparse(response.headers["location"])
     params = dict(parse_qsl(consent.query))
     assert response.headers["location"].startswith("https://acme.test/authorize?")
-    assert params["scope"] == "posts.write profile.read"
+    assert params["scope"] == "openid posts.write profile.read"
     assert params["redirect_uri"] == "https://druks.example/api/oauth/callback"
 
 
@@ -626,6 +649,7 @@ async def record(name, **kwargs):
 
         [connection] = OauthConnection.list_for_provider("acme")
         assert connection.refresh_token.decrypt() == "rt-1"
+        assert connection.identity == {"email": "op@acme.test"}
         assert connection.scopes == ["profile.read", "posts.write"]
 
         # Reconsent through the same connection replaces its tokens and
@@ -744,7 +768,7 @@ def entry(client, name="acme"):
         before = entry(client)
         assert before["isOauth"] is True
         assert before["connections"] == []
-        assert before["requiredScopes"] == ["posts.write", "profile.read"]
+        assert before["requiredScopes"] == ["openid", "posts.write", "profile.read"]
         assert before["usedBy"] == ["night_watch.acme"]
 
         row = OauthConnection.create(
@@ -756,6 +780,7 @@ def entry(client, name="acme"):
         [connection] = entry(client)["connections"]
         assert connection["id"] == row.id
         assert connection["scopes"] == ["profile.read"]
+        assert connection["identity"] == {}
         assert connection["connectedAt"]