ref(control_silo): Move over Authenticator model to Users module (#75857)

Part of moving control silo user related resources into the users module. Includes adding of types for the user model functions.

Apart of (#73856)

---------

Co-authored-by: Mark Story <[email protected]>
Co-authored-by: getsantry[bot] <66042841+getsantry[bot]@users.noreply.github.com>

Author: 3 people

src/sentry/api/endpoints/auth_index.py

@@ -21,8 +21,8 @@
 from sentry.auth.providers.saml2.provider import handle_saml_single_logout
 from sentry.auth.services.auth.impl import promote_request_rpc_user
 from sentry.auth.superuser import SUPERUSER_ORG_ID
-from sentry.models.authenticator import Authenticator
 from sentry.organizations.services.organization import organization_service
+from sentry.users.models.authenticator import Authenticator
 from sentry.utils import auth, json, metrics
 from sentry.utils.auth import DISABLE_SSO_CHECK_FOR_LOCAL_DEV, has_completed_sso, initiate_login
 from sentry.utils.settings import is_self_hosted
@@ -74,7 +74,8 @@ def _verify_user_via_inputs(validator: AuthVerifyValidator, request: Request) ->
         # See if we have a u2f challenge/response
         if "challenge" in validator.validated_data and "response" in validator.validated_data:
             try:
-                interface: U2fInterface = Authenticator.objects.get_interface(request.user, "u2f")
+                interface = Authenticator.objects.get_interface(request.user, "u2f")
+                assert isinstance(interface, U2fInterface)
                 if not interface.is_enrolled():
                     raise LookupError()
                 challenge = json.loads(validator.validated_data["challenge"])

src/sentry/api/endpoints/authenticator_index.py

@@ -7,7 +7,8 @@
 from sentry.api.api_owners import ApiOwner
 from sentry.api.api_publish_status import ApiPublishStatus
 from sentry.api.base import Endpoint, control_silo_endpoint
-from sentry.models.authenticator import Authenticator
+from sentry.auth.authenticators.base import ActivationChallengeResult
+from sentry.users.models.authenticator import Authenticator
 
 
 @control_silo_endpoint
@@ -30,9 +31,11 @@ def get(self, request: Request) -> Response:
         except LookupError:
             return Response([])
 
-        challenge = interface.activate(request._request).challenge
+        activation_result = interface.activate(request._request)
+        assert isinstance(activation_result, ActivationChallengeResult)
+        challenge_bytes = activation_result.challenge
 
-        webAuthnAuthenticationData = b64encode(challenge)
+        webAuthnAuthenticationData = b64encode(challenge_bytes)
         challenge = {}
         challenge["webAuthnAuthenticationData"] = webAuthnAuthenticationData
 

src/sentry/api/endpoints/user_authenticator_details.py

@@ -10,12 +10,14 @@
 from sentry.api.bases.user import OrganizationUserPermission, UserEndpoint
 from sentry.api.decorators import sudo_required
 from sentry.api.serializers import serialize
-from sentry.auth.authenticators.u2f import decode_credential_id
+from sentry.auth.authenticators.recovery_code import RecoveryCodeInterface
+from sentry.auth.authenticators.sms import SmsInterface
+from sentry.auth.authenticators.u2f import U2fInterface, decode_credential_id
 from sentry.auth.staff import has_staff_option, is_active_staff
 from sentry.auth.superuser import is_active_superuser
-from sentry.models.authenticator import Authenticator
 from sentry.models.user import User
 from sentry.security.utils import capture_security_activity
+from sentry.users.models.authenticator import Authenticator
 from sentry.utils.auth import MFA_SESSION_KEY
 
 
@@ -95,10 +97,19 @@ def get(self, request: Request, user, auth_id) -> Response:
         response = serialize(interface)
 
         if interface.interface_id == "recovery":
+            assert isinstance(
+                interface, RecoveryCodeInterface
+            ), "Interace must be RecoveryCodeInterface to get unused codes"
             response["codes"] = interface.get_unused_codes()
         if interface.interface_id == "sms":
+            assert isinstance(
+                interface, SmsInterface
+            ), "Interace must be SmsInterface to get phone number"
             response["phone"] = interface.phone_number
         if interface.interface_id == "u2f":
+            assert isinstance(
+                interface, U2fInterface
+            ), "Interace must be U2fInterface to get registered devices"
             response["devices"] = interface.get_registered_devices()
 
         return Response(response)
@@ -147,10 +158,15 @@ def delete(self, request: Request, user: User, auth_id, interface_device_id=None
         interface = authenticator.interface
         # Remove a single device and not entire authentication method
         if interface.interface_id == "u2f" and interface_device_id is not None:
+            assert isinstance(
+                interface, U2fInterface
+            ), "Interace must be U2fInterface to get registered devices"
             device_name = interface.get_device_name(interface_device_id)
             # Can't remove if this is the last device, will return False if so
             if not interface.remove_u2f_device(interface_device_id):
                 return Response(status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+            assert interface.authenticator, "Interface must have an authenticator to save"
             interface.authenticator.save()
 
             capture_security_activity(
@@ -198,6 +214,7 @@ def delete(self, request: Request, user: User, auth_id, interface_device_id=None
                 backup_interfaces = [x for x in interfaces if x.is_backup_interface]
                 if len(backup_interfaces) == len(interfaces):
                     for iface in backup_interfaces:
+                        assert iface.authenticator, "Interface must have an authenticator to delete"
                         iface.authenticator.delete()
 
                     # wait to generate entries until all pending writes

src/sentry/api/endpoints/user_authenticator_enroll.py

@@ -17,11 +17,13 @@
 from sentry.api.invite_helper import ApiInviteHelper, remove_invite_details_from_session
 from sentry.api.serializers import serialize
 from sentry.auth.authenticators.base import EnrollmentStatus, NewEnrollmentDisallowed
-from sentry.auth.authenticators.sms import SMSRateLimitExceeded
-from sentry.models.authenticator import Authenticator
+from sentry.auth.authenticators.sms import SmsInterface, SMSRateLimitExceeded
+from sentry.auth.authenticators.totp import TotpInterface
+from sentry.auth.authenticators.u2f import U2fInterface
 from sentry.models.user import User
 from sentry.organizations.services.organization import organization_service
 from sentry.security.utils import capture_security_activity
+from sentry.users.models.authenticator import Authenticator
 from sentry.utils.auth import MFA_SESSION_KEY
 
 logger = logging.getLogger(__name__)
@@ -149,15 +151,19 @@ def get(self, request: Request, user, interface_id) -> HttpResponse:
         response["form"] = get_serializer_field_metadata(serializer_map[interface_id]())
 
         # U2fInterface has no 'secret' attribute
-        try:
+        if hasattr(interface, "secret"):
             response["secret"] = interface.secret
-        except AttributeError:
-            pass
 
         if interface_id == "totp":
+            assert isinstance(
+                interface, TotpInterface
+            ), "Interface must be a TotpInterface to get provision URL"
             response["qrcode"] = interface.get_provision_url(user.email)
 
         if interface_id == "u2f":
+            assert isinstance(
+                interface, U2fInterface
+            ), "Interface must be a U2fInterface to start enrollement"
             publicKeyCredentialCreate, state = interface.start_enrollment(user)
             response["challenge"] = {}
             response["challenge"]["webAuthnRegisterData"] = b64encode(publicKeyCredentialCreate)
@@ -224,14 +230,15 @@ def post(self, request: Request, user, interface_id) -> HttpResponse:
             else:
                 return Response(ALREADY_ENROLLED_ERR, status=status.HTTP_400_BAD_REQUEST)
 
-        try:
+        if hasattr(interface, "secret"):
             interface.secret = request.data["secret"]
-        except KeyError:
-            pass
 
         context = {}
         # Need to update interface with phone number before validating OTP
         if "phone" in request.data:
+            assert isinstance(
+                interface, SmsInterface
+            ), "Interface must be a SmsInterface to get phone number"
             interface.phone_number = serializer.data["phone"]
 
             # Disregarding value of 'otp', if no OTP was provided,
@@ -263,6 +270,8 @@ def post(self, request: Request, user, interface_id) -> HttpResponse:
             if "webauthn_register_state" not in request.session:
                 return Response(INVALID_AUTH_STATE, status=status.HTTP_400_BAD_REQUEST)
             state = request.session["webauthn_register_state"]
+
+            assert isinstance(interface, U2fInterface), "Interface must be a U2fInterface to enroll"
             interface.try_enroll(
                 serializer.data["challenge"],
                 serializer.data["response"],

src/sentry/api/endpoints/user_authenticator_index.py

@@ -6,7 +6,7 @@
 from sentry.api.base import control_silo_endpoint
 from sentry.api.bases.user import UserEndpoint
 from sentry.api.serializers import serialize
-from sentry.models.authenticator import Authenticator
+from sentry.users.models.authenticator import Authenticator
 
 
 @control_silo_endpoint

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

@@ -17,7 +17,6 @@
 from sentry.app import env
 from sentry.auth.elevated_mode import has_elevated_mode
 from sentry.hybridcloud.services.organization_mapping import organization_mapping_service
-from sentry.models.authenticator import Authenticator
 from sentry.models.authidentity import AuthIdentity
 from sentry.models.avatars.user_avatar import UserAvatar
 from sentry.models.options.user_option import UserOption
@@ -29,6 +28,7 @@
 from sentry.models.userpermission import UserPermission
 from sentry.models.userrole import UserRoleUser
 from sentry.organizations.services.organization import RpcOrganizationSummary
+from sentry.users.models.authenticator import Authenticator
 from sentry.users.services.user import RpcUser
 from sentry.utils.avatar import get_gravatar_url
 

src/sentry/auth/authenticators/base.py

@@ -4,6 +4,7 @@
 from typing import TYPE_CHECKING, Any, Literal, Self
 
 from django.core.cache import cache
+from django.http import HttpRequest
 from django.utils import timezone
 from django.utils.translation import gettext_lazy as _
 from rest_framework.request import Request
@@ -14,8 +15,8 @@
 if TYPE_CHECKING:
     from django.utils.functional import _StrPromise
 
-    from sentry.models.authenticator import Authenticator
     from sentry.models.user import User
+    from sentry.users.models.authenticator import Authenticator
 
 
 class ActivationResult:
@@ -32,8 +33,8 @@ def __init__(
         self.type = type
         self.message = message
 
-    def __str__(self):
-        return self.message
+    def __str__(self) -> str:
+        return str(self.message)
 
     def __repr__(self) -> str:
         return f"<{type(self).__name__}: {self.message}>"
@@ -70,6 +71,9 @@ class AuthenticatorInterface:
     is_available = True
     allow_multi_enrollment = False
     allow_rotation_in_place = False
+    authenticator: Authenticator | None
+    status: EnrollmentStatus
+    _unbound_config: dict[Any, Any]
 
     def __init__(
         self, authenticator=None, status: EnrollmentStatus = EnrollmentStatus.EXISTING
@@ -136,21 +140,22 @@ def generate_new_config(self) -> dict[str, Any]:
         """This method is invoked if a new config is required."""
         return {}
 
-    def activate(self, request: Request):
+    def activate(self, request: HttpRequest) -> ActivationResult | None:
         """If an authenticator overrides this then the method is called
         when the dialog for authentication is brought up.  The returned string
         is then rendered in the UI.
         """
         # This method needs to be empty for the default
         # `requires_activation` property to make sense.
+        return None
 
     def enroll(self, user: User) -> None:
         """Invoked to enroll a user for this interface.  If already enrolled
         an error is raised.
 
         If `disallow_new_enrollment` is `True`, raises exception: `NewEnrollmentDisallowed`.
         """
-        from sentry.models.authenticator import Authenticator
+        from sentry.users.models.authenticator import Authenticator
 
         if self.disallow_new_enrollment:
             raise NewEnrollmentDisallowed
@@ -192,7 +197,7 @@ def validate_response(self, request: Request, challenge, response):
         return False
 
 
-class OtpMixin:
+class OtpMixin(AuthenticatorInterface):
     # mixed in from base class
     config: dict[str, Any]
     authenticator: Authenticator | None

src/sentry/auth/authenticators/recovery_code.py

@@ -46,6 +46,8 @@ def regenerate_codes(self, save=True):
         if not self.is_enrolled():
             raise RuntimeError("Interface is not enrolled")
         self.config.update(self.generate_new_config())
+
+        assert self.authenticator, "Cannot regenerate codes without self.authenticator"
         self.authenticator.reset_fields(save=False)
         if save:
             self.authenticator.save()

src/sentry/auth/authenticators/sms.py

@@ -12,7 +12,7 @@
 from sentry.utils.otp import TOTP
 from sentry.utils.sms import phone_number_as_e164, send_sms, sms_available
 
-from .base import ActivationMessageResult, AuthenticatorInterface, OtpMixin
+from .base import ActivationMessageResult, OtpMixin
 
 if TYPE_CHECKING:
     from django.utils.functional import _StrPromise
@@ -28,7 +28,7 @@ def __init__(self, phone_number, user_id, remote_ip):
         self.remote_ip = remote_ip
 
 
-class SmsInterface(OtpMixin, AuthenticatorInterface):
+class SmsInterface(OtpMixin):
     """This interface sends OTP codes via text messages to the user."""
 
     type = 2

src/sentry/auth/authenticators/totp.py

@@ -1,9 +1,9 @@
 from django.utils.translation import gettext_lazy as _
 
-from .base import AuthenticatorInterface, OtpMixin
+from .base import OtpMixin
 
 
-class TotpInterface(OtpMixin, AuthenticatorInterface):
+class TotpInterface(OtpMixin):
     """This interface uses TOTP with an authenticator."""
 
     type = 1

src/sentry/migrations/0001_squashed_0484_break_org_member_user_fk.py

@@ -28,14 +28,14 @@
 import sentry.models.apiapplication
 import sentry.models.apigrant
 import sentry.models.apitoken
-import sentry.models.authenticator
 import sentry.models.broadcast
 import sentry.models.groupshare
 import sentry.models.integrations.sentry_app
 import sentry.models.integrations.sentry_app_installation
 import sentry.models.scheduledeletion
 import sentry.models.servicehook
 import sentry.models.user
+import sentry.users.models.authenticator
 import sentry.utils.security.hash
 from sentry.new_migrations.migrations import CheckedMigration
 
@@ -9149,7 +9149,7 @@ class Migration(CheckedMigration):
         migrations.AlterField(
             model_name="authenticator",
             name="config",
-            field=sentry.models.authenticator.AuthenticatorConfig(editable=False),
+            field=sentry.users.models.authenticator.AuthenticatorConfig(editable=False),
         ),
         migrations.CreateModel(
             name="MonitorEnvironment",

src/sentry/models/__init__.py

@@ -1,3 +1,4 @@
+from sentry.users.models.authenticator import *  # NOQA
 from sentry.users.models.email import *  # NOQA
 
 from .activity import *  # NOQA
@@ -10,7 +11,6 @@
 from .artifactbundle import *  # NOQA
 from .assistant import *  # NOQA
 from .auditlogentry import *  # NOQA
-from .authenticator import *  # NOQA
 from .authidentity import *  # NOQA
 from .authidentityreplica import *  # NOQA
 from .authprovider import *  # NOQA