first commit

apps/user_auth/admin.py

@@ -0,0 +1,14 @@
+from django.contrib import admin
+from apps.user_auth.models import SmsOtp as SmsOtpModel, JwtToken as JwtTokenModel
+# Register your models here.
+
+
+@admin.register(SmsOtpModel)
+class SmsOtpAdmin(admin.ModelAdmin):
+    pass
+
+
+@admin.register(JwtTokenModel)
+class JwtTokenAdmin(admin.ModelAdmin):
+    pass
+

apps/user_auth/apps.py

@@ -0,0 +1,6 @@
+from django.apps import AppConfig
+
+
+class AuthConfig(AppConfig):
+    default_auto_field = 'django.db.models.BigAutoField'
+    name = 'apps.user_auth'

apps/user_auth/auth_service.py

@@ -0,0 +1,64 @@
+import datetime
+from abc import ABC, abstractmethod
+from utils.services.sms import SmsProviderABC
+from django.db import models
+from django.contrib.auth import authenticate
+from typing import Protocol, Any
+import random
+
+
+class DjangoAuthProtocol(Protocol):
+    def authenticate(self, username:str, password:str) -> Any:
+        pass
+
+
+class SmsOtpService:
+
+        def __init__(self, sms_provider: SmsProviderABC):
+            self.provider = sms_provider
+
+        def send_code(self, mobile: str, otp_code: int):
+            message = f'کد ورود شما در باشگاه مجریان لوله گستر گلپایگان: {otp_code} @mojrian.loolegostar.com #{otp_code}'
+            return self.provider.send_sms(mobile, message)
+
+
+class AuthenticatorABC(ABC):
+    @abstractmethod
+    def authenticate(self) -> bool:
+        pass
+
+
+class SMSAuth(AuthenticatorABC):
+
+    def __init__(self, mobile: str, otp_code: int, otp_model: models.Model):
+        self.mobile = int(mobile)
+        self.otp_code = otp_code
+        self.otp_model = otp_model
+
+    def authenticate(self) -> models.Model | bool:
+        # check code is in DB
+        try:
+            datetime_now = datetime.datetime.now()
+            otp_obj = self.otp_model.objects.get(
+                mobile=self.mobile, otp=self.otp_code,
+                expire__gte=datetime_now
+            )
+        except Exception as exp:
+            self.otp_model.objects.filter(mobile=self.mobile).delete()
+            return False
+
+        return otp_obj
+
+
+class UsernameAuth(AuthenticatorABC):
+
+    def __init__(self, username: str, password: str, auth_class: DjangoAuthProtocol):
+        self.username = username
+        self.password = password
+        self.auth_class = auth_class
+
+    def authenticate(self) -> bool:
+        user = self.auth_class.authenticate(username=self.username, password=self.password)
+        if user:
+            return True
+

apps/user_auth/migrations/0001_initial.py

@@ -0,0 +1,47 @@
+# Generated by Django 4.2.1 on 2024-03-30 19:59
+
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='SmsOtp',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('created_at', models.DateTimeField(auto_now_add=True)),
+                ('updated_at', models.DateTimeField(auto_now=True)),
+                ('mobile', models.BigIntegerField()),
+                ('otp', models.IntegerField()),
+                ('expire', models.DateTimeField(blank=True, null=True)),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+        migrations.CreateModel(
+            name='JwtToken',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('created_at', models.DateTimeField(auto_now_add=True)),
+                ('updated_at', models.DateTimeField(auto_now=True)),
+                ('refresh', models.CharField(max_length=500)),
+                ('access', models.CharField(blank=True, max_length=500, null=True)),
+                ('expire', models.DateTimeField(blank=True, null=True)),
+                ('is_blocked', models.BooleanField(default=False)),
+                ('cuser', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
+            ],
+            options={
+                'abstract': False,
+            },
+        ),
+    ]

apps/user_auth/models.py

