357 lines
18 KiB
Python
357 lines
18 KiB
Python
# Create your tests here.
|
|
from unittest.mock import patch
|
|
|
|
from django.core.cache import cache
|
|
from django.test import TestCase
|
|
from rest_framework.test import APIClient
|
|
from rest_framework.throttling import ScopedRateThrottle
|
|
|
|
from accounts.authentication import issue_access_token
|
|
from accounts.models import User
|
|
|
|
from .models import BugReport, Ticket, TicketMessage
|
|
|
|
|
|
class BugReportCreateThrottleTests(TestCase):
|
|
"""
|
|
🔒 اثبات محدودیتِ جدیدِ per-user روی BugReportCreateView
|
|
(throttle_scope='support_create'): هیچ idempotency ای روی این
|
|
endpoint نیست، پس بدونِ throttle یک کاربرِ احرازهویت‌شده می‌تونست
|
|
تعدادِ نامحدودی BugReport با درخواست‌های پرتکرار بسازه.
|
|
ScopedRateThrottle با pk کاربر کلید می‌خوره، نه IP.
|
|
"""
|
|
|
|
def setUp(self):
|
|
cache.clear()
|
|
self.client = APIClient()
|
|
self.user = User.objects.create(device_id='device-bugreport-throttle-1', is_guest=True, name='')
|
|
self.token = issue_access_token(self.user.id)
|
|
|
|
def _create(self, i=0, remote_addr=None):
|
|
kwargs = {'HTTP_AUTHORIZATION': f'Bearer {self.token}'}
|
|
if remote_addr:
|
|
kwargs['REMOTE_ADDR'] = remote_addr
|
|
return self.client.post(
|
|
'/api/support/bug-reports',
|
|
{'title': f'باگ {i}', 'description': f'توضیحات {i}'},
|
|
format='json',
|
|
**kwargs,
|
|
)
|
|
|
|
def test_requests_within_the_limit_still_succeed(self):
|
|
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '3/hour'}):
|
|
for i in range(3):
|
|
response = self._create(i)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
def test_excessive_requests_from_same_user_return_429(self):
|
|
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '3/hour'}):
|
|
for i in range(3):
|
|
response = self._create(i)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
blocked_response = self._create('extra')
|
|
self.assertEqual(blocked_response.status_code, 429)
|
|
|
|
def test_changing_client_ip_does_not_bypass_the_user_limit(self):
|
|
# کاربرِ غیرمهمان (نه self.user که مهمانه)، چون کلیدِ throttleِ
|
|
# مهمان‌ها عمداً بر اساسِ IP است؛ این تست فقط رفتارِ per-user رو می‌سنجه.
|
|
registered_user = User.objects.create(is_guest=False, name='', phone='+989120000901')
|
|
registered_token = issue_access_token(registered_user.id)
|
|
|
|
def _create_as(i, remote_addr):
|
|
return self.client.post(
|
|
'/api/support/bug-reports',
|
|
{'title': f'باگ {i}', 'description': f'توضیحات {i}'},
|
|
format='json',
|
|
HTTP_AUTHORIZATION=f'Bearer {registered_token}',
|
|
REMOTE_ADDR=remote_addr,
|
|
)
|
|
|
|
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '3/hour'}):
|
|
for i in range(3):
|
|
response = _create_as(i, f'10.4.0.{i + 1}')
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
blocked_response = _create_as('extra', '10.4.0.99')
|
|
self.assertEqual(blocked_response.status_code, 429)
|
|
|
|
def test_throttled_request_creates_no_additional_bug_report(self):
|
|
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '1/hour'}):
|
|
first_response = self._create()
|
|
self.assertEqual(first_response.status_code, 200)
|
|
|
|
count_before = BugReport.objects.filter(user_id=self.user.id).count()
|
|
|
|
blocked_response = self._create('blocked')
|
|
self.assertEqual(blocked_response.status_code, 429)
|
|
self.assertEqual(BugReport.objects.filter(user_id=self.user.id).count(), count_before)
|
|
|
|
|
|
class TicketCreateThrottleTests(TestCase):
|
|
"""
|
|
🔒 اثبات محدودیتِ جدیدِ per-user روی TicketCreateView
|
|
(throttle_scope='support_create'، مشترک با BugReport/TicketMessage).
|
|
"""
|
|
|
|
def setUp(self):
|
|
cache.clear()
|
|
self.client = APIClient()
|
|
self.user = User.objects.create(device_id='device-ticket-throttle-1', is_guest=True, name='')
|
|
self.token = issue_access_token(self.user.id)
|
|
|
|
def _create(self, i=0, remote_addr=None):
|
|
kwargs = {'HTTP_AUTHORIZATION': f'Bearer {self.token}'}
|
|
if remote_addr:
|
|
kwargs['REMOTE_ADDR'] = remote_addr
|
|
return self.client.post(
|
|
'/api/support/tickets',
|
|
{'subject': f'موضوع {i}', 'category': 'billing'},
|
|
format='json',
|
|
**kwargs,
|
|
)
|
|
|
|
def test_requests_within_the_limit_still_succeed(self):
|
|
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '3/hour'}):
|
|
for i in range(3):
|
|
response = self._create(i)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
def test_excessive_requests_from_same_user_return_429(self):
|
|
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '3/hour'}):
|
|
for i in range(3):
|
|
response = self._create(i)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
blocked_response = self._create('extra')
|
|
self.assertEqual(blocked_response.status_code, 429)
|
|
|
|
def test_changing_client_ip_does_not_bypass_the_user_limit(self):
|
|
# کاربرِ غیرمهمان (نه self.user که مهمانه)، چون کلیدِ throttleِ
|
|
# مهمان‌ها عمداً بر اساسِ IP است؛ این تست فقط رفتارِ per-user رو می‌سنجه.
|
|
registered_user = User.objects.create(is_guest=False, name='', phone='+989120000902')
|
|
registered_token = issue_access_token(registered_user.id)
|
|
|
|
def _create_as(i, remote_addr):
|
|
return self.client.post(
|
|
'/api/support/tickets',
|
|
{'subject': f'موضوع {i}', 'category': 'billing'},
|
|
format='json',
|
|
HTTP_AUTHORIZATION=f'Bearer {registered_token}',
|
|
REMOTE_ADDR=remote_addr,
|
|
)
|
|
|
|
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '3/hour'}):
|
|
for i in range(3):
|
|
response = _create_as(i, f'10.5.0.{i + 1}')
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
blocked_response = _create_as('extra', '10.5.0.99')
|
|
self.assertEqual(blocked_response.status_code, 429)
|
|
|
|
def test_throttled_request_creates_no_additional_ticket(self):
|
|
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '1/hour'}):
|
|
first_response = self._create()
|
|
self.assertEqual(first_response.status_code, 200)
|
|
|
|
count_before = Ticket.objects.filter(user_id=self.user.id).count()
|
|
|
|
blocked_response = self._create('blocked')
|
|
self.assertEqual(blocked_response.status_code, 429)
|
|
self.assertEqual(Ticket.objects.filter(user_id=self.user.id).count(), count_before)
|
|
|
|
|
|
class TicketMessageCreateThrottleTests(TestCase):
|
|
"""
|
|
🔒 اثبات محدودیتِ جدیدِ per-user روی TicketMessageCreateView
|
|
(throttle_scope='support_create'، مشترک با BugReport/Ticket).
|
|
"""
|
|
|
|
def setUp(self):
|
|
cache.clear()
|
|
self.client = APIClient()
|
|
self.user = User.objects.create(device_id='device-ticketmsg-throttle-1', is_guest=True, name='')
|
|
self.token = issue_access_token(self.user.id)
|
|
self.ticket = Ticket.objects.create(user_id=self.user.id, subject='موضوع', category='billing')
|
|
|
|
def _create(self, i=0, remote_addr=None):
|
|
kwargs = {'HTTP_AUTHORIZATION': f'Bearer {self.token}'}
|
|
if remote_addr:
|
|
kwargs['REMOTE_ADDR'] = remote_addr
|
|
return self.client.post(
|
|
f'/api/support/tickets/{self.ticket.id}/messages',
|
|
{'text': f'پیام {i}'},
|
|
format='json',
|
|
**kwargs,
|
|
)
|
|
|
|
def test_requests_within_the_limit_still_succeed(self):
|
|
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '3/hour'}):
|
|
for i in range(3):
|
|
response = self._create(i)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
def test_excessive_requests_from_same_user_return_429(self):
|
|
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '3/hour'}):
|
|
for i in range(3):
|
|
response = self._create(i)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
blocked_response = self._create('extra')
|
|
self.assertEqual(blocked_response.status_code, 429)
|
|
|
|
def test_changing_client_ip_does_not_bypass_the_user_limit(self):
|
|
# کاربرِ غیرمهمان (نه self.user که مهمانه)، چون کلیدِ throttleِ
|
|
# مهمان‌ها عمداً بر اساسِ IP است؛ این تست فقط رفتارِ per-user رو می‌سنجه.
|
|
registered_user = User.objects.create(is_guest=False, name='', phone='+989120000903')
|
|
registered_token = issue_access_token(registered_user.id)
|
|
registered_ticket = Ticket.objects.create(
|
|
user_id=registered_user.id, subject='موضوع', category='billing',
|
|
)
|
|
|
|
def _create_as(i, remote_addr):
|
|
return self.client.post(
|
|
f'/api/support/tickets/{registered_ticket.id}/messages',
|
|
{'text': f'پیام {i}'},
|
|
format='json',
|
|
HTTP_AUTHORIZATION=f'Bearer {registered_token}',
|
|
REMOTE_ADDR=remote_addr,
|
|
)
|
|
|
|
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '3/hour'}):
|
|
for i in range(3):
|
|
response = _create_as(i, f'10.6.0.{i + 1}')
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
blocked_response = _create_as('extra', '10.6.0.99')
|
|
self.assertEqual(blocked_response.status_code, 429)
|
|
|
|
def test_throttled_request_creates_no_additional_message(self):
|
|
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '1/hour'}):
|
|
first_response = self._create()
|
|
self.assertEqual(first_response.status_code, 200)
|
|
|
|
count_before = TicketMessage.objects.filter(ticket=self.ticket).count()
|
|
|
|
blocked_response = self._create('blocked')
|
|
self.assertEqual(blocked_response.status_code, 429)
|
|
self.assertEqual(TicketMessage.objects.filter(ticket=self.ticket).count(), count_before)
|
|
|
|
|
|
class SupportCreateGuestThrottleBypassTests(TestCase):
|
|
"""
|
|
🔒 اثبات فیکسِ امنیتی: قبلاً کلیدِ throttle_scope='support_create'
|
|
(مشترک بینِ BugReportCreateView، TicketCreateView و
|
|
TicketMessageCreateView) برای هر کاربرِ احرازهویت‌شده (مهمان یا نه)
|
|
user.pk بود. چون GuestSessionView با هر deviceId جدید یک کاربرِ
|
|
مهمانِ تازه (pk جدید) می‌سازه، یک مهاجم می‌تونست با ساختنِ پی‌درپیِ
|
|
حساب‌های مهمان، هر بار یک bucket throttle تازه برای این scopeِ
|
|
مشترک بگیره و سقفِ ۳/ساعتِ ساختِ رکوردهای پشتیبانی (BugReport/
|
|
Ticket/TicketMessage) رو چند برابر کنه. حالا GuestAwareSupportThrottle
|
|
برای کاربرهای مهمان کلید رو بر اساسِ IP (همون get_ident که
|
|
guest_session هم استفاده می‌کنه) می‌سازه، پس چرخشِ حساب مهمان دیگه
|
|
bucket تازه نمی‌ده — این هم روی bucket مشترکِ بینِ هر سه Endpoint و
|
|
هم روی رفتارِ per-user برای کاربرهای غیرمهمان تست می‌شه.
|
|
"""
|
|
|
|
def setUp(self):
|
|
cache.clear()
|
|
self.client = APIClient()
|
|
|
|
def _create_bug_report(self, token, i=0, remote_addr=None):
|
|
kwargs = {'HTTP_AUTHORIZATION': f'Bearer {token}'}
|
|
if remote_addr:
|
|
kwargs['REMOTE_ADDR'] = remote_addr
|
|
return self.client.post(
|
|
'/api/support/bug-reports',
|
|
{'title': f'باگ {i}', 'description': f'توضیحات {i}'},
|
|
format='json',
|
|
**kwargs,
|
|
)
|
|
|
|
def _create_ticket(self, token, i=0):
|
|
return self.client.post(
|
|
'/api/support/tickets',
|
|
{'subject': f'موضوع {i}', 'category': 'billing'},
|
|
format='json',
|
|
HTTP_AUTHORIZATION=f'Bearer {token}',
|
|
)
|
|
|
|
def test_rotating_guest_accounts_share_the_same_throttle_bucket(self):
|
|
# سه حساب مهمانِ متفاوت (شبیه‌سازیِ چرخشِ حساب برای دورزدنِ throttle)،
|
|
# همه از همون کلاینت/IP — باید همه یک bucket مشترک رو مصرف کنند،
|
|
# حتی وقتی از دو Endpointِ متفاوتِ همون scope استفاده می‌کنن.
|
|
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '2/hour'}):
|
|
guest1 = User.objects.create(device_id='device-support-guest-rot-1', is_guest=True, name='')
|
|
response1 = self._create_bug_report(issue_access_token(guest1.id), 1)
|
|
self.assertEqual(response1.status_code, 200)
|
|
|
|
guest2 = User.objects.create(device_id='device-support-guest-rot-2', is_guest=True, name='')
|
|
response2 = self._create_ticket(issue_access_token(guest2.id), 2)
|
|
self.assertEqual(response2.status_code, 200)
|
|
|
|
# سومین حسابِ مهمانِ تازه (pk جدید) باید throttle بشه، چون
|
|
# bucket بر اساسِ IP به‌اشتراک گذاشته شده، نه user.pk — یعنی
|
|
# سقفِ ۲/ساعت هنوز اعمال می‌شه، صرف‌نظر از تعدادِ حساب‌های
|
|
# مهمانِ ساخته‌شده یا این‌که کدوم Endpoint صدا زده شده.
|
|
guest3 = User.objects.create(device_id='device-support-guest-rot-3', is_guest=True, name='')
|
|
response3 = self._create_bug_report(issue_access_token(guest3.id), 3)
|
|
|
|
self.assertEqual(response3.status_code, 429)
|
|
|
|
def test_rotated_guest_account_cannot_get_a_fresh_bucket_after_limit(self):
|
|
# حتی وقتی حسابِ مهمانِ جدید به سقف رسیده، یک حسابِ مهمانِ دیگر
|
|
# (چهارمین pk) هم نباید bucket تازه‌ای بگیره — سقف هنوز به IP
|
|
# مشترک متصل است.
|
|
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '1/hour'}):
|
|
guest1 = User.objects.create(device_id='device-support-guest-rot-4', is_guest=True, name='')
|
|
response1 = self._create_bug_report(issue_access_token(guest1.id), 4)
|
|
self.assertEqual(response1.status_code, 200)
|
|
|
|
guest2 = User.objects.create(device_id='device-support-guest-rot-5', is_guest=True, name='')
|
|
blocked = self._create_bug_report(issue_access_token(guest2.id), 5)
|
|
self.assertEqual(blocked.status_code, 429)
|
|
|
|
guest3 = User.objects.create(device_id='device-support-guest-rot-6', is_guest=True, name='')
|
|
still_blocked = self._create_bug_report(issue_access_token(guest3.id), 6)
|
|
|
|
self.assertEqual(still_blocked.status_code, 429)
|
|
|
|
def test_registered_users_keep_independent_per_user_buckets(self):
|
|
# عدمِ رگرسیون: کاربرهای غیرمهمان (ثبت‌نام‌شده با شماره) باید مثلِ
|
|
# قبل هرکدوم bucket جداگانه‌ی خودشون رو داشته باشن — throttle
|
|
# برای اونا هنوز بر اساسِ user.pk است، نه IP.
|
|
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '1/hour'}):
|
|
user_a = User.objects.create(is_guest=False, name='', phone='+989120001001')
|
|
response_a = self._create_bug_report(issue_access_token(user_a.id), 'a')
|
|
self.assertEqual(response_a.status_code, 200)
|
|
|
|
user_b = User.objects.create(is_guest=False, name='', phone='+989120001002')
|
|
response_b = self._create_bug_report(issue_access_token(user_b.id), 'b')
|
|
|
|
self.assertEqual(response_b.status_code, 200)
|
|
|
|
def test_blocked_guest_request_does_not_create_support_record(self):
|
|
# درخواستِ بلاک‌شده (از حسابِ مهمانِ چرخشی) نباید هیچ BugReport یا
|
|
# Ticket جدیدی بسازه — throttle باید *قبل* از رسیدن به منطقِ
|
|
# ساختِ رکورد رد کنه.
|
|
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'support_create': '1/hour'}):
|
|
guest1 = User.objects.create(device_id='device-support-guest-rot-7', is_guest=True, name='')
|
|
first_response = self._create_bug_report(issue_access_token(guest1.id), 7)
|
|
self.assertEqual(first_response.status_code, 200)
|
|
|
|
# حسابِ مهمانِ تازه‌ی دوم باید بلاک بشه (bucket مشترکِ IP از
|
|
# قبل مصرف شده) و نباید BugReport یا Ticket جدیدی بسازد.
|
|
guest2 = User.objects.create(device_id='device-support-guest-rot-8', is_guest=True, name='')
|
|
bug_count_before = BugReport.objects.count()
|
|
ticket_count_before = Ticket.objects.count()
|
|
|
|
blocked_bug_response = self._create_bug_report(issue_access_token(guest2.id), 8)
|
|
self.assertEqual(blocked_bug_response.status_code, 429)
|
|
|
|
blocked_ticket_response = self._create_ticket(issue_access_token(guest2.id), 8)
|
|
self.assertEqual(blocked_ticket_response.status_code, 429)
|
|
|
|
self.assertEqual(BugReport.objects.count(), bug_count_before)
|
|
self.assertEqual(Ticket.objects.count(), ticket_count_before)
|