Mint narrows a token and refreshes past the cache

backend/druks/services/base.py

@@ -32,8 +32,10 @@ def scopes(self) -> list[str]:
     def connected_at(self):
         return self.row.connected_at
 
-    async def mint_access_token(self) -> str:
-        return await self.service.get_oauth_client().mint_access_token(connection=self.row)
+    async def mint_access_token(self, scopes: tuple[str, ...] = (), cached: bool = True) -> str:
+        return await self.service.get_oauth_client().mint_access_token(
+            connection=self.row, scopes=scopes, cached=cached
+        )
 
     async def disconnect(self) -> None:
         await OauthClient(provider=self.service.name).disconnect(self.row)

backend/druks/services/oauth.py

@@ -156,23 +156,49 @@ async def begin_connect(
         query.update(extra_authorize_params or {})
         return f"{self.authorization_endpoint}?{urlencode(query)}"
 
-    async def mint_access_token(self, *, connection: OauthConnection) -> str:
+    async def mint_access_token(
+        self,
+        *,
+        connection: OauthConnection,
+        scopes: tuple[str, ...] = (),
+        cached: bool = True,
+    ) -> str:
         """The delivery-side token for one connection: the cached access
         token while it lives, else one refreshed through the stored refresh
         token. The provider may rotate the refresh token on use — two
         concurrent refreshes trip its reuse detection and can revoke the
-        whole connection — so Redis elects one refresher per connection (SET
-        NX; the TTL is a crash backstop a live refresh cannot outlive).
-        Losers poll for the winner's cache fill, for about one token-endpoint
-        round trip, then fail loudly. The engine reads the connection fresh
-        under the lock and commits a rotated token before the cache fills."""
+        whole connection — so Redis elects one refresher per (connection,
+        scope set) (SET NX; the TTL is a crash backstop a live refresh cannot
+        outlive). Losers poll for the winner's cache fill, for about one
+        token-endpoint round trip, then fail loudly. The engine reads the
+        connection fresh under the lock and commits a rotated token before
+        the cache fills.
+
+        ``scopes`` asks the provider for a token narrower than the grant
+        (RFC 6749 §6) — a server-side ceiling for a token handed to
+        untrusted compute; it must be a subset of the connection's granted
+        scopes. ``cached=False`` skips the cache read for a full-lifetime
+        token, still electing one refresher and filling the cache for later
+        callers."""
+        requested = tuple(sorted(scopes))
+        if requested and not set(requested) <= set(connection.scopes):
+            missing = ", ".join(sorted(set(requested) - set(connection.scopes)))
+            raise OauthRefreshError(
+                self.provider, f"the connection does not grant scope(s) {missing}"
+            )
         redis = get_client()
-        token_key = f"{self.provider}:access_token:{connection.id}"
-        lock_key = f"{self.provider}:refresh_lock:{connection.id}"
+        # A down-scoped token must never serve a full-scope caller, or the
+        # reverse — the cache and the refresher election key on the scope set.
+        suffix = ""
+        if requested:
+            suffix = ":" + hashlib.sha256(" ".join(requested).encode()).hexdigest()[:16]
+        token_key = f"{self.provider}:access_token:{connection.id}{suffix}"
+        lock_key = f"{self.provider}:refresh_lock:{connection.id}{suffix}"
         for _ in range(self.mint_wait_attempts):
-            cached = await redis.get(token_key)
             if cached:
-                return cast(bytes, cached).decode()
+                cached_token = await redis.get(token_key)
+                if cached_token:
+                    return cast(bytes, cached_token).decode()
             if await redis.set(lock_key, "1", nx=True, ex=OAUTH_REFRESH_LOCK_TTL_SECONDS):
                 break
             await asyncio.sleep(self.mint_wait_interval_seconds)
@@ -186,6 +212,8 @@ async def mint_access_token(self, *, connection: OauthConnection) -> str:
                 "refresh_token": connection._load_refresh_token(),
                 **self.extra_token_params,
             }
+            if requested:
+                data["scope"] = " ".join(requested)
             async with _http() as http:
                 try:
                     response = await _post_token(
@@ -215,6 +243,14 @@ async def mint_access_token(self, *, connection: OauthConnection) -> str:
                 )
             if tokens.get("refresh_token"):
                 connection._save_refresh_token(tokens["refresh_token"])
+            if requested and tokens.get("scope") and set(tokens["scope"].split()) != set(requested):
+                # A provider that ignores the narrowing hands back a token the
+                # sandbox must never hold — fail rather than cache it.
+                raise OauthRefreshError(
+                    self.provider,
+                    f"asked for scope(s) {' '.join(requested)}; "
+                    f"the token came back with {tokens['scope']!r}",
+                )
             try:
                 ttl = int(tokens.get("expires_in", 3600)) - OAUTH_TOKEN_TTL_SKEW_SECONDS
             except (TypeError, ValueError) as error:
@@ -228,7 +264,10 @@ async def mint_access_token(self, *, connection: OauthConnection) -> str:
             await redis.delete(lock_key)
 
     async def evict_access_token(self, connection_id: str) -> None:
-        await get_client().delete(f"{self.provider}:access_token:{connection_id}")
+        # Down-scoped variants ride the same prefix; one sweep drops them all.
+        redis = get_client()
+        async for key in redis.scan_iter(match=f"{self.provider}:access_token:{connection_id}*"):
+            await redis.delete(key)
 
     async def disconnect(self, connection: OauthConnection) -> None:
         """Delete the connection and evict its cached access token."""

backend/tests/test_oauth_client.py

@@ -1,4 +1,5 @@
 import asyncio
+import hashlib
 from urllib.parse import parse_qsl, urlparse
 
 import httpx
@@ -52,19 +53,23 @@ def _client(**overrides) -> OauthClient:
     return OauthClient(**kwargs)
 
 
-def _connection(refresh_token: str = "rt-old") -> OauthConnection:
+def _connection(refresh_token: str = "rt-old", scopes: list[str] | None = None) -> OauthConnection:
     return OauthConnection.create(
         provider=_PROVIDER,
         account_id=SYSTEM_ACCOUNT_ID,
         refresh_token=refresh_token,
-        scopes=[],
+        scopes=scopes or [],
     )
 
 
 def _token_key(connection: OauthConnection) -> str:
     return f"{_PROVIDER}:access_token:{connection.id}"
 
 
+def _scoped_suffix(scopes: tuple[str, ...]) -> str:
+    return ":" + hashlib.sha256(" ".join(sorted(scopes)).encode()).hexdigest()[:16]
+
+
 def _lock_key(connection: OauthConnection) -> str:
     return f"{_PROVIDER}:refresh_lock:{connection.id}"
 
@@ -214,3 +219,102 @@ async def test_complete_connect_requires_a_refresh_token(token_endpoint):
 
     with pytest.raises(OauthExchangeError, match="no refresh token"):
         await complete_connect(state=state, code="code-1")
+
+
+async def test_downscoped_mint_asks_and_caches_apart_from_the_full_grant(token_endpoint):
+    connection = _connection(scopes=["posts.write", "profile.read"])
+    token_endpoint.response = {
+        "access_token": "at-narrow",
+        "refresh_token": "rt-1",
+        "expires_in": 3600,
+        "scope": "profile.read",
+    }
+
+    narrow = await _client().mint_access_token(connection=connection, scopes=("profile.read",))
+
+    assert narrow == "at-narrow"
+    assert token_endpoint.requests[0]["scope"] == "profile.read"
+    redis = get_client()
+    scoped_key = _token_key(connection) + _scoped_suffix(("profile.read",))
+    assert await redis.get(scoped_key) == b"at-narrow"
+    assert not await redis.get(_token_key(connection))
+
+    token_endpoint.response = {
+        "access_token": "at-full",
+        "refresh_token": "rt-1",
+        "expires_in": 3600,
+    }
+    full = await _client().mint_access_token(connection=connection)
+
+    assert full == "at-full"
+    assert "scope" not in token_endpoint.requests[1]
+    assert await redis.get(_token_key(connection)) == b"at-full"
+    assert await redis.get(scoped_key) == b"at-narrow"
+
+    # The scoped cache serves the scoped ask without another refresh.
+    assert (
+        await _client().mint_access_token(connection=connection, scopes=("profile.read",))
+        == "at-narrow"
+    )
+    assert len(token_endpoint.requests) == 2
+
+
+async def test_downscoped_mint_rejects_scopes_outside_the_grant(token_endpoint):
+    connection = _connection(scopes=["profile.read"])
+
+    with pytest.raises(OauthRefreshError, match="does not grant scope"):
+        await _client().mint_access_token(connection=connection, scopes=("posts.write",))
+
+    assert not token_endpoint.requests
+
+
+async def test_downscoped_mint_rejects_a_provider_that_ignores_the_ask(token_endpoint):
+    connection = _connection(scopes=["posts.write", "profile.read"])
+    token_endpoint.response = {
+        "access_token": "at-broad",
+        "refresh_token": "rt-1",
+        "expires_in": 3600,
+        "scope": "posts.write profile.read",
+    }
+
+    with pytest.raises(OauthRefreshError, match="came back with"):
+        await _client().mint_access_token(connection=connection, scopes=("profile.read",))
+
+    scoped_key = _token_key(connection) + _scoped_suffix(("profile.read",))
+    assert not await get_client().get(scoped_key)
+
+
+async def test_uncached_mint_refreshes_past_a_live_cache_and_refills_it(token_endpoint):
+    connection = _connection()
+    redis = get_client()
+    await redis.set(_token_key(connection), "at-tail")
+
+    token = await _client().mint_access_token(connection=connection, cached=False)
+
+    assert token == "at-1"
+    assert len(token_endpoint.requests) == 1
+    assert await redis.get(_token_key(connection)) == b"at-1"
+    assert not await redis.get(_lock_key(connection))
+
+
+async def test_refresher_election_is_per_scope_set(token_endpoint):
+    connection = _connection(scopes=["profile.read"])
+    redis = get_client()
+    await redis.set(_lock_key(connection) + _scoped_suffix(("profile.read",)), "1")
+
+    # The scoped variant's lock never blocks the full-grant mint.
+    assert await _client().mint_access_token(connection=connection) == "at-1"
+    assert len(token_endpoint.requests) == 1
+
+
+async def test_disconnect_evicts_the_scope_variant_keys(token_endpoint):
+    connection = _connection(scopes=["profile.read"])
+    redis = get_client()
+    scoped_key = _token_key(connection) + _scoped_suffix(("profile.read",))
+    await redis.set(_token_key(connection), "at-full")
+    await redis.set(scoped_key, "at-narrow")
+
+    await _client().disconnect(connection)
+
+    assert not await redis.get(_token_key(connection))
+    assert not await redis.get(scoped_key)

docs/writing-an-extension.md

@@ -791,10 +791,15 @@ hold, across services. Replacing a service's client credentials deletes its
 connections: a new client can never refresh the old client's tokens.
 
 `mint_access_token` serves a Redis-cached access token and lets only one
-refresher run per connection. This is necessary: two refreshes at the same
-time can make the provider revoke the whole connection. It raises
-`OauthRefreshError` when the refresh fails. Then ask the user to
-reconnect.
+refresher run per connection and scope set. This is necessary: two
+refreshes at the same time can make the provider revoke the whole
+connection. It raises `OauthRefreshError` when the refresh fails. Then ask
+the user to reconnect.
+
+`mint_access_token(scopes=("profile.read",))` asks the provider for a token
+narrower than the grant — pass it when the token goes to untrusted compute,
+with a subset of the connection's scopes. `cached=False` refreshes past the
+cache for a full-lifetime token.
 
 ## Extension settings and checks