feat(oauth): First-party CLI client and per-client scope policy

api/app/settings/common.py

@@ -954,10 +954,16 @@
     "ACCESS_TOKEN_EXPIRE_SECONDS": 60 * 15,  # 15 minutes
     "REFRESH_TOKEN_EXPIRE_SECONDS": 60 * 60 * 24 * 30,  # 30 days
     "ROTATE_REFRESH_TOKEN": True,
+    "REFRESH_TOKEN_REUSE_PROTECTION": True,
+    "REFRESH_TOKEN_GRACE_PERIOD_SECONDS": 60 * 2,
     "PKCE_REQUIRED": True,
     "ALLOWED_CODE_CHALLENGE_METHODS": ["S256"],
-    "SCOPES": {"mcp": "MCP access"},
+    "SCOPES": {
+        "mcp": "MCP access",
+        "admin-api": "Admin API access",
+    },
     "DEFAULT_SCOPES": ["mcp"],
+    "SCOPES_BACKEND_CLASS": "oauth2_metadata.scopes.FlagsmithScopes",
     "ALLOWED_GRANT_TYPES": [
         "authorization_code",
         "refresh_token",

api/oauth2_metadata/constants.py

@@ -0,0 +1,8 @@
+FLAGSMITH_CLI_CLIENT_ID = "flagsmith-cli"
+
+SCOPE_MCP = "mcp"
+SCOPE_ADMIN_API = "admin-api"
+
+FIRST_PARTY_CLIENT_IDS = frozenset({FLAGSMITH_CLI_CLIENT_ID})
+FIRST_PARTY_SCOPES = frozenset({SCOPE_ADMIN_API})
+THIRD_PARTY_SCOPES = frozenset({SCOPE_MCP})

api/oauth2_metadata/migrations/0001_create_flagsmith_cli_application.py

@@ -0,0 +1,37 @@
+from django.apps.registry import Apps
+from django.db import migrations
+from django.db.backends.base.schema import BaseDatabaseSchemaEditor
+
+
+
+def create_flagsmith_cli_application(
+    apps: Apps,
+    schema_editor: BaseDatabaseSchemaEditor,
+) -> None:
+    Application = apps.get_model("oauth2_provider", "Application")
+    Application.objects.get_or_create(
+        client_id="flagsmith-cli",
+        defaults={
+            "name": "Flagsmith CLI",
+            "client_type": "public",
+            "authorization_grant_type": "authorization-code",
+            "client_secret": "",
+            "redirect_uris": "http://127.0.0.1/callback http://[::1]/callback",
+            "skip_authorization": True,
+        },
+    )
+
+
+class Migration(migrations.Migration):
+    initial = True
+
+    dependencies = [
+        ("oauth2_provider", "0012_add_token_checksum"),
+    ]
+
+    operations = [
+        migrations.RunPython(
+            create_flagsmith_cli_application,
+            migrations.RunPython.noop,
+        ),
+    ]

api/oauth2_metadata/scopes.py

@@ -0,0 +1,51 @@
+from typing import Any
+
+from oauth2_provider.models import Application
+from oauth2_provider.scopes import SettingsScopes
+
+from oauth2_metadata.constants import (
+    FIRST_PARTY_CLIENT_IDS,
+    FIRST_PARTY_SCOPES,
+    THIRD_PARTY_SCOPES,
+)
+
+
+class FlagsmithScopes(SettingsScopes):  # type: ignore[misc]
+    """Per-client scope issuance policy."""
+
+    def get_allowed_scopes(
+        self,
+        application: Application | None = None,
+    ) -> frozenset[str]:
+        if application is not None:
+            if application.client_id in FIRST_PARTY_CLIENT_IDS:
+                return FIRST_PARTY_SCOPES
+        return THIRD_PARTY_SCOPES
+
+    def get_available_scopes(
+        self,
+        application: Application | None = None,
+        request: Any = None,
+        *args: Any,
+        **kwargs: Any,
+    ) -> list[str]:
+        """What may a client request."""
+        allowed_scopes = self.get_allowed_scopes(application)
+        scopes: list[str] = super().get_available_scopes(
+            application, request, *args, **kwargs
+        )
+        return [scope for scope in scopes if scope in allowed_scopes]
+
+    def get_default_scopes(
+        self,
+        application: Application | None = None,
+        request: Any = None,
+        *args: Any,
+        **kwargs: Any,
+    ) -> list[str]:
+        """What does a client get when it doesn't request for any particular scopes."""
+        allowed_scopes = self.get_allowed_scopes(application)
+        scopes: list[str] = super().get_default_scopes(
+            application, request, *args, **kwargs
+        )
+        return [scope for scope in scopes if scope in allowed_scopes]

api/tests/unit/oauth2_metadata/test_authorize_view.py

@@ -5,12 +5,11 @@
 
 import pytest
 from django.contrib.auth.models import AbstractUser
-from django.urls import reverse
 from oauth2_provider.models import Application
 from rest_framework import status
 from rest_framework.test import APIClient
 
-AUTHORIZE_URL = "oauth-authorize"
+from oauth2_metadata.constants import FLAGSMITH_CLI_CLIENT_ID
 
 
 def _pkce_pair() -> tuple[str, str]:
@@ -63,7 +62,7 @@ def test_get__valid_params__returns_application_info(
 ) -> None:
     # Given
     _verifier, challenge = pkce_pair
-    url = reverse(AUTHORIZE_URL)
+    url = "/api/v1/oauth/authorize/"
 
     # When
     response = auth_client.get(
@@ -95,7 +94,7 @@ def test_get__verified_application__returns_is_verified_true(
 ) -> None:
     # Given
     _verifier, challenge = pkce_pair
-    url = reverse(AUTHORIZE_URL)
+    url = "/api/v1/oauth/authorize/"
 
     # When
     response = auth_client.get(
@@ -122,11 +121,10 @@ def test_get__invalid_client_id__returns_400(
 ) -> None:
     # Given
     _verifier, challenge = pkce_pair
-    url = reverse(AUTHORIZE_URL)
 
     # When
     response = auth_client.get(
-        url,
+        "/api/v1/oauth/authorize/",
         {
             "client_id": "nonexistent-client-id",
             "response_type": "code",
@@ -149,7 +147,7 @@ def test_post__invalid_client_id__returns_400(
 ) -> None:
     # Given
     _verifier, challenge = pkce_pair
-    url = reverse(AUTHORIZE_URL)
+    url = "/api/v1/oauth/authorize/"
 
     # When
     response = auth_client.post(
@@ -178,11 +176,10 @@ def test_authorize__unauthenticated__returns_401(
 ) -> None:
     # Given
     client = APIClient()
-    url = reverse(AUTHORIZE_URL)
 
     # When
     response = getattr(client, method)(
-        url,
+        "/api/v1/oauth/authorize/",
         {"client_id": "some-id", "response_type": "code"},
     )
 
@@ -207,11 +204,10 @@ def test_post__consent_decision__returns_redirect(
 ) -> None:
     # Given
     _verifier, challenge = pkce_pair
-    url = reverse(AUTHORIZE_URL)
 
     # When
     response = auth_client.post(
-        url,
+        "/api/v1/oauth/authorize/",
         {
             "allow": allow,
             "client_id": oauth_application.client_id,
@@ -240,11 +236,10 @@ def test_post__pkce_params_preserved__code_exchangeable(
 ) -> None:
     # Given
     code_verifier, code_challenge = _pkce_pair()
-    authorize_url = reverse(AUTHORIZE_URL)
 
     # When
     response = auth_client.post(
-        authorize_url,
+        "/api/v1/oauth/authorize/",
         {
             "allow": True,
             "client_id": oauth_application.client_id,
@@ -263,10 +258,9 @@ def test_post__pkce_params_preserved__code_exchangeable(
     query_params = parse_qs(parsed.query)
     code = query_params["code"][0]
 
-    token_url = reverse("oauth2_provider:token")
     token_client = APIClient()
     token_response = token_client.post(
-        token_url,
+        "/o/token/",
         {
             "grant_type": "authorization_code",
             "code": code,
@@ -282,3 +276,59 @@ def test_post__pkce_params_preserved__code_exchangeable(
     assert "access_token" in token_data
     assert "refresh_token" in token_data
     assert token_data["token_type"] == "Bearer"
+
+
+def test_get__third_party_application_requests_admin_api__returns_invalid_scope(
+    auth_client: APIClient,
+    oauth_application: Application,
+    pkce_pair: tuple[str, str],
+) -> None:
+    # Given
+    _verifier, challenge = pkce_pair
+
+    # When
+    response = auth_client.get(
+        "/api/v1/oauth/authorize/",
+        {
+            "client_id": oauth_application.client_id,
+            "response_type": "code",
+            "redirect_uri": "https://example.com/callback",
+            "scope": "admin-api",
+            "code_challenge": challenge,
+            "code_challenge_method": "S256",
+        },
+    )
+
+    # Then
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert response.json()["error"] == "invalid_scope"
+
+
+def test_get__flagsmith_cli_requests_admin_api__returns_application_info(
+    auth_client: APIClient,
+    pkce_pair: tuple[str, str],
+    db: None,
+) -> None:
+    # Given
+    application = Application.objects.get(client_id=FLAGSMITH_CLI_CLIENT_ID)
+    _verifier, challenge = pkce_pair
+
+    # When
+    response = auth_client.get(
+        "/api/v1/oauth/authorize/",
+        {
+            "client_id": application.client_id,
+            "response_type": "code",
+            "redirect_uri": "http://127.0.0.1:53682/callback",
+            "scope": "admin-api",
+            "code_challenge": challenge,
+            "code_challenge_method": "S256",
+        },
+    )
+
+    # Then
+    assert response.status_code == status.HTTP_200_OK
+    data = response.json()
+    assert data["application"]["client_id"] == FLAGSMITH_CLI_CLIENT_ID
+    assert "admin-api" in data["scopes"]
+    assert data["is_verified"] is True

api/tests/unit/oauth2_metadata/test_migrations.py

@@ -0,0 +1,63 @@
+from django.contrib.auth.hashers import check_password
+from django_test_migrations.migrator import Migrator
+
+
+def test_0001__fresh_install__creates_flagsmith_cli_application(
+    migrator: Migrator,
+) -> None:
+    # Given
+    old_state = migrator.apply_initial_migration(("oauth2_metadata", None))
+    OldApplication = old_state.apps.get_model("oauth2_provider", "Application")
+    OldApplication.objects.filter(client_id="flagsmith-cli").delete()
+
+    # When
+    new_state = migrator.apply_tested_migration(
+        ("oauth2_metadata", "0001_create_flagsmith_cli_application")
+    )
+
+    # Then
+    Application = new_state.apps.get_model("oauth2_provider", "Application")
+    application = Application.objects.get(client_id="flagsmith-cli")
+    assert application.name == "Flagsmith CLI"
+    assert application.client_type == "public"
+    assert application.authorization_grant_type == "authorization-code"
+    # The client_secret field hashes on save; the stored value must be the
+    # hash of an empty secret (public client, token_endpoint_auth "none").
+    assert check_password("", application.client_secret)
+    assert (
+        application.redirect_uris == "http://127.0.0.1/callback http://[::1]/callback"
+    )
+    assert application.skip_authorization is True
+
+
+def test_0001__application_already_exists__does_not_overwrite(
+    migrator: Migrator,
+) -> None:
+    # Given
+    old_state = migrator.apply_initial_migration(("oauth2_metadata", None))
+    OldApplication = old_state.apps.get_model("oauth2_provider", "Application")
+    OldApplication.objects.filter(client_id="flagsmith-cli").delete()
+    OldApplication.objects.create(
+        client_id="flagsmith-cli",
+        name="Pre-existing Application",
+        client_type="confidential",
+        authorization_grant_type="client-credentials",
+        client_secret="pre-existing-secret",
+        redirect_uris="https://example.com/callback",
+        skip_authorization=False,
+    )
+
+    # When
+    new_state = migrator.apply_tested_migration(
+        ("oauth2_metadata", "0001_create_flagsmith_cli_application")
+    )
+
+    # Then
+    Application = new_state.apps.get_model("oauth2_provider", "Application")
+    application = Application.objects.get(client_id="flagsmith-cli")
+    assert application.name == "Pre-existing Application"
+    assert application.client_type == "confidential"
+    assert application.authorization_grant_type == "client-credentials"
+    assert check_password("pre-existing-secret", application.client_secret)
+    assert application.redirect_uris == "https://example.com/callback"
+    assert application.skip_authorization is False