feat(service-accounts): Add an endpoint to list service account tokens

src/sentry/api/urls.py

@@ -694,6 +694,9 @@
 from sentry.sentry_apps.api.endpoints.sentry_internal_app_tokens import (
     SentryInternalAppTokensEndpoint,
 )
+from sentry.service_accounts.endpoints.organization_service_account_tokens import (
+    OrganizationServiceAccountTokenIndexEndpoint,
+)
 from sentry.service_accounts.endpoints.organization_service_account_viewer_context import (
     OrganizationServiceAccountViewerContextEndpoint,
 )
@@ -1819,6 +1822,11 @@ def create_group_urls(
         OrganizationServiceAccountDetailsEndpoint.as_view(),
         name="sentry-api-0-organization-service-account-details",
     ),
+    re_path(
+        r"^(?P<organization_id_or_slug>[^/]+)/service-accounts/(?P<service_account_id>\d+)/tokens/$",
+        OrganizationServiceAccountTokenIndexEndpoint.as_view(),
+        name="sentry-api-0-organization-service-account-tokens",
+    ),
     re_path(
         r"^(?P<organization_id_or_slug>[^/]+)/service-accounts/(?P<service_account_id>\d+)/viewer-context/$",
         OrganizationServiceAccountViewerContextEndpoint.as_view(),

src/sentry/service_accounts/endpoints/organization_service_account_tokens.py

@@ -0,0 +1,104 @@
+"""Service account token management APIs.
+
+Tokens live in control next to `ServiceAccountToken`; each save writes the token's outbox,
+which replicates it to the org's cell.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from datetime import datetime
+from typing import Any, TypedDict
+
+from drf_spectacular.utils import extend_schema
+from rest_framework.request import Request
+from rest_framework.response import Response
+
+from sentry.api.api_publish_status import ApiPublishStatus
+from sentry.api.base import control_silo_endpoint
+from sentry.api.paginator import OffsetPaginator
+from sentry.api.serializers import Serializer, serialize
+from sentry.apidocs.constants import RESPONSE_FORBIDDEN, RESPONSE_NOT_FOUND, RESPONSE_UNAUTHORIZED
+from sentry.apidocs.parameters import GlobalParams
+from sentry.apidocs.utils import inline_sentry_response_serializer
+from sentry.organizations.services.organization import (
+    RpcOrganization,
+    RpcUserOrganizationContext,
+)
+from sentry.service_accounts.endpoints.organization_service_accounts import (
+    SERVICE_ACCOUNT_ID,
+    ServiceAccountScopedEndpointBase,
+)
+from sentry.service_accounts.models import ServiceAccount, ServiceAccountToken
+
+
+class ServiceAccountTokenResponse(TypedDict):
+    id: str
+    name: str
+    scopes: list[str]
+    tokenLastCharacters: str
+    dateCreated: datetime
+    dateLastUsed: datetime | None
+    dateExpires: datetime | None
+
+
+class ServiceAccountTokenSerializer(Serializer[ServiceAccountTokenResponse]):
+    def serialize(
+        self, obj: ServiceAccountToken, attrs: Mapping[str, Any], user: Any, **kwargs: Any
+    ) -> ServiceAccountTokenResponse:
+        return {
+            "id": str(obj.id),
+            "name": obj.name,
+            "scopes": sorted(obj.scope_list),
+            "tokenLastCharacters": obj.token_last_characters,
+            "dateCreated": obj.date_added,
+            "dateLastUsed": obj.date_last_used,
+            "dateExpires": obj.date_expires,
+        }
+
+
+class ServiceAccountTokenEndpointBase(ServiceAccountScopedEndpointBase):
+    # A token outlives whoever minted it, so only a person signed in as themselves may
+    # manage tokens, not a service account.
+    requires_user = True
+
+
+@control_silo_endpoint
+class OrganizationServiceAccountTokenIndexEndpoint(ServiceAccountTokenEndpointBase):
+    publish_status = {
+        "GET": ApiPublishStatus.PRIVATE,
+    }
+
+    @extend_schema(
+        operation_id="List a Service Account's Tokens",
+        parameters=[GlobalParams.ORG_ID_OR_SLUG, SERVICE_ACCOUNT_ID],
+        responses={
+            200: inline_sentry_response_serializer(
+                "ListServiceAccountTokensResponse", list[ServiceAccountTokenResponse]
+            ),
+            401: RESPONSE_UNAUTHORIZED,
+            403: RESPONSE_FORBIDDEN,
+            404: RESPONSE_NOT_FOUND,
+        },
+    )
+    def get(
+        self,
+        request: Request,
+        organization_context: RpcUserOrganizationContext,
+        organization: RpcOrganization,
+        service_account: ServiceAccount,
+    ) -> Response:
+        """Return the account's active tokens, newest first, without their values."""
+        return self.paginate(
+            request=request,
+            queryset=ServiceAccountToken.objects.filter(
+                service_account=service_account,
+                organization_id=organization.id,
+                date_deactivated__isnull=True,
+            ),
+            order_by="-id",
+            paginator_cls=OffsetPaginator,
+            on_results=lambda results: serialize(
+                results, request.user, serializer=ServiceAccountTokenSerializer()
+            ),
+        )

src/sentry/service_accounts/endpoints/organization_service_accounts.py

@@ -331,13 +331,8 @@ def post(
         return Response(response, status=201)
 
 
-@control_silo_endpoint
-class OrganizationServiceAccountDetailsEndpoint(OrganizationServiceAccountEndpointBase):
-    publish_status = {
-        "GET": ApiPublishStatus.PRIVATE,
-        "PUT": ApiPublishStatus.PRIVATE,
-        "DELETE": ApiPublishStatus.PRIVATE,
-    }
+class ServiceAccountScopedEndpointBase(OrganizationServiceAccountEndpointBase):
+    """Resolves `service_account_id` to an active account in the URL's organization."""
 
     def convert_args(
         self, request: Request, *args: Any, **kwargs: Any
@@ -353,6 +348,15 @@ def convert_args(
             raise ResourceDoesNotExist
         return args, kwargs
 
+
+@control_silo_endpoint
+class OrganizationServiceAccountDetailsEndpoint(ServiceAccountScopedEndpointBase):
+    publish_status = {
+        "GET": ApiPublishStatus.PRIVATE,
+        "PUT": ApiPublishStatus.PRIVATE,
+        "DELETE": ApiPublishStatus.PRIVATE,
+    }
+
     @extend_schema(
         operation_id="Retrieve a Service Account",
         parameters=[GlobalParams.ORG_ID_OR_SLUG, SERVICE_ACCOUNT_ID],

static/app/data/controlsiloUrlPatterns.ts

@@ -95,6 +95,7 @@ export const controlsiloUrlPatterns: RegExp[] = [
   new RegExp('^api/0/organizations/[^/]+/audit-logs/$'),
   new RegExp('^api/0/organizations/[^/]+/service-accounts/$'),
   new RegExp('^api/0/organizations/[^/]+/service-accounts/[^/]+/$'),
+  new RegExp('^api/0/organizations/[^/]+/service-accounts/[^/]+/tokens/$'),
   new RegExp('^api/0/organizations/[^/]+/monitoring-providers/$'),
   new RegExp('^api/0/organizations/[^/]+/monitoring-providers/[^/]+/$'),
   new RegExp('^api/0/organizations/[^/]+/integrations/$'),

static/app/utils/api/knownSentryApiUrls.generated.ts

@@ -396,6 +396,7 @@ export type KnownSentryApiUrls =
   | '/organizations/$organizationIdOrSlug/sentry-apps/'
   | '/organizations/$organizationIdOrSlug/service-accounts/'
   | '/organizations/$organizationIdOrSlug/service-accounts/$serviceAccountId/'
+  | '/organizations/$organizationIdOrSlug/service-accounts/$serviceAccountId/tokens/'
   | '/organizations/$organizationIdOrSlug/service-accounts/$serviceAccountId/viewer-context/'
   | '/organizations/$organizationIdOrSlug/sessions/'
   | '/organizations/$organizationIdOrSlug/shared/groups/$shareId/'

tests/sentry/service_accounts/endpoints/test_organization_service_account_tokens.py

@@ -0,0 +1,138 @@
+from typing import Any
+from unittest import mock
+
+from django.urls import reverse
+from rest_framework.authentication import BaseAuthentication
+from rest_framework.request import Request
+
+from sentry.auth.services.auth import AuthenticatedToken
+from sentry.service_accounts.auth import SERVICE_ACCOUNT_TOKEN_KIND
+from sentry.service_accounts.endpoints.organization_service_account_tokens import (
+    OrganizationServiceAccountTokenIndexEndpoint,
+)
+from sentry.service_accounts.models import ServiceAccount, ServiceAccountToken
+from sentry.service_accounts.principal import ServiceAccountPrincipal
+from sentry.service_accounts.tokens import stored_token_fields
+from sentry.silo.base import SiloMode
+from sentry.testutils.cases import APITestCase
+from sentry.testutils.silo import assume_test_silo_mode, control_silo_test
+
+FEATURE_FLAG = "organizations:service-accounts"
+
+
+class ServiceAccountTokenTestCase(APITestCase):
+    def setUp(self) -> None:
+        super().setUp()
+        self.login_as(self.user)
+        self.service_account = self.create_service_account(
+            organization=self.organization,
+            name="ci",
+            scope_list=["org:read", "project:read", "project:releases"],
+        )
+
+    def create_token(
+        self,
+        service_account: ServiceAccount | None = None,
+        name: str = "deploys",
+        **kwargs: Any,
+    ) -> ServiceAccountToken:
+        service_account = service_account or self.service_account
+        stored = stored_token_fields(f"plaintext-{name}-{ServiceAccountToken.objects.count()}")
+        return ServiceAccountToken.objects.create(
+            service_account=service_account,
+            organization_id=service_account.organization_id,
+            name=name,
+            token_hashed=stored.token_hashed,
+            token_last_characters=stored.token_last_characters,
+            scope_list=["project:read"],
+            created_by=self.user,
+            **kwargs,
+        )
+
+    def login_as_role(self, role: str) -> None:
+        user = self.create_user()
+        self.create_member(organization=self.organization, user=user, role=role)
+        self.login_as(user)
+
+
+class ServiceAccountAuthentication(BaseAuthentication):
+    def authenticate(self, request: Request) -> tuple[Any, AuthenticatedToken]:
+        principal = ServiceAccountPrincipal(service_account_id=1, organization_id=1, name="sa")
+        token = AuthenticatedToken(
+            kind=SERVICE_ACCOUNT_TOKEN_KIND,
+            entity_id=1,
+            organization_id=1,
+            scopes=["service_account:read", "service_account:write"],
+        )
+        return (principal, token)
+
+
+@control_silo_test
+class OrganizationServiceAccountTokenIndexTest(ServiceAccountTokenTestCase):
+    endpoint = "sentry-api-0-organization-service-account-tokens"
+
+    def test_flag_off_returns_404(self) -> None:
+        self.get_error_response(self.organization.slug, self.service_account.id, status_code=404)
+
+    def test_list_returns_active_tokens_newest_first(self) -> None:
+        older = self.create_token(name="older")
+        newer = self.create_token(name="newer")
+        self.create_token(name="revoked", date_deactivated=older.date_added)
+        other_account = self.create_service_account(organization=self.organization, name="other")
+        self.create_token(service_account=other_account, name="other-account")
+
+        with self.feature(FEATURE_FLAG):
+            response = self.get_success_response(self.organization.slug, self.service_account.id)
+
+        assert [row["name"] for row in response.data] == ["newer", "older"]
+        assert response.data[0] == {
+            "id": str(newer.id),
+            "name": "newer",
+            "scopes": ["project:read"],
+            "tokenLastCharacters": newer.token_last_characters,
+            "dateCreated": newer.date_added,
+            "dateLastUsed": None,
+            "dateExpires": None,
+        }
+
+    def test_member_can_list(self) -> None:
+        self.create_token()
+        self.login_as_role("member")
+
+        with self.feature(FEATURE_FLAG):
+            response = self.get_success_response(self.organization.slug, self.service_account.id)
+
+        assert len(response.data) == 1
+
+    def test_service_account_caller_cannot_list(self) -> None:
+        url = reverse(self.endpoint, args=[self.organization.slug, self.service_account.id])
+
+        with (
+            self.feature(FEATURE_FLAG),
+            mock.patch.object(
+                OrganizationServiceAccountTokenIndexEndpoint,
+                "authentication_classes",
+                (ServiceAccountAuthentication,),
+            ),
+        ):
+            response = self.client.get(url)
+
+        assert response.status_code == 403
+        assert response.data["detail"]["code"] == "user-required"
+
+    def test_other_org_returns_404(self) -> None:
+        with assume_test_silo_mode(SiloMode.CELL):
+            other_organization = self.create_organization()
+        other = self.create_service_account(organization=other_organization, name="other")
+        self.create_token(service_account=other)
+
+        with self.feature(FEATURE_FLAG):
+            self.get_error_response(self.organization.slug, other.id, status_code=404)
+
+    def test_deleted_account_returns_404(self) -> None:
+        self.service_account.deactivate()
+
+        with self.feature(FEATURE_FLAG):
+            self.get_error_response(
+                self.organization.slug, self.service_account.id, status_code=404
+            )