[ENG-158] feat: Abuse protection for OTP based login

.env.example

@@ -311,6 +311,34 @@ FACILITY_S3_BUCKET=facility-bucket
 # Default: "sms/otp_sms.txt"
 # OTP_SMS_TEMPLATE=sms/otp_sms.txt
 
+# ============================================================================
+# OTP RATE LIMITING
+# ============================================================================
+
+# Time window (in minutes) for tracking OTP request limits
+# Default: 60
+# OTP_SEND_WINDOW_MINUTES=60
+
+# Maximum number of OTPs that can be generated within OTP_SEND_WINDOW_MINUTES
+# Default: 10
+# OTP_MAX_SENDS_PER_WINDOW=10
+
+# Number of failed verification attempts allowed before an OTP is invalidated
+# Default: 3
+# OTP_MAX_VERIFY_ATTEMPTS=3
+
+# Maximum total failures allowed before the phone number is restricted
+# Default: 5
+# OTP_MAX_FAILURES=5
+
+# Duration (in minutes) the account remains locked after reaching maximum failures
+# Default: 60
+# OTP_LOCKOUT_MINUTES=60
+
+# Duration (in minutes) an OTP remains valid for verification after it is generated
+# Default: 10
+# OTP_VALIDITY_MINUTES=10
+
 # ============================================================================
 # EMAIL TEMPLATES
 # ============================================================================

care/emr/api/otp_viewsets/login.py

@@ -4,25 +4,34 @@
 from datetime import timedelta
 
 from django.conf import settings
+from django.db import transaction
+from django.db.models import Sum
 from django.utils import timezone
 from drf_spectacular.utils import extend_schema
 from pydantic import BaseModel, Field, field_validator
 from rest_framework.decorators import action
-from rest_framework.exceptions import ValidationError
+from rest_framework.exceptions import APIException, Throttled, ValidationError
 from rest_framework.response import Response
 
 from care.emr.api.viewsets.base import EMRBaseViewSet
+from care.emr.locks.otp import OTPSendLock
 from care.facility.models.patient import MobileOTP
 from care.utils import sms
 from care.utils.models.validators import mobile_validator
+from care.utils.time_util import care_now
 from config.patient_otp_token import PatientToken
 
 logger = logging.getLogger(__name__)
 
 
+def generate_otp(size):
+    return "".join(secrets.choice(string.digits) for _ in range(size))
+
+
 class BaseOTPType:
-    def render_content(self, otp: str) -> str:
-        pass
+    @classmethod
+    def render_content(cls, otp: str) -> str:
+        raise NotImplementedError
 
 
 class LoginOTP(BaseOTPType):
@@ -31,10 +40,6 @@ def render_content(cls, otp: str) -> str:
         return settings.OTP_SMS_LOGIN_CONTENT.format(otp=otp)
 
 
