feat(oauth): Let users narrow granted scopes on consent

src/sentry/api/endpoints/oauth_authorize.py

@@ -1,5 +1,5 @@
 from collections.abc import Mapping, Sequence
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
 from typing import Any, Literal, TypedDict
 
 from django.contrib.auth.models import AnonymousUser
@@ -27,7 +27,12 @@
     prune_authorization_transactions,
     save_authorization_transaction,
 )
-from sentry.api.helpers.oauth_consent import get_organization_options, has_existing_consent
+from sentry.api.helpers.oauth_consent import (
+    InvalidScopeSelection,
+    get_organization_options,
+    has_existing_consent,
+    select_granted_scopes,
+)
 from sentry.api.permissions import DisallowImpersonatedTokenCreation, SentryIsAuthenticated
 from sentry.api.serializers import Serializer, serialize
 from sentry.api.serializers.models.oauth_consent import (
@@ -199,8 +204,21 @@ def post(self, request: AuthenticatedHttpRequest) -> Response:
                 client_id=authorization.application.client_id,
             )
 
+        try:
+            granted_scopes = select_granted_scopes(authorization.scopes, decision["scopes"])
+        except InvalidScopeSelection as exc:
+            raise authorization_error(
+                response_type=authorization.response_type,
+                redirect_uri=authorization.redirect_uri,
+                name="invalid_scope",
+                state=authorization.state,
+                client_id=authorization.application.client_id,
+            ) from exc
+
         return self.redirect_response(
             approve_authorization(
-                authorization, request.user.id, decision["selected_organization_id"]
+                replace(authorization, scopes=granted_scopes),
+                request.user.id,
+                decision["selected_organization_id"],
             )
         )

src/sentry/api/endpoints/oauth_device.py

@@ -15,6 +15,7 @@
 from sentry.api.base import Endpoint, control_silo_endpoint
 from sentry.api.helpers.oauth_consent import (
     InvalidOrganizationSelection,
+    InvalidScopeSelection,
     get_organization_options,
 )
 from sentry.api.helpers.oauth_device import (
@@ -172,10 +173,16 @@ def post(self, request: AuthenticatedHttpRequest) -> Response:
         )
         try:
             status = complete_device_request(
-                device.id, request.user.id, decision["op"], decision["selected_organization_id"]
+                device.id,
+                request.user.id,
+                decision["op"],
+                decision["selected_organization_id"],
+                decision["scopes"],
             )
         except InvalidOrganizationSelection as exc:
             raise ValidationError({"selectedOrganizationId": exc.public_message}) from exc
