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

.env.example

@@ -291,6 +291,10 @@ FACILITY_S3_BUCKET=facility-bucket
 # Default: False (True in staging/production)
 # USE_SMS=False
 
+# SMS backend dotted path (string)
+# Default: "care.utils.sms.backend.sns.SnsBackend"
+# SMS_BACKEND=care.utils.sms.backend.sns.SnsBackend
+
 # AWS SNS access key for SMS (string)
 # Default: ""
 # SNS_ACCESS_KEY=
@@ -311,6 +315,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,63 +4,54 @@
 from datetime import timedelta
 
 from django.conf import settings
-from django.utils import timezone
+from django.db.models import Sum
 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, OTPVerifyLock
 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__)
 
 
-class BaseOTPType:
-    def render_content(self, otp: str) -> str:
-        pass
+def generate_otp(size):
+    return "".join(secrets.choice(string.digits) for _ in range(size))
 
 
-class LoginOTP(BaseOTPType):
+class BaseOTPType:
     @classmethod
     def render_content(cls, otp: str) -> str:
-        return settings.OTP_SMS_LOGIN_CONTENT.format(otp=otp)
+        raise NotImplementedError
 
+    @classmethod
+    def send_window(cls) -> timedelta:
+        raise NotImplementedError
 
-def rand_pass(size):
-    return "".join(secrets.choice(string.digits) for _ in range(size))
+    @classmethod
+    def max_sends(cls) -> int:
+        raise NotImplementedError
 
 
-def send_otp(phone_number, otp_type: BaseOTPType):
-    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 ValueError("Max Retries has exceeded")
+class LoginOTP(BaseOTPType):
+    @classmethod
+    def render_content(cls, otp: str) -> str:
+        return settings.OTP_SMS_LOGIN_CONTENT.format(otp=otp)
 
-    random_otp = ""
-    if settings.USE_SMS:
-        random_otp = rand_pass(settings.OTP_LENGTH)
-        try:
-            content = otp_type.render_content(random_otp)
-            sms.send_text_message(
-                content=content,
-                recipients=[phone_number],
-            )
-        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)
-    else:
-        random_otp = "45612"
+    @classmethod
+    def send_window(cls) -> timedelta:
+        return timedelta(minutes=settings.OTP_SEND_WINDOW_MINUTES)
 
-    MobileOTP.objects.create(phone_number=phone_number, otp=random_otp)
+    @classmethod
+    def max_sends(cls) -> int:
+        return settings.OTP_MAX_SENDS_PER_WINDOW
 
 
 class OTPRequestBaseSpec(BaseModel):
@@ -81,6 +72,53 @@ class OTPLoginSpec(OTPRequestBaseSpec):
     otp: str = Field(min_length=settings.OTP_LENGTH, max_length=settings.OTP_LENGTH)
 
 
+def failure_count(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
+
+
+def send_otp(phone_number, otp_type: type[BaseOTPType]):
+    with OTPSendLock(phone_number):
+        if failure_count(phone_number) >= settings.OTP_MAX_FAILURES:
+            raise Throttled(detail="Too many failed login attempts. Try again later.")
+
+        sent_otps = MobileOTP.objects.filter(
+            created_date__gte=care_now() - otp_type.send_window(),
+            phone_number=phone_number,
+        )
+        if sent_otps.count() >= otp_type.max_sends():
+            raise ValidationError({"phone_number": "Max Retries has exceeded"})
+
+        otp_value = (
+            generate_otp(settings.OTP_LENGTH) if settings.IS_PRODUCTION else "45612"
+        )
+        if settings.USE_SMS:
+            try:
+                content = otp_type.render_content(otp_value)
+                sms.send_text_message(
+                    content=content,
+                    recipients=[phone_number],
+                )
+            except Exception as e:
+                logger.error(e)
+                raise ValidationError(
+                    {"error": "Error while sending OTP. Contact admin."}
+                ) from e
+        elif settings.IS_PRODUCTION:
+            raise APIException("SMS Backend not configured")
+
+        # disable all other existing otp before creating a new one
+        MobileOTP.objects.filter(phone_number=phone_number, is_used=False).update(
+            is_used=True
+        )
+        MobileOTP.objects.create(phone_number=phone_number, otp=otp_value)
+
+
 class OTPLoginView(EMRBaseViewSet):
     authentication_classes = []
     permission_classes = []
@@ -91,12 +129,7 @@ class OTPLoginView(EMRBaseViewSet):
     @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)
+        send_otp(data.phone_number, otp_type=LoginOTP)
         return Response({"otp": "generated"})
 
     @extend_schema(
@@ -105,16 +138,42 @@ 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
+        expired = False
+        with OTPVerifyLock(data.phone_number):
+            if failure_count(data.phone_number) >= settings.OTP_MAX_FAILURES:
+                raise Throttled(
+                    detail="Too many failed login attempts. Try again later."
+                )
+
+            otp_object = (
+                MobileOTP.objects.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,15 @@
+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
+
+
+class OTPVerifyLock(Lock):
+    def __init__(self, phone_number, timeout=settings.LOCK_TIMEOUT):
+        self.key = f"lock:otp_verify:{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():
+    """
+    Hard-deletes MobileOTP rows older than the lockout window
+    """
+    cutoff = care_now() - timedelta(minutes=settings.OTP_LOCKOUT_MINUTES)
+    count, _ = MobileOTP.objects.filter(modified_date__lt=cutoff).delete()
+    logger.info("Deleted %d expired OTP rows", count)