604 lines
34 KiB
Python
604 lines
34 KiB
Python
from django.core.cache import cache
|
||
from django.test import TestCase
|
||
from django.utils import timezone
|
||
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 payment.models import Order
|
||
from .models import DiscountCode, Offer
|
||
|
||
|
||
class ValidateDiscountPermissionTests(TestCase):
|
||
"""
|
||
🔒 اثبات فیکسِ امنیتی: ValidateDiscountView قبلاً permission_classes
|
||
نداشت (یعنی دیفالتِ پروژه AllowAny اعمال می‌شد) و یک کلاینتِ بدونِ
|
||
احرازهویت می‌تونست بی‌نهایت کد رو probe کنه و valid/type/value رو
|
||
ببینه — درحالی‌که RedeemDiscountView (اندپوینتِ واقعیِ مصرف) از قبل
|
||
IsAuthenticated داره. این تست هم رفتارِ ۴۰۱ برای anonymous، هم رفتارِ
|
||
صحیحِ اعتبارسنجی برای کاربرِ لاگین‌کرده رو تضمین می‌کنه.
|
||
"""
|
||
|
||
def setUp(self):
|
||
self.client = APIClient()
|
||
self.user = User.objects.create(device_id='device-validate-1', is_guest=True, name='')
|
||
self.token = issue_access_token(self.user.id)
|
||
self.discount = DiscountCode.objects.create(
|
||
code='SUMMER10', type='percent', value=10, max_uses=5, expires_at=None,
|
||
)
|
||
|
||
def test_unauthenticated_request_is_rejected(self):
|
||
response = self.client.get('/api/discount/validate', {'code': self.discount.code})
|
||
self.assertEqual(response.status_code, 401)
|
||
|
||
def test_authenticated_request_validates_usable_code(self):
|
||
response = self.client.get(
|
||
'/api/discount/validate',
|
||
{'code': self.discount.code},
|
||
HTTP_AUTHORIZATION=f'Bearer {self.token}',
|
||
)
|
||
self.assertEqual(response.status_code, 200)
|
||
self.assertTrue(response.data['valid'])
|
||
self.assertEqual(response.data['type'], 'percent')
|
||
self.assertEqual(response.data['value'], 10)
|
||
|
||
def test_authenticated_request_reports_unusable_code_as_invalid(self):
|
||
expired = DiscountCode.objects.create(
|
||
code='EXPIRED5', type='fixed', value=5000, max_uses=5,
|
||
expires_at=timezone.now() - timezone.timedelta(days=1),
|
||
)
|
||
response = self.client.get(
|
||
'/api/discount/validate',
|
||
{'code': expired.code},
|
||
HTTP_AUTHORIZATION=f'Bearer {self.token}',
|
||
)
|
||
self.assertEqual(response.status_code, 200)
|
||
self.assertFalse(response.data['valid'])
|
||
|
||
|
||
class RedeemOfferCodeThrottleTests(TestCase):
|
||
"""
|
||
🔒 اثبات محدودیتِ جدیدِ throttle_scope='offer_redeem' روی
|
||
RedeemOfferCodeView: قبلاً این endpoint هیچ throttle ای نداشت، پس
|
||
یک کاربرِ احرازهویت‌شده می‌تونست بی‌نهایت کدِ آفر رو probe کنه (چون
|
||
کدهای نامعتبر فقط یک ValidationError ساده‌ی ۴۰۰ برمی‌گردوندن، نه
|
||
۴۲۹). این تست‌ها هم رفتارِ throttle (per-user، نه IP) و هم عدمِ
|
||
رگرسیون در رفتارِ موفقِ redemption رو تضمین می‌کنن.
|
||
"""
|
||
|
||
def setUp(self):
|
||
cache.clear()
|
||
self.client = APIClient()
|
||
self.user = User.objects.create(device_id='device-offer-redeem-1', is_guest=True, name='')
|
||
self.token = issue_access_token(self.user.id)
|
||
|
||
def _redeem(self, code):
|
||
return self.client.post(
|
||
'/api/discount/offers/redeem',
|
||
{'code': code},
|
||
format='json',
|
||
HTTP_AUTHORIZATION=f'Bearer {self.token}',
|
||
)
|
||
|
||
def test_unauthenticated_request_is_rejected(self):
|
||
# رفتارِ IsAuthenticated موجود دست‌نخورده مونده؛ throttle جایگزینِ
|
||
# authentication نیست.
|
||
response = self.client.post(
|
||
'/api/discount/offers/redeem', {'code': 'ANYCODE'}, format='json',
|
||
)
|
||
self.assertEqual(response.status_code, 401)
|
||
|
||
def test_invalid_code_attempts_are_throttled_after_configured_limit(self):
|
||
# این دقیقاً همون سناریوی probing هست: کدهای نامعتبرِ پشتِ‌سرِهم.
|
||
# قبل از این فیکس، همه‌شون ۴۰۰ می‌گرفتن و هیچ‌وقت بلاک نمی‌شدن.
|
||
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'offer_redeem': '3/hour'}):
|
||
for i in range(3):
|
||
response = self._redeem(f'INVALID-{i}')
|
||
self.assertEqual(response.status_code, 400)
|
||
|
||
blocked_response = self._redeem('INVALID-EXTRA')
|
||
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='+989120000701')
|
||
other_token = issue_access_token(other_user.id)
|
||
|
||
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'offer_redeem': '1/hour'}):
|
||
first_user_response = self._redeem('INVALID-A')
|
||
self.assertEqual(first_user_response.status_code, 400)
|
||
|
||
first_user_blocked = self._redeem('INVALID-B')
|
||
self.assertEqual(first_user_blocked.status_code, 429)
|
||
|
||
# کاربرِ دومِ متفاوت هنوز باید بتونه تلاش کنه — throttle
|
||
# per-user است، نه یک شمارنده‌ی سراسری روی endpoint.
|
||
second_user_response = self.client.post(
|
||
'/api/discount/offers/redeem',
|
||
{'code': 'INVALID-C'},
|
||
format='json',
|
||
HTTP_AUTHORIZATION=f'Bearer {other_token}',
|
||
)
|
||
self.assertEqual(second_user_response.status_code, 400)
|
||
|
||
def test_successful_redemption_still_works_within_the_limit(self):
|
||
# عدمِ رگرسیون در رفتارِ اصلیِ redemption: با یک کدِ معتبر و در
|
||
# محدوده‌ی مجاز، endpoint باید دقیقاً مثلِ قبل کار کنه.
|
||
offer = Offer.objects.create(
|
||
name='Test Offer', kind='free_full', code='VALIDCODE1', is_active=True,
|
||
)
|
||
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'offer_redeem': '10/hour'}):
|
||
response = self._redeem(offer.code)
|
||
|
||
self.assertEqual(response.status_code, 200)
|
||
self.assertEqual(response.data['offerName'], offer.name)
|
||
self.assertEqual(response.data['kind'], offer.kind)
|
||
|
||
def test_blocked_request_does_not_call_redeem_offer_code(self):
|
||
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'offer_redeem': '1/hour'}):
|
||
first_response = self._redeem('INVALID-X')
|
||
self.assertEqual(first_response.status_code, 400)
|
||
|
||
with patch('discount.views.redeem_offer_code') as mock_redeem:
|
||
blocked_response = self._redeem('INVALID-Y')
|
||
|
||
self.assertEqual(blocked_response.status_code, 429)
|
||
mock_redeem.assert_not_called()
|
||
|
||
|
||
class RedeemOfferCodeGuestThrottleBypassTests(TestCase):
|
||
"""
|
||
🔒 اثبات فیکسِ امنیتی: قبلاً کلیدِ throttle_scope='offer_redeem' برای هر
|
||
کاربرِ احرازهویت‌شده (مهمان یا نه) user.pk بود. چون GuestSessionView با
|
||
هر deviceId جدید یک کاربرِ مهمانِ تازه (pk جدید) می‌سازه، یک مهاجم
|
||
می‌تونست با ساختنِ پی‌درپیِ حساب‌های مهمان، هر بار یک bucket throttle
|
||
تازه برای probe/brute-force کردنِ کدهای آفر بگیره. چون Offer.code
|
||
سراسری/یکتاست و redeem_offer_code() فقط جلوی تکرارِ *همون* کاربر رو
|
||
می‌گیره، این دورزدن می‌تونست به یک حسابِ مهمانِ چرخشی اجازه بده همون
|
||
آفرِ فعال رو دوباره از apply_offer() بگیره و اشتراکِ رایگان/آزمایشیِ
|
||
مربوطه رو دوباره اعمال کنه. حالا GuestAwareDiscountThrottle برای
|
||
کاربرهای مهمان کلید رو بر اساسِ IP (همون get_ident که guest_session هم
|
||
استفاده می‌کنه) می‌سازه، پس چرخشِ حساب مهمان دیگه bucket تازه نمی‌ده.
|
||
رفتارِ throttle برای کاربرِ غیرمهمان (per-user، مثلِ قبل) دست‌نخورده
|
||
مونده.
|
||
"""
|
||
|
||
def setUp(self):
|
||
cache.clear()
|
||
self.client = APIClient()
|
||
|
||
def _redeem(self, code, token):
|
||
return self.client.post(
|
||
'/api/discount/offers/redeem',
|
||
{'code': 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, {'offer_redeem': '2/hour'}):
|
||
guest1 = User.objects.create(device_id='device-offer-guest-rot-1', is_guest=True, name='')
|
||
response1 = self._redeem('INVALID-ROT-1', issue_access_token(guest1.id))
|
||
self.assertEqual(response1.status_code, 400)
|
||
|
||
guest2 = User.objects.create(device_id='device-offer-guest-rot-2', is_guest=True, name='')
|
||
response2 = self._redeem('INVALID-ROT-2', issue_access_token(guest2.id))
|
||
self.assertEqual(response2.status_code, 400)
|
||
|
||
# سومین حسابِ مهمانِ تازه (pk جدید) باید throttle بشه، چون
|
||
# bucket بر اساسِ IP به‌اشتراک گذاشته شده، نه user.pk — یعنی
|
||
# سقفِ ۲/ساعت هنوز اعمال می‌شه، صرف‌نظر از تعدادِ حساب‌های
|
||
# مهمانِ ساخته‌شده.
|
||
guest3 = User.objects.create(device_id='device-offer-guest-rot-3', is_guest=True, name='')
|
||
response3 = self._redeem('INVALID-ROT-3', issue_access_token(guest3.id))
|
||
|
||
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, {'offer_redeem': '1/hour'}):
|
||
guest1 = User.objects.create(device_id='device-offer-guest-rot-4', is_guest=True, name='')
|
||
response1 = self._redeem('INVALID-ROT-4', issue_access_token(guest1.id))
|
||
self.assertEqual(response1.status_code, 400)
|
||
|
||
guest2 = User.objects.create(device_id='device-offer-guest-rot-5', is_guest=True, name='')
|
||
blocked = self._redeem('INVALID-ROT-5', issue_access_token(guest2.id))
|
||
self.assertEqual(blocked.status_code, 429)
|
||
|
||
guest3 = User.objects.create(device_id='device-offer-guest-rot-6', is_guest=True, name='')
|
||
still_blocked = self._redeem('INVALID-ROT-6', issue_access_token(guest3.id))
|
||
|
||
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, {'offer_redeem': '1/hour'}):
|
||
user_a = User.objects.create(is_guest=False, name='', phone='+989120000801')
|
||
response_a = self._redeem('INVALID-REG-A', issue_access_token(user_a.id))
|
||
self.assertEqual(response_a.status_code, 400)
|
||
|
||
user_b = User.objects.create(is_guest=False, name='', phone='+989120000802')
|
||
response_b = self._redeem('INVALID-REG-B', issue_access_token(user_b.id))
|
||
|
||
self.assertEqual(response_b.status_code, 400)
|
||
|
||
def test_blocked_guest_request_does_not_call_redeem_offer_code(self):
|
||
# درخواستِ بلاک‌شده (از حسابِ مهمانِ چرخشی) نباید redeem_offer_code()
|
||
# رو صدا بزنه — throttle باید *قبل* از رسیدن به منطقِ redemption/
|
||
# apply_offer رد کنه، حتی با کدِ درست.
|
||
offer = Offer.objects.create(
|
||
name='Guest Rotation Offer', kind='free_full', code='GUESTROTBLOCK', is_active=True,
|
||
)
|
||
|
||
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'offer_redeem': '1/hour'}):
|
||
guest1 = User.objects.create(device_id='device-offer-guest-rot-7', is_guest=True, name='')
|
||
first_response = self._redeem('INVALID-ROT-7', issue_access_token(guest1.id))
|
||
self.assertEqual(first_response.status_code, 400)
|
||
|
||
# حسابِ مهمانِ تازه‌ی دوم، حتی با کدِ درست، باید بلاک بشه
|
||
# (bucket مشترکِ IP از قبل مصرف شده) و نباید به
|
||
# redeem_offer_code() برسه.
|
||
guest2 = User.objects.create(device_id='device-offer-guest-rot-8', is_guest=True, name='')
|
||
with patch('discount.views.redeem_offer_code') as mock_redeem:
|
||
blocked_response = self._redeem(offer.code, issue_access_token(guest2.id))
|
||
self.assertEqual(blocked_response.status_code, 429)
|
||
mock_redeem.assert_not_called()
|
||
|
||
|
||
class ValidateDiscountThrottleTests(TestCase):
|
||
"""
|
||
🔒 اثبات محدودیتِ جدیدِ throttle_scope='discount_validate' روی
|
||
ValidateDiscountView: قبلاً این endpoint هیچ throttle ای نداشت، در حالی
|
||
که type/value کدهای معتبر رو هم برمی‌گردوند — یعنی یک کاربرِ
|
||
احرازهویت‌شده می‌تونست بی‌نهایت کدهای تخفیف رو probe کنه. این تست‌ها
|
||
هم رفتارِ throttle (per-user، نه IP)، هم عدمِ رگرسیون در پاسخِ
|
||
validation و رفتارِ authentication رو تضمین می‌کنن.
|
||
"""
|
||
|
||
def setUp(self):
|
||
cache.clear()
|
||
self.client = APIClient()
|
||
self.user = User.objects.create(device_id='device-validate-throttle-1', is_guest=True, name='')
|
||
self.token = issue_access_token(self.user.id)
|
||
self.discount = DiscountCode.objects.create(
|
||
code='THROTTLE10', type='percent', value=10, max_uses=5, expires_at=None,
|
||
)
|
||
|
||
def _validate(self, code, token=None):
|
||
headers = {}
|
||
if token:
|
||
headers['HTTP_AUTHORIZATION'] = f'Bearer {token}'
|
||
return self.client.get('/api/discount/validate', {'code': code}, **headers)
|
||
|
||
def test_unauthenticated_request_is_still_rejected(self):
|
||
# رفتارِ IsAuthenticated موجود دست‌نخورده مونده؛ throttle جایگزینِ
|
||
# authentication نیست.
|
||
response = self._validate('ANYCODE')
|
||
self.assertEqual(response.status_code, 401)
|
||
|
||
def test_repeated_code_probing_is_throttled_after_configured_limit(self):
|
||
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'discount_validate': '3/hour'}):
|
||
for i in range(3):
|
||
response = self._validate(f'PROBE-{i}', token=self.token)
|
||
self.assertEqual(response.status_code, 200)
|
||
|
||
blocked_response = self._validate('PROBE-EXTRA', token=self.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='+989120000731')
|
||
other_token = issue_access_token(other_user.id)
|
||
|
||
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'discount_validate': '1/hour'}):
|
||
first_user_response = self._validate('PROBE-A', token=self.token)
|
||
self.assertEqual(first_user_response.status_code, 200)
|
||
|
||
first_user_blocked = self._validate('PROBE-B', token=self.token)
|
||
self.assertEqual(first_user_blocked.status_code, 429)
|
||
|
||
# کاربرِ دومِ متفاوت هنوز باید بتونه چک کنه — throttle
|
||
# per-user است، نه یک شمارنده‌ی سراسری روی endpoint.
|
||
second_user_response = self._validate('PROBE-C', token=other_token)
|
||
self.assertEqual(second_user_response.status_code, 200)
|
||
|
||
def test_valid_code_response_unchanged_within_the_limit(self):
|
||
# عدمِ رگرسیون: پاسخِ اعتبارسنجیِ کدِ معتبر دقیقاً مثلِ قبل باید کار کنه.
|
||
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'discount_validate': '10/hour'}):
|
||
response = self._validate(self.discount.code, token=self.token)
|
||
|
||
self.assertEqual(response.status_code, 200)
|
||
self.assertTrue(response.data['valid'])
|
||
self.assertEqual(response.data['type'], 'percent')
|
||
self.assertEqual(response.data['value'], 10)
|
||
|
||
def test_invalid_code_response_unchanged_within_the_limit(self):
|
||
# عدمِ رگرسیون: پاسخ برای کدِ نامعتبر باید همچنان ۲۰۰ با valid=False
|
||
# باشه (نه ۴۲۹)، تا وقتی که به سقفِ throttle نرسیده باشیم.
|
||
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'discount_validate': '10/hour'}):
|
||
response = self._validate('NOT-A-REAL-CODE', token=self.token)
|
||
|
||
self.assertEqual(response.status_code, 200)
|
||
self.assertFalse(response.data['valid'])
|
||
|
||
|
||
class RedeemDiscountThrottleTests(TestCase):
|
||
"""
|
||
🔒 اثبات محدودیتِ جدیدِ throttle_scope='discount_redeem' روی
|
||
RedeemDiscountView: این endpoint همون oracle ای که ValidateDiscountView
|
||
و RedeemOfferCodeView رو throttle کرد داره (۴۰۴ برای کدِ ناموجود،
|
||
۴۰۰ برای کدِ نامعتبر)، با این تفاوت که حدسِ درست یک کدِ واقعی رو
|
||
مصرف می‌کنه، نه فقط اطلاعات لو می‌ده. این تست‌ها هم رفتارِ throttle
|
||
(per-user، نه IP)، هم عدمِ رگرسیون در ردیمِ موفق رو تضمین می‌کنن.
|
||
"""
|
||
|
||
def setUp(self):
|
||
cache.clear()
|
||
self.client = APIClient()
|
||
self.user = User.objects.create(device_id='device-redeem-throttle-1', is_guest=True, name='')
|
||
self.token = issue_access_token(self.user.id)
|
||
self.order = Order.objects.create(
|
||
user_id=self.user.id, amount=10000, idempotency_key='order-redeem-throttle-1',
|
||
)
|
||
|
||
def _redeem(self, code, order_id=None, token=None):
|
||
headers = {}
|
||
if token:
|
||
headers['HTTP_AUTHORIZATION'] = f'Bearer {token}'
|
||
return self.client.post(
|
||
'/api/discount/redeem',
|
||
{'code': code, 'orderId': order_id or self.order.id},
|
||
format='json',
|
||
**headers,
|
||
)
|
||
|
||
def test_unauthenticated_request_is_still_rejected(self):
|
||
# رفتارِ IsAuthenticated موجود دست‌نخورده مونده؛ throttle جایگزینِ
|
||
# authentication نیست.
|
||
response = self._redeem('ANYCODE')
|
||
self.assertEqual(response.status_code, 401)
|
||
|
||
def test_repeated_code_probing_is_throttled_after_configured_limit(self):
|
||
# سناریوی brute-force: کدهای ناموجودِ پشتِ‌سرِهم روی همون سفارش.
|
||
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'discount_redeem': '3/hour'}):
|
||
for i in range(3):
|
||
response = self._redeem(f'PROBE-{i}', token=self.token)
|
||
self.assertEqual(response.status_code, 404)
|
||
|
||
blocked_response = self._redeem('PROBE-EXTRA', token=self.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='+989120000301')
|
||
other_token = issue_access_token(other_user.id)
|
||
other_order = Order.objects.create(
|
||
user_id=other_user.id, amount=5000, idempotency_key='order-redeem-throttle-2',
|
||
)
|
||
|
||
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'discount_redeem': '1/hour'}):
|
||
first_user_response = self._redeem('PROBE-A', token=self.token)
|
||
self.assertEqual(first_user_response.status_code, 404)
|
||
|
||
first_user_blocked = self._redeem('PROBE-B', token=self.token)
|
||
self.assertEqual(first_user_blocked.status_code, 429)
|
||
|
||
# کاربرِ دومِ متفاوت هنوز باید بتونه تلاش کنه — throttle
|
||
# per-user است، نه یک شمارنده‌ی سراسری روی endpoint.
|
||
second_user_response = self._redeem('PROBE-C', order_id=other_order.id, token=other_token)
|
||
self.assertEqual(second_user_response.status_code, 404)
|
||
|
||
def test_successful_redemption_still_works_within_the_limit(self):
|
||
# عدمِ رگرسیون: رفتارِ اصلیِ redemption باید دقیقاً مثلِ قبل کار کنه.
|
||
discount = DiscountCode.objects.create(
|
||
code='REDEEMTHROTTLE10', type='percent', value=10, max_uses=5, expires_at=None,
|
||
)
|
||
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'discount_redeem': '10/hour'}):
|
||
response = self._redeem(discount.code, token=self.token)
|
||
|
||
self.assertEqual(response.status_code, 200)
|
||
self.assertTrue(response.data['success'])
|
||
self.assertEqual(response.data['discountAmount'], 1000)
|
||
self.order.refresh_from_db()
|
||
self.assertEqual(self.order.amount, 9000)
|
||
|
||
def test_blocked_request_does_not_mutate_order(self):
|
||
# درخواستِ بلاک‌شده نباید هیچ Redemption ای بسازه یا مبلغِ سفارش
|
||
# رو تغییر بده — throttle باید *قبل* از رسیدن به منطقِ ویو رد کنه.
|
||
discount = DiscountCode.objects.create(
|
||
code='REDEEMTHROTTLEBLOCK', type='fixed', value=1000, max_uses=5, expires_at=None,
|
||
)
|
||
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'discount_redeem': '1/hour'}):
|
||
first_response = self._redeem('PROBE-X', token=self.token)
|
||
self.assertEqual(first_response.status_code, 404)
|
||
|
||
blocked_response = self._redeem(discount.code, token=self.token)
|
||
|
||
self.assertEqual(blocked_response.status_code, 429)
|
||
self.order.refresh_from_db()
|
||
self.assertEqual(self.order.amount, 10000)
|
||
|
||
|
||
class RedeemDiscountGuestThrottleBypassTests(TestCase):
|
||
"""
|
||
🔒 اثبات فیکسِ امنیتی: قبلاً کلیدِ throttle_scope='discount_redeem' برای
|
||
هر کاربرِ احرازهویت‌شده (مهمان یا نه) user.pk بود. چون GuestSessionView
|
||
با هر deviceId جدید یک کاربرِ مهمانِ تازه (pk جدید) می‌سازه، یک مهاجم
|
||
می‌تونست با ساختنِ پی‌درپیِ حساب‌های مهمان، هر بار یک bucket throttle
|
||
تازه برای probe/brute-force کردنِ کدهای تخفیف بگیره. چون DiscountCode
|
||
یک رکوردِ سراسری با max_uses مشترک است، این دورزدن صرفاً افشای
|
||
اطلاعات نبود؛ به مهاجم guessهای واقعیِ بیشتری روی یک کدِ سراسری
|
||
می‌داد که هرکدوم می‌تونست یک Redemption واقعی بسازه و مبلغِ سفارش رو
|
||
کم کنه. حالا GuestAwareDiscountThrottle برای کاربرهای مهمان کلید رو
|
||
بر اساسِ IP (همون get_ident که guest_session هم استفاده می‌کنه)
|
||
می‌سازه، پس چرخشِ حساب مهمان دیگه bucket تازه نمی‌ده. رفتارِ throttle
|
||
برای کاربرِ غیرمهمان (per-user، مثلِ قبل) دست‌نخورده مونده.
|
||
"""
|
||
|
||
def setUp(self):
|
||
cache.clear()
|
||
self.client = APIClient()
|
||
|
||
def _make_guest_with_order(self, device_id, amount=10000):
|
||
guest = User.objects.create(device_id=device_id, is_guest=True, name='')
|
||
order = Order.objects.create(
|
||
user_id=guest.id, amount=amount, idempotency_key=f'order-{device_id}',
|
||
)
|
||
return guest, order
|
||
|
||
def _redeem(self, code, order_id, token):
|
||
return self.client.post(
|
||
'/api/discount/redeem',
|
||
{'code': code, 'orderId': order_id},
|
||
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, {'discount_redeem': '2/hour'}):
|
||
guest1, order1 = self._make_guest_with_order('device-redeem-guest-rot-1')
|
||
response1 = self._redeem('PROBE-ROT-1', order1.id, issue_access_token(guest1.id))
|
||
self.assertEqual(response1.status_code, 404)
|
||
|
||
guest2, order2 = self._make_guest_with_order('device-redeem-guest-rot-2')
|
||
response2 = self._redeem('PROBE-ROT-2', order2.id, issue_access_token(guest2.id))
|
||
self.assertEqual(response2.status_code, 404)
|
||
|
||
# سومین حسابِ مهمانِ تازه (pk جدید) باید throttle بشه، چون
|
||
# bucket بر اساسِ IP به‌اشتراک گذاشته شده، نه user.pk — یعنی
|
||
# سقفِ ۲/ساعت هنوز اعمال می‌شه، صرف‌نظر از تعدادِ حساب‌های
|
||
# مهمانِ ساخته‌شده.
|
||
guest3, order3 = self._make_guest_with_order('device-redeem-guest-rot-3')
|
||
response3 = self._redeem('PROBE-ROT-3', order3.id, issue_access_token(guest3.id))
|
||
|
||
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, {'discount_redeem': '1/hour'}):
|
||
guest1, order1 = self._make_guest_with_order('device-redeem-guest-rot-4')
|
||
response1 = self._redeem('PROBE-ROT-4', order1.id, issue_access_token(guest1.id))
|
||
self.assertEqual(response1.status_code, 404)
|
||
|
||
guest2, order2 = self._make_guest_with_order('device-redeem-guest-rot-5')
|
||
blocked = self._redeem('PROBE-ROT-5', order2.id, issue_access_token(guest2.id))
|
||
self.assertEqual(blocked.status_code, 429)
|
||
|
||
guest3, order3 = self._make_guest_with_order('device-redeem-guest-rot-6')
|
||
still_blocked = self._redeem('PROBE-ROT-6', order3.id, issue_access_token(guest3.id))
|
||
|
||
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, {'discount_redeem': '1/hour'}):
|
||
user_a = User.objects.create(is_guest=False, name='', phone='+989120000401')
|
||
order_a = Order.objects.create(
|
||
user_id=user_a.id, amount=10000, idempotency_key='order-redeem-reg-a',
|
||
)
|
||
response_a = self._redeem('PROBE-REG-A', order_a.id, issue_access_token(user_a.id))
|
||
self.assertEqual(response_a.status_code, 404)
|
||
|
||
user_b = User.objects.create(is_guest=False, name='', phone='+989120000402')
|
||
order_b = Order.objects.create(
|
||
user_id=user_b.id, amount=10000, idempotency_key='order-redeem-reg-b',
|
||
)
|
||
response_b = self._redeem('PROBE-REG-B', order_b.id, issue_access_token(user_b.id))
|
||
|
||
self.assertEqual(response_b.status_code, 404)
|
||
|
||
def test_blocked_guest_request_does_not_mutate_order_or_create_redemption(self):
|
||
# درخواستِ بلاک‌شده (از حسابِ مهمانِ چرخشی) نباید مبلغِ سفارش رو
|
||
# تغییر بده یا Redemption بسازه — throttle باید *قبل* از رسیدن به
|
||
# select_for_update()/منطقِ redemption رد کنه، حتی با کدِ درست.
|
||
discount = DiscountCode.objects.create(
|
||
code='REDEEMGUESTROTBLOCK', type='fixed', value=1000, max_uses=5, expires_at=None,
|
||
)
|
||
|
||
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'discount_redeem': '1/hour'}):
|
||
guest1, order1 = self._make_guest_with_order('device-redeem-guest-rot-7')
|
||
first_response = self._redeem('PROBE-ROT-7', order1.id, issue_access_token(guest1.id))
|
||
self.assertEqual(first_response.status_code, 404)
|
||
|
||
# حسابِ مهمانِ تازه‌ی دوم، حتی با کدِ درست روی سفارشِ خودش، باید
|
||
# بلاک بشه (bucket مشترکِ IP از قبل مصرف شده).
|
||
guest2, order2 = self._make_guest_with_order('device-redeem-guest-rot-8')
|
||
blocked_response = self._redeem(discount.code, order2.id, issue_access_token(guest2.id))
|
||
|
||
self.assertEqual(blocked_response.status_code, 429)
|
||
order2.refresh_from_db()
|
||
self.assertEqual(order2.amount, 10000)
|
||
self.assertEqual(discount.redemptions.count(), 0)
|
||
|
||
|
||
class ValidateDiscountGuestThrottleBypassTests(TestCase):
|
||
"""
|
||
🔒 اثبات فیکسِ امنیتی: قبلاً کلیدِ throttle_scope='discount_validate' برای
|
||
هر کاربرِ احرازهویت‌شده (مهمان یا نه) user.pk بود، دقیقاً همون مشکلی که
|
||
RedeemDiscountGuestThrottleBypassTests برای discount_redeem اثبات کرده.
|
||
چون GuestSessionView با هر deviceId جدید یک کاربرِ مهمانِ تازه (pk جدید)
|
||
می‌سازه، یک مهاجم می‌تونست با ساختنِ پی‌درپیِ حساب‌های مهمان، هر بار یک
|
||
bucket throttle تازه برای probe کردنِ type/value کدهای تخفیف بگیره.
|
||
حالا GuestAwareDiscountThrottle برای کاربرهای مهمان کلید رو بر اساسِ IP
|
||
می‌سازه، پس چرخشِ حساب مهمان دیگه bucket تازه نمی‌ده. رفتارِ throttle
|
||
برای کاربرِ غیرمهمان (per-user، مثلِ قبل) دست‌نخورده مونده.
|
||
"""
|
||
|
||
def setUp(self):
|
||
cache.clear()
|
||
self.client = APIClient()
|
||
self.discount = DiscountCode.objects.create(
|
||
code='GUESTROTVALIDATE10', type='percent', value=10, max_uses=5, expires_at=None,
|
||
)
|
||
|
||
def _validate(self, code, token):
|
||
return self.client.get(
|
||
'/api/discount/validate', {'code': code}, 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, {'discount_validate': '2/hour'}):
|
||
guest1 = User.objects.create(device_id='device-validate-guest-rot-1', is_guest=True, name='')
|
||
response1 = self._validate(self.discount.code, issue_access_token(guest1.id))
|
||
self.assertEqual(response1.status_code, 200)
|
||
|
||
guest2 = User.objects.create(device_id='device-validate-guest-rot-2', is_guest=True, name='')
|
||
response2 = self._validate(self.discount.code, issue_access_token(guest2.id))
|
||
self.assertEqual(response2.status_code, 200)
|
||
|
||
# سومین حسابِ مهمانِ تازه (pk جدید) باید throttle بشه، چون
|
||
# bucket بر اساسِ IP به‌اشتراک گذاشته شده، نه user.pk.
|
||
guest3 = User.objects.create(device_id='device-validate-guest-rot-3', is_guest=True, name='')
|
||
response3 = self._validate(self.discount.code, issue_access_token(guest3.id))
|
||
|
||
self.assertEqual(response3.status_code, 429)
|
||
|
||
def test_registered_users_keep_independent_per_user_buckets(self):
|
||
# عدمِ رگرسیون: کاربرهای غیرمهمان (ثبت‌نام‌شده با شماره) باید مثلِ
|
||
# قبل هرکدوم bucket جداگانه‌ی خودشون رو داشته باشن.
|
||
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'discount_validate': '1/hour'}):
|
||
user_a = User.objects.create(is_guest=False, name='', phone='+989120000741')
|
||
response_a = self._validate(self.discount.code, issue_access_token(user_a.id))
|
||
self.assertEqual(response_a.status_code, 200)
|
||
|
||
user_b = User.objects.create(is_guest=False, name='', phone='+989120000742')
|
||
response_b = self._validate(self.discount.code, issue_access_token(user_b.id))
|
||
|
||
self.assertEqual(response_b.status_code, 200)
|