mohd/Salon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

I proceeded with Phase 1 step 1 (phone-first auth + OTP hardening) and updated the plan context. Here’s the current state and the next steps.

Browse Source 
What I implemented

Phone-first auth endpoints with OTP → JWT issuance.
views.py
urls.py
Phone normalization (KSA-focused, E.164 or Saudi mobile) and validation.
phone.py
serializers.py
OTP protections: rate limit, resend cooldown, attempt counting.
otp.py
models.py
Email is now optional to allow phone-only users.
models.py
0002_phone_auth_fields.py
Admin OTP visibility improved.
admin.py
Risks updated.
risks.md
This commit is contained in:
mohammad
2026-02-27 15:57:09 +03:00
parent dc68ecfe4c
commit 46af911a06
12 changed files with 292 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/apps/accounts/admin.py
+11 -2
View File
@@ -28,6 +28,15 @@ class UserAdmin(DjangoUserAdmin):
@admin.register(PhoneOTP)
class PhoneOTPAdmin(admin.ModelAdmin):
list_display = ("phone_number", "channel", "provider", "created_at", "expires_at", "verified_at")
list_filter = ("channel", "provider")
list_display = (
"phone_number",
"channel",
"purpose",
"provider",
"created_at",
"expires_at",
"verified_at",
"attempt_count",
)
list_filter = ("channel", "purpose", "provider")
search_fields = ("phone_number",)
backend/apps/accounts/migrations/0002_phone_auth_fields.py
+34
View File
@@ -0,0 +1,34 @@
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
("accounts", "0001_initial"),
]
operations = [
migrations.AlterField(
model_name="user",
name="email",
field=models.EmailField(blank=True, max_length=254, null=True, unique=True),
),
migrations.AddField(
model_name="phoneotp",
name="attempt_count",
field=models.PositiveSmallIntegerField(default=0),
),
migrations.AddField(
model_name="phoneotp",
name="max_attempts",
field=models.PositiveSmallIntegerField(default=5),
),
migrations.AddField(
model_name="phoneotp",
name="purpose",
field=models.CharField(
choices=[("auth", "Authentication"), ("verify", "Phone Verification")],
default="auth",
max_length=20,
),
),
]
backend/apps/accounts/models.py
+20 -7
View File
@@ -15,12 +15,17 @@ class UserRole(models.TextChoices):
class UserManager(BaseUserManager):
def create_user(self, email, password=None, **extra_fields):
if not email:
raise ValueError("Email is required")
email = self.normalize_email(email)
def create_user(self, email=None, password=None, **extra_fields):
phone_number = extra_fields.get("phone_number")
if not email and not phone_number:
raise ValueError("Email or phone number is required")
if email:
email = self.normalize_email(email)
user = self.model(email=email, **extra_fields)
user.set_password(password)
if password:
user.set_password(password)
else:
user.set_unusable_password()
user.save(using=self._db)
return user
@@ -36,7 +41,7 @@ class UserManager(BaseUserManager):
class User(AbstractBaseUser, PermissionsMixin):
email = models.EmailField(unique=True)
email = models.EmailField(unique=True, null=True, blank=True)
phone_number = models.CharField(max_length=20, unique=True, null=True, blank=True)
role = models.CharField(max_length=20, choices=UserRole.choices, default=UserRole.CUSTOMER)
first_name = models.CharField(max_length=150, blank=True)
@@ -52,7 +57,7 @@ class User(AbstractBaseUser, PermissionsMixin):
USERNAME_FIELD = "email"
def __str__(self):
return self.email
return self.email or self.phone_number or str(self.id)
class OtpChannel(models.TextChoices):
@@ -60,15 +65,23 @@ class OtpChannel(models.TextChoices):
WHATSAPP = "whatsapp", "WhatsApp"
class OtpPurpose(models.TextChoices):
AUTH = "auth", "Authentication"
VERIFY = "verify", "Phone Verification"
class PhoneOTP(models.Model):
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
phone_number = models.CharField(max_length=20)
channel = models.CharField(max_length=20, choices=OtpChannel.choices)
purpose = models.CharField(max_length=20, choices=OtpPurpose.choices, default=OtpPurpose.AUTH)
provider = models.CharField(max_length=50)
code_hash = models.CharField(max_length=128)
created_at = models.DateTimeField(auto_now_add=True)
expires_at = models.DateTimeField()
verified_at = models.DateTimeField(null=True, blank=True)
attempt_count = models.PositiveSmallIntegerField(default=0)
max_attempts = models.PositiveSmallIntegerField(default=5)
def is_expired(self):
return timezone.now() >= self.expires_at
backend/apps/accounts/serializers.py
+34
View File
@@ -2,6 +2,7 @@ from django.contrib.auth import get_user_model
from rest_framework import serializers
from apps.accounts.models import OtpChannel, PhoneOTP
from apps.accounts.services.phone import normalize_phone_number
User = get_user_model()
@@ -20,6 +21,14 @@ class RegisterSerializer(serializers.ModelSerializer):
model = User
fields = ["email", "password", "phone_number", "first_name", "last_name"]
def validate_phone_number(self, value):
if not value:
return value
try:
return normalize_phone_number(value)
except ValueError as exc:
raise serializers.ValidationError(str(exc)) from exc
def create(self, validated_data):
return User.objects.create_user(**validated_data)
@@ -28,12 +37,37 @@ class OTPRequestSerializer(serializers.Serializer):
phone_number = serializers.CharField(max_length=20)
channel = serializers.ChoiceField(choices=OtpChannel.choices)
def validate_phone_number(self, value):
try:
return normalize_phone_number(value)
except ValueError as exc:
raise serializers.ValidationError(str(exc)) from exc
class OTPVerifySerializer(serializers.Serializer):
request_id = serializers.UUIDField()
code = serializers.CharField(max_length=6)
class PhoneAuthRequestSerializer(serializers.Serializer):
phone_number = serializers.CharField(max_length=20)
channel = serializers.ChoiceField(choices=OtpChannel.choices)
email = serializers.EmailField(required=False, allow_null=True, allow_blank=True)
first_name = serializers.CharField(required=False, allow_blank=True)
last_name = serializers.CharField(required=False, allow_blank=True)
def validate_phone_number(self, value):
try:
return normalize_phone_number(value)
except ValueError as exc:
raise serializers.ValidationError(str(exc)) from exc
class PhoneAuthVerifySerializer(serializers.Serializer):
request_id = serializers.UUIDField()
code = serializers.CharField(max_length=6)
class OTPStatusSerializer(serializers.ModelSerializer):
class Meta:
model = PhoneOTP
backend/apps/accounts/services/otp.py
+47 -3
View File
@@ -1,13 +1,14 @@
import logging
import os
import secrets
from datetime import timedelta
from dataclasses import dataclass
from django.conf import settings
from django.contrib.auth.hashers import check_password, make_password
from django.utils import timezone
from apps.accounts.models import OtpChannel, PhoneOTP
from apps.accounts.models import OtpChannel, OtpPurpose, PhoneOTP
logger = logging.getLogger(__name__)
@@ -18,6 +19,18 @@ class OtpSendResult:
expires_at: str
class OtpRateLimitError(RuntimeError):
def __init__(self, retry_after_seconds: int):
super().__init__("Too many OTP requests. Try again later.")
self.retry_after_seconds = retry_after_seconds
class OtpCooldownError(RuntimeError):
def __init__(self, retry_after_seconds: int):
super().__init__("Please wait before requesting another code.")
self.retry_after_seconds = retry_after_seconds
class BaseOtpProvider:
def send_sms(self, to_number: str, message: str) -> None:
raise NotImplementedError
@@ -97,12 +110,38 @@ def generate_code(length: int = 6) -> str:
return "".join(secrets.choice(digits) for _ in range(length))
def create_and_send_otp(phone_number: str, channel: str) -> OtpSendResult:
def create_and_send_otp(phone_number: str, channel: str, purpose: str = OtpPurpose.AUTH) -> OtpSendResult:
provider = get_provider()
now = timezone.now()
window_minutes = getattr(settings, "OTP_WINDOW_MINUTES", 15)
max_per_window = getattr(settings, "OTP_MAX_PER_WINDOW", 5)
cooldown_seconds = getattr(settings, "OTP_RESEND_COOLDOWN_SECONDS", 60)
window_start = now - timedelta(minutes=window_minutes)
recent_qs = PhoneOTP.objects.filter(phone_number=phone_number, created_at__gte=window_start)
if recent_qs.count() >= max_per_window:
oldest_recent = recent_qs.order_by("created_at").first()
if oldest_recent:
retry_after = int((oldest_recent.created_at + timedelta(minutes=window_minutes) - now).total_seconds())
else:
retry_after = cooldown_seconds
raise OtpRateLimitError(retry_after_seconds=max(retry_after, cooldown_seconds))
latest = (
PhoneOTP.objects.filter(phone_number=phone_number, channel=channel)
.order_by("-created_at")
.first()
)
if latest and latest.verified_at is None:
elapsed = (now - latest.created_at).total_seconds()
if elapsed < cooldown_seconds:
raise OtpCooldownError(retry_after_seconds=int(cooldown_seconds - elapsed))
code = generate_code()
otp = PhoneOTP.objects.create(
phone_number=phone_number,
channel=channel,
purpose=purpose,
provider=settings.OTP_PROVIDER,
code_hash=make_password(code),
expires_at=PhoneOTP.expiry_at(),
@@ -122,8 +161,13 @@ def create_and_send_otp(phone_number: str, channel: str) -> OtpSendResult:
def verify_otp(otp: PhoneOTP, code: str) -> bool:
if otp.verified_at or otp.is_expired():
return False
otp.attempt_count += 1
if otp.attempt_count > otp.max_attempts:
otp.save(update_fields=["attempt_count"])
return False
if not check_password(code, otp.code_hash):
otp.save(update_fields=["attempt_count"])
return False
otp.verified_at = timezone.now()
otp.save(update_fields=["verified_at"])
otp.save(update_fields=["verified_at", "attempt_count"])
return True
backend/apps/accounts/services/phone.py
+26
View File
@@ -0,0 +1,26 @@
import re
def normalize_phone_number(raw_phone: str) -> str:
if not raw_phone:
raise ValueError("Phone number is required")
phone = re.sub(r"[\s\-\(\)]", "", raw_phone)
if phone.startswith("00"):
phone = "+" + phone[2:]
if phone.startswith("+"):
digits = phone[1:]
if not digits.isdigit() or not (8 <= len(digits) <= 15):
raise ValueError("Invalid phone number format")
return "+" + digits
digits = re.sub(r"\D", "", phone)
if digits.startswith("0") and len(digits) == 10 and digits[1] == "5":
return "+966" + digits[1:]
if digits.startswith("5") and len(digits) == 9:
return "+966" + digits
raise ValueError("Phone number must be in E.164 format or a valid Saudi mobile")
backend/apps/accounts/urls.py
+11 -1
View File
@@ -1,7 +1,15 @@
from django.urls import path
from rest_framework_simplejwt.views import TokenObtainPairView, TokenRefreshView
from apps.accounts.views import MeView, OTPRequestView, OTPVerifyView, RegisterView, SocialLoginPlaceholderView
from apps.accounts.views import (
MeView,
OTPRequestView,
OTPVerifyView,
PhoneAuthRequestView,
PhoneAuthVerifyView,
RegisterView,
SocialLoginPlaceholderView,
)
urlpatterns = [
path("register/", RegisterView.as_view(), name="register"),
@@ -10,5 +18,7 @@ urlpatterns = [
path("token/refresh/", TokenRefreshView.as_view(), name="token_refresh"),
path("otp/request/", OTPRequestView.as_view(), name="otp_request"),
path("otp/verify/", OTPVerifyView.as_view(), name="otp_verify"),
path("phone/request/", PhoneAuthRequestView.as_view(), name="phone_auth_request"),
path("phone/verify/", PhoneAuthVerifyView.as_view(), name="phone_auth_verify"),
path("social/<str:provider>/", SocialLoginPlaceholderView.as_view(), name="social_login"),
]
backend/apps/accounts/views.py
+96 -3
View File
@@ -3,15 +3,23 @@ from django.shortcuts import get_object_or_404
from rest_framework import generics, permissions, status
from rest_framework.response import Response
from rest_framework.views import APIView
from rest_framework_simplejwt.tokens import RefreshToken
from apps.accounts.models import PhoneOTP
from apps.accounts.models import OtpPurpose, PhoneOTP
from apps.accounts.serializers import (
OTPRequestSerializer,
OTPVerifySerializer,
PhoneAuthRequestSerializer,
PhoneAuthVerifySerializer,
RegisterSerializer,
UserSerializer,
)
from apps.accounts.services.otp import create_and_send_otp, verify_otp
from apps.accounts.services.otp import (
OtpCooldownError,
OtpRateLimitError,
create_and_send_otp,
verify_otp,
)
User = get_user_model()
@@ -36,7 +44,18 @@ class OTPRequestView(APIView):
serializer = OTPRequestSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
data = serializer.validated_data
result = create_and_send_otp(data["phone_number"], data["channel"])
try:
result = create_and_send_otp(data["phone_number"], data["channel"], purpose=OtpPurpose.VERIFY)
except OtpCooldownError as exc:
return Response(
{"detail": str(exc), "retry_after_seconds": exc.retry_after_seconds},
status=status.HTTP_429_TOO_MANY_REQUESTS,
)
except OtpRateLimitError as exc:
return Response(
{"detail": str(exc), "retry_after_seconds": exc.retry_after_seconds},
status=status.HTTP_429_TOO_MANY_REQUESTS,
)
return Response(
{"request_id": result.request_id, "expires_at": result.expires_at},
status=status.HTTP_201_CREATED,
@@ -62,6 +81,80 @@ class OTPVerifyView(APIView):
return Response({"detail": "Phone verified"}, status=status.HTTP_200_OK)
class PhoneAuthRequestView(APIView):
permission_classes = [permissions.AllowAny]
def post(self, request):
serializer = PhoneAuthRequestSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
data = serializer.validated_data
phone_number = data["phone_number"]
email = data.get("email") or None
user = User.objects.filter(phone_number=phone_number).first()
if not user:
if email and User.objects.filter(email=email).exists():
return Response(
{"detail": "Email already in use."},
status=status.HTTP_400_BAD_REQUEST,
)
user = User.objects.create_user(
email=email,
phone_number=phone_number,
first_name=data.get("first_name", ""),
last_name=data.get("last_name", ""),
role="customer",
)
try:
result = create_and_send_otp(phone_number, data["channel"], purpose=OtpPurpose.AUTH)
except OtpCooldownError as exc:
return Response(
{"detail": str(exc), "retry_after_seconds": exc.retry_after_seconds},
status=status.HTTP_429_TOO_MANY_REQUESTS,
)
except OtpRateLimitError as exc:
return Response(
{"detail": str(exc), "retry_after_seconds": exc.retry_after_seconds},
status=status.HTTP_429_TOO_MANY_REQUESTS,
)
return Response(
{"request_id": result.request_id, "expires_at": result.expires_at},
status=status.HTTP_201_CREATED,
)
class PhoneAuthVerifyView(APIView):
permission_classes = [permissions.AllowAny]
def post(self, request):
serializer = PhoneAuthVerifySerializer(data=request.data)
serializer.is_valid(raise_exception=True)
data = serializer.validated_data
otp = get_object_or_404(PhoneOTP, id=data["request_id"])
if not verify_otp(otp, data["code"]):
return Response({"detail": "Invalid or expired code"}, status=status.HTTP_400_BAD_REQUEST)
user = User.objects.filter(phone_number=otp.phone_number).first()
if not user:
return Response({"detail": "User not found"}, status=status.HTTP_404_NOT_FOUND)
if not user.is_phone_verified:
user.is_phone_verified = True
user.save(update_fields=["is_phone_verified"])
refresh = RefreshToken.for_user(user)
return Response(
{
"refresh": str(refresh),
"access": str(refresh.access_token),
"user": UserSerializer(user).data,
},
status=status.HTTP_200_OK,
)
class SocialLoginPlaceholderView(APIView):
permission_classes = [permissions.AllowAny]