[Eng-32] feat :create otp based reset for password

care/templates/sms/otp_reset_sms.txt

@@ -0,0 +1 @@
+Your CARE password reset OTP is {{random_otp}}. Do not share this with anyone.

care/users/api/otp_viewset/reset_password.py

@@ -0,0 +1,156 @@
+import logging
+import secrets
+import string
+from datetime import timedelta
+
+from django.conf import settings
+from django.contrib.auth.password_validation import (
+    get_password_validators,
+    validate_password,
+)
+from django.utils import timezone
+from drf_spectacular.utils import extend_schema
+from pydantic import BaseModel, Field, field_validator
+from pydantic import ValidationError as PydanticValidationError
+from rest_framework.exceptions import ValidationError
+from rest_framework.generics import GenericAPIView
+from rest_framework.response import Response
+
+from care.users.models import User, UserMobileOTP
+from care.utils import sms
+from care.utils.models.validators import mobile_validator
+from care.utils.sms.utils import get_sms_content
+from config.ratelimit import ratelimit
+
+logger = logging.getLogger(__name__)
+
+
+def rand_pass(size):
+    if not settings.USE_SMS:
+        return "45612"
+
+    return "".join(secrets.choice(string.digits) for _ in range(size))
+
+
+class OTPBaseSpec(BaseModel):
+    phone_number: str
+
+    @field_validator("phone_number")
+    @classmethod
+    def validate_phone_number(cls, value):
+        try:
+            mobile_validator(value)
+        except Exception as e:
+            msg = "Invalid phone number"
+            raise ValueError(msg) from e
+        return value
+
+
+class OTPResetSendSpec(OTPBaseSpec):
+    pass
+
+
+class OTPResetConfirmSpec(OTPBaseSpec):
+    otp: str = Field(min_length=settings.OTP_LENGTH, max_length=settings.OTP_LENGTH)
+    password: str = Field(min_length=8)
+
+
+class OTPResetSendView(GenericAPIView):
+    authentication_classes = []
+    permission_classes = []
+
+    @extend_schema(request=OTPResetSendSpec)
+    def post(self, request):
+        try:
+            data = OTPResetSendSpec(**request.data)
+        except PydanticValidationError as e:
+            raise ValidationError(e.errors()) from e
+
+        if ratelimit(request, "otp-password-reset", ["ip"]):
+            error_message = "Too many requests. Please try again later."
+            return Response(
+                {"detail": error_message},
+                status=429,
+            )
+
+        sent_otps = UserMobileOTP.objects.filter(
+            created_date__gte=(
+                timezone.now() - timedelta(hours=settings.OTP_REPEAT_WINDOW)
+            ),
+            is_used=False,
+            phone_number=data.phone_number,
+        )
+        if sent_otps.count() >= settings.OTP_MAX_REPEATS_WINDOW:
+            raise ValidationError(
+                {"error": "Max OTP requests exceeded. Try again later."}
+            )
+        if not User.objects.filter(phone_number=data.phone_number).exists():
+            return Response({"otp": "generated"})
+
+        random_otp = rand_pass(settings.OTP_LENGTH)
+        if settings.USE_SMS:
+            try:
+                content = get_sms_content(
+                    settings.OTP_SMS_RESET_PASSWORD_TEMPLATE_PATH,
+                    {"random_otp": random_otp},
+                )
+                sms.send_text_message(
+                    content=content,
+                    recipients=[data.phone_number],
+                )
+            except Exception as e:
+                logger.error(e)
+                return Response(
+                    {"error": "Error while sending OTP. Contact admin."}, status=400
+                )
+
+        UserMobileOTP.objects.create(phone_number=data.phone_number, otp=random_otp)
+        return Response({"otp": "generated"})
+
+
+class OTPResetConfirmView(GenericAPIView):
+    authentication_classes = []
+    permission_classes = []
+
+    @extend_schema(request=OTPResetConfirmSpec)
+    def post(self, request):
+        try:
+            data = OTPResetConfirmSpec(**request.data)
+        except PydanticValidationError as e:
+            raise ValidationError(e.errors()) from e
+        if ratelimit(request, "otp-password-confirm", ["ip"]):
+            error_message = "Too many requests. Please try again later."
+            return Response(
+                {"detail": error_message},
+                status=429,
+            )
+        user = User.objects.filter(phone_number=data.phone_number).first()
+        if not user:
+            raise ValidationError({"error": "No User linked to this phone number"})
+        otp_obj = (
+            UserMobileOTP.objects.filter(
+                phone_number=data.phone_number,
+                is_used=False,
+                created_date__gte=(
+                    timezone.now() - timedelta(hours=settings.OTP_REPEAT_WINDOW)
+                ),
+            )
+            .order_by("-created_date")
+            .first()
+        )
+        if not otp_obj or otp_obj.otp != data.otp:
+            raise ValidationError({"otp": "Invalid OTP"})
+
+        validate_password(
+            data.password,
+            user=user,
+            password_validators=get_password_validators(
+                settings.AUTH_PASSWORD_VALIDATORS
+            ),
+        )
+        user.set_password(data.password)
+        user.save()
+        UserMobileOTP.objects.filter(
+            phone_number=data.phone_number,
+        ).delete()
+        return Response({"message": "Password reset successful"})

