167 lines
8.6 KiB
Python
167 lines
8.6 KiB
Python
"""
|
||
نرمال‌سازیِ شماره و سقف‌های دیتابیسیِ پیامکِ OTP (۲۰۲۶-۰۹-۲۵، همان روزی که
|
||
کلیدِ کاوه‌نگار وصل شد). security-before-production.md بندِ ۱۷.
|
||
|
||
کش در هر تست پاک می‌شود تا رفتارِ «workerِ دیگر» (کشِ خالی، دیتابیسِ مشترک)
|
||
شبیه‌سازی شود. هم‌زمانیِ واقعیِ دو پروسه‌ی gunicorn تست نشده است.
|
||
"""
|
||
from unittest.mock import patch
|
||
|
||
import requests
|
||
from django.core.cache import cache
|
||
from django.test import TestCase, override_settings
|
||
from django.utils import timezone
|
||
from rest_framework.test import APIClient
|
||
|
||
from accounts.models import OTPCode, User
|
||
from accounts.services import send_sms
|
||
|
||
|
||
def _request(phone):
|
||
return APIClient().post('/api/auth/otp/request', {'phone': phone}, format='json')
|
||
|
||
|
||
@override_settings(DEBUG=True, TEST_LOGIN_PHONES={})
|
||
class PhoneNormalizationTests(TestCase):
|
||
def setUp(self):
|
||
cache.clear()
|
||
|
||
def test_other_spellings_of_the_same_number_share_one_cooldown(self):
|
||
# همان سنجه‌ای که security-before-production.md برای بستنِ بندِ ۱۷ نوشته.
|
||
with patch('accounts.views.send_otp_sms') as send:
|
||
self.assertEqual(_request('09121112233').status_code, 200)
|
||
for spelling in ('+989121112233', '989121112233', '۰۹۱۲۱۱۱۲۲۳۳', '0912 111 2233'):
|
||
self.assertEqual(_request(spelling).status_code, 429, spelling)
|
||
send.assert_called_once()
|
||
self.assertEqual(set(OTPCode.objects.values_list('phone', flat=True)), {'09121112233'})
|
||
|
||
def test_login_with_another_spelling_reaches_the_same_account(self):
|
||
User.objects.create(phone='09121112244', is_guest=False)
|
||
with patch('accounts.views.secrets.randbelow', return_value=246810), \
|
||
patch('accounts.views.send_otp_sms'):
|
||
_request('+989121112244')
|
||
body = APIClient().post('/api/auth/login', {'phone': '۰۹۱۲۱۱۱۲۲۴۴', 'otp': '246810'},
|
||
format='json').json()
|
||
self.assertEqual(User.objects.filter(phone__endswith='9121112244').count(), 1)
|
||
self.assertEqual(body['userId'], str(User.objects.get(phone='09121112244').id))
|
||
|
||
def test_non_mobile_numbers_are_rejected_before_any_sms(self):
|
||
with patch('accounts.views.send_otp_sms') as send:
|
||
for bad in ('+14155552671', '02188888888', '0912123', 'abc'):
|
||
self.assertEqual(_request(bad).status_code, 400, bad)
|
||
send.assert_not_called()
|
||
self.assertFalse(OTPCode.objects.exists())
|
||
|
||
@override_settings(TEST_LOGIN_PHONES={'09120000101': '4321'})
|
||
def test_demo_phone_matches_in_any_spelling(self):
|
||
response = APIClient().post('/api/auth/login', {'phone': '+989120000101', 'otp': '4321'},
|
||
format='json')
|
||
self.assertEqual(response.status_code, 200)
|
||
|
||
|
||
@override_settings(DEBUG=True, TEST_LOGIN_PHONES={}, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=120)
|
||
class DatabaseBackedLimitTests(TestCase):
|
||
"""کشِ LocMem مالِ هر worker است؛ این سقف‌ها باید با کشِ خالی هم بمانند."""
|
||
|
||
def setUp(self):
|
||
cache.clear()
|
||
|
||
def test_cooldown_holds_on_a_worker_with_an_empty_cache(self):
|
||
with patch('accounts.views.send_otp_sms') as send:
|
||
self.assertEqual(_request('09121112255').status_code, 200)
|
||
cache.clear() # درخواستِ بعدی به workerِ دیگری می‌رسد
|
||
response = _request('09121112255')
|
||
self.assertEqual(response.status_code, 429)
|
||
self.assertTrue(1 <= response.json()['retryAfter'] <= 120)
|
||
send.assert_called_once()
|
||
|
||
@override_settings(OTP_REQUEST_PHONE_COOLDOWN_SECONDS=0, OTP_REQUEST_PHONE_DAILY_MAX=3)
|
||
def test_daily_cap_per_phone(self):
|
||
with patch('accounts.views.send_otp_sms') as send:
|
||
statuses = []
|
||
for _ in range(5):
|
||
cache.clear() # سقفِ ساعتیِ کش را کنار می‌گذارد؛ فقط سقفِ روزانه بماند
|
||
statuses.append(_request('09121112266').status_code)
|
||
self.assertEqual(statuses, [200, 200, 200, 429, 429])
|
||
self.assertEqual(send.call_count, 3)
|
||
|
||
@override_settings(OTP_REQUEST_PHONE_COOLDOWN_SECONDS=0)
|
||
def test_codes_older_than_a_day_do_not_count(self):
|
||
OTPCode.objects.bulk_create([
|
||
OTPCode(phone='09121112277', code='000000',
|
||
expires_at=timezone.now() - timezone.timedelta(days=2))
|
||
for _ in range(20)
|
||
])
|
||
with patch('accounts.views.send_otp_sms'):
|
||
self.assertEqual(_request('09121112277').status_code, 200)
|
||
|
||
@override_settings(OTP_REQUEST_PHONE_COOLDOWN_SECONDS=0, OTP_SMS_GLOBAL_DAILY_MAX=2)
|
||
def test_global_daily_cap_stops_sms_pumping(self):
|
||
with patch('accounts.views.send_otp_sms') as send:
|
||
self.assertEqual(_request('09121110001').status_code, 200)
|
||
self.assertEqual(_request('09121110002').status_code, 200)
|
||
with self.assertLogs('accounts.views', level='ERROR'):
|
||
response = _request('09121110003')
|
||
self.assertEqual(response.status_code, 503)
|
||
self.assertEqual(send.call_count, 2)
|
||
self.assertFalse(OTPCode.objects.filter(phone='09121110003').exists())
|
||
|
||
@override_settings(DEBUG=False, KAVENEGAR_API_KEY='k')
|
||
def test_failed_delivery_frees_the_cooldown_and_quota(self):
|
||
with patch('accounts.services.requests.post', side_effect=requests.ConnectionError('down')):
|
||
self.assertEqual(_request('09121112288').status_code, 503)
|
||
cache.clear()
|
||
ok = type('R', (), {'status_code': 200, 'json': lambda self: {
|
||
'return': {'status': 200}, 'entries': [{'messageid': 1, 'status': 1}]}})()
|
||
with patch('accounts.services.requests.post', return_value=ok):
|
||
self.assertEqual(_request('09121112288').status_code, 200)
|
||
|
||
|
||
class ProviderLoggingTests(TestCase):
|
||
@override_settings(KAVENEGAR_API_KEY='k')
|
||
def test_accepted_message_logs_its_id_not_its_text(self):
|
||
ok = type('R', (), {'status_code': 200, 'json': lambda self: {
|
||
'return': {'status': 200}, 'entries': [{'messageid': 8792343, 'status': 1}]}})()
|
||
with patch('accounts.services.requests.post', return_value=ok), \
|
||
self.assertLogs('accounts.services', level='INFO') as logs:
|
||
send_sms(phone='09121112299', message='کد ورود شما: 123456')
|
||
joined = '\n'.join(logs.output)
|
||
self.assertIn('messageid=8792343', joined)
|
||
self.assertNotIn('123456', joined)
|
||
|
||
|
||
def _kavenegar_reply(http=200, status=200):
|
||
return type('R', (), {'status_code': http, 'json': lambda self: {
|
||
'return': {'status': status}, 'entries': [{'messageid': 1, 'status': 1}] if status == 200 else None}})()
|
||
|
||
|
||
@override_settings(DEBUG=False, KAVENEGAR_API_KEY='k', TEST_LOGIN_PHONES={})
|
||
class VerifyLookupTests(TestCase):
|
||
"""خطِ حسابِ فعلی متنِ آزاد را لغو می‌کند (وضعیتِ ۱۳)؛ کدِ ورود باید با الگو برود."""
|
||
|
||
def setUp(self):
|
||
cache.clear()
|
||
|
||
@override_settings(KAVENEGAR_OTP_TEMPLATE='tadrisino-otp')
|
||
def test_otp_goes_through_the_template_with_only_the_code(self):
|
||
with patch('accounts.views.secrets.randbelow', return_value=4321), \
|
||
patch('accounts.services.requests.post', return_value=_kavenegar_reply()) as post:
|
||
self.assertEqual(_request('09121113301').status_code, 200)
|
||
url, data = post.call_args.args[0], post.call_args.kwargs['data']
|
||
self.assertTrue(url.endswith('/verify/lookup.json'))
|
||
self.assertEqual(data, {'receptor': '09121113301', 'template': 'tadrisino-otp', 'token': '004321'})
|
||
|
||
@override_settings(KAVENEGAR_OTP_TEMPLATE='')
|
||
def test_without_a_template_the_plain_send_is_kept(self):
|
||
with patch('accounts.services.requests.post', return_value=_kavenegar_reply()) as post:
|
||
self.assertEqual(_request('09121113302').status_code, 200)
|
||
self.assertTrue(post.call_args.args[0].endswith('/sms/send.json'))
|
||
self.assertIn('message', post.call_args.kwargs['data'])
|
||
|
||
@override_settings(KAVENEGAR_OTP_TEMPLATE='misspelled')
|
||
def test_unknown_template_is_an_explicit_503_and_frees_the_cooldown(self):
|
||
# کاوه‌نگار برای الگوی ناموجود/تأییدنشده HTTP ۴۲۴ می‌دهد (۲۰۲۶-۰۹-۲۵ زنده دیده شد).
|
||
with patch('accounts.services.requests.post', return_value=_kavenegar_reply(424, 424)):
|
||
self.assertEqual(_request('09121113303').status_code, 503)
|
||
self.assertFalse(OTPCode.objects.filter(phone='09121113303').exists())
|