from django.core.cache import cache from django.test import TestCase from rest_framework.test import APIClient from rest_framework.throttling import ScopedRateThrottle from unittest.mock import patch from accounts.authentication import issue_access_token from accounts.models import User from .models import FamilyLink class FamilyLinkAcceptThrottleTests(TestCase): """ 🔒 اثبات محدودیتِ جدیدِ throttle_scope='family_invite_accept' روی FamilyLinkAcceptView: inviteCode یک رازِ عددیِ ۶رقمی است که این endpoint با select_for_update() جست‌وجو می‌کنه؛ بدونِ throttle یک کاربرِ احرازهویت‌شده می‌تونست بی‌نهایت کد رو حدس بزنه. این تست‌ها هم رفتارِ throttle (per-user، نه IP)، هم عدمِ رگرسیون در ردیمِ موفق، هم عدمِ تغییرِ FamilyLink توسطِ درخواستِ بلاک‌شده رو تضمین می‌کنن. """ def setUp(self): cache.clear() self.client = APIClient() self.parent = User.objects.create(device_id='device-family-parent-1', is_guest=True, name='') self.child = User.objects.create( device_id='device-family-child-1', is_guest=True, name='', phone='+989120000001', ) self.child_token = issue_access_token(self.child.id) self.registered_user = User.objects.create(is_guest=False, name='', phone='+989120000098') self.registered_token = issue_access_token(self.registered_user.id) self.link = FamilyLink.objects.create( parent_user_id=self.parent.id, child_phone=self.child.phone, invite_code='123456', ) def _accept(self, code, token=None): headers = {} if token: headers['HTTP_AUTHORIZATION'] = f'Bearer {token}' return self.client.post( '/api/family/link/accept', {'inviteCode': code}, format='json', **headers, ) def test_unauthenticated_request_is_still_rejected(self): # رفتارِ IsAuthenticated موجود دست‌نخورده مونده؛ throttle جایگزینِ # authentication نیست. response = self._accept('123456') self.assertEqual(response.status_code, 401) def test_repeated_code_guessing_is_throttled_after_configured_limit(self): # سناریوی brute-force: کدهای نادرستِ پشتِ‌سرِهم. with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'family_invite_accept': '3/hour'}): for i in range(3): response = self._accept(f'00000{i}', token=self.child_token) self.assertEqual(response.status_code, 404) blocked_response = self._accept('999999', token=self.child_token) self.assertEqual(blocked_response.status_code, 429) def test_throttle_is_scoped_per_user_not_shared_globally(self): # کاربرِ غیرمهمان (نه یک حسابِ مهمانِ دوم، که دیگه bucket جدا نداره). other_user = User.objects.create( is_guest=False, name='', phone='+989120000099', ) other_token = issue_access_token(other_user.id) with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'family_invite_accept': '1/hour'}): first_user_response = self._accept('000000', token=self.registered_token) self.assertEqual(first_user_response.status_code, 404) first_user_blocked = self._accept('111111', token=self.registered_token) self.assertEqual(first_user_blocked.status_code, 429) # کاربرِ دومِ غیرمهمانِ متفاوت هنوز باید بتونه تلاش کنه — # throttle برای کاربرهای ثبت‌نام‌شده هنوز per-user است، نه IP. second_user_response = self._accept('000000', token=other_token) self.assertEqual(second_user_response.status_code, 404) def test_successful_accept_still_works_within_the_limit(self): # عدمِ رگرسیون: رفتارِ اصلیِ accept باید دقیقاً مثلِ قبل کار کنه. with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'family_invite_accept': '10/hour'}): response = self._accept(self.link.invite_code, token=self.child_token) self.assertEqual(response.status_code, 200) self.assertEqual(response.data['status'], 'accepted') self.link.refresh_from_db() self.assertEqual(self.link.status, 'accepted') self.assertEqual(self.link.child_user_id, self.child.id) def test_blocked_request_does_not_mutate_family_link(self): # درخواستِ بلاک‌شده نباید FamilyLink رو تغییر بده — throttle باید # *قبل* از رسیدنِ به select_for_update() رد کنه. with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'family_invite_accept': '1/hour'}): first_response = self._accept('000000', token=self.child_token) self.assertEqual(first_response.status_code, 404) blocked_response = self._accept(self.link.invite_code, token=self.child_token) self.assertEqual(blocked_response.status_code, 429) self.link.refresh_from_db() self.assertEqual(self.link.status, 'pending') self.assertIsNone(self.link.child_user_id) class FamilyLinkAcceptGuestThrottleBypassTests(TestCase): """ 🔒 اثبات فیکسِ امنیتی: قبلاً کلیدِ throttle_scope='family_invite_accept' برای هر کاربرِ احرازهویت‌شده (مهمان یا نه) user.pk بود. چون GuestSessionView با هر deviceId جدید یک کاربرِ مهمانِ تازه (pk جدید) می‌سازه، یک مهاجم می‌تونست با ساختنِ پی‌درپیِ حساب‌های مهمان، هر بار یک bucket throttle تازه برای brute-force کردنِ inviteCode ۶رقمی بگیره. حالا GuestAwareFamilyThrottle برای کاربرهای مهمان کلید رو بر اساسِ IP (همون get_ident که guest_session هم استفاده می‌کنه) می‌سازه، پس چرخشِ حساب مهمان دیگه bucket تازه نمی‌ده. رفتارِ throttle برای کاربرِ غیرمهمان (per-user، مثلِ قبل) دست‌نخورده مونده. """ def setUp(self): cache.clear() self.client = APIClient() def _accept(self, code, token): return self.client.post( '/api/family/link/accept', {'inviteCode': code}, format='json', HTTP_AUTHORIZATION=f'Bearer {token}', ) def test_rotating_guest_accounts_share_the_same_throttle_bucket(self): # سه حساب مهمانِ متفاوت (شبیه‌سازیِ چرخشِ حساب برای دورزدنِ throttle)، # همه از همون کلاینت/IP — باید همه یک bucket مشترک رو مصرف کنند. with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'family_invite_accept': '2/hour'}): guest1 = User.objects.create(device_id='device-family-guest-rot-1', is_guest=True, name='') response1 = self._accept('000001', issue_access_token(guest1.id)) self.assertEqual(response1.status_code, 404) guest2 = User.objects.create(device_id='device-family-guest-rot-2', is_guest=True, name='') response2 = self._accept('000002', issue_access_token(guest2.id)) self.assertEqual(response2.status_code, 404) # سومین حسابِ مهمانِ تازه (pk جدید) باید throttle بشه، چون # bucket بر اساسِ IP به‌اشتراک گذاشته شده، نه user.pk — یعنی # سقفِ ۲/ساعت هنوز اعمال می‌شه، صرف‌نظر از تعدادِ حساب‌های # مهمانِ ساخته‌شده. guest3 = User.objects.create(device_id='device-family-guest-rot-3', is_guest=True, name='') response3 = self._accept('000003', issue_access_token(guest3.id)) self.assertEqual(response3.status_code, 429) def test_registered_users_keep_independent_per_user_buckets(self): # عدمِ رگرسیون: کاربرهای غیرمهمان (ثبت‌نام‌شده با شماره) باید مثلِ # قبل هرکدوم bucket جداگانه‌ی خودشون رو داشته باشن — throttle # برای اونا هنوز بر اساسِ user.pk است، نه IP. with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'family_invite_accept': '1/hour'}): user_a = User.objects.create(is_guest=False, name='', phone='+989120000201') response_a = self._accept('000001', issue_access_token(user_a.id)) self.assertEqual(response_a.status_code, 404) user_b = User.objects.create(is_guest=False, name='', phone='+989120000202') response_b = self._accept('000002', issue_access_token(user_b.id)) self.assertEqual(response_b.status_code, 404) def test_blocked_guest_request_does_not_reach_or_mutate_family_link(self): # درخواستِ بلاک‌شده (چه از همون حساب، چه از حسابِ مهمانِ چرخشی) # نباید FamilyLink رو تغییر بده و نباید حتی به select_for_update() # برسه — throttle باید *قبل* از منطقِ accept رد کنه. parent = User.objects.create(device_id='device-family-guest-rot-parent', is_guest=True, name='') child = User.objects.create( device_id='device-family-guest-rot-child', is_guest=True, name='', phone='+989120000203', ) link = FamilyLink.objects.create( parent_user_id=parent.id, child_phone=child.phone, invite_code='654321', ) with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'family_invite_accept': '1/hour'}): guest1 = User.objects.create(device_id='device-family-guest-rot-4', is_guest=True, name='') first_response = self._accept('000000', issue_access_token(guest1.id)) self.assertEqual(first_response.status_code, 404) # حسابِ مهمانِ تازه‌ی دوم، حتی با inviteCode درست، باید بلاک بشه # (bucket مشترکِ IP از قبل مصرف شده) و FamilyLink رو دست نزنه. guest2 = User.objects.create(device_id='device-family-guest-rot-5', is_guest=True, name='') blocked_response = self._accept(link.invite_code, issue_access_token(guest2.id)) self.assertEqual(blocked_response.status_code, 429) link.refresh_from_db() self.assertEqual(link.status, 'pending') self.assertIsNone(link.child_user_id) class FamilyLinkInviteThrottleTests(TestCase): """ 🔒 اثبات محدودیتِ جدیدِ throttle_scope='family_invite_create' روی FamilyLinkInviteView: این endpoint قبلاً هیچ throttle مخصوصی نداشت، درحالی‌که هر درخواستِ پذیرفته‌شده یک FamilyLink ماندگار و یک کدِ دعوتِ ۶رقمیِ تازه می‌سازه. GuestAwareFamilyThrottle با pk کاربرِ احرازهویت‌شده کلید می‌خوره (نه IP)، پس تغییرِ IP این محدودیت رو دور نمی‌زنه. """ def setUp(self): cache.clear() self.client = APIClient() self.parent = User.objects.create(device_id='device-family-invite-throttle-1', is_guest=True, name='') self.token = issue_access_token(self.parent.id) def _invite(self, child_phone, token=None, remote_addr=None): headers = {} if token: headers['HTTP_AUTHORIZATION'] = f'Bearer {token}' if remote_addr: headers['REMOTE_ADDR'] = remote_addr return self.client.post( '/api/family/link/invite', {'childPhone': child_phone}, format='json', **headers, ) def test_unauthenticated_request_is_rejected(self): response = self._invite('+989120001001') self.assertEqual(response.status_code, 401) def test_requests_within_the_limit_still_succeed(self): with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'family_invite_create': '3/hour'}): for i in range(3): response = self._invite(f'+98912000110{i}', token=self.token) self.assertEqual(response.status_code, 200) self.assertIn('inviteCode', response.data) def test_excessive_invite_requests_from_same_user_are_blocked(self): with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'family_invite_create': '3/hour'}): for i in range(3): response = self._invite(f'+98912000120{i}', token=self.token) self.assertEqual(response.status_code, 200) blocked_response = self._invite('+989120001299', token=self.token) self.assertEqual(blocked_response.status_code, 429) def test_changing_client_ip_does_not_bypass_the_user_limit(self): # کاربرِ غیرمهمان (نه self.parent که مهمانه)، چون کلیدِ throttleِ # مهمان‌ها عمداً بر اساسِ IP است؛ این تست فقط رفتارِ per-user رو می‌سنجه. registered_parent = User.objects.create(is_guest=False, name='', phone='+989120001300') registered_token = issue_access_token(registered_parent.id) with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'family_invite_create': '3/hour'}): for i in range(3): response = self._invite( f'+98912000131{i}', token=registered_token, remote_addr=f'10.5.0.{i + 1}', ) self.assertEqual(response.status_code, 200) blocked_response = self._invite('+989120001399', token=registered_token, remote_addr='10.5.0.99') self.assertEqual(blocked_response.status_code, 429) def test_throttled_request_does_not_create_family_link(self): with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'family_invite_create': '1/hour'}): first_response = self._invite('+989120001401', token=self.token) self.assertEqual(first_response.status_code, 200) link_count_before = FamilyLink.objects.filter(parent_user_id=self.parent.id).count() blocked_response = self._invite('+989120001402', token=self.token) self.assertEqual(blocked_response.status_code, 429) self.assertEqual( FamilyLink.objects.filter(parent_user_id=self.parent.id).count(), link_count_before, ) self.assertFalse( FamilyLink.objects.filter(parent_user_id=self.parent.id, child_phone='+989120001402').exists(), ) class FamilyLinkInviteGuestThrottleBypassTests(TestCase): """ 🔒 اثبات فیکسِ امنیتی: FamilyLinkInviteView تازه throttle گرفته و از همون ابتدا با GuestAwareFamilyThrottle محافظت می‌شه (نه یک ScopedRateThrottleِ ساده که بعداً فیکس بشه). چون GuestSessionView با هر deviceId جدید یک کاربرِ مهمانِ تازه (pk جدید) می‌سازه، بدونِ این فیکس یک مهاجم می‌تونست با ساختنِ پی‌درپیِ حساب‌های مهمان، هر بار یک bucket throttle تازه برای ساختنِ FamilyLink/کدِ دعوتِ نامحدود بگیره. GuestAwareFamilyThrottle برای کاربرهای مهمان کلید رو بر اساسِ IP می‌سازه، پس چرخشِ حساب مهمان دیگه bucket تازه نمی‌ده. رفتارِ throttle برای کاربرِ غیرمهمان (per-user) دست‌نخورده مونده. """ def setUp(self): cache.clear() self.client = APIClient() def _invite(self, child_phone, token): return self.client.post( '/api/family/link/invite', {'childPhone': child_phone}, format='json', HTTP_AUTHORIZATION=f'Bearer {token}', ) def test_rotating_guest_accounts_share_the_same_throttle_bucket(self): with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'family_invite_create': '2/hour'}): guest1 = User.objects.create(device_id='device-family-invite-guest-rot-1', is_guest=True, name='') response1 = self._invite('+989120001501', issue_access_token(guest1.id)) self.assertEqual(response1.status_code, 200) guest2 = User.objects.create(device_id='device-family-invite-guest-rot-2', is_guest=True, name='') response2 = self._invite('+989120001502', issue_access_token(guest2.id)) self.assertEqual(response2.status_code, 200) # سومین حسابِ مهمانِ تازه (pk جدید) باید throttle بشه، چون # bucket بر اساسِ IP به‌اشتراک گذاشته شده، نه user.pk. guest3 = User.objects.create(device_id='device-family-invite-guest-rot-3', is_guest=True, name='') response3 = self._invite('+989120001503', issue_access_token(guest3.id)) self.assertEqual(response3.status_code, 429) def test_registered_users_keep_independent_per_user_buckets(self): with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'family_invite_create': '1/hour'}): user_a = User.objects.create(is_guest=False, name='', phone='+989120001601') response_a = self._invite('+989120001701', issue_access_token(user_a.id)) self.assertEqual(response_a.status_code, 200) user_b = User.objects.create(is_guest=False, name='', phone='+989120001602') response_b = self._invite('+989120001702', issue_access_token(user_b.id)) self.assertEqual(response_b.status_code, 200) def test_blocked_guest_request_does_not_create_family_link(self): with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'family_invite_create': '1/hour'}): guest1 = User.objects.create(device_id='device-family-invite-guest-rot-4', is_guest=True, name='') first_response = self._invite('+989120001801', issue_access_token(guest1.id)) self.assertEqual(first_response.status_code, 200) guest2 = User.objects.create(device_id='device-family-invite-guest-rot-5', is_guest=True, name='') blocked_response = self._invite('+989120001802', issue_access_token(guest2.id)) self.assertEqual(blocked_response.status_code, 429) self.assertFalse(FamilyLink.objects.filter(parent_user_id=guest2.id).exists())