@@ -0,0 +1,27 @@
+from django.contrib.auth import get_user_model
+from django.db import models
+from utils.base_model import BaseModel
+
+# Create your models here.
+
+CUserModel = get_user_model()
+
+
+class JwtToken(BaseModel):
+    refresh = models.CharField(max_length=500, null=False, blank=False)
+    access = models.CharField(max_length=500, null=True, blank=True)
+    expire = models.DateTimeField(null=True, blank=True)
+    cuser = models.ForeignKey(CUserModel, null=False, blank=False, on_delete=models.CASCADE)
+    is_blocked = models.BooleanField(default=False)
+
+    def __str__(self):
+        return f'{self.cuser}'
+
+
+class SmsOtp(BaseModel):
+    mobile = models.BigIntegerField(null=False, blank=False)
+    otp = models.IntegerField(null=False, blank=False)
+    expire = models.DateTimeField(null=True, blank=True)
+
+    def __str__(self):
+        return f'{self.mobile} - {self.otp}'

apps/user_auth/services.py

@@ -0,0 +1,91 @@
+from datetime import datetime, timedelta
+from django.contrib.auth import get_user_model
+from rest_framework import serializers
+from apps.user_auth.auth_service import SmsOtpService
+from utils.services.sms import DefaultSmsProvider
+from apps.user_auth.models import SmsOtp as SmsOtpModel
+from apps.user_auth.auth_service import SMSAuth
+from apps.user_auth.token_service import get_tokens_for_user
+from rest_framework import exceptions
+from django.db import transaction
+from apps.user_auth.models import JwtToken as JwtTokenModel
+from django.contrib.auth import authenticate
+from django.conf import settings
+from typing import Annotated
+import random
+
+
+CUserModel = get_user_model()
+
+
+def login_by_user_pass(username: str, password: str):
+    user_obj = authenticate(username=username, password=password)
+
+    if user_obj:
+        try:
+            token = get_tokens_for_user(user_obj)
+            JwtTokenModel.objects.create(
+                refresh=token['refresh'],
+                access=token['access'],
+                cuser=user_obj,
+            )
+        except Exception as exp:
+            raise exceptions.ValidationError({"error": str(exp)})
+        return token
+    else:
+        raise exceptions.ValidationError({"error": "invalid user or password"})
+
+
+@transaction.atomic
+def send_otp_code(user_mobile: int) -> bool:
+    try:
+        CUserModel.objects.get(mobile=int(user_mobile))
+    except CUserModel.DoesNotExist:
+        raise exceptions.ValidationError({'error': 'User not found'})
+
+    default_sms_provider = DefaultSmsProvider()
+    otp_service = SmsOtpService(sms_provider=default_sms_provider)
+
+    if settings.OTP_BYPASS:
+        random_otp = 123456
+    else:
+        random_otp = random.randint(100000, 999999)
+        otp_is_send = otp_service.send_code(mobile=str(user_mobile), otp_code=random_otp)
+        if not otp_is_send:
+            raise exceptions.ValidationError({'error': 'Cannot Send Otp Code'})
+
+    # remove old otps
+    SmsOtpModel.objects.filter(mobile=str(user_mobile)).delete()
+
+    # save otp to database
+    expire_time = datetime.now() + timedelta(minutes=4)
+    otp_obj = SmsOtpModel.objects.create(
+        mobile=str(user_mobile), otp=random_otp,
+        expire=expire_time,
+    )
+
+    return True
+
+
+def sms_auth(mobile, otp_code: int) -> dict | None:
+
+    sms_auth = SMSAuth(
+        mobile=str(mobile), otp_code=otp_code,
+        otp_model=SmsOtpModel,
+    )
+    otp_obj = sms_auth.authenticate()
+
+    if otp_obj:
+        try:
+            user_obj = CUserModel.objects.get(mobile=otp_obj.mobile)
+            token = get_tokens_for_user(user_obj)
+            JwtTokenModel.objects.create(
+                refresh=token['refresh'],
+                access=token['access'],
+                cuser=user_obj,
+            )
+        except Exception as exp:
+            raise exceptions.ValidationError({"error": str(exp)})
+        return token
+    else:
+        raise exceptions.ValidationError({"error": "invalid otp code or mobile"})

apps/user_auth/tests.py

@@ -0,0 +1,3 @@
+from django.test import TestCase
+
+# Create your tests here.

apps/user_auth/token_service.py

