"""Authentication and registration. 1. POST /api/auth/start/ {mobile} -> {"next": "password"} the account signs in with a password -> {"next": "otp", ...} a code was sent by SMS 2. POST /api/auth/login/password/ {mobile, password} -> {"next": "dashboard", "token", "expiry", "user"} POST /api/auth/verify/ {mobile, code} -> {"next": "dashboard", "token", "expiry", "user"} -> {"next": "register", "registration_token", "mobile", "expires_at"} 3. POST /api/auth/register/ {registration_token, ...form} -> 201 {"status": "pending", ...} Whether a number has an account is only answered once its owner has confirmed a code, so `start` looks the same for every valid number. """ import logging from django.conf import settings from django.contrib.auth import authenticate from django.utils import timezone from knox.views import LoginView as KnoxLoginView from rest_framework import status from rest_framework.permissions import AllowAny, IsAuthenticated from rest_framework.response import Response from rest_framework.views import APIView from .models import OtpCode, RegistrationTicket, User from .serializers import ( AuthStartSerializer, OtpVerifySerializer, PasswordLoginSerializer, RegistrationSerializer, UserSerializer, ) from .sms import send_otp, send_registration_submitted logger = logging.getLogger(__name__) CODE_LENGTH = 6 def _issue_code(mobile: str, user: User | None) -> OtpCode: purpose = OtpCode.Purpose.LOGIN if user else OtpCode.Purpose.REGISTRATION otp, code = OtpCode.issue(mobile=mobile, purpose=purpose, user=user) send_otp(mobile, code, purpose) return otp def _code_payload(mobile: str, otp: OtpCode | None = None) -> dict: payload = {"next": "otp", "mobile": mobile, "code_length": CODE_LENGTH} if otp is not None: payload["expires_at"] = otp.expires_at payload["resend_available_at"] = otp.resend_available_at return payload class AuthStartView(APIView): """Step 1: password box, or a code sent by SMS.""" authentication_classes = [] permission_classes = [AllowAny] throttle_scope = "auth_start" def post(self, request): serializer = AuthStartSerializer(data=request.data) serializer.is_valid(raise_exception=True) mobile = serializer.validated_data["mobile"] user = User.objects.filter(mobile=mobile).first() if user is not None and user.effective_auth_method == User.AuthMethod.PASSWORD: return Response({"next": "password", "mobile": mobile}) return Response(_code_payload(mobile, _issue_code(mobile, user))) class OtpResendView(APIView): """Send a fresh code, honouring the cooldown.""" authentication_classes = [] permission_classes = [AllowAny] throttle_scope = "otp_request" def post(self, request): serializer = AuthStartSerializer(data=request.data) serializer.is_valid(raise_exception=True) mobile = serializer.validated_data["mobile"] user = User.objects.filter(mobile=mobile).first() if user is not None and user.effective_auth_method == User.AuthMethod.PASSWORD: return Response(_code_payload(mobile)) latest = OtpCode.objects.active().filter(mobile=mobile).first() if latest and latest.resend_available_at > timezone.now(): return Response( { "detail": "A code was just sent. Try again shortly.", "code": "resend_cooldown", "resend_available_at": latest.resend_available_at, }, status=status.HTTP_429_TOO_MANY_REQUESTS, ) return Response(_code_payload(mobile, _issue_code(mobile, user))) class BaseLoginView(KnoxLoginView): """Issues a knox token, with the account serialised alongside it.""" authentication_classes = [] permission_classes = [AllowAny] def issue_token(self, request, user): request.user = user return super().post(request, format=None) def get_post_response_data(self, request, token, instance): data = super().get_post_response_data(request, token, instance) data["next"] = "dashboard" data["user"] = UserSerializer(request.user).data return data class PasswordLoginView(BaseLoginView): """Step 2, password.""" throttle_scope = "auth_login" def post(self, request, format=None): serializer = PasswordLoginSerializer(data=request.data) serializer.is_valid(raise_exception=True) user = authenticate( request, username=serializer.validated_data["mobile"], password=serializer.validated_data["password"], ) if user is None: return Response( { "detail": "Incorrect mobile number or password.", "code": "invalid_credentials", }, status=status.HTTP_400_BAD_REQUEST, ) return self.issue_token(request, user) class VerifyCodeView(BaseLoginView): """Step 2, SMS code. A confirmed code either signs the account in or, when the number has no account, hands back the ticket the registration form needs. """ throttle_scope = "auth_login" def post(self, request, format=None): serializer = OtpVerifySerializer(data=request.data) serializer.is_valid(raise_exception=True) mobile = serializer.validated_data["mobile"] otp = ( OtpCode.objects.active() .filter(mobile=mobile) .select_related("user") .first() ) if otp is None: return Response( { "detail": "That code has expired. Request a new one.", "code": "code_expired", }, status=status.HTTP_400_BAD_REQUEST, ) if otp.is_exhausted: return Response( { "detail": "Too many incorrect attempts. Request a new code.", "code": "code_exhausted", }, status=status.HTTP_429_TOO_MANY_REQUESTS, ) if not otp.verify(serializer.validated_data["code"]): return Response( { "detail": "That code is not correct.", "code": "invalid_code", "attempts_left": max(0, settings.OTP_MAX_ATTEMPTS - otp.attempts), }, status=status.HTTP_400_BAD_REQUEST, ) user = otp.user or User.objects.filter(mobile=mobile).first() if user is None: ticket = RegistrationTicket.issue(mobile) return Response( { "next": "register", "registration_token": ticket.key, "mobile": mobile, "expires_at": ticket.expires_at, } ) if not user.is_active: return Response( {"detail": "This account is disabled.", "code": "account_disabled"}, status=status.HTTP_403_FORBIDDEN, ) if not user.mobile_verified: user.mobile_verified = True user.save(update_fields=["mobile_verified", "updated_at"]) return self.issue_token(request, user) class RegisterView(APIView): """Submit the registration form. Creates a pending account - no token.""" authentication_classes = [] permission_classes = [AllowAny] throttle_scope = "register" def post(self, request): serializer = RegistrationSerializer(data=request.data) serializer.is_valid(raise_exception=True) user = serializer.save() send_registration_submitted(user.mobile) logger.info("Registration submitted for %s (user %s)", user.mobile, user.pk) return Response( { "status": user.status, "mobile": user.mobile, "full_name": user.full_name, "detail": ( "Your registration is pending approval. We'll text you " "once it has been reviewed." ), }, status=status.HTTP_201_CREATED, ) class MeView(APIView): """The signed-in account, for the SPA to hydrate its session.""" permission_classes = [IsAuthenticated] def get(self, request): return Response(UserSerializer(request.user).data)