+        except InvalidScopeSelection as exc:
+            raise ValidationError({"scopes": exc.public_message}) from exc
 
         request.session.pop(f"oa2:{device.user_code}", None)
         return Response(

src/sentry/api/helpers/oauth_consent.py

@@ -296,6 +296,30 @@ def resolve_organization(
     return organization_id
 
 
+class InvalidScopeSelection(Exception):
+    public_message = "Selected permissions must be a subset of the requested permissions."
+
+    def __init__(self) -> None:
+        super().__init__(self.public_message)
+
+
+def select_granted_scopes(requested: Sequence[str], selected: Sequence[str] | None) -> list[str]:
+    """Narrow the requested scopes to the ones the user chose to grant.
+
+    Omitting a selection grants every requested scope. A selection may only
+    remove scopes; granting a scope the application did not request is rejected.
+    """
+    if selected is None:
+        return list(requested)
+
+    selected_scopes = set(selected)
+    if not selected_scopes.issubset(requested):
+        raise InvalidScopeSelection()
+
+    # Preserve request order so issued grants read like the original request.
+    return [scope for scope in dict.fromkeys(requested) if scope in selected_scopes]
+
+
 def has_existing_consent(application: ApiApplication, user_id: int, scopes: Sequence[str]) -> bool:
     # Organization-scoped consent requires an explicit selection on each request.
     if application.requires_org_level_access:

src/sentry/api/helpers/oauth_device.py

@@ -1,4 +1,5 @@
 import logging
+from collections.abc import Sequence
 from enum import Enum
 from typing import Literal, TypedDict
 
@@ -9,6 +10,7 @@
     OAuthDecision,
     persist_authorization,
     resolve_organization,
+    select_granted_scopes,
 )
 from sentry.models.apiapplication import ApiApplicationStatus
 from sentry.models.apidevicecode import (
@@ -97,6 +99,7 @@ def complete_device_request(
     user_id: int,
     decision: OAuthDecision,
     selected_organization_id: str | None = None,
+    selected_scopes: Sequence[str] | None = None,
 ) -> Literal["approved", "denied"]:
     # Hold the pending request until both its decision and authorization commit.
     with transaction.atomic(router.db_for_write(ApiDeviceCode)):
@@ -111,17 +114,21 @@ def complete_device_request(
 
         validate_device_request(device)
         organization_id = None
+        scope_list = device.scope_list
         if decision == "approve":
             organization_id = resolve_organization(
                 device.application, user_id, selected_organization_id
             )
-            persist_authorization(device.application, user_id, device.get_scopes(), organization_id)
+            scope_list = select_granted_scopes(device.get_scopes(), selected_scopes)
+            persist_authorization(device.application, user_id, scope_list, organization_id)
 
         status: Literal["approved", "denied"] = "approved" if decision == "approve" else "denied"
         device.update(
             status=status,
             user_id=user_id if decision == "approve" else None,
             organization_id=organization_id,
+            # Token exchange issues the device token from this list, so store what was granted.
+            scope_list=scope_list,
         )
 
     if decision == "approve":

src/sentry/api/serializers/rest_framework/oauth_consent.py

@@ -1,5 +1,6 @@
 from typing import TypedDict, TypeVar
 
+from django.conf import settings
 from rest_framework import serializers
 
 from sentry.api.helpers.oauth_consent import OAuthDecision
@@ -9,6 +10,7 @@
 class ConsentDecision(TypedDict):
     op: OAuthDecision
     selected_organization_id: str | None
+    scopes: list[str] | None
 
 
 DecisionT = TypeVar("DecisionT", bound=ConsentDecision)
@@ -24,3 +26,11 @@ class ConsentDecisionSerializer(CamelSnakeSerializer[DecisionT]):
         default=None,
         help_text="The organization receiving access for an organization-scoped application.",
     )
+    scopes = serializers.ListField(
+        child=serializers.ChoiceField(choices=sorted(settings.SENTRY_SCOPES)),
+        required=False,
+        allow_null=True,
+        default=None,
+        max_length=len(settings.SENTRY_SCOPES),
+        help_text=("The subset of requested scopes to grant. Omit to grant every requested scope."),
+    )

tests/sentry/api/endpoints/test_oauth_authorize.py

@@ -390,3 +390,68 @@ def test_existing_consent_requires_decision_to_issue_grant(self) -> None:
 
         grant = ApiGrant.objects.get(application=self.application, user=self.user)
         assert parse_qs(urlsplit(response.data["redirectUrl"]).query)["code"] == [grant.code]
+
+    def test_selected_scopes_narrow_grant_and_token(self) -> None:
+        transaction = self.prepare(scope="project:read project:write org:read")
+        response = self.get_success_response(
+            transactionId=transaction, op="approve", scopes=["project:read", "org:read"]
+        )
+        code = parse_qs(urlsplit(response.data["redirectUrl"]).query)["code"][0]
+        grant = ApiGrant.objects.get(application=self.application, user=self.user)
+        assert grant.scope_list == ["project:read", "org:read"]
+        authorization = ApiAuthorization.objects.get(application=self.application, user=self.user)
+        assert set(authorization.scope_list) == {"project:read", "org:read"}
+
+        exchanged = self.client.post(
+            reverse("sentry-oauth-token"),
+            {
+                "grant_type": "authorization_code",
+                "client_id": self.application.client_id,
+                "client_secret": self.application.client_secret,
+                "code": code,
+                "redirect_uri": self.application.get_default_redirect_uri(),
+            },
+            format="multipart",
+        )
+        assert exchanged.status_code == 200
+        assert exchanged.json()["scope"] == "org:read project:read"
+
+    def test_selected_scopes_narrow_implicit_token(self) -> None:
+        transaction = self.prepare(response_type="token", scope="project:read project:write")
+        response = self.get_success_response(
+            transactionId=transaction, op="approve", scopes=["project:read"]
+        )
+        token = ApiToken.objects.get(application=self.application, user=self.user)
+        assert token.scope_list == ["project:read"]
+        fragment = parse_qs(urlsplit(response.data["redirectUrl"]).fragment)
+        assert fragment["scope"] == ["project:read"]
+
+    def test_narrowed_consent_prompts_for_full_request(self) -> None:
+        transaction = self.prepare(scope="project:read project:write")
+        self.get_success_response(transactionId=transaction, op="approve", scopes=["project:read"])
+        consent = self.get_success_response(method="get", **self.query)
+        assert consent.data["hasExistingConsent"] is True
+        consent = self.get_success_response(
+            method="get", **{**self.query, "scope": "project:read project:write"}
+        )
+        assert consent.data["hasExistingConsent"] is False
+
+    def test_unrequested_selected_scope(self) -> None:
+        transaction = self.prepare()
+        response = self.get_success_response(
+            transactionId=transaction, op="approve", scopes=["project:read", "project:write"]
+        )
+        assert parse_qs(urlsplit(response.data["redirectUrl"]).query) == {
+            "error": ["invalid_scope"],
+            "state": [self.query["state"]],
+        }
+        assert not ApiGrant.objects.filter(application=self.application).exists()
+        assert not ApiAuthorization.objects.filter(application=self.application).exists()
+
+    def test_unknown_selected_scope_preserves_transaction(self) -> None:
+        transaction = self.prepare()
+        self.get_error_response(
+            transactionId=transaction, op="approve", scopes=["not:a-scope"], status_code=400
+        )
+        self.get_success_response(transactionId=transaction, op="approve")
+        assert ApiGrant.objects.get(application=self.application).scope_list == ["project:read"]

tests/sentry/api/endpoints/test_oauth_device.py

@@ -275,3 +275,50 @@ def test_expiry_after_consent_preparation(self) -> None:
         self.get_error_response(userCode=self.device.user_code, op="approve", status_code=400)
         self.device.refresh_from_db()
         assert self.device.status == DeviceCodeStatus.PENDING
+
+    def test_selected_scopes_narrow_device_token(self) -> None:
+        self.device.update(scope_list=["project:read", "project:write"])
+        self.prepare()
+        self.get_success_response(
+            userCode=self.device.user_code, op="approve", scopes=["project:read"]
+        )
+        self.device.refresh_from_db()
+        assert self.device.scope_list == ["project:read"]
+        assert ApiAuthorization.objects.get(
+            application=self.application, user=self.user
+        ).scope_list == ["project:read"]
+
+        polled = self.client.post(
+            reverse("sentry-oauth-token"),
+            {
+                "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
+                "device_code": self.device.device_code,
+                "client_id": self.application.client_id,
+                "client_secret": self.application.client_secret,
+            },
+            format="multipart",
+        )
+        assert polled.status_code == 200
+        assert polled.json()["scope"] == "project:read"
+        token = ApiToken.objects.get(application=self.application, user=self.user)
+        assert token.scope_list == ["project:read"]
+
+    def test_unrequested_selected_scope_can_be_corrected(self) -> None:
+        self.prepare()
+        response = self.get_error_response(
+            userCode=self.device.user_code,
+            op="approve",
+            scopes=["project:read", "project:write"],
+            status_code=400,
+        )
+        assert "scopes" in response.data
+        self.device.refresh_from_db()
+        assert self.device.status == DeviceCodeStatus.PENDING
+        assert self.device.scope_list == ["project:read"]
+        assert not ApiAuthorization.objects.filter(application=self.application).exists()
+
+        self.get_success_response(
+            userCode=self.device.user_code, op="approve", scopes=["project:read"]
+        )
+        self.device.refresh_from_db()
+        assert self.device.status == DeviceCodeStatus.APPROVED

tests/sentry/api/helpers/test_oauth_consent.py

@@ -9,13 +9,15 @@
 
 from sentry.api.helpers.oauth_consent import (
     InvalidOrganizationSelection,
+    InvalidScopeSelection,
     OrganizationSelectionError,
     get_organization_options,
     get_permission_descriptions,
     get_permission_groups,
     has_existing_consent,
     persist_authorization,
     resolve_organization,
+    select_granted_scopes,
 )
 from sentry.hybridcloud.services.organization_mapping.model import RpcOrganizationMapping
 from sentry.models.apiauthorization import ApiAuthorization
@@ -242,3 +244,21 @@ def test_resource_access_descriptions() -> None:
     (ci,) = get_permission_groups(["org:ci"])
     assert ci["accessDescriptions"] == {}
     assert ci["description"]
+
+
+def test_select_granted_scopes_defaults_to_requested() -> None:
+    assert select_granted_scopes(["project:read", "org:read"], None) == ["project:read", "org:read"]
+
+
+def test_select_granted_scopes_narrows_in_request_order() -> None:
+    requested = ["org:read", "project:read", "project:write"]
+    assert select_granted_scopes(requested, ["project:read", "org:read"]) == [
+        "org:read",
+        "project:read",
+    ]
+    assert select_granted_scopes(requested, []) == []
+
+
+def test_select_granted_scopes_rejects_unrequested_scope() -> None:
+    with pytest.raises(InvalidScopeSelection):
+        select_granted_scopes(["project:read"], ["project:read", "project:write"])