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

@@ -30,9 +30,8 @@ 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["refreshToken"] = obj.refresh_token
+                data["token"] = obj._plaintext_token
+                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
@@ -32,6 +33,57 @@ def generate_token():
     return secrets.token_hex(nbytes=32)
 
 
+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 = f"{token_type}{generate_token()}"
+            plaintext_refresh_token = None  # user auth tokens do not have refresh tokens
+        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 is not None:
+                kwargs["hashed_refresh_token"] = hashlib.sha256(
+                    plaintext_refresh_token.encode()
+                ).hexdigest()
+
+        kwargs["token"] = plaintext_token
+        kwargs["refresh_token"] = plaintext_refresh_token
+
+        if plaintext_refresh_token is not None:
+            kwargs["refresh_token"] = plaintext_refresh_token
+            kwargs["hashed_refresh_token"] = hashlib.sha256(
+                plaintext_refresh_token.encode()
+            ).hexdigest()
+
+        api_token = super().create(*args, **kwargs)
+
+        # Store the plaintext tokens for one-time retrieval
+        api_token.__plaintext_token = plaintext_token
+        api_token.__plaintext_refresh_token = plaintext_refresh_token
+
+        return api_token
+
+
 @control_silo_only_model
 class ApiToken(ReplicatedControlModel, HasApiScopes):
     __relocation_scope__ = {RelocationScope.Global, RelocationScope.Config}
@@ -50,7 +102,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,6 +115,40 @@ class Meta:
     def __str__(self):
         return force_str(self.token)
 
+    @property
+    def _plaintext_token(self):
+        """
+        To be called immediately after creation of a new token to return the
+        plaintext token to the user. After reading the token, it will be set
+        to `None` to prevent future accidental leaking of the token in logs,
+        exceptions, etc.
+        """
+        manager_class_name = self.objects.__class__.__name__
+        plaintext_token: str | None = getattr(self, f"_{manager_class_name}__plaintext_token", None)
+
+        if plaintext_token is not None:
+            setattr(self, f"_{manager_class_name}__plaintext_token", None)
+
+        return plaintext_token
+
+    @property
+    def _plaintext_refresh_token(self):
+        """
+        To be called immediately after creation of a new token to return the
+        plaintext refresh token to the user. After reading the refresh token, it will be set
+        to `None` to prevent future accidental leaking of the refresh token in logs,
+        exceptions, etc.
+        """
+        manager_class_name = self.objects.__class__.__name__
+        plaintext_refresh_token: str | None = getattr(
+            self, f"_{manager_class_name}__plaintext_refresh_token", None
+        )
+
+        if plaintext_refresh_token is not None:
+            setattr(self, f"_{manager_class_name}__plaintext_refresh_token", None)
+
+        return plaintext_refresh_token
+
     def save(self, *args: Any, **kwargs: Any) -> None:
         if options.get("apitoken.auto-add-last-chars"):
             token_last_characters = self.token[-4:]

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/models/test_apitoken.py

@@ -1,3 +1,4 @@
+import hashlib
 from datetime import timedelta
 
 from django.utils import timezone
@@ -12,6 +13,7 @@
 from sentry.testutils.helpers import override_options
 from sentry.testutils.outbox import outbox_runner
 from sentry.testutils.silo import assume_test_silo_mode, control_silo_test
+from sentry.types.token import AuthTokenType
 
 
 @control_silo_test
@@ -74,6 +76,41 @@ def test_last_chars_are_not_set(self):
         token = ApiToken.objects.create(user_id=user.id)
         assert token.token_last_characters is None
 
+    @override_options({"apitoken.save-hash-on-create": True})
+    def test_hash_exists_on_user_token(self):
+        user = self.create_user()
+        token = ApiToken.objects.create(user_id=user.id, token_type=AuthTokenType.USER)
+        assert token.hashed_token is not None
+        assert len(token.hashed_token) == 64  # sha256 hash
+        assert token.hashed_refresh_token is None  # user auth tokens don't have refresh tokens
+
+    @override_options({"apitoken.save-hash-on-create": False})
+    def test_hash_does_not_exist_on_user_token_with_option_off(self):
+        user = self.create_user()
+        token = ApiToken.objects.create(user_id=user.id, token_type=AuthTokenType.USER)
+        assert token.hashed_token is None
+        assert token.hashed_refresh_token is None  # user auth tokens don't have refresh tokens
+
+    @override_options({"apitoken.save-hash-on-create": True})
+    def test_plaintext_values_only_available_immediately_after_create(self):
+        user = self.create_user()
+        token = ApiToken.objects.create(user_id=user.id, token_type=AuthTokenType.USER)
+        assert token._plaintext_token is not None
+        assert token._plaintext_refresh_token is None  # user auth tokens don't have refresh tokens
+
+        _ = token._plaintext_token
+
+        # we read the value above so now it should
+        # now be None as it is a "read once" property
+        assert token._plaintext_token is None
+
+    @override_options({"apitoken.save-hash-on-create": True})
+    def test_user_auth_token_hash(self):
+        user = self.create_user()
+        token = ApiToken.objects.create(user_id=user.id, token_type=AuthTokenType.USER)
+        expected_hash = hashlib.sha256(token._plaintext_token.encode()).hexdigest()
+        assert expected_hash == token.hashed_token
+
 
 @control_silo_test
 class ApiTokenInternalIntegrationTest(TestCase):