Merge pull request #260 from spoo-me/feat/device-auth-pkce-scopes

feat(auth): PKCE and grant-scoped tokens for connected apps

Author: Zingzy

config/apps.yaml

@@ -82,6 +82,16 @@ apps:
     verified: true
     status: coming_soon
     type: device_auth
+    redirect_uris:
+      - https://raycast.com/redirect?packageName=Extension
+      - https://raycast.com/redirect/extension
+    links:
+      store: https://www.raycast.com/zingzy/spoo
+    scopes:
+      - shorten:create
+      - urls:read
+      - urls:manage
+      - stats:read
 
   spoo-vscode:
     name: VS Code Extension
@@ -126,8 +136,11 @@ apps:
       - http://127.0.0.1:53682/callback
     links:
       github: https://github.com/spoo-me/spoo-cli
-    permissions:
-      - Access your spoo.me account
-      - Create and manage your short URLs
-      - View your analytics
-      - Manage your API keys
+    scopes:
+      - shorten:create
+      - urls:read
+      - urls:manage
+      - stats:read
+      - domains:read
+      - domains:manage
+      - keys:manage

dependencies/__init__.py

@@ -8,6 +8,7 @@
 from dependencies.auth import (
     DOMAIN_MANAGE_SCOPES,
     DOMAIN_READ_SCOPES,
+    KEYS_MANAGE_SCOPES,
     REPORTS_SCOPES,
     SHORTEN_SCOPES,
     STATS_SCOPES,
@@ -17,17 +18,21 @@
     CurrentUser,
     JwtUser,
     JwtVerifiedUser,
+    KeysAccessUser,
     OptionalUser,
     VerifiedUser,
     check_api_key_scope,
+    check_credential_scopes,
     get_current_user,
     optional_scopes,
     optional_scopes_verified,
     require_auth,
     require_jwt,
     require_jwt_verified,
+    require_keys_access,
     require_scopes,
     require_scopes_verified,
+    require_session_or_scopes,
     require_verified_email,
 )
 from dependencies.infra import (
@@ -94,6 +99,7 @@
 __all__ = [
     "DOMAIN_MANAGE_SCOPES",
     "DOMAIN_READ_SCOPES",
+    "KEYS_MANAGE_SCOPES",
     "REPORTS_SCOPES",
     "SHORTEN_SCOPES",
     "STATS_SCOPES",
@@ -120,6 +126,7 @@
     "JwtConfig",
     "JwtUser",
     "JwtVerifiedUser",
+    "KeysAccessUser",
     "OAuthProviders",
     "OAuthSvc",
     "OnboardingCacheDep",
@@ -135,6 +142,7 @@
     "VerificationSvc",
     "VerifiedUser",
     "check_api_key_scope",
+    "check_credential_scopes",
     # services (getters)
     "fetch_user_profile",
     "get_api_key_service",
@@ -172,7 +180,9 @@
     "require_auth",
     "require_jwt",
     "require_jwt_verified",
+    "require_keys_access",
     "require_scopes",
     "require_scopes_verified",
+    "require_session_or_scopes",
     "require_verified_email",
 ]

dependencies/auth.py

@@ -34,13 +34,19 @@ class CurrentUser:
 
     ``api_key_doc`` is set when the request was authenticated via API key
     (``Authorization: Bearer spoo_<raw>``).  It is ``None`` for JWT auth.
-    Scope checks inspect ``api_key_doc.scopes`` when present.
+    Scope checks inspect ``api_key_doc.scopes`` when present, otherwise
+    ``scopes`` (the JWT ``scp`` claim) when not None.
     """
 
     user_id: ObjectId
     email_verified: bool
     api_key_doc: ApiKeyDoc | None = field(default=None)
     amr: str = "pwd"
+    # Scope slugs from the JWT "scp" claim (device-auth app tokens).
+    # None = unrestricted interactive session; [] would mean "no scopes".
+    scopes: list[str] | None = field(default=None)
+    # Connected-app id from the JWT "app_id" claim (device auth flow).
+    app_id: str | None = field(default=None)
     # Lowercased user email — consumed by FeatureFlagService's ALLOWLIST
     # rollout (allowlist_emails). Populated from the "email" claim on the
     # JWT path and from the owning UserDoc on the API-key path. None for
@@ -149,6 +155,17 @@ async def get_current_user(
             if isinstance(raw_email, str) and raw_email.strip()
             else None
         )
+        # Device-auth app tokens carry scp + app_id; session tokens carry
+        # neither. A malformed scp claim fails closed (empty scope list)
+        # rather than falling back to unrestricted.
+        raw_scopes = claims.get("scp")
+        scopes: list[str] | None = None
+        if raw_scopes is not None:
+            scopes = (
+                [s for s in raw_scopes if isinstance(s, str)]
+                if isinstance(raw_scopes, list)
+                else []
+            )
         structlog.contextvars.bind_contextvars(user_id=str(user_id), auth_method="jwt")
         return CurrentUser(
             user_id=user_id,
@@ -158,6 +175,8 @@ async def get_current_user(
             # Not issued yet — the paid-plans launch adds the claim; TIER
             # flag rollouts become a pure data change at that point.
             tier=claims.get("plan"),
+            scopes=scopes,
+            app_id=claims.get("app_id"),
         )
     except Exception:
         return None
@@ -184,14 +203,18 @@ async def require_verified_email(
 async def require_jwt(
     user: CurrentUser = Depends(require_auth),
 ) -> CurrentUser:
-    """Raise 403 if the request was authenticated via API key.
+    """Raise 403 unless the request comes from an interactive session.
 
-    Use on endpoints where API key auth must be explicitly prohibited —
-    e.g. key management routes (an API key must not be able to create,
-    list, or delete other API keys).
+    Rejects API keys AND scoped device-app tokens (``scp`` claim). Use on
+    account-security surfaces — profile, app management, device revoke —
+    where a delegated credential must never act.
     """
     if user.api_key_doc is not None:
         raise ForbiddenError("API keys cannot be used to manage API keys")
+    # Delegation is marked by app_id, not scp: a legacy grant mints an
+    # app token with app_id but no scp, and it must be barred here too.
+    if user.app_id is not None:
+        raise ForbiddenError("This operation requires an interactive session")
     return user
 
 
@@ -204,19 +227,35 @@ async def require_jwt_verified(
     return user
 
 
-def check_api_key_scope(user: CurrentUser | None, required_scopes: set[str]) -> None:
-    """Raise ForbiddenError if an API-key-authenticated user lacks a required scope.
+def _granted_scopes(user: CurrentUser) -> set[str] | None:
+    """The scope set a credential holds, or None for unrestricted sessions."""
+    if user.api_key_doc is not None:
+        return set(user.api_key_doc.scopes)
+    if user.scopes is not None:
+        return set(user.scopes)
+    return None
+
 
-    JWT-authenticated and anonymous requests are not scope-restricted.
+def check_credential_scopes(
+    user: CurrentUser | None, required_scopes: set[str]
+) -> None:
+    """Raise ForbiddenError if a scoped credential lacks a required scope.
+
+    Fires for API keys (key scopes) and device-app tokens (``scp`` claim);
+    interactive sessions and anonymous requests are not scope-restricted.
+    A single match against ``required_scopes`` suffices (OR semantics).
     """
-    if (
-        user is not None
-        and user.api_key_doc is not None
-        and not set(user.api_key_doc.scopes) & required_scopes
-    ):
+    if user is None:
+        return
+    granted = _granted_scopes(user)
+    if granted is not None and not granted & required_scopes:
         raise ForbiddenError("Insufficient scope for this operation")
 
 
+# Back-compat alias — the check now covers app tokens too.
+check_api_key_scope = check_credential_scopes
+
+
 # ── Named scope sets ─────────────────────────────────────────────────────────
 
 STATS_SCOPES: set[str] = {
@@ -238,6 +277,9 @@ def check_api_key_scope(user: CurrentUser | None, required_scopes: set[str]) ->
     ApiKeyScope.DOMAINS_READ,
     ApiKeyScope.ADMIN_ALL,
 }
+# Deliberately NOT satisfied by admin:all: API keys can hold admin:all but
+# must never manage keys, and app tokens declare keys:manage explicitly.
+KEYS_MANAGE_SCOPES: set[str] = {ApiKeyScope.KEYS_MANAGE}
 
 
 # ── Parameterised scope dependency factories ─────────────────────────────────
@@ -255,7 +297,30 @@ def require_scopes(scopes: set[str]):
     """
 
     async def _dep(user: CurrentUser = Depends(require_auth)) -> CurrentUser:
-        check_api_key_scope(user, scopes)
+        check_credential_scopes(user, scopes)
+        return user
+
+    return _dep
+
+
+def require_session_or_scopes(scopes: set[str]):
+    """Dependency factory: interactive session OR a credential holding *scopes*.
+
+    Passes for an interactive session (no API key, no ``app_id``), or for a
+    delegated credential whose scope set intersects *scopes*. A delegated
+    credential is identified by ``api_key_doc``/``app_id``, not by ``scp``,
+    so a legacy app token (``app_id`` but no ``scp``) is treated as delegated
+    and denied — it predates ``keys:manage`` and never held it. API keys can
+    never be created with ``keys:manage`` (see ALLOWED_SCOPES), so the
+    anti-self-propagation guard holds.
+    """
+
+    async def _dep(user: CurrentUser = Depends(require_auth)) -> CurrentUser:
+        if user.api_key_doc is None and user.app_id is None:
+            return user  # interactive session — unrestricted
+        granted = _granted_scopes(user)
+        if not granted or not granted & scopes:
+            raise ForbiddenError("Insufficient scope for this operation")
         return user
 
     return _dep
@@ -275,7 +340,7 @@ def optional_scopes(scopes: set[str]):
     async def _dep(
         user: CurrentUser | None = Depends(get_current_user),
     ) -> CurrentUser | None:
-        check_api_key_scope(user, scopes)
+        check_credential_scopes(user, scopes)
         return user
 
     return _dep
@@ -321,10 +386,19 @@ async def _dep(
     return _dep
 
 
+# ── Named dependency instances ────────────────────────────────────────────────
+
+# Key listing/deletion: interactive session OR a credential holding
+# keys:manage. Key *creation* is not here — minting a new credential is a
+# first-party act, so create uses JwtVerifiedUser (interactive session only).
+require_keys_access = require_session_or_scopes(KEYS_MANAGE_SCOPES)
+
+
 # ── Annotated type aliases — community-standard Depends shortcuts ─────────────
 
 AuthUser = Annotated[CurrentUser, Depends(require_auth)]
 VerifiedUser = Annotated[CurrentUser, Depends(require_verified_email)]
 OptionalUser = Annotated[CurrentUser | None, Depends(get_current_user)]
 JwtUser = Annotated[CurrentUser, Depends(require_jwt)]
 JwtVerifiedUser = Annotated[CurrentUser, Depends(require_jwt_verified)]
+KeysAccessUser = Annotated[CurrentUser, Depends(require_keys_access)]

dependencies/wiring.py

@@ -298,6 +298,7 @@ def wire_services(app: FastAPI, settings: AppSettings, redis_client) -> None:
         user_repo,
         token_repo,
         token_factory,
+        app_grant_repo,
         app_registry=getattr(app.state, "app_registry", None),
     )
     app.state.oauth_service = OAuthService(

infrastructure/crypto.py

@@ -1,11 +1,13 @@
 """
-Cryptographic helpers — password hashing and token hashing.
+Cryptographic helpers — password hashing, token hashing, and PKCE.
 
-Uses argon2 for passwords (via argon2-cffi) and SHA-256 for token hashing.
+Uses argon2 for passwords (via argon2-cffi) and SHA-256 for token hashing
+and PKCE code-challenge derivation (RFC 7636).
 """
 
 from __future__ import annotations
 
+import base64
 import hashlib
 
 from argon2 import PasswordHasher
@@ -49,3 +51,16 @@ def hash_token(token: str) -> str:
         64-character lowercase hex string.
     """
     return hashlib.sha256(token.encode("utf-8")).hexdigest()
+
+
+def pkce_s256_challenge(code_verifier: str) -> str:
+    """Derive the S256 PKCE code challenge for *code_verifier* (RFC 7636 §4.2).
+
+    ``code_challenge = BASE64URL-ENCODE(SHA256(ASCII(code_verifier)))``
+    with no ``=`` padding.
+
+    Returns:
+        43-character base64url string.
+    """
+    digest = hashlib.sha256(code_verifier.encode("ascii")).digest()
+    return base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii")

repositories/app_grant_repository.py

@@ -58,11 +58,26 @@ async def find_all_for_user(self, user_id: ObjectId) -> list[AppGrantDoc]:
             )
             raise
 
-    async def create_or_reactivate(self, user_id: ObjectId, app_id: str) -> AppGrantDoc:
+    async def find_by_id_for_user(
+        self, user_id: ObjectId, grant_id: ObjectId
+    ) -> AppGrantDoc | None:
+        """Find a grant by document id, scoped to its owner."""
+        return await self._find_one({"_id": grant_id, "user_id": user_id})
+
+    async def create_or_reactivate(
+        self,
+        user_id: ObjectId,
+        app_id: str,
+        *,
+        scopes: list[str] | None,
+    ) -> AppGrantDoc:
         """Create a new grant or reactivate a revoked one.
 
         Uses upsert: if a document exists for (user_id, app_id), clears
         revoked_at and updates granted_at. Otherwise inserts a new document.
+
+        ``scopes`` snapshots the registry scopes at approval time; passing
+        None preserves a legacy (unrestricted) grant shape.
         """
         now = datetime.now(timezone.utc)
         try:
@@ -72,6 +87,7 @@ async def create_or_reactivate(self, user_id: ObjectId, app_id: str) -> AppGrant
                     "$set": {
                         "granted_at": now,
                         "revoked_at": None,
+                        "scopes": scopes,
                     },
                     "$setOnInsert": {
                         "user_id": user_id,

routes/api_v1/apps.py

@@ -14,6 +14,7 @@
 from middleware.openapi import AUTH_RESPONSES
 from middleware.rate_limiter import Limits, limiter
 from schemas.dto.responses.app_grant import AppGrantResponse, AppGrantsListResponse
+from services.auth.device import effective_scopes_for
 
 router = APIRouter(tags=["Apps"])
 
@@ -48,6 +49,11 @@ async def list_app_grants(
     grants.sort(key=lambda g: g.granted_at, reverse=True)
     return AppGrantsListResponse(
         items=[
-            AppGrantResponse.from_grant(g, app_registry.get(g.app_id)) for g in grants
+            AppGrantResponse.from_grant(
+                g,
+                app_registry.get(g.app_id),
+                effective_scopes_for(g, app_registry.get(g.app_id)),
+            )
+            for g in grants
         ]
     )