62 lines
1.8 KiB
Python
62 lines
1.8 KiB
Python
from django.conf import settings
|
|
from django.core.cache import cache
|
|
from django.utils.translation import gettext_lazy as _
|
|
|
|
from ..models import OTPCode
|
|
from .sms import SMSError, get_sms_backend
|
|
|
|
|
|
class OTPCooldownError(Exception):
|
|
def __init__(self, retry_after: int):
|
|
self.retry_after = retry_after
|
|
super().__init__(f"Wait {retry_after} seconds before requesting another code")
|
|
|
|
|
|
class OTPVerificationError(Exception):
|
|
pass
|
|
|
|
|
|
def _cooldown_key(phone: str) -> str:
|
|
return f"otp:cooldown:{phone}"
|
|
|
|
|
|
def send_otp(phone_number: str) -> OTPCode:
|
|
cooldown = getattr(settings, "OTP_SEND_COOLDOWN_SECONDS", 60)
|
|
if cache.get(_cooldown_key(phone_number)):
|
|
raise OTPCooldownError(retry_after=cooldown)
|
|
|
|
ttl_seconds = getattr(settings, "OTP_TTL_SECONDS", 120)
|
|
length = getattr(settings, "OTP_CODE_LENGTH", 4)
|
|
|
|
otp = OTPCode.issue(phone_number=phone_number, ttl_seconds=ttl_seconds, length=length)
|
|
|
|
try:
|
|
get_sms_backend().send_otp(phone_number, otp.code)
|
|
except SMSError:
|
|
otp.delete()
|
|
raise
|
|
|
|
cache.set(_cooldown_key(phone_number), "1", timeout=cooldown)
|
|
return otp
|
|
|
|
|
|
def verify_otp(phone_number: str, code: str) -> OTPCode:
|
|
otp = (
|
|
OTPCode.objects.filter(phone_number=phone_number, is_used=False)
|
|
.order_by("-created_at")
|
|
.first()
|
|
)
|
|
if otp is None:
|
|
raise OTPVerificationError(_("No active code for this phone number"))
|
|
if otp.is_expired:
|
|
raise OTPVerificationError(_("Code has expired"))
|
|
if otp.attempts >= OTPCode.MAX_ATTEMPTS:
|
|
raise OTPVerificationError(_("Too many attempts; request a new code"))
|
|
|
|
if otp.code != str(code):
|
|
OTPCode.objects.filter(pk=otp.pk).update(attempts=otp.attempts + 1)
|
|
raise OTPVerificationError(_("Invalid code"))
|
|
|
|
otp.mark_consumed()
|
|
return otp
|