-def rand_pass(size):
-    return "".join(secrets.choice(string.digits) for _ in range(size))
-
-
 def send_otp(phone_number, otp_type: BaseOTPType):
     sent_otps = MobileOTP.objects.filter(
         created_date__gte=(timezone.now() - timedelta(settings.OTP_REPEAT_WINDOW)),
@@ -46,7 +51,7 @@ def send_otp(phone_number, otp_type: BaseOTPType):
 
     random_otp = ""
     if settings.USE_SMS:
-        random_otp = rand_pass(settings.OTP_LENGTH)
+        random_otp = generate_otp(settings.OTP_LENGTH)
         try:
             content = otp_type.render_content(random_otp)
             sms.send_text_message(
@@ -56,7 +61,7 @@ def send_otp(phone_number, otp_type: BaseOTPType):
         except Exception as e:
             raise Exception("Error while sending OTP. Contact admin.") from e
     elif settings.IS_PRODUCTION:
-        random_otp = rand_pass(settings.OTP_LENGTH)
+        random_otp = generate_otp(settings.OTP_LENGTH)
     else:
         random_otp = "45612"
 
@@ -85,18 +90,60 @@ class OTPLoginView(EMRBaseViewSet):
     authentication_classes = []
     permission_classes = []
 
+    def failure_count(self, phone_number: str) -> int:
+        since = care_now() - timedelta(minutes=settings.OTP_LOCKOUT_MINUTES)
+        total = MobileOTP.objects.filter(
+            phone_number=phone_number,
+            modified_date__gte=since,
+            failed_attempts__gt=0,
+        ).aggregate(total=Sum("failed_attempts"))["total"]
+        return total or 0
+
     @extend_schema(
         request=OTPRequestBaseSpec,
     )
     @action(detail=False, methods=["POST"])
     def send(self, request):
         data = OTPRequestBaseSpec(**request.data)
-        try:
-            send_otp(data.phone_number, otp_type=LoginOTP)
-        except ValueError as e:
-            raise ValidationError({"phone_number": "Unable to send OTP"}) from e
-        except Exception:
-            return Response({"error": "Unable to send OTP"}, status=400)
+
+        if self.failure_count(data.phone_number) >= settings.OTP_MAX_FAILURES:
+            raise Throttled(detail="Too many failed login attempts. Try again later.")
+
+        with OTPSendLock(data.phone_number):
+            sent_otps = MobileOTP.objects.filter(
+                created_date__gte=(
+                    care_now() - timedelta(minutes=settings.OTP_SEND_WINDOW_MINUTES)
+                ),
+                phone_number=data.phone_number,
+            )
+            if sent_otps.count() >= settings.OTP_MAX_SENDS_PER_WINDOW:
+                raise ValidationError({"phone_number": "Max Retries has exceeded"})
+
+            random_otp = ""
+            if settings.USE_SMS:
+                random_otp = generate_otp(settings.OTP_LENGTH)
+                try:
+                    content = LoginOTP.render_content(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 not settings.IS_PRODUCTION:
+                random_otp = "45612"
+            else:
+                raise APIException("SMS Backend not configured")
+
+            # disable all other existing otp before creating a new one
+            MobileOTP.objects.filter(
+                phone_number=data.phone_number, is_used=False
+            ).update(is_used=True)
+            MobileOTP.objects.create(phone_number=data.phone_number, otp=random_otp)
         return Response({"otp": "generated"})
 
     @extend_schema(
@@ -105,16 +152,41 @@ def send(self, request):
     @action(detail=False, methods=["POST"])
     def login(self, request):
         data = OTPLoginSpec(**request.data)
-        otp_object = MobileOTP.objects.filter(
-            phone_number=data.phone_number, otp=data.otp, is_used=False
-        ).first()
-        if not otp_object:
-            raise ValidationError({"otp": "Invalid OTP"})
-
-        otp_object.is_used = True
-        otp_object.save()
 
-        token = PatientToken()
-        token["phone_number"] = data.phone_number
+        if self.failure_count(data.phone_number) >= settings.OTP_MAX_FAILURES:
+            raise Throttled(detail="Too many failed login attempts. Try again later.")
+
+        expired = False
+        with transaction.atomic():
+            otp_object = (
+                MobileOTP.objects.select_for_update()
+                .filter(
+                    phone_number=data.phone_number,
+                    is_used=False,
+                    created_date__gte=care_now()
+                    - timedelta(minutes=settings.OTP_VALIDITY_MINUTES),
+                )
+                .order_by("-created_date")
+                .first()
+            )
 
-        return Response({"access": str(token)})
+            if otp_object:
+                if otp_object.otp == data.otp:
+                    otp_object.is_used = True
+                    otp_object.save(update_fields=["is_used", "modified_date"])
+                    token = PatientToken()
+                    token["phone_number"] = data.phone_number
+                    return Response({"access": str(token)})
+                otp_object.failed_attempts += 1
+                if otp_object.failed_attempts >= settings.OTP_MAX_VERIFY_ATTEMPTS:
+                    otp_object.is_used = True
+                    expired = True
+                otp_object.save(
+                    update_fields=["failed_attempts", "is_used", "modified_date"]
+                )
+
+        if expired:
+            raise ValidationError(
+                {"otp": "Too many wrong attempts. Please request a new OTP."}
+            )
+        raise ValidationError({"otp": "Invalid OTP"})

care/emr/locks/otp.py

@@ -0,0 +1,9 @@
+from django.conf import settings
+
+from care.utils.lock import Lock
+
+
+class OTPSendLock(Lock):
+    def __init__(self, phone_number, timeout=settings.LOCK_TIMEOUT):
+        self.key = f"lock:otp_send:{phone_number}"
+        self.timeout = timeout

care/emr/tasks/__init__.py

@@ -2,6 +2,7 @@
 from celery.schedules import crontab
 from django.conf import settings
 
+from care.emr.tasks.cleanup_expired_otps import cleanup_expired_otps
 from care.emr.tasks.cleanup_expired_token_slots import cleanup_expired_token_slots
 from care.emr.tasks.cleanup_incomplete_file_uploads import (
     cleanup_incomplete_file_uploads,
@@ -16,6 +17,12 @@ def setup_periodic_tasks(sender: Celery, **kwargs):
         name="cleanup_expired_token_slots",
     )
 
+    sender.add_periodic_task(
+        crontab(hour="0", minute="0"),
+        cleanup_expired_otps.s(),
+        name="cleanup_expired_otps",
+    )
+
     if cleanup_file_upload_hours := settings.FILE_UPLOAD_EXPIRY_HOURS:
         sender.add_periodic_task(
             cleanup_file_upload_hours * 3600,  # convert hours to seconds

care/emr/tasks/cleanup_expired_otps.py

@@ -0,0 +1,21 @@
+from datetime import timedelta
+from logging import Logger
+
+from celery import shared_task
+from celery.utils.log import get_task_logger
+from django.conf import settings
+
+from care.facility.models.patient import MobileOTP
+from care.utils.time_util import care_now
+
+logger: Logger = get_task_logger(__name__)
+
+
+@shared_task
+def cleanup_expired_otps():
+    """
+    Soft-deletes MobileOTP rows older than the lockout window
+    """
+    cutoff = care_now() - timedelta(minutes=settings.OTP_LOCKOUT_MINUTES)
+    count = MobileOTP.objects.filter(created_date__lt=cutoff).update(deleted=True)
+    logger.info("Soft-deleted %d expired OTP rows", count)