care/users/migrations/0028_usermobileotp.py

@@ -0,0 +1,31 @@
+# Generated by Django 6.0 on 2026-04-27 08:54
+
+import care.utils.models.validators
+import uuid
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('users', '0027_user_cached_role_orgs'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='UserMobileOTP',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('external_id', models.UUIDField(db_index=True, default=uuid.uuid4, unique=True)),
+                ('created_date', models.DateTimeField(auto_now_add=True, db_index=True, null=True)),
+                ('modified_date', models.DateTimeField(auto_now=True, db_index=True, null=True)),
+                ('deleted', models.BooleanField(db_index=True, default=False)),
+                ('is_used', models.BooleanField(default=False)),
+                ('phone_number', models.CharField(max_length=14, validators=[care.utils.models.validators.PhoneNumberValidator(types=('mobile', 'landline'))])),
+                ('otp', models.CharField(max_length=10)),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+    ]

care/users/models.py

@@ -266,3 +266,11 @@ def check_user_has_flag(cls, user_id: int, flag_name: FlagName) -> bool:
     @classmethod
     def get_all_flags(cls, user_id: int) -> tuple[FlagName]:
         return super().get_all_flags(user_id)
+
+
+class UserMobileOTP(BaseModel):
+    is_used = models.BooleanField(default=False)
+    phone_number = models.CharField(
+        max_length=14, validators=[mobile_or_landline_number_validator]
+    )
+    otp = models.CharField(max_length=10)

config/settings/base.py

@@ -692,6 +692,10 @@
 
 OTP_SMS_TEMPLATE_PATH = env("OTP_SMS_TEMPLATE", default="sms/otp_sms.txt")
 
+OTP_SMS_RESET_PASSWORD_TEMPLATE_PATH = env(
+    "OTP_SMS_RESET_PASSWORD_TEMPLATE", default="sms/otp_reset_sms.txt"
+)
+
 USER_CREATE_PASSWORD_EMAIL_TEMPLATE_PATH = env(
     "USER_CREATE_PASSWORD_TEMPLATE_PATH", default="email/user_create_password.html"
 )

config/urls.py

@@ -10,6 +10,10 @@
     SpectacularSwaggerView,
 )
 
+from care.users.api.otp_viewset.reset_password import (
+    OTPResetConfirmView,
+    OTPResetSendView,
+)
 from care.users.api.viewsets.change_password import ChangePasswordView
 from care.users.reset_password_views import (
     ResetPasswordCheck,
@@ -62,6 +66,16 @@
         ChangePasswordView.as_view(),
         name="change_password_view",
     ),
+    path(
+        "api/v1/otp/password_reset/send/",
+        OTPResetSendView.as_view(),
+        name="otp_password_reset_send",
+    ),
+    path(
+        "api/v1/otp/password_reset/confirm/",
+        OTPResetConfirmView.as_view(),
+        name="otp_password_reset_confirm",
+    ),
     path("api/v1/", include(api_router.urlpatterns)),
     *static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT),
 ]