@@ -0,0 +1,48 @@
+from rest_framework_simplejwt.tokens import RefreshToken, AccessToken, Token, TokenError
+from datetime import datetime, timedelta
+from django.contrib.auth import get_user_model
+from rest_framework import exceptions
+
+from apps.user_auth.models import JwtToken as JwtTokenModel
+
+
+
+CUserModel = get_user_model()
+
+
+def get_tokens_for_user(user):
+    token = RefreshToken.for_user(user)
+    return {
+        'refresh': str(token),
+        'access': str(token.access_token),
+    }
+
+
+def get_refresh_token(ref_token: str) -> dict:
+
+    try:
+        user_token = RefreshToken(token=ref_token)
+    except TokenError as exp:
+        JwtTokenModel.objects.filter(refresh=ref_token).delete()
+        raise exceptions.ValidationError({'error': str(exp)})
+
+    if not user_token:
+        raise exceptions.ValidationError({'error': 'Invalid Token in Verify'})
+
+    try:
+        jwt_token = JwtTokenModel.objects.get(refresh=ref_token)
+    except JwtTokenModel.DoesNotExist:
+        raise exceptions.ValidationError({'error': 'Invalid Token'})
+
+    if jwt_token.is_blocked:
+        raise exceptions.ValidationError({'error': 'Token is blocked'})
+
+    token = get_tokens_for_user(jwt_token.cuser)
+    JwtTokenModel.objects.create(
+        refresh=token['refresh'],
+        access=token['access'],
+        cuser=jwt_token.cuser,
+    )
+    # remove old token
+    jwt_token.delete()
+    return token

apps/user_auth/urls.py

@@ -0,0 +1,11 @@
+from django.urls import path
+from apps.user_auth.views import (MojriByMobileApiView, MojriMobileVerificationApiView,
+                                  CustomRefreshToken, LoginAdmin)
+
+
+urlpatterns = [
+    path('login-admin/', LoginAdmin.as_view(), name='login-admin'),
+    path('mojri/mobile/', MojriByMobileApiView.as_view(), name='mojri-mobile'),
+    path('mojri/mobile/verify/', MojriMobileVerificationApiView.as_view(), name='mojri-mobile-verify'),
+    path('refresh-token/', CustomRefreshToken.as_view(), name='refresh-token')
+]

apps/user_auth/views.py

@@ -0,0 +1,81 @@
+from datetime import datetime, timedelta
+from django.contrib.auth import get_user_model
+from django.shortcuts import render
+from rest_framework.views import APIView
+from rest_framework import serializers
+from apps.user_auth.auth_service import SMSAuth
+from rest_framework.response import Response
+from apps.user_auth.models import SmsOtp as SmsOtpModel
+from apps.user_auth.services import send_otp_code, sms_auth, login_by_user_pass
+from apps.user_auth.token_service import get_refresh_token
+from rest_framework import status
+
+from rest_framework import exceptions
+CUserModel = get_user_model()
+
+
+class MojriByMobileApiView(APIView):
+
+    class InputSerializer(serializers.Serializer):
+        mobile = serializers.IntegerField()
+
+    def post(self, request):
+
+        serializer = self.InputSerializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+
+        user_mobile = serializer.validated_data.get('mobile', None)
+
+        send_otp_code(user_mobile=user_mobile)
+
+        return Response({'status': 'otp code send'}, status=status.HTTP_200_OK)
+
+
+class MojriMobileVerificationApiView(APIView):
+
+    class InputSerializer(serializers.Serializer):
+        mobile = serializers.IntegerField()
+        otp_code = serializers.IntegerField()
+
+    def post(self, request):
+
+        serializer = self.InputSerializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+
+        token = sms_auth(**serializer.validated_data)
+
+        return Response(token)
+
+
+class LoginAdmin(APIView):
+
+    class InputSerializers(serializers.Serializer):
+        username = serializers.CharField(required=True)
+        password = serializers.CharField(required=True)
+
+    def post(self, request):
+        serializer = self.InputSerializers(data=request.data)
+        serializer.is_valid(raise_exception=True)
+
+        token = login_by_user_pass(**serializer.validated_data)
+
+        return Response(token, status=status.HTTP_200_OK)
+
+class CustomRefreshToken(APIView):
+
+        class InputSerializer(serializers.Serializer):
+            refresh = serializers.CharField()
+
+        def post(self, request):
+
+            serializer = self.InputSerializer(data=request.data)
+            serializer.is_valid(raise_exception=True)
+
+            ref_token = serializer.validated_data.get('refresh')
+
+            token = get_refresh_token(ref_token=ref_token)
+
+            return Response(token)
+
+
+