From 48d0ecba79b248689c67ad86f67a601f94e427ac Mon Sep 17 00:00:00 2001 From: javad Date: Fri, 9 Oct 2026 10:07:31 +0330 Subject: [PATCH] fix login and jwt --- apps/jwt_auth/services.py | 16 ++++++ apps/jwt_auth/test_customer_login.py | 83 ++++++++++++++++++++++++++++ apps/jwt_auth/test_regressions.py | 4 +- apps/jwt_auth/views.py | 4 +- 4 files changed, 104 insertions(+), 3 deletions(-) create mode 100644 apps/jwt_auth/test_customer_login.py diff --git a/apps/jwt_auth/services.py b/apps/jwt_auth/services.py index abcb33b..6a83cf2 100644 --- a/apps/jwt_auth/services.py +++ b/apps/jwt_auth/services.py @@ -13,6 +13,14 @@ from utils.services.sms import DefaultSmsProvider from utils.validations.mobile import normalize_mobile CUser = get_user_model() +SMS_RESEND_INTERVAL = 60 + + +class SmsRequestThrottled(exceptions.Throttled): + def __init__(self, wait): + super().__init__(wait=wait) + self.detail = {'detail': 'کد ورود قبلاً ارسال شده است. کمی صبر کنید و کد قبلی را وارد کنید.', + 'retry_after': wait} class SmsDeliveryUnavailable(exceptions.APIException): @@ -77,6 +85,14 @@ def login_by_mobile(mobile: int): # Serialize code requests for this account, retaining the old valid code if # the provider refuses the new message. CUser.objects.select_for_update().get(pk=user.pk) + # Recheck eligibility after waiting for concurrent requests/administrative edits. + user, _ = mobile_user(user.mobile) + current = SmsOpt.objects.filter(mobile=user.mobile, expire__gt=timezone.now()).order_by('-created_at').first() + if current: + remaining = SMS_RESEND_INTERVAL - (timezone.now() - current.created_at).total_seconds() + if remaining > 0: + from math import ceil + raise SmsRequestThrottled(ceil(remaining)) code = secrets.randbelow(900000) + 100000 if not DefaultSmsProvider().send_sms(str(user.mobile), f'کد ورود شما: {code}'): raise SmsDeliveryUnavailable() diff --git a/apps/jwt_auth/test_customer_login.py b/apps/jwt_auth/test_customer_login.py new file mode 100644 index 0000000..d08e606 --- /dev/null +++ b/apps/jwt_auth/test_customer_login.py @@ -0,0 +1,83 @@ +"""Customer OTP journey, repeat submits, and recoverable failure contracts.""" +from datetime import timedelta +from unittest.mock import patch + +from django.utils import timezone + +from apps.cuser.models import SmsOpt +from apps.jwt_auth.test_regressions import RegressionFixtures + + +class CustomerLoginTests(RegressionFixtures): + def test_complete_customer_journey_login_refresh_and_logout(self): + with patch('apps.jwt_auth.services.DefaultSmsProvider.send_sms', return_value=True) as send: + sent = self.post('login/mobile/', {'mobile': '۰۹۱۲۳۴۵۶۷۸۹'}) + self.assertEqual(sent.status_code, 200) + self.assertEqual(sent.data['result']['mobile'], self.user.mobile) + self.assertEqual(sent.data['result']['retry_after'], 60) + send.assert_called_once() + code = SmsOpt.objects.get(mobile=self.user.mobile).opt + options = self.post('login/mobile/verify/', {'mobile': self.user.mobile, 'opt': code}) + self.assertEqual(options.status_code, 200) + self.assertEqual(len(options.data), 1) + self.assertEqual(options.data[0]['cuser_role']['role_type']['role_name_en'], 'workspace_customer') + tokens = self.post('token/', {'uuid_opt': options.data[0]['uuid']}) + self.assertEqual(tokens.status_code, 200) + self.client.credentials(HTTP_AUTHORIZATION='Bearer ' + tokens.data['access']) + self.assertEqual(self.client.get('/cuser/info/').status_code, 200) + rotated = self.post('token/refresh/', {'refresh': tokens.data['refresh']}) + self.assertEqual(rotated.status_code, 200) + self.assertNotEqual(rotated.data['refresh'], tokens.data['refresh']) + self.assertEqual(self.post('logout/', {'refresh': tokens.data['refresh']}).status_code, 204) + self.client.credentials(HTTP_AUTHORIZATION='Bearer ' + rotated.data['access']) + self.assertEqual(self.client.get('/cuser/info/').status_code, 401) + self.assertEqual(self.post('token/refresh/', {'refresh': rotated.data['refresh']}).status_code, 401) + + def test_repeat_send_does_not_invalidate_code_or_send_another_sms(self): + with patch('apps.jwt_auth.services.DefaultSmsProvider.send_sms', return_value=True) as send: + self.assertEqual(self.post('login/mobile/', {'mobile': self.user.mobile}).status_code, 200) + original = SmsOpt.objects.get(mobile=self.user.mobile) + for _ in range(3): + repeated = self.post('login/mobile/', {'mobile': self.user.mobile}) + self.assertEqual(repeated.status_code, 429) + self.assertIn('قبلاً ارسال', repeated.data['detail']) + self.assertGreater(int(repeated['Retry-After']), 0) + self.assertEqual(SmsOpt.objects.get().pk, original.pk) + self.assertEqual(SmsOpt.objects.get().opt, original.opt) + send.assert_called_once() + verified = self.post('login/mobile/verify/', {'mobile': self.user.mobile, 'opt': original.opt}) + self.assertEqual(verified.status_code, 200) + + def test_resend_after_interval_replaces_code_once(self): + original = self.otp() + SmsOpt.objects.filter(pk=original.pk).update(created_at=timezone.now() - timedelta(seconds=61)) + with patch('apps.jwt_auth.services.DefaultSmsProvider.send_sms', return_value=True) as send: + response = self.post('login/mobile/', {'mobile': self.user.mobile}) + self.assertEqual(response.status_code, 200) + self.assertFalse(SmsOpt.objects.filter(pk=original.pk).exists()) + self.assertEqual(SmsOpt.objects.filter(mobile=self.user.mobile).count(), 1) + send.assert_called_once() + + def test_expired_code_does_not_prevent_requesting_a_new_one(self): + self.otp(expired=True) + with patch('apps.jwt_auth.services.DefaultSmsProvider.send_sms', return_value=True): + self.assertEqual(self.post('login/mobile/', {'mobile': self.user.mobile}).status_code, 200) + + def test_panel_disabled_between_send_and_verify_has_clear_error(self): + code = self.otp() + self.role.customer_panel = False + self.role.save(update_fields=['customer_panel']) + response = self.post('login/mobile/verify/', {'mobile': self.user.mobile, 'opt': code.opt}) + self.assertEqual(response.status_code, 403) + self.assertIn('پنل مشتریان', str(response.data)) + + def test_double_verify_and_double_token_exchange_cannot_create_multiple_sessions(self): + from apps.jwt_auth.models import JwtToken + code = self.otp() + first = self.post('login/mobile/verify/', {'mobile': self.user.mobile, 'opt': code.opt}) + self.assertEqual(first.status_code, 200) + self.assertEqual(self.post('login/mobile/verify/', {'mobile': self.user.mobile, 'opt': code.opt}).status_code, 400) + option = first.data[0]['uuid'] + self.assertEqual(self.post('token/', {'uuid_opt': option}).status_code, 200) + self.assertEqual(self.post('token/', {'uuid_opt': option}).status_code, 401) + self.assertEqual(JwtToken.objects.count(), 1) diff --git a/apps/jwt_auth/test_regressions.py b/apps/jwt_auth/test_regressions.py index f2c4570..0defed5 100644 --- a/apps/jwt_auth/test_regressions.py +++ b/apps/jwt_auth/test_regressions.py @@ -80,6 +80,7 @@ class AuthRegressionTests(RegressionFixtures): def test_sms_failure_has_503_and_preserves_old_otp(self): old = self.otp() + SmsOpt.objects.filter(pk=old.pk).update(created_at=timezone.now() - timedelta(minutes=2)) with patch('apps.jwt_auth.services.DefaultSmsProvider.send_sms', return_value=False): response = self.post('login/mobile/', {'mobile': self.user.mobile}) self.assertEqual(response.status_code, 503) @@ -87,7 +88,8 @@ class AuthRegressionTests(RegressionFixtures): self.assertTrue(SmsOpt.objects.filter(pk=old.pk).exists()) def test_successful_sms_saves_only_one_code(self): - self.otp() + old = self.otp() + SmsOpt.objects.filter(pk=old.pk).update(created_at=timezone.now() - timedelta(minutes=2)) with patch('apps.jwt_auth.services.DefaultSmsProvider.send_sms', return_value=True): response = self.post('login/mobile/', {'mobile': '+98 912-345-6789'}) self.assertEqual(response.status_code, 200) diff --git a/apps/jwt_auth/views.py b/apps/jwt_auth/views.py index 51abd35..29344bb 100644 --- a/apps/jwt_auth/views.py +++ b/apps/jwt_auth/views.py @@ -9,7 +9,7 @@ from rest_framework_simplejwt.tokens import RefreshToken from apps.cuser.models import CUser, CuserRole, RoleType from apps.jwt_auth.models import JwtToken, CUserRoleLoginOpt -from apps.jwt_auth.services import login_by_username, login_by_mobile, verify_mobile_and_get_token +from apps.jwt_auth.services import login_by_username, login_by_mobile, verify_mobile_and_get_token, SMS_RESEND_INTERVAL from apps.jwt_auth.session import issue_tokens, validate_session_claims from apps.workspace.models import Workspace @@ -70,7 +70,7 @@ class LoginByMobileView(PublicAuthView): serializer = self.InputSerializer(data=request.data) serializer.is_valid(raise_exception=True) mobile = login_by_mobile(**serializer.validated_data) - return Response({'result': {'mobile': mobile, 'message': 'opt code send'}}) + return Response({'result': {'mobile': mobile, 'message': 'opt code send', 'retry_after': SMS_RESEND_INTERVAL}}) class LoginByUserView(PublicAuthView):