870 lines
43 KiB
Python
870 lines
43 KiB
Python
from unittest.mock import Mock, patch
|
||
|
||
from django.conf import settings
|
||
from django.core.cache import cache
|
||
from django.test import TestCase, TransactionTestCase, override_settings
|
||
from django.utils import timezone
|
||
from rest_framework.test import APIClient
|
||
from rest_framework.throttling import ScopedRateThrottle
|
||
|
||
from .models import OTPCode, Session, User
|
||
from .services import OTPDeliveryError, send_otp_sms
|
||
|
||
|
||
class SendOtpSmsTests(TestCase):
|
||
"""
|
||
تست‌های واحد برای accounts.services.send_otp_sms — تمرکز روی همون رفتاری
|
||
که این fix اضافه می‌کنه: در dev فقط لاگ، در پروداکشن یا واقعاً ارسال یا
|
||
خطای صریح (هیچ‌وقت موفقیتِ خاموش/ساختگی).
|
||
"""
|
||
|
||
@override_settings(DEBUG=True)
|
||
def test_dev_mode_does_not_call_provider(self):
|
||
with patch('accounts.services.requests.post') as mock_post:
|
||
send_otp_sms(phone='+989120000000', code='1234')
|
||
mock_post.assert_not_called()
|
||
|
||
@override_settings(DEBUG=False, KAVENEGAR_API_KEY='')
|
||
def test_production_without_api_key_raises_instead_of_silent_success(self):
|
||
with patch('accounts.services.requests.post') as mock_post:
|
||
with self.assertRaises(OTPDeliveryError):
|
||
send_otp_sms(phone='+989120000000', code='1234')
|
||
mock_post.assert_not_called()
|
||
|
||
@override_settings(DEBUG=False, KAVENEGAR_API_KEY='test-key', KAVENEGAR_SENDER='')
|
||
def test_production_success_calls_provider(self):
|
||
# پاسخِ موفقِ کاوه‌نگار: HTTP 200 + return.status == 200 در بدنه.
|
||
mock_response = Mock(
|
||
status_code=200,
|
||
text='{"return": {"status": 200, "message": "OK"}, "entries": []}',
|
||
)
|
||
mock_response.json.return_value = {
|
||
'return': {'status': 200, 'message': 'OK'},
|
||
'entries': [{'messageid': 1, 'status': 1}],
|
||
}
|
||
with patch('accounts.services.requests.post', return_value=mock_response) as mock_post:
|
||
send_otp_sms(phone='+989120000000', code='1234')
|
||
mock_post.assert_called_once()
|
||
args, kwargs = mock_post.call_args
|
||
self.assertIn('test-key', args[0])
|
||
self.assertEqual(kwargs['data']['receptor'], '+989120000000')
|
||
self.assertIn('1234', kwargs['data']['message'])
|
||
|
||
@override_settings(DEBUG=False, KAVENEGAR_API_KEY='test-key')
|
||
def test_production_provider_error_status_raises(self):
|
||
mock_response = Mock(status_code=400, text='bad request')
|
||
with patch('accounts.services.requests.post', return_value=mock_response):
|
||
with self.assertRaises(OTPDeliveryError):
|
||
send_otp_sms(phone='+989120000000', code='1234')
|
||
|
||
@override_settings(DEBUG=False, KAVENEGAR_API_KEY='test-key')
|
||
def test_production_http_200_with_application_level_rejection_raises(self):
|
||
# کاوه‌نگار حتی رد کردنِ اپلیکیشنی (مثلاً کلیدِ نامعتبر) رو با
|
||
# HTTP 200 برمی‌گردونه؛ return.status غیر از 200 باید شکست باشه.
|
||
mock_response = Mock(
|
||
status_code=200,
|
||
text='{"return": {"status": 401, "message": "Invalid API key"}, "entries": []}',
|
||
)
|
||
mock_response.json.return_value = {
|
||
'return': {'status': 401, 'message': 'Invalid API key'},
|
||
'entries': [],
|
||
}
|
||
with patch('accounts.services.requests.post', return_value=mock_response):
|
||
with self.assertRaises(OTPDeliveryError):
|
||
send_otp_sms(phone='+989120000000', code='1234')
|
||
|
||
@override_settings(DEBUG=False, KAVENEGAR_API_KEY='test-key')
|
||
def test_production_malformed_provider_body_raises(self):
|
||
# بدنه‌ی نامعتبر/غیرمنتظره (نه JSON، یا بدونِ ساختارِ return.status)
|
||
# باید شکستِ تحویل در نظر گرفته بشه، نه موفقیتِ خاموش.
|
||
mock_response = Mock(status_code=200, text='not json')
|
||
mock_response.json.side_effect = ValueError('No JSON object could be decoded')
|
||
with patch('accounts.services.requests.post', return_value=mock_response):
|
||
with self.assertRaises(OTPDeliveryError):
|
||
send_otp_sms(phone='+989120000000', code='1234')
|
||
|
||
@override_settings(DEBUG=False, KAVENEGAR_API_KEY='test-key')
|
||
def test_production_network_failure_raises(self):
|
||
import requests as requests_module
|
||
with patch('accounts.services.requests.post', side_effect=requests_module.ConnectionError('boom')):
|
||
with self.assertRaises(OTPDeliveryError):
|
||
send_otp_sms(phone='+989120000000', code='1234')
|
||
|
||
@override_settings(DEBUG=True)
|
||
def test_dev_mode_log_never_contains_full_phone_only_safe_suffix(self):
|
||
# 🔒 اثبات فیکسِ این دور: لاگِ OTP نباید هیچ‌وقت شماره‌تلفنِ کامل رو
|
||
# نشون بده؛ فقط ۴ رقمِ آخر (شناسه‌ی امن) باید در پیامِ لاگ باشه.
|
||
phone = '+989120000000'
|
||
with self.assertLogs('accounts.services', level='INFO') as captured:
|
||
send_otp_sms(phone=phone, code='1234')
|
||
|
||
self.assertEqual(len(captured.records), 1)
|
||
message = captured.records[0].getMessage()
|
||
|
||
self.assertNotIn(phone, message)
|
||
self.assertIn('phone_suffix=0000', message)
|
||
# رفتار/سطح/کدِ قبلی حفظ شده: کدِ OTP هیچ‌وقت لاگ نمی‌شه، فقط طولش.
|
||
self.assertNotIn('1234', message)
|
||
self.assertIn('code_len=4', message)
|
||
|
||
@override_settings(DEBUG=False, KAVENEGAR_API_KEY='test-key')
|
||
def test_provider_error_logs_never_contain_raw_response_body(self):
|
||
# 🔒 اثبات فیکسِ این دور: لاگ‌های خطای provider دیگه response.text
|
||
# خام رو شامل نمی‌شن؛ فقط context مفید و بی‌خطر (phone_suffix،
|
||
# status، provider_status) باقی می‌مونه.
|
||
|
||
# حالتِ ۱: HTTP غیرِ ۲۰۰.
|
||
mock_response = Mock(status_code=400, text='bad request body secret-ish')
|
||
with patch('accounts.services.requests.post', return_value=mock_response):
|
||
with self.assertLogs('accounts.services', level='ERROR') as captured:
|
||
with self.assertRaises(OTPDeliveryError):
|
||
send_otp_sms(phone='+989120000000', code='1234')
|
||
message = captured.records[0].getMessage()
|
||
self.assertNotIn('bad request body secret-ish', message)
|
||
self.assertIn('phone_suffix=0000', message)
|
||
self.assertIn('status=400', message)
|
||
|
||
# حالتِ ۲: بدنه‌ی نامعتبر/غیرقابلparse.
|
||
mock_response = Mock(status_code=200, text='not json body secret-ish')
|
||
mock_response.json.side_effect = ValueError('No JSON object could be decoded')
|
||
with patch('accounts.services.requests.post', return_value=mock_response):
|
||
with self.assertLogs('accounts.services', level='ERROR') as captured:
|
||
with self.assertRaises(OTPDeliveryError):
|
||
send_otp_sms(phone='+989120000000', code='1234')
|
||
message = captured.records[0].getMessage()
|
||
self.assertNotIn('not json body secret-ish', message)
|
||
self.assertIn('phone_suffix=0000', message)
|
||
self.assertIn('status=200', message)
|
||
|
||
# حالتِ ۳: ردِ اپلیکیشنی (HTTP 200، return.status != 200).
|
||
mock_response = Mock(
|
||
status_code=200,
|
||
text='{"return": {"status": 401, "message": "secret-ish detail"}, "entries": []}',
|
||
)
|
||
mock_response.json.return_value = {
|
||
'return': {'status': 401, 'message': 'secret-ish detail'},
|
||
'entries': [],
|
||
}
|
||
with patch('accounts.services.requests.post', return_value=mock_response):
|
||
with self.assertLogs('accounts.services', level='ERROR') as captured:
|
||
with self.assertRaises(OTPDeliveryError):
|
||
send_otp_sms(phone='+989120000000', code='1234')
|
||
message = captured.records[0].getMessage()
|
||
self.assertNotIn('secret-ish detail', message)
|
||
self.assertIn('phone_suffix=0000', message)
|
||
self.assertIn('provider_status=401', message)
|
||
|
||
|
||
|
||
class RequestOTPViewTests(TestCase):
|
||
"""
|
||
تست‌های سطح API برای RequestOTPView: مطمئن می‌شیم رفتارِ dev قبلی حفظ
|
||
شده و در پروداکشن، شکستِ تحویلِ پیامک باعثِ یک خطای صریح می‌شه (نه
|
||
success=True دروغین) در حالی که ردیفِ OTPCode همچنان ساخته شده.
|
||
"""
|
||
|
||
def setUp(self):
|
||
self.client = APIClient()
|
||
|
||
@override_settings(DEBUG=True)
|
||
def test_dev_mode_returns_otp_in_response_without_provider_call(self):
|
||
with patch('accounts.views.send_otp_sms') as mock_send:
|
||
response = self.client.post(
|
||
'/api/auth/otp/request', {'phone': '+989120000001'}, format='json',
|
||
)
|
||
self.assertEqual(response.status_code, 200)
|
||
self.assertTrue(response.data['success'])
|
||
self.assertTrue(response.data['devOtp'])
|
||
mock_send.assert_called_once()
|
||
self.assertTrue(OTPCode.objects.filter(phone='+989120000001').exists())
|
||
|
||
@override_settings(DEBUG=False, KAVENEGAR_API_KEY='')
|
||
def test_production_delivery_failure_does_not_report_success(self):
|
||
response = self.client.post(
|
||
'/api/auth/otp/request', {'phone': '+989120000002'}, format='json',
|
||
)
|
||
self.assertEqual(response.status_code, 503)
|
||
# OTP row is created before delivery is attempted; only the SMS leg fails.
|
||
self.assertTrue(OTPCode.objects.filter(phone='+989120000002').exists())
|
||
|
||
@override_settings(DEBUG=True)
|
||
def test_generated_otp_is_always_a_4_digit_string_with_leading_zeros(self):
|
||
# ⚠️ فیکسِ این تست: تولیدِ OTP از random.randint به secrets.randbelow
|
||
# تغییر کرده؛ این تست مطمئن می‌شه فرمتِ ۴ رقمی (با صفرهای ابتدایی
|
||
# حفظ‌شده، نه یک عددِ کوتاه‌تر) دست‌نخورده مونده.
|
||
with patch('accounts.views.secrets.randbelow', return_value=7):
|
||
with patch('accounts.views.send_otp_sms'):
|
||
response = self.client.post(
|
||
'/api/auth/otp/request', {'phone': '+989120000003'}, format='json',
|
||
)
|
||
self.assertEqual(response.status_code, 200)
|
||
self.assertEqual(response.data['devOtp'], '0007')
|
||
otp = OTPCode.objects.get(phone='+989120000003')
|
||
self.assertEqual(otp.code, '0007')
|
||
self.assertEqual(len(otp.code), 4)
|
||
self.assertTrue(otp.code.isdigit())
|
||
|
||
|
||
class RequestOTPStaleInvalidationTests(TestCase):
|
||
"""
|
||
تست‌های متمرکز روی فیکسِ stale OTP: قبل از ساختِ OTPCode جدید، تمامِ
|
||
OTPCode های is_used=False قبلیِ همون شماره باید invalid (is_used=True)
|
||
بشن — طوری که فقط آخرین کدِ صادرشده برای همون شماره برای ورود کار کنه.
|
||
"""
|
||
|
||
def setUp(self):
|
||
self.client = APIClient()
|
||
# throttle (ScopedRateThrottle) از همون cache پیش‌فرضِ جنگو استفاده
|
||
# می‌کنه؛ بین تست‌ها پاکش می‌کنیم تا شمارنده‌ی یک تست روی تستِ بعدی
|
||
# اثر نذاره.
|
||
cache.clear()
|
||
|
||
def _request_otp(self, phone, code):
|
||
# با patch کردنِ secrets.randbelow، کدِ هر درخواست معلوم و
|
||
# قابل‌پیش‌بینیه (به‌جای اتکا به شانسِ برخورد نکردنِ دو کدِ تصادفی).
|
||
with patch('accounts.views.secrets.randbelow', return_value=int(code)):
|
||
with patch('accounts.views.send_otp_sms'):
|
||
response = self.client.post(
|
||
'/api/auth/otp/request', {'phone': phone}, format='json',
|
||
)
|
||
self.assertEqual(response.status_code, 200)
|
||
self.assertEqual(response.data['devOtp'], code)
|
||
return code
|
||
|
||
# ⚠️ این کلاس چندین درخواست پشتِ‌سرِهم برای *همون* شماره می‌فرسته (بدونِ
|
||
# فاصله)، چون داره رفتارِ invalidation رو می‌سنجه، نه cooldown رو. پس
|
||
# cooldown جدیدِ بین‌درخواستی رو اینجا خاموش می‌کنیم تا این تست‌ها با
|
||
# اون تداخل نکنن؛ cooldown خودش تستِ اختصاصیِ جدا داره
|
||
# (RequestOTPPhoneCooldownTests).
|
||
@override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=0)
|
||
def test_second_request_invalidates_first_otp(self):
|
||
phone = '+989121110001'
|
||
first_code = self._request_otp(phone, '1111')
|
||
first_otp = OTPCode.objects.get(phone=phone, code=first_code)
|
||
self.assertFalse(first_otp.is_used)
|
||
|
||
self._request_otp(phone, '2222')
|
||
|
||
first_otp.refresh_from_db()
|
||
self.assertTrue(first_otp.is_used)
|
||
|
||
@override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=0)
|
||
def test_newest_otp_still_works_for_login(self):
|
||
phone = '+989121110002'
|
||
self._request_otp(phone, '1111')
|
||
second_code = self._request_otp(phone, '2222')
|
||
|
||
response = self.client.post(
|
||
'/api/auth/login', {'phone': phone, 'otp': second_code}, format='json',
|
||
)
|
||
self.assertEqual(response.status_code, 200)
|
||
|
||
@override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=0)
|
||
def test_older_otp_is_rejected_after_new_request(self):
|
||
phone = '+989121110003'
|
||
first_code = self._request_otp(phone, '1111')
|
||
self._request_otp(phone, '2222')
|
||
|
||
response = self.client.post(
|
||
'/api/auth/login', {'phone': phone, 'otp': first_code}, format='json',
|
||
)
|
||
self.assertEqual(response.status_code, 400)
|
||
|
||
|
||
class RefreshTokenReplayTests(TestCase):
|
||
"""
|
||
🔒 اثبات فیکسِ replay: یک refresh token معتبر باید فقط یک‌بار قابل
|
||
استفاده باشه. بعد از یک رفرشِ موفق، همون توکن دیگه نباید کار کنه —
|
||
حتی بدون rotation، مصرفِ همون توکن دوباره باید 401 بگیره.
|
||
"""
|
||
|
||
def setUp(self):
|
||
self.client = APIClient()
|
||
self.user = User.objects.create(is_guest=True, device_id='device-refresh-test')
|
||
self.session = Session.objects.create(
|
||
user=self.user,
|
||
refresh_token='test-refresh-token-abc123',
|
||
device_id='device-refresh-test',
|
||
expires_at=timezone.now() + timezone.timedelta(days=30),
|
||
revoked=False,
|
||
)
|
||
|
||
def test_valid_refresh_token_succeeds(self):
|
||
response = self.client.post(
|
||
'/api/auth/refresh', {'refreshToken': self.session.refresh_token}, format='json',
|
||
)
|
||
self.assertEqual(response.status_code, 200)
|
||
self.assertTrue(response.data['token'])
|
||
|
||
def test_same_refresh_token_rejected_on_second_use(self):
|
||
first_response = self.client.post(
|
||
'/api/auth/refresh', {'refreshToken': self.session.refresh_token}, format='json',
|
||
)
|
||
self.assertEqual(first_response.status_code, 200)
|
||
|
||
second_response = self.client.post(
|
||
'/api/auth/refresh', {'refreshToken': self.session.refresh_token}, format='json',
|
||
)
|
||
self.assertEqual(second_response.status_code, 401)
|
||
|
||
self.session.refresh_from_db()
|
||
self.assertTrue(self.session.revoked)
|
||
|
||
def test_successful_refresh_returns_new_refresh_token(self):
|
||
response = self.client.post(
|
||
'/api/auth/refresh', {'refreshToken': self.session.refresh_token}, format='json',
|
||
)
|
||
self.assertEqual(response.status_code, 200)
|
||
new_refresh_token = response.data['refreshToken']
|
||
self.assertTrue(new_refresh_token)
|
||
self.assertNotEqual(new_refresh_token, self.session.refresh_token)
|
||
self.assertTrue(
|
||
Session.objects.filter(
|
||
refresh_token=new_refresh_token, user=self.user, revoked=False,
|
||
).exists()
|
||
)
|
||
|
||
|
||
class LoginPhoneRateLimitTests(TestCase):
|
||
"""
|
||
🔒 اثبات محدودیتِ جدیدِ بر اساسِ شماره‌تلفن روی LoginView: تلاش‌های
|
||
تکراریِ ورود برای همون شماره باید بعد از حدِ مجاز بلاک بشن، حتی اگه
|
||
throttle_scope='otp_login' (IP-محور) رو دور بزنیم.
|
||
"""
|
||
|
||
def setUp(self):
|
||
self.client = APIClient()
|
||
cache.clear()
|
||
|
||
@override_settings(DEBUG=True, OTP_LOGIN_PHONE_MAX_ATTEMPTS=3)
|
||
def test_repeated_attempts_for_same_phone_are_blocked(self):
|
||
phone = '+989123334444'
|
||
|
||
for _ in range(3):
|
||
response = self.client.post(
|
||
'/api/auth/login', {'phone': phone, 'otp': 'wrong'}, format='json',
|
||
)
|
||
self.assertEqual(response.status_code, 400)
|
||
|
||
blocked_response = self.client.post(
|
||
'/api/auth/login', {'phone': phone, 'otp': 'wrong'}, format='json',
|
||
)
|
||
self.assertEqual(blocked_response.status_code, 429)
|
||
|
||
|
||
class RequestOTPPhoneRateLimitTests(TestCase):
|
||
"""
|
||
🔒 اثبات محدودیتِ جدیدِ بر اساسِ شماره‌تلفن روی RequestOTPView: درخواست‌های
|
||
تکراریِ صدورِ OTP برای همون شماره باید بعد از حدِ مجاز بلاک بشن، حتی اگه
|
||
throttle_scope='otp_request' (IP-محور) رو با تغییرِ IP دور بزنیم. همچنین
|
||
درخواستِ بلاک‌شده نباید OTPCode جدید بسازه یا پیامک بفرسته.
|
||
"""
|
||
|
||
def setUp(self):
|
||
self.client = APIClient()
|
||
cache.clear()
|
||
|
||
# cooldown بین‌درخواستیِ جدید رو اینجا خاموش می‌کنیم چون این تست‌ها دارن
|
||
# سقفِ *ساعتی* رو می‌سنجن (با چند درخواستِ پشتِ‌سرِهم بدونِ فاصله)، نه
|
||
# cooldown رو؛ cooldown تستِ اختصاصیِ خودش رو داره
|
||
# (RequestOTPPhoneCooldownTests).
|
||
@override_settings(DEBUG=True, OTP_REQUEST_PHONE_MAX_ATTEMPTS=3, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=0)
|
||
def test_repeated_requests_for_same_phone_are_eventually_blocked(self):
|
||
phone = '+989127778888'
|
||
|
||
# otp_request (IP-محور) پیش‌فرض 5/hour است؛ برای اینکه فقط
|
||
# محدودیتِ شماره‌تلفن رو بسنجیم، throttle IP-محور رو موقتاً باز
|
||
# می‌کنیم تا تداخلی با شمارشِ این تست نداشته باشه.
|
||
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'otp_request': '1000/hour'}):
|
||
with patch('accounts.views.send_otp_sms'):
|
||
for _ in range(3):
|
||
response = self.client.post(
|
||
'/api/auth/otp/request', {'phone': phone}, format='json',
|
||
)
|
||
self.assertEqual(response.status_code, 200)
|
||
|
||
blocked_response = self.client.post(
|
||
'/api/auth/otp/request', {'phone': phone}, format='json',
|
||
)
|
||
self.assertEqual(blocked_response.status_code, 429)
|
||
|
||
@override_settings(DEBUG=True, OTP_REQUEST_PHONE_MAX_ATTEMPTS=3, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=0)
|
||
def test_changing_client_ip_does_not_bypass_the_phone_limit(self):
|
||
phone = '+989127778889'
|
||
|
||
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'otp_request': '1000/hour'}):
|
||
with patch('accounts.views.send_otp_sms'):
|
||
for i in range(3):
|
||
# هر درخواست از یک IP متفاوت میاد؛ throttle IP-محور
|
||
# (که به‌هرحال بالا باز شده) اینجا اهمیتی نداره — نکته
|
||
# اینه که محدودیتِ شماره‌تلفن مستقل از IP باید اعمال بشه.
|
||
response = self.client.post(
|
||
'/api/auth/otp/request', {'phone': phone}, format='json',
|
||
REMOTE_ADDR=f'10.0.0.{i + 1}',
|
||
)
|
||
self.assertEqual(response.status_code, 200)
|
||
|
||
blocked_response = self.client.post(
|
||
'/api/auth/otp/request', {'phone': phone}, format='json',
|
||
REMOTE_ADDR='10.0.0.99',
|
||
)
|
||
self.assertEqual(blocked_response.status_code, 429)
|
||
|
||
@override_settings(DEBUG=True, OTP_REQUEST_PHONE_MAX_ATTEMPTS=1, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=0)
|
||
def test_throttled_request_creates_no_new_otp_and_sends_no_sms(self):
|
||
phone = '+989127778890'
|
||
|
||
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'otp_request': '1000/hour'}):
|
||
with patch('accounts.views.send_otp_sms') as mock_send:
|
||
first_response = self.client.post(
|
||
'/api/auth/otp/request', {'phone': phone}, format='json',
|
||
)
|
||
self.assertEqual(first_response.status_code, 200)
|
||
self.assertEqual(mock_send.call_count, 1)
|
||
|
||
otp_count_before = OTPCode.objects.filter(phone=phone).count()
|
||
|
||
blocked_response = self.client.post(
|
||
'/api/auth/otp/request', {'phone': phone}, format='json',
|
||
)
|
||
self.assertEqual(blocked_response.status_code, 429)
|
||
|
||
# نه OTPCode جدیدی ساخته شده، نه پیامکِ جدیدی ارسال شده.
|
||
self.assertEqual(mock_send.call_count, 1)
|
||
self.assertEqual(
|
||
OTPCode.objects.filter(phone=phone).count(), otp_count_before,
|
||
)
|
||
|
||
|
||
class RequestOTPPhoneRateLimitConcurrencyTests(TransactionTestCase):
|
||
"""
|
||
🔒 اثباتِ ایمنی محدودیتِ شماره‌تلفن زیرِ همزمانی: دو درخواستِ *واقعاً*
|
||
هم‌زمان (با threading.Barrier هم‌زمان‌شده، نه صرفاً پشتِ‌سرِهم) برای یک
|
||
شماره، با سقفِ مجازِ ۱، نباید هر دو موفق بشن. الگویِ cache.add که یک
|
||
عملیاتِ اتمیکِ سطحِ backend cache است (نه یک read-then-write ناامن)
|
||
دقیقاً همینو تضمین می‌کنه. از TransactionTestCase استفاده می‌شه چون
|
||
TestCase هر تست رو داخلِ یک تراکنشِ بیرونیِ خودش می‌پیچه که برای
|
||
درخواست‌هایِ واقعاً هم‌زمان از threadهایِ جدا مناسب نیست.
|
||
"""
|
||
|
||
def setUp(self):
|
||
cache.clear()
|
||
|
||
@override_settings(DEBUG=True, OTP_REQUEST_PHONE_MAX_ATTEMPTS=1)
|
||
def test_two_simultaneous_requests_do_not_both_pass_the_phone_limit(self):
|
||
import threading
|
||
|
||
phone = '+989127778891'
|
||
barrier = threading.Barrier(2)
|
||
results = []
|
||
results_lock = threading.Lock()
|
||
|
||
def _make_request():
|
||
client = APIClient()
|
||
barrier.wait() # هر دو thread دقیقاً هم‌زمان درخواست بفرستن
|
||
response = client.post(
|
||
'/api/auth/otp/request', {'phone': phone}, format='json',
|
||
)
|
||
with results_lock:
|
||
results.append(response.status_code)
|
||
|
||
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'otp_request': '1000/hour'}):
|
||
with patch('accounts.views.send_otp_sms'):
|
||
threads = [threading.Thread(target=_make_request) for _ in range(2)]
|
||
for t in threads:
|
||
t.start()
|
||
for t in threads:
|
||
t.join(timeout=10)
|
||
|
||
self.assertEqual(len(results), 2)
|
||
# دقیقاً یکی باید موفق بشه (200) و اون یکی باید بلاک بشه (429)؛
|
||
# هیچ‌وقت هر دو نباید 200 بگیرن.
|
||
self.assertEqual(sorted(results), [200, 429])
|
||
self.assertEqual(OTPCode.objects.filter(phone=phone).count(), 1)
|
||
|
||
|
||
class RequestOTPPhoneCooldownTests(TestCase):
|
||
"""
|
||
🔒 اثباتِ محدودیتِ جدیدِ cooldown بین‌درخواستی برای RequestOTPView: هدفِ
|
||
این محدودیت جلوگیریِ burst (چند درخواست برای همون شماره در عرضِ چند
|
||
ثانیه/بدونِ فاصله) است — چیزی که سقفِ ساعتیِ per-phone موجود
|
||
(OTP_REQUEST_PHONE_MAX_ATTEMPTS) به‌تنهایی جلوش رو نمی‌گیره، چون فقط
|
||
حجمِ کلی رو محدود می‌کنه نه فاصله‌ی بینِ تلاش‌ها رو.
|
||
"""
|
||
|
||
def setUp(self):
|
||
self.client = APIClient()
|
||
cache.clear()
|
||
|
||
@override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=60)
|
||
def test_immediate_resend_for_same_phone_is_blocked_by_cooldown(self):
|
||
phone = '+989127779001'
|
||
|
||
with patch('accounts.views.send_otp_sms'):
|
||
first_response = self.client.post(
|
||
'/api/auth/otp/request', {'phone': phone}, format='json',
|
||
)
|
||
self.assertEqual(first_response.status_code, 200)
|
||
|
||
second_response = self.client.post(
|
||
'/api/auth/otp/request', {'phone': phone}, format='json',
|
||
)
|
||
self.assertEqual(second_response.status_code, 429)
|
||
|
||
@override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=60, OTP_REQUEST_PHONE_MAX_ATTEMPTS=5)
|
||
def test_cooldown_blocks_burst_without_consuming_the_hourly_quota(self):
|
||
# پنج درخواستِ پشتِ‌سرِهم بدونِ فاصله برای همون شماره: فقط اولی باید
|
||
# موفق بشه، بقیه باید با cooldown (429) بلاک بشن — و چون
|
||
# درخواست‌هایِ بلاک‌شده توسطِ cooldown هیچ‌وقت به شمارنده‌ی ساعتی
|
||
# نمی‌رسن، سهمیه‌ی ساعتی مصرف نمی‌شه.
|
||
phone = '+989127779002'
|
||
|
||
# throttle IP-محورِ otp_request (پیش‌فرض 5/hour) رو موقتاً باز
|
||
# می‌کنیم تا اینجا فقط cooldown/سهمیه‌ی ساعتیِ per-phone سنجیده بشه،
|
||
# نه throttle IP-محورِ جدا (که تستِ اختصاصیِ خودش رو داره).
|
||
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'otp_request': '1000/hour'}):
|
||
with patch('accounts.views.send_otp_sms') as mock_send:
|
||
statuses = []
|
||
for _ in range(5):
|
||
response = self.client.post(
|
||
'/api/auth/otp/request', {'phone': phone}, format='json',
|
||
)
|
||
statuses.append(response.status_code)
|
||
|
||
self.assertEqual(statuses, [200, 429, 429, 429, 429])
|
||
mock_send.assert_called_once()
|
||
self.assertEqual(OTPCode.objects.filter(phone=phone).count(), 1)
|
||
|
||
# سهمیه‌ی ساعتی هنوز دست‌نخورده است: بعد از پایانِ cooldown،
|
||
# همچنان می‌شه تا سقفِ مجاز درخواست داد.
|
||
cache.delete(f'otp_request_phone_cooldown:{phone}')
|
||
with patch('accounts.views.send_otp_sms') as mock_send_again:
|
||
allowed_again = self.client.post(
|
||
'/api/auth/otp/request', {'phone': phone}, format='json',
|
||
)
|
||
self.assertEqual(allowed_again.status_code, 200)
|
||
mock_send_again.assert_called_once()
|
||
|
||
@override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=60)
|
||
def test_cooldown_blocked_request_creates_no_otp_and_sends_no_sms(self):
|
||
phone = '+989127779003'
|
||
|
||
with patch('accounts.views.send_otp_sms') as mock_send:
|
||
self.client.post('/api/auth/otp/request', {'phone': phone}, format='json')
|
||
otp_count_before = OTPCode.objects.filter(phone=phone).count()
|
||
|
||
blocked_response = self.client.post(
|
||
'/api/auth/otp/request', {'phone': phone}, format='json',
|
||
)
|
||
|
||
self.assertEqual(blocked_response.status_code, 429)
|
||
self.assertEqual(mock_send.call_count, 1)
|
||
self.assertEqual(OTPCode.objects.filter(phone=phone).count(), otp_count_before)
|
||
|
||
@override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=1)
|
||
def test_cooldown_expires_after_configured_window(self):
|
||
import time
|
||
|
||
phone = '+989127779004'
|
||
|
||
with patch('accounts.views.send_otp_sms'):
|
||
first_response = self.client.post(
|
||
'/api/auth/otp/request', {'phone': phone}, format='json',
|
||
)
|
||
self.assertEqual(first_response.status_code, 200)
|
||
|
||
time.sleep(1.2)
|
||
|
||
second_response = self.client.post(
|
||
'/api/auth/otp/request', {'phone': phone}, format='json',
|
||
)
|
||
self.assertEqual(second_response.status_code, 200)
|
||
|
||
@override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=60)
|
||
def test_cooldown_is_per_phone_not_global(self):
|
||
with patch('accounts.views.send_otp_sms'):
|
||
first_response = self.client.post(
|
||
'/api/auth/otp/request', {'phone': '+989127779005'}, format='json',
|
||
)
|
||
second_response = self.client.post(
|
||
'/api/auth/otp/request', {'phone': '+989127779006'}, format='json',
|
||
)
|
||
self.assertEqual(first_response.status_code, 200)
|
||
self.assertEqual(second_response.status_code, 200)
|
||
|
||
def test_default_cooldown_is_120_seconds(self):
|
||
# ⚠️ تصمیمِ محصول/امنیت: کاربر باید هر ۲ دقیقه بتونه OTP دوباره
|
||
# بگیره؛ این باید دقیقاً همون مقداریه که enforcement سمتِ سرور
|
||
# اعمال می‌کنه (نه صرفاً یک عددِ hardcodeشده در فرانت).
|
||
self.assertEqual(settings.OTP_REQUEST_PHONE_COOLDOWN_SECONDS, 120)
|
||
|
||
@override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=120)
|
||
def test_cooldown_response_reports_retry_after_header_and_body(self):
|
||
phone = '+989127779008'
|
||
|
||
with patch('accounts.views.send_otp_sms'):
|
||
first_response = self.client.post(
|
||
'/api/auth/otp/request', {'phone': phone}, format='json',
|
||
)
|
||
self.assertEqual(first_response.status_code, 200)
|
||
|
||
blocked_response = self.client.post(
|
||
'/api/auth/otp/request', {'phone': phone}, format='json',
|
||
)
|
||
|
||
self.assertEqual(blocked_response.status_code, 429)
|
||
# هدرِ استانداردِ Retry-After — فرانت باید countdown رو دقیقاً از
|
||
# همین مقدار شروع کنه، نه از یک عددِ ثابتِ سمتِ کلاینت.
|
||
self.assertIn('Retry-After', blocked_response.headers)
|
||
retry_after_header = int(blocked_response.headers['Retry-After'])
|
||
self.assertGreaterEqual(retry_after_header, 1)
|
||
self.assertLessEqual(retry_after_header, 120)
|
||
|
||
# همون مقدار در بدنه‌ی پاسخ هم برگردونده می‌شه (retryAfter، هم‌راستا
|
||
# با استایلِ camelCase بقیه‌ی فیلدهای response در این پروژه).
|
||
self.assertIn('retryAfter', blocked_response.data)
|
||
self.assertEqual(blocked_response.data['retryAfter'], retry_after_header)
|
||
|
||
@override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=120)
|
||
def test_retry_after_counts_down_as_time_passes(self):
|
||
import time
|
||
|
||
phone = '+989127779009'
|
||
|
||
with patch('accounts.views.send_otp_sms'):
|
||
self.client.post('/api/auth/otp/request', {'phone': phone}, format='json')
|
||
|
||
immediate_retry = self.client.post(
|
||
'/api/auth/otp/request', {'phone': phone}, format='json',
|
||
)
|
||
self.assertEqual(immediate_retry.status_code, 429)
|
||
immediate_retry_after = immediate_retry.data['retryAfter']
|
||
|
||
time.sleep(1.1)
|
||
|
||
later_retry = self.client.post(
|
||
'/api/auth/otp/request', {'phone': phone}, format='json',
|
||
)
|
||
self.assertEqual(later_retry.status_code, 429)
|
||
later_retry_after = later_retry.data['retryAfter']
|
||
|
||
# با گذشتِ زمان، عددِ retryAfter باید کم بشه (نه ثابت بمونه) — این
|
||
# همون چیزیه که فرانت برای countdown دقیق بهش نیاز داره.
|
||
self.assertLess(later_retry_after, immediate_retry_after)
|
||
|
||
|
||
class RequestOTPPhoneCooldownConcurrencyTests(TransactionTestCase):
|
||
"""
|
||
🔒 اثباتِ ایمنیِ cooldown زیرِ همزمانیِ واقعی (نه صرفاً پشتِ‌سرِهم):
|
||
دقیقاً همون الگویِ RequestOTPPhoneRateLimitConcurrencyTests، این‌بار
|
||
برایِ کلیدِ cooldown. از TransactionTestCase استفاده می‌شه چون
|
||
TestCase هر تست رو داخلِ یک تراکنشِ بیرونیِ خودش می‌پیچه که برایِ
|
||
درخواست‌هایِ واقعاً هم‌زمان از threadهایِ جدا مناسب نیست.
|
||
"""
|
||
|
||
def setUp(self):
|
||
cache.clear()
|
||
|
||
@override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=60)
|
||
def test_two_simultaneous_requests_for_same_phone_only_one_passes_cooldown(self):
|
||
import threading
|
||
|
||
phone = '+989127779007'
|
||
barrier = threading.Barrier(2)
|
||
results = []
|
||
results_lock = threading.Lock()
|
||
|
||
def _make_request():
|
||
client = APIClient()
|
||
barrier.wait()
|
||
response = client.post(
|
||
'/api/auth/otp/request', {'phone': phone}, format='json',
|
||
)
|
||
with results_lock:
|
||
results.append(response.status_code)
|
||
|
||
with patch('accounts.views.send_otp_sms'):
|
||
threads = [threading.Thread(target=_make_request) for _ in range(2)]
|
||
for t in threads:
|
||
t.start()
|
||
for t in threads:
|
||
t.join(timeout=10)
|
||
|
||
self.assertEqual(len(results), 2)
|
||
self.assertEqual(sorted(results), [200, 429])
|
||
self.assertEqual(OTPCode.objects.filter(phone=phone).count(), 1)
|
||
|
||
|
||
@override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=120)
|
||
def test_cooldown_is_cleared_when_sms_delivery_fails(self):
|
||
# ⚠️ فیکسِ باگ: قبلاً cooldown *قبل از* فراخوانیِ send_otp_sms ثبت
|
||
# می‌شد. اگه ارسالِ پیامک با OTPDeliveryError شکست می‌خورد، کاربر
|
||
# هیچ کدی دریافت نکرده بود ولی همچنان تا ۱۲۰ ثانیه بلاک می‌موند.
|
||
# این تست تضمین می‌کنه که با شکستِ ارسال، cooldown پاک می‌شه و
|
||
# کاربر می‌تونه فوراً دوباره تلاش کنه.
|
||
phone = '+989127779010'
|
||
|
||
with patch('accounts.views.send_otp_sms', side_effect=OTPDeliveryError('boom')):
|
||
first_response = self.client.post(
|
||
'/api/auth/otp/request', {'phone': phone}, format='json',
|
||
)
|
||
self.assertEqual(first_response.status_code, 503)
|
||
|
||
# اثباتِ اینکه cooldown دیگه فعال نیست: بلافاصله یک تلاشِ دیگه (این‌بار
|
||
# موفق) باید 429 نگیره.
|
||
with patch('accounts.views.send_otp_sms') as mock_send:
|
||
retry_response = self.client.post(
|
||
'/api/auth/otp/request', {'phone': phone}, format='json',
|
||
)
|
||
self.assertEqual(retry_response.status_code, 200)
|
||
mock_send.assert_called_once()
|
||
|
||
@override_settings(DEBUG=True, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=120)
|
||
def test_cooldown_still_active_after_successful_delivery(self):
|
||
# اثباتِ عدمِ رگرسیون: وقتی ارسال موفق باشه، cooldown همچنان طبقِ
|
||
# رفتارِ قبلی فعال می‌مونه (فقط شکستِ تحویل باید cooldown رو پاک کنه).
|
||
phone = '+989127779011'
|
||
|
||
with patch('accounts.views.send_otp_sms'):
|
||
first_response = self.client.post(
|
||
'/api/auth/otp/request', {'phone': phone}, format='json',
|
||
)
|
||
self.assertEqual(first_response.status_code, 200)
|
||
|
||
blocked_response = self.client.post(
|
||
'/api/auth/otp/request', {'phone': phone}, format='json',
|
||
)
|
||
self.assertEqual(blocked_response.status_code, 429)
|
||
self.assertIn('Retry-After', blocked_response.headers)
|
||
|
||
|
||
class RefreshTokenThrottleTests(TestCase):
|
||
"""
|
||
🔒 اثبات محدودیتِ جدیدِ throttle_scope='token_refresh' روی
|
||
RefreshTokenView: درخواست‌های عادی و کافی موفق باشن، درخواست‌های
|
||
بیش‌ازحد با ۴۲۹ رد بشن، و یک درخواستِ بلاک‌شده هیچ Session جدیدی
|
||
نسازه/revoke نکنه و هیچ توکنِ جدیدی تولید نکنه.
|
||
"""
|
||
|
||
def setUp(self):
|
||
cache.clear()
|
||
self.client = APIClient()
|
||
self.user = User.objects.create(is_guest=True, device_id='device-refresh-throttle-test')
|
||
|
||
def _create_session(self, token_suffix):
|
||
return Session.objects.create(
|
||
user=self.user,
|
||
refresh_token=f'test-refresh-token-{token_suffix}',
|
||
device_id='device-refresh-throttle-test',
|
||
expires_at=timezone.now() + timezone.timedelta(days=30),
|
||
revoked=False,
|
||
)
|
||
|
||
def test_requests_within_the_limit_still_succeed(self):
|
||
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'token_refresh': '3/hour'}):
|
||
for i in range(3):
|
||
session = self._create_session(i)
|
||
response = self.client.post(
|
||
'/api/auth/refresh', {'refreshToken': session.refresh_token}, format='json',
|
||
)
|
||
self.assertEqual(response.status_code, 200)
|
||
|
||
def test_excessive_refresh_requests_are_eventually_rejected(self):
|
||
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'token_refresh': '3/hour'}):
|
||
for i in range(3):
|
||
session = self._create_session(i)
|
||
response = self.client.post(
|
||
'/api/auth/refresh', {'refreshToken': session.refresh_token}, format='json',
|
||
)
|
||
self.assertEqual(response.status_code, 200)
|
||
|
||
extra_session = self._create_session('extra')
|
||
blocked_response = self.client.post(
|
||
'/api/auth/refresh', {'refreshToken': extra_session.refresh_token}, format='json',
|
||
)
|
||
self.assertEqual(blocked_response.status_code, 429)
|
||
|
||
def test_throttled_request_does_not_revoke_or_create_session_or_token(self):
|
||
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'token_refresh': '1/hour'}):
|
||
first_session = self._create_session('first')
|
||
first_response = self.client.post(
|
||
'/api/auth/refresh', {'refreshToken': first_session.refresh_token}, format='json',
|
||
)
|
||
self.assertEqual(first_response.status_code, 200)
|
||
|
||
session_count_before = Session.objects.count()
|
||
blocked_session = self._create_session('blocked')
|
||
|
||
with patch('accounts.views.issue_access_token') as mock_issue_token:
|
||
blocked_response = self.client.post(
|
||
'/api/auth/refresh', {'refreshToken': blocked_session.refresh_token}, format='json',
|
||
)
|
||
self.assertEqual(blocked_response.status_code, 429)
|
||
mock_issue_token.assert_not_called()
|
||
|
||
# هیچ Session جدیدی ساخته نشده (فقط همون session ای که قبل از
|
||
# درخواستِ بلاک‌شده دستی ساختیم) و session ای که با آن تلاش
|
||
# کردیم هنوز revoked=False و دست‌نخورده است.
|
||
self.assertEqual(Session.objects.count(), session_count_before + 1)
|
||
blocked_session.refresh_from_db()
|
||
self.assertFalse(blocked_session.revoked)
|
||
|
||
|
||
class GuestSessionThrottleTests(TestCase):
|
||
"""
|
||
🔒 اثبات محدودیتِ جدیدِ ScopedRateThrottle (scope='guest_session') روی
|
||
GuestSessionView: درخواست‌های تکراری از همون کلاینت (IP) باید بعد از
|
||
حدِ مجاز با ۴۲۹ رد بشن.
|
||
"""
|
||
|
||
def setUp(self):
|
||
self.client = APIClient()
|
||
cache.clear()
|
||
|
||
@override_settings(
|
||
REST_FRAMEWORK={
|
||
**settings.REST_FRAMEWORK,
|
||
'DEFAULT_THROTTLE_RATES': {
|
||
**settings.REST_FRAMEWORK['DEFAULT_THROTTLE_RATES'],
|
||
'guest_session': '3/hour',
|
||
},
|
||
}
|
||
)
|
||
def test_repeated_guest_session_requests_are_eventually_throttled(self):
|
||
# ⚠️ اصلاح (باگِ خودِ تست، نه اپلیکیشن): این تست دو مشکلِ جدا
|
||
# داشت:
|
||
# ۱) `type` رو نمی‌فرستاد (پایین‌تر فیکس شده) — قراردادِ واقعیِ
|
||
# API طبقِ README نیازش داره، پس هر درخواست قبل از throttle
|
||
# با 400 رد می‌شد.
|
||
# ۲) override_settings(REST_FRAMEWORK=...) به‌تنهایی رفتارِ
|
||
# ScopedRateThrottle رو تغییر نمی‌ده: DRF مقدارِ
|
||
# THROTTLE_RATES رو یک‌بار موقعِ import ماژولِ throttling
|
||
# به‌عنوانِ یک class attribute از api_settings.
|
||
# DEFAULT_THROTTLE_RATES می‌گیره (همون دیکشنریِ اصلیِ
|
||
# settings.py، با reference). override_settings یک دیکشنریِ
|
||
# *جدید* برای settings.REST_FRAMEWORK می‌سازه، اما این
|
||
# class attribute هنوز به همون دیکشنریِ قدیمی اشاره می‌کنه —
|
||
# سیگنالِ setting_changed فقط api_settings رو reload می‌کنه،
|
||
# نه این class attribute رو. نتیجه: نرخِ واقعیِ اعمال‌شده
|
||
# همیشه همون دیفالتِ 20/hour در settings.py می‌موند، نه
|
||
# 3/hour، پس ۵ درخواست هیچ‌وقت به سقف نمی‌رسیدن. برای اینکه
|
||
# این تست واقعاً نرخِ محدودشده رو بسنجه، مستقیماً همون
|
||
# دیکشنریِ زنده‌ی ScopedRateThrottle.THROTTLE_RATES رو
|
||
# patch.dict می‌کنیم (که override_settings بالا هم برای
|
||
# سازگاری/مستندسازیِ نیت نگه داشته می‌شه).
|
||
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'guest_session': '3/hour'}):
|
||
for i in range(3):
|
||
response = self.client.post(
|
||
'/api/auth/session',
|
||
{'type': 'guest', 'deviceId': f'device-{i}'},
|
||
format='json',
|
||
)
|
||
self.assertEqual(response.status_code, 200)
|
||
|
||
blocked_response = self.client.post(
|
||
'/api/auth/session',
|
||
{'type': 'guest', 'deviceId': 'device-blocked'},
|
||
format='json',
|
||
)
|
||
self.assertEqual(blocked_response.status_code, 429)
|