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

care/emr/api/otp_viewsets/login.py

@@ -2,6 +2,7 @@
 import secrets
 import string
 from datetime import timedelta
+from enum import Enum
 
 from django.conf import settings
 from django.utils import timezone
@@ -12,20 +13,60 @@
 from rest_framework.response import Response
 
 from care.emr.api.viewsets.base import EMRBaseViewSet
-from care.facility.models.patient import PatientMobileOTP
+from care.facility.models.patient import MobileOTP
 from care.utils import sms
 from care.utils.models.validators import mobile_validator
-from care.utils.sms.utils import get_sms_content
 from config.patient_otp_token import PatientToken
 
 logger = logging.getLogger(__name__)
 
 
+class OTPType(str, Enum):
+    login = "login"
+    reset_password = "reset_password"
+
+
 def rand_pass(size):
     return "".join(secrets.choice(string.digits) for _ in range(size))
 
 
-class OTPLoginRequestSpec(BaseModel):
+def send_otp(phone_number, purpose):
+    sent_otps = MobileOTP.objects.filter(
+        created_date__gte=(timezone.now() - timedelta(settings.OTP_REPEAT_WINDOW)),
+        is_used=False,
+        phone_number=phone_number,
+    )
+    if sent_otps.count() >= settings.OTP_MAX_REPEATS_WINDOW:
+        raise ValidationError({"phone_number": "Max Retries has exceeded"})
+
+    random_otp = ""
+    if settings.USE_SMS:
+        random_otp = rand_pass(settings.OTP_LENGTH)
+        try:
+            if purpose == OTPType.login:
+                content = settings.OTP_SMS_CONTENT.format(otp=random_otp)
+            elif purpose == OTPType.reset_password:
+                content = settings.OTP_SMS_RESET_PASSWORD_CONTENT.format(otp=random_otp)
+
+            sms.send_text_message(
+                content=content,
+                recipients=[phone_number],
+            )
+        except Exception as e:
+            logger.error(e)
+            return Response(
+                {"error": "Error while sending OTP. Contact admin."}, status=400
+            )
+    elif settings.IS_PRODUCTION:
+        random_otp = rand_pass(settings.OTP_LENGTH)
+    else:
+        random_otp = "45612"
+
+    MobileOTP.objects.create(phone_number=phone_number, otp=random_otp)
+    return None
+
+
+class OTPRequestBaseSpec(BaseModel):
     phone_number: str
 
     @field_validator("phone_number")
@@ -39,7 +80,7 @@ def validate_phone_number(cls, value):
         return value
 
 
-class OTPLoginSpec(OTPLoginRequestSpec):
+class OTPLoginSpec(OTPRequestBaseSpec):
     otp: str = Field(min_length=settings.OTP_LENGTH, max_length=settings.OTP_LENGTH)
 
 
@@ -48,41 +89,14 @@ class OTPLoginView(EMRBaseViewSet):
     permission_classes = []
 
     @extend_schema(
-        request=OTPLoginRequestSpec,
+        request=OTPRequestBaseSpec,
     )
     @action(detail=False, methods=["POST"])
     def send(self, request):
-        data = OTPLoginRequestSpec(**request.data)
-        sent_otps = PatientMobileOTP.objects.filter(
-            created_date__gte=(timezone.now() - timedelta(settings.OTP_REPEAT_WINDOW)),
-            is_used=False,
-            phone_number=data.phone_number,
-        )
-        if sent_otps.count() >= settings.OTP_MAX_REPEATS_WINDOW:
-            raise ValidationError({"phone_number": "Max Retries has exceeded"})
-        random_otp = ""
-        if settings.USE_SMS:
-            random_otp = rand_pass(settings.OTP_LENGTH)
-            try:
-                content = get_sms_content(
-                    settings.OTP_SMS_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
-                )
-        elif settings.IS_PRODUCTION:
-            random_otp = rand_pass(settings.OTP_LENGTH)
-        else:
-            random_otp = "45612"
-
-        otp_obj = PatientMobileOTP(phone_number=data.phone_number, otp=random_otp)
-        otp_obj.save()
+        data = OTPRequestBaseSpec(**request.data)
+        error_response = send_otp(data.phone_number, purpose=OTPType.login)
+        if error_response:
+            return error_response
         return Response({"otp": "generated"})
 
     @extend_schema(
@@ -91,7 +105,7 @@ def send(self, request):
     @action(detail=False, methods=["POST"])
     def login(self, request):
         data = OTPLoginSpec(**request.data)
-        otp_object = PatientMobileOTP.objects.filter(
+        otp_object = MobileOTP.objects.filter(
             phone_number=data.phone_number, otp=data.otp, is_used=False
         ).first()
         if not otp_object:

care/facility/migrations/0485_rename_patientmobileotp_mobileotp.py

@@ -0,0 +1,17 @@
+# Generated by Django 6.0 on 2026-05-05 17:41
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('facility', '0484_remove_facility_discount_codes_and_more'),
+    ]
+
+    operations = [
+        migrations.RenameModel(
+            old_name='PatientMobileOTP',
+            new_name='MobileOTP',
+        ),
+    ]

care/facility/models/patient.py

@@ -4,7 +4,7 @@
 from care.utils.models.validators import mobile_or_landline_number_validator
 
 
-class PatientMobileOTP(BaseModel):
+class MobileOTP(BaseModel):
     is_used = models.BooleanField(default=False)
     phone_number = models.CharField(
         max_length=14, validators=[mobile_or_landline_number_validator]

care/users/api/otp_viewset/reset_password.py

@@ -0,0 +1,100 @@
+import logging
+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 Field
+from rest_framework.decorators import action
+from rest_framework.exceptions import ValidationError
+from rest_framework.response import Response
+
+from care.emr.api.otp_viewsets.login import OTPRequestBaseSpec, OTPType, send_otp
+from care.emr.api.viewsets.base import EMRBaseViewSet
+from care.facility.models.patient import MobileOTP
+from care.users.models import User
+from config.ratelimit import ratelimit
+
+logger = logging.getLogger(__name__)
+
+
+class OTPResetSendSpec(OTPRequestBaseSpec):
+    pass
+
+
+class OTPResetConfirmSpec(OTPRequestBaseSpec):
+    otp: str = Field(min_length=settings.OTP_LENGTH, max_length=settings.OTP_LENGTH)
+    password: str = Field(min_length=8)
+
+
+class OTPResetPasswordView(EMRBaseViewSet):
+    authentication_classes = []
+    permission_classes = []
+
+    @action(detail=False, methods=["POST"])
+    @extend_schema(request=OTPResetSendSpec)
+    def send(self, request):
+        data = OTPResetSendSpec(**request.data)
+
+        if ratelimit(request, "otp-password-reset", ["ip"]):
+            error_message = "Too many requests. Please try again later."
+            return Response(
+                {"detail": error_message},
+                status=429,
+            )
+
+        if not User.objects.filter(phone_number=data.phone_number).exists():
+            return Response({"otp": "generated"})
+
+        error_response = send_otp(
+            data.phone_number,
+            purpose=OTPType.reset_password,
+        )
+        if error_response:
+            return error_response
+        return Response({"otp": "generated"})
+
+    @action(detail=False, methods=["POST"])
+    @extend_schema(request=OTPResetConfirmSpec)
+    def confirm(self, request):
+        data = OTPResetConfirmSpec(**request.data)
+        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 = (
+            MobileOTP.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()
+        MobileOTP.objects.filter(
+            phone_number=data.phone_number,
+        ).delete()
+        return Response({"message": "Password reset successful"})

config/api_router.py

@@ -109,6 +109,7 @@
 from care.emr.api.viewsets.valueset import ValueSetViewSet
 from care.security.api.viewsets.permissions import PermissionViewSet
 from care.security.api.viewsets.roles import RoleViewSet
+from care.users.api.otp_viewset.reset_password import OTPResetPasswordView
 from care.users.api.viewsets.plug_config import PlugConfigViewset
 
 router = DefaultRouter() if settings.DEBUG else SimpleRouter()
@@ -123,6 +124,9 @@
 router.register("meta_artifacts", MetaArtifactViewSet, basename="meta_artifacts")
 
 router.register("otp", OTPLoginView, basename="otp-login")
+router.register(
+    "otp/password_reset", OTPResetPasswordView, basename="otp-password-reset"
+)
 
 router.register("otp/patient", PatientOTPView, basename="otp-patient")
 

config/settings/base.py

@@ -690,7 +690,15 @@
 
 SMS_BACKEND = "care.utils.sms.backend.sns.SnsBackend"
 
-OTP_SMS_TEMPLATE_PATH = env("OTP_SMS_TEMPLATE", default="sms/otp_sms.txt")
+OTP_SMS_CONTENT = env(
+    "OTP_SMS_CONTENT",
+    default="Care OTP for login is {otp}. Please do not share this with anyone.",
+)
+
+OTP_SMS_RESET_PASSWORD_CONTENT = env(
+    "OTP_SMS_RESET_PASSWORD_CONTENT",
+    default="Care OTP for password reset is {otp}. Please do not share this with anyone.",
+)
 
 USER_CREATE_PASSWORD_EMAIL_TEMPLATE_PATH = env(
     "USER_CREATE_PASSWORD_TEMPLATE_PATH", default="email/user_create_password.html"