feat: improved user auth tokens

src/sentry/api/endpoints/api_tokens.py

@@ -19,6 +19,7 @@
 from sentry.models.apitoken import ApiToken
 from sentry.models.outbox import outbox_context
 from sentry.security.utils import capture_security_activity
+from sentry.types.token import AuthTokenType
 
 
 class ApiTokenSerializer(serializers.Serializer):
@@ -78,8 +79,8 @@ def post(self, request: Request) -> Response:
             token = ApiToken.objects.create(
                 user_id=request.user.id,
                 name=result.get("name", None),
+                token_type=AuthTokenType.USER,
                 scope_list=result["scopes"],
-                refresh_token=None,
                 expires_at=None,
             )
 

src/sentry/api/serializers/models/apitoken.py

@@ -1,5 +1,6 @@
 from sentry.api.serializers import Serializer, register, serialize
 from sentry.models.apitoken import ApiToken
+from sentry.types.token import AuthTokenType
 
 
 @register(ApiToken)
@@ -30,9 +31,10 @@ def serialize(self, obj, attrs, user, **kwargs):
         if not attrs["application"]:
             include_token = kwargs.get("include_token", True)
             if include_token:
-                data["token"] = obj.token
+                data["token"] = obj.plaintext_token
 
-            data["refreshToken"] = obj.refresh_token
+                if not obj.token_type == AuthTokenType.USER:
+                    data["refreshToken"] = obj.plaintext_refresh_token
 
         """
         While this is a nullable column at the db level, this should never be empty. If it is, it's a sign that the

src/sentry/models/apitoken.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import hashlib
 import secrets
 from collections.abc import Collection
 from datetime import timedelta
@@ -22,16 +23,82 @@
 from sentry.types.token import AuthTokenType
 
 DEFAULT_EXPIRATION = timedelta(days=30)
+TOKEN_REDACTED = "***REDACTED***"
 
 
 def default_expiration():
     return timezone.now() + DEFAULT_EXPIRATION
 
 
-def generate_token():
+def generate_token(token_type: AuthTokenType | str | None = AuthTokenType.__empty__) -> str:
+    if token_type:
+        return f"{token_type}{secrets.token_hex(nbytes=32)}"
+
     return secrets.token_hex(nbytes=32)
 
 
+class PlaintextSecretAlreadyRead(Exception):
+    """the secret you are trying to read is read-once and cannot be accessed directly again"""
+
+    pass
+
+
+class NotSupported(Exception):
+    """the method you called is not supported by this token type"""
+
+    pass
+
+
+class ApiTokenManager(ControlOutboxProducingManager):
+    def create(self, *args, **kwargs):
+        token_type: AuthTokenType | None = kwargs.get("token_type", None)
+
+        # Typically the .create() method is called with `refresh_token=None` as an
+        # argument when we specifically do not want a refresh_token.
+        #
+        # But if it is not None or not specified, we should generate a token since
+        # that is the expected behavior... the refresh_token field on ApiToken has
+        # a default of generate_token()
+        #
+        # TODO(mdtro): All of these if/else statements will be cleaned up at a later time
+        #   to use a match statment on the AuthTokenType. Move each of the various token type
+        #   create calls one at a time.
+        if "refresh_token" in kwargs:
+            plaintext_refresh_token = kwargs["refresh_token"]
+        else:
+            plaintext_refresh_token = generate_token()
+
+        if token_type == AuthTokenType.USER:
+            plaintext_token = generate_token(token_type=AuthTokenType.USER)
+            plaintext_refresh_token = None  # user auth tokens do not have refresh tokens
+        else:
+            # to maintain compatibility with current
+            # code that currently calls create with token= specified
+            if "token" in kwargs:
+                plaintext_token = kwargs["token"]
+            else:
+                plaintext_token = generate_token()
+
+        if options.get("apitoken.save-hash-on-create"):
+            kwargs["hashed_token"] = hashlib.sha256(plaintext_token.encode()).hexdigest()
+
+            if plaintext_refresh_token:
+                kwargs["hashed_refresh_token"] = hashlib.sha256(
+                    plaintext_refresh_token.encode()
+                ).hexdigest()
+
+        kwargs["token"] = plaintext_token
+        kwargs["refresh_token"] = plaintext_refresh_token
+
+        api_token = super().create(*args, **kwargs)
+
+        # Store the plaintext tokens for one-time retrieval
+        api_token._set_plaintext_token(token=plaintext_token)
+        api_token._set_plaintext_refresh_token(token=plaintext_refresh_token)
+
+        return api_token
+
+
 @control_silo_only_model
 class ApiToken(ReplicatedControlModel, HasApiScopes):
     __relocation_scope__ = {RelocationScope.Global, RelocationScope.Config}
@@ -50,7 +117,7 @@ class ApiToken(ReplicatedControlModel, HasApiScopes):
     expires_at = models.DateTimeField(null=True, default=default_expiration)
     date_added = models.DateTimeField(default=timezone.now)
 
-    objects: ClassVar[ControlOutboxProducingManager[ApiToken]] = ControlOutboxProducingManager(
+    objects: ClassVar[ControlOutboxProducingManager[ApiToken]] = ApiTokenManager(
         cache_fields=("token",)
     )
 
@@ -63,12 +130,117 @@ class Meta:
     def __str__(self):
         return force_str(self.token)
 
+    def _set_plaintext_token(self, token: str) -> None:
+        """Set the plaintext token for one-time reading
+        This function should only be called from the model's
+        manager class.
+
+        :param token: A plaintext string of the token
+        :raises PlaintextSecretAlreadyRead: when the token has already been read once
+        """
+        existing_token: str | None = None
+        try:
+            existing_token = self.__plaintext_token
+        except AttributeError:
+            self.__plaintext_token: str = token
+
+        if existing_token == TOKEN_REDACTED:
+            raise PlaintextSecretAlreadyRead()
+
+    def _set_plaintext_refresh_token(self, token: str) -> None:
+        """Set the plaintext refresh token for one-time reading
+        This function should only be called from the model's
+        manager class.
+
+        :param token: A plaintext string of the refresh token
+        :raises PlaintextSecretAlreadyRead: if the token has already been read once
+        """
+        existing_refresh_token: str | None = None
+        try:
+            existing_refresh_token = self.__plaintext_refresh_token
+        except AttributeError:
+            self.__plaintext_refresh_token: str = token
+
+        if existing_refresh_token == TOKEN_REDACTED:
+            raise PlaintextSecretAlreadyRead()
+
+    @property
+    def plaintext_token(self) -> str:
+        """The plaintext value of the token
+        To be called immediately after creation of a new `ApiToken` to return the
+        plaintext token to the user. After reading the token, the plaintext token
+        string will be set to `TOKEN_REDACTED` to prevent future accidental leaking
+        of the token in logs, exceptions, etc.
+
+        :raises PlaintextSecretAlreadyRead: if the token has already been read once
+        :return: the plaintext value of the token
+        """
+        token = self.__plaintext_token
+        if token == TOKEN_REDACTED:
+            raise PlaintextSecretAlreadyRead()
+
+        self.__plaintext_token = TOKEN_REDACTED
+
+        return token
+
+    @property
+    def plaintext_refresh_token(self) -> str:
+        """The plaintext value of the refresh token
+        To be called immediately after creation of a new `ApiToken` to return the
+        plaintext token to the user. After reading the token, the plaintext token
+        string will be set to `TOKEN_REDACTED` to prevent future accidental leaking
+        of the token in logs, exceptions, etc.
+
+        :raises PlaintextSecretAlreadyRead: if the refresh token has already been read once
+        :raises NotSupported: if called on a User Auth Token
+        :return: the plaintext value of the refresh token
+        """
+        if not self.refresh_token and not self.hashed_refresh_token:
+            raise NotSupported("This API token type does not support refresh tokens")
+
+        token = self.__plaintext_refresh_token
+        if token == TOKEN_REDACTED:
+            raise PlaintextSecretAlreadyRead()
+
+        self.__plaintext_refresh_token = TOKEN_REDACTED
+
+        return token
+
     def save(self, *args: Any, **kwargs: Any) -> None:
+        if options.get("apitoken.save-hash-on-create"):
+            self.hashed_token = hashlib.sha256(self.token.encode()).hexdigest()
+
+            if self.refresh_token:
+                self.hashed_refresh_token = hashlib.sha256(self.refresh_token.encode()).hexdigest()
+            else:
+                # The backup tests create a token with a refresh_token and then clear it out.
+                # So if the refresh_token is None, wipe out any hashed value that may exist too.
+                # https://github.com/getsentry/sentry/blob/1fc699564e79c62bff6cc3c168a49bfceadcac52/tests/sentry/backup/test_imports.py#L1306
+                self.hashed_refresh_token = None
+
         if options.get("apitoken.auto-add-last-chars"):
             token_last_characters = self.token[-4:]
             self.token_last_characters = token_last_characters
 
-        return super().save(**kwargs)
+        return super().save(*args, **kwargs)
+
+    def update(self, *args: Any, **kwargs: Any) -> int:
+        # if the token or refresh_token was updated, we need to
+        # re-calculate the hashed values
+        if options.get("apitoken.save-hash-on-create"):
+            if "token" in kwargs:
+                kwargs["hashed_token"] = hashlib.sha256(kwargs["token"].encode()).hexdigest()
+
+            if "refresh_token" in kwargs:
+                kwargs["hashed_refresh_token"] = hashlib.sha256(
+                    kwargs["refresh_token"].encode()
+                ).hexdigest()
+
+        if options.get("apitoken.auto-add-last-chars"):
+            if "token" in kwargs:
+                kwargs["token_last_characters"] = kwargs["token"][-4:]
+
+        return super().update(*args, **kwargs)
 
     def outbox_region_names(self) -> Collection[str]:
         return list(find_all_region_names())
@@ -104,10 +276,16 @@ def get_allowed_origins(self):
         return ()
 
     def refresh(self, expires_at=None):
+        if self.token_type == AuthTokenType.USER:
+            raise NotSupported("User auth tokens do not support refreshing the token")
+
         if expires_at is None:
             expires_at = timezone.now() + DEFAULT_EXPIRATION
 
-        self.update(token=generate_token(), refresh_token=generate_token(), expires_at=expires_at)
+        new_token = generate_token(token_type=self.token_type)
+        new_refresh_token = generate_token(token_type=self.token_type)
+
+        self.update(token=new_token, refresh_token=new_refresh_token, expires_at=expires_at)
 
     def get_relocation_scope(self) -> RelocationScope:
         if self.application_id is not None:
@@ -125,9 +303,9 @@ def write_relocation_import(
         )
         existing = self.__class__.objects.filter(query).first()
         if existing:
-            self.token = generate_token()
+            self.token = generate_token(token_type=self.token_type)
             if self.refresh_token is not None:
-                self.refresh_token = generate_token()
+                self.refresh_token = generate_token(token_type=self.token_type)
             if self.expires_at is not None:
                 self.expires_at = timezone.now() + DEFAULT_EXPIRATION
 

src/sentry/testutils/factories.py

@@ -148,6 +148,7 @@
 from sentry.types.activity import ActivityType
 from sentry.types.integrations import ExternalProviders
 from sentry.types.region import Region, get_local_region, get_region_by_name
+from sentry.types.token import AuthTokenType
 from sentry.utils import json, loremipsum
 from sentry.utils.performance_issues.performance_problem import PerformanceProblem
 from social_auth.models import UserSocialAuth
@@ -423,6 +424,7 @@ def create_user_auth_token(user, scope_list: list[str] | None = None, **kwargs)
         return ApiToken.objects.create(
             user=user,
             scope_list=scope_list,
+            token_type=AuthTokenType.USER,
             **kwargs,
         )
 

src/sentry/testutils/helpers/backups.py

@@ -99,6 +99,7 @@
 from sentry.testutils.cases import TransactionTestCase
 from sentry.testutils.factories import get_fixture_path
 from sentry.testutils.silo import assume_test_silo_mode
+from sentry.types.token import AuthTokenType
 from sentry.utils import json
 from sentry.utils.json import JSONData
 
@@ -632,7 +633,10 @@ def create_exhaustive_global_configs(self, owner: User):
         ControlOption.objects.create(key="bar", value="b")
         ApiAuthorization.objects.create(user=owner)
         ApiToken.objects.create(
-            user=owner, expires_at=None, name="create_exhaustive_global_configs"
+            user=owner,
+            expires_at=None,
+            name="create_exhaustive_global_configs",
+            token_type=AuthTokenType.USER,
         )
 
     @assume_test_silo_mode(SiloMode.REGION)

src/sentry/web/frontend/setup_wizard.py

@@ -24,6 +24,7 @@
 from sentry.services.hybrid_cloud.project_key.model import ProjectKeyRole
 from sentry.services.hybrid_cloud.project_key.service import project_key_service
 from sentry.services.hybrid_cloud.user.model import RpcUser
+from sentry.types.token import AuthTokenType
 from sentry.utils.http import absolute_uri
 from sentry.utils.security.orgauthtoken_token import (
     SystemUrlPrefixMissingException,
@@ -159,7 +160,7 @@ def get_token(mappings: list[OrganizationMapping], user: RpcUser):
         token = ApiToken.objects.create(
             user_id=user.id,
             scope_list=["project:releases"],
-            refresh_token=None,
+            token_type=AuthTokenType.USER,
             expires_at=None,
         )
     return serialize(token)

tests/sentry/api/serializers/test_apitoken.py

@@ -1,5 +1,9 @@
 from sentry.api.serializers import ApiTokenSerializer
+from sentry.models.apitoken import ApiToken
+from sentry.silo.base import SiloMode
 from sentry.testutils.cases import TestCase
+from sentry.testutils.helpers.options import override_options
+from sentry.testutils.silo import assume_test_silo_mode
 
 
 class TestApiTokenSerializer(TestCase):
@@ -38,6 +42,33 @@ def test_when_flag_is_false(self) -> None:
         assert "token" not in serialized_object
 
 
+class TestRefreshTokens(TestApiTokenSerializer):
+    def setUp(self) -> None:
+        super().setUp()
+        attrs = self._serializer.get_attrs(item_list=[self._token], user=self._user)
+        attrs["application"] = None
+        self._attrs = attrs
+
+    def test_no_refresh_token_on_user_token(self) -> None:
+        serialized_object = self._serializer.serialize(
+            obj=self._token, user=self._user, attrs=self._attrs
+        )
+
+        assert "refreshToken" not in serialized_object
+
+    @override_options({"apitoken.save-hash-on-create": True})
+    def test_refresh_token_on_non_user_token(self) -> None:
+        with assume_test_silo_mode(SiloMode.CONTROL):
+            token = ApiToken.objects.create(user=self._user)
+            assert token.hashed_refresh_token is not None
+
+            serialized_object = self._serializer.serialize(
+                obj=token, user=self._user, attrs=self._attrs
+            )
+
+            assert "refreshToken" in serialized_object
+
+
 class TestLastTokenCharacters(TestApiTokenSerializer):
     def test_field_is_returned(self) -> None:
         attrs = self._serializer.get_attrs(item_list=[self._token], user=self._user)

tests/sentry/api/test_authentication.py

@@ -181,11 +181,11 @@ def setUp(self):
 
         self.auth = UserAuthTokenAuthentication()
         self.org = self.create_organization(owner=self.user)
-        self.token = "abc123"
         self.api_token = ApiToken.objects.create(
-            token=self.token,
+            token_type=AuthTokenType.USER,
             user=self.user,
         )
+        self.token = self.api_token.plaintext_token
 
     def test_authenticate(self):
         request = HttpRequest()