367 lines
19 KiB
Python
367 lines
19 KiB
Python
import importlib
|
|
import os
|
|
from unittest.mock import patch
|
|
|
|
from django.core.cache import cache
|
|
from django.test import TestCase, override_settings
|
|
from rest_framework.test import APIClient, APIRequestFactory
|
|
from rest_framework.throttling import ScopedRateThrottle
|
|
|
|
from accounts.authentication import issue_access_token
|
|
from accounts.models import User
|
|
|
|
from . import bazaar
|
|
from .bazaar import BazaarPurchaseResult
|
|
from .models import Order, Transaction
|
|
from .permissions import HasInternalServiceKey, HasWebhookSecret
|
|
|
|
|
|
class SecretDefaultsFailClosedTests(TestCase):
|
|
"""
|
|
تست‌های متمرکز روی این فیکس: INTERNAL_SERVICE_KEY و PAYMENT_WEBHOOK_SECRET
|
|
دیگه در پروداکشن (DEBUG=False) دیفالتِ قابل‌حدس‌زدن ندارن. اگه env تنظیم
|
|
نشده باشه، مقدار خالی می‌مونه و پرمیشن‌های مربوطه صراحتاً رد می‌کنن
|
|
(fail closed)، نه اینکه با یک مقدارِ شناخته‌شده قبول کنن.
|
|
"""
|
|
|
|
def setUp(self):
|
|
self.factory = APIRequestFactory()
|
|
|
|
@override_settings(DEBUG=False, INTERNAL_SERVICE_KEY='')
|
|
def test_production_missing_internal_key_rejects(self):
|
|
request = self.factory.post('/api/payment/orders')
|
|
# حتی اگه کلاینت هیچ هدری هم نفرسته یا مقدارِ خالی بفرسته، چون
|
|
# settings.INTERNAL_SERVICE_KEY خودش خالیه، درخواست باید رد بشه.
|
|
self.assertFalse(HasInternalServiceKey().has_permission(request, None))
|
|
|
|
@override_settings(DEBUG=False, PAYMENT_WEBHOOK_SECRET='')
|
|
def test_production_missing_webhook_secret_rejects(self):
|
|
request = self.factory.post('/api/payment/webhook')
|
|
self.assertFalse(HasWebhookSecret().has_permission(request, None))
|
|
|
|
@override_settings(
|
|
DEBUG=False,
|
|
INTERNAL_SERVICE_KEY='prod-internal-key-xyz',
|
|
PAYMENT_WEBHOOK_SECRET='prod-webhook-secret-xyz',
|
|
)
|
|
def test_production_configured_secrets_accepts(self):
|
|
internal_request = self.factory.post(
|
|
'/api/payment/orders', HTTP_X_INTERNAL_KEY='prod-internal-key-xyz',
|
|
)
|
|
self.assertTrue(HasInternalServiceKey().has_permission(internal_request, None))
|
|
|
|
webhook_request = self.factory.post(
|
|
'/api/payment/webhook', HTTP_X_WEBHOOK_SECRET='prod-webhook-secret-xyz',
|
|
)
|
|
self.assertTrue(HasWebhookSecret().has_permission(webhook_request, None))
|
|
|
|
def test_debug_mode_local_fallback_still_valid(self):
|
|
# settings.py این مقادیر رو یک‌بار موقعِ import محاسبه می‌کنه، نه
|
|
# چیزی که override_settings بعداً toggle کنه؛ پس برای تستِ این
|
|
# منطق باید خودِ ماژول رو reload کرد.
|
|
from config import settings as settings_module
|
|
|
|
env_overrides = {'DJANGO_DEBUG': '1'}
|
|
env_removed = {'INTERNAL_SERVICE_KEY', 'PAYMENT_WEBHOOK_SECRET'}
|
|
clean_env = {
|
|
k: v for k, v in os.environ.items() if k not in env_removed
|
|
}
|
|
clean_env.update(env_overrides)
|
|
|
|
try:
|
|
with patch.dict(os.environ, clean_env, clear=True):
|
|
importlib.reload(settings_module)
|
|
self.assertTrue(settings_module.DEBUG)
|
|
self.assertEqual(
|
|
settings_module.INTERNAL_SERVICE_KEY,
|
|
'dev-internal-service-key-change-me',
|
|
)
|
|
self.assertEqual(
|
|
settings_module.PAYMENT_WEBHOOK_SECRET,
|
|
'dev-webhook-secret-change-me',
|
|
)
|
|
finally:
|
|
# همیشه ماژول رو به حالتِ واقعیِ محیطِ تست برمی‌گردونیم تا روی
|
|
# بقیه‌ی تست‌ها اثر نذاره.
|
|
importlib.reload(settings_module)
|
|
|
|
|
|
class BazaarVerifyPurchaseFailClosedTests(TestCase):
|
|
"""
|
|
تست‌های متمرکز روی فیکسِ باگ fail-open در bazaar.verify_purchase: قبلاً
|
|
وقتی BAZAAR_REFRESH_TOKEN (کلید سرور-به-سرورِ بازار) ست نشده بود، هر
|
|
purchaseTokenِ غیرخالیِ ارسالی از کلاینت valid فرض می‌شد. حالا باید در
|
|
این حالت همیشه invalid برگرده، و مسیر واقعیِ صحت‌سنجی با بازار (وقتی
|
|
کلیدها ست شدن) دست‌نخورده کار کنه.
|
|
"""
|
|
|
|
def setUp(self):
|
|
# کش access_token رو بین تست‌ها پاک می‌کنیم تا یک تست روی تست بعدی
|
|
# اثر نذاره.
|
|
cache.clear()
|
|
|
|
@override_settings(BAZAAR_REFRESH_TOKEN='')
|
|
def test_missing_credentials_never_trusts_client_token(self):
|
|
# حتی با یک purchaseToken دلخواهِ غیرخالی، بدون کلیدهای بازار نباید
|
|
# valid برگرده.
|
|
result = bazaar.verify_purchase(
|
|
product_sku='some_sku', purchase_token='arbitrary-client-supplied-token',
|
|
)
|
|
self.assertFalse(result.is_valid)
|
|
|
|
@override_settings(
|
|
BAZAAR_REFRESH_TOKEN='refresh-token-xyz',
|
|
BAZAAR_CLIENT_ID='client-id',
|
|
BAZAAR_CLIENT_SECRET='client-secret',
|
|
BAZAAR_PACKAGE_NAME='ir.tadrisino.app',
|
|
BAZAAR_API_BASE='https://pardakht.cafebazaar.ir/devapi/v2',
|
|
)
|
|
@patch('payment.bazaar.requests.get')
|
|
@patch('payment.bazaar.requests.post')
|
|
def test_configured_credentials_and_successful_response_is_valid(self, mock_post, mock_get):
|
|
mock_post.return_value.raise_for_status = lambda: None
|
|
mock_post.return_value.json.return_value = {
|
|
'access_token': 'real-access-token', 'expires_in': 3600,
|
|
}
|
|
mock_get.return_value.status_code = 200
|
|
mock_get.return_value.json.return_value = {
|
|
'purchaseState': 0, 'consumptionState': 0, 'developerPayload': '',
|
|
}
|
|
|
|
result = bazaar.verify_purchase(product_sku='sku1', purchase_token='real-token')
|
|
|
|
self.assertTrue(result.is_valid)
|
|
mock_get.assert_called_once()
|
|
_, get_kwargs = mock_get.call_args
|
|
self.assertEqual(get_kwargs['headers']['Authorization'], 'Bearer real-access-token')
|
|
|
|
@override_settings(
|
|
BAZAAR_REFRESH_TOKEN='refresh-token-xyz',
|
|
BAZAAR_CLIENT_ID='client-id',
|
|
BAZAAR_CLIENT_SECRET='client-secret',
|
|
BAZAAR_PACKAGE_NAME='ir.tadrisino.app',
|
|
BAZAAR_API_BASE='https://pardakht.cafebazaar.ir/devapi/v2',
|
|
)
|
|
@patch('payment.bazaar.requests.get')
|
|
@patch('payment.bazaar.requests.post')
|
|
def test_configured_credentials_provider_reports_invalid(self, mock_post, mock_get):
|
|
mock_post.return_value.raise_for_status = lambda: None
|
|
mock_post.return_value.json.return_value = {
|
|
'access_token': 'real-access-token', 'expires_in': 3600,
|
|
}
|
|
mock_get.return_value.status_code = 200
|
|
# purchaseState=1 یعنی لغوشده (طبق سند بازار)، یعنی خودِ سرور بازار
|
|
# می‌گه این خرید معتبر نیست.
|
|
mock_get.return_value.json.return_value = {
|
|
'purchaseState': 1, 'consumptionState': 0, 'developerPayload': '',
|
|
}
|
|
|
|
result = bazaar.verify_purchase(product_sku='sku1', purchase_token='cancelled-token')
|
|
|
|
self.assertFalse(result.is_valid)
|
|
|
|
|
|
class BazaarVerifyPurchaseThrottleTests(TestCase):
|
|
"""
|
|
🔒 اثبات محدودیتِ جدیدِ per-user روی BazaarVerifyPurchaseView
|
|
(throttle_scope='bazaar_verify'): هر درخواستِ پذیرفته‌شده یک تماسِ
|
|
سرور-به-سرورِ واقعی با API بازار می‌زنه؛ بدونِ throttle مخصوص، یک
|
|
کاربرِ احرازهویت‌شده می‌تونست با ارسالِ مکررِ purchaseToken بارِ زیادی
|
|
روی سرویسِ بازار/بک‌اند بندازه. GuestAwareBazaarThrottle با pk کاربرِ
|
|
احرازهویت‌شده کلید می‌خوره (نه IP)، پس تغییرِ IP این محدودیت رو دور
|
|
نمی‌زنه.
|
|
"""
|
|
|
|
def setUp(self):
|
|
cache.clear()
|
|
self.client = APIClient()
|
|
self.user = User.objects.create(device_id='device-bazaar-throttle-1', is_guest=True, name='')
|
|
self.token = issue_access_token(self.user.id)
|
|
|
|
def _make_order(self, key):
|
|
return Order.objects.create(
|
|
user_id=self.user.id, amount=10000, provider='bazaar',
|
|
bazaar_product_sku='sku_one_time', idempotency_key=key,
|
|
)
|
|
|
|
def _verify(self, order_id, purchase_token, 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/payment/bazaar/verify',
|
|
{'orderId': order_id, 'purchaseToken': purchase_token},
|
|
format='json',
|
|
**headers,
|
|
)
|
|
|
|
def test_unauthenticated_request_is_still_rejected(self):
|
|
response = self._verify(order_id='1', purchase_token='tok')
|
|
self.assertEqual(response.status_code, 401)
|
|
|
|
def test_requests_within_the_limit_still_succeed(self):
|
|
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'bazaar_verify': '3/hour'}):
|
|
with patch('payment.views.bazaar.verify_purchase') as mock_verify:
|
|
mock_verify.return_value = BazaarPurchaseResult(is_valid=True, purchase_state=0)
|
|
for i in range(3):
|
|
order = self._make_order(f'order-bazaar-within-{i}')
|
|
response = self._verify(order.id, f'token-within-{i}', token=self.token)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
def test_excessive_verify_attempts_from_same_user_are_blocked(self):
|
|
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'bazaar_verify': '3/hour'}):
|
|
with patch('payment.views.bazaar.verify_purchase') as mock_verify:
|
|
mock_verify.return_value = BazaarPurchaseResult(is_valid=True, purchase_state=0)
|
|
for i in range(3):
|
|
order = self._make_order(f'order-bazaar-block-{i}')
|
|
response = self._verify(order.id, f'token-block-{i}', token=self.token)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
extra_order = self._make_order('order-bazaar-block-extra')
|
|
blocked_response = self._verify(extra_order.id, 'token-block-extra', token=self.token)
|
|
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='+989120000751')
|
|
registered_token = issue_access_token(registered_user.id)
|
|
|
|
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'bazaar_verify': '3/hour'}):
|
|
with patch('payment.views.bazaar.verify_purchase') as mock_verify:
|
|
mock_verify.return_value = BazaarPurchaseResult(is_valid=True, purchase_state=0)
|
|
for i in range(3):
|
|
order = Order.objects.create(
|
|
user_id=registered_user.id, amount=10000, provider='bazaar',
|
|
bazaar_product_sku='sku_one_time', idempotency_key=f'order-bazaar-ip-{i}',
|
|
)
|
|
response = self._verify(
|
|
order.id, f'token-ip-{i}', token=registered_token, remote_addr=f'10.4.0.{i + 1}',
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
extra_order = Order.objects.create(
|
|
user_id=registered_user.id, amount=10000, provider='bazaar',
|
|
bazaar_product_sku='sku_one_time', idempotency_key='order-bazaar-ip-extra',
|
|
)
|
|
blocked_response = self._verify(
|
|
extra_order.id, 'token-ip-extra', token=registered_token, remote_addr='10.4.0.99',
|
|
)
|
|
self.assertEqual(blocked_response.status_code, 429)
|
|
|
|
def test_throttled_request_does_not_call_bazaar_or_mutate_order(self):
|
|
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'bazaar_verify': '1/hour'}):
|
|
with patch('payment.views.bazaar.verify_purchase') as mock_verify:
|
|
mock_verify.return_value = BazaarPurchaseResult(is_valid=True, purchase_state=0)
|
|
first_order = self._make_order('order-bazaar-mutate-1')
|
|
first_response = self._verify(first_order.id, 'token-mutate-1', token=self.token)
|
|
self.assertEqual(first_response.status_code, 200)
|
|
|
|
mock_verify.reset_mock()
|
|
second_order = self._make_order('order-bazaar-mutate-2')
|
|
blocked_response = self._verify(second_order.id, 'token-mutate-2', token=self.token)
|
|
self.assertEqual(blocked_response.status_code, 429)
|
|
mock_verify.assert_not_called()
|
|
|
|
second_order.refresh_from_db()
|
|
self.assertEqual(second_order.status, 'pending')
|
|
self.assertEqual(second_order.bazaar_purchase_token, '')
|
|
self.assertFalse(Transaction.objects.filter(order=second_order).exists())
|
|
|
|
|
|
class BazaarVerifyPurchaseGuestThrottleBypassTests(TestCase):
|
|
"""
|
|
🔒 اثبات فیکسِ امنیتی: قبلاً کلیدِ throttle_scope='bazaar_verify' برای
|
|
هر کاربرِ احرازهویت‌شده (مهمان یا نه) user.pk بود. چون GuestSessionView
|
|
با هر deviceId جدید یک کاربرِ مهمانِ تازه (pk جدید) می‌سازه، یک مهاجم
|
|
می‌تونست با ساختنِ پی‌درپیِ حساب‌های مهمان، هر بار یک bucket throttle
|
|
تازه برای تریگر کردنِ تماسِ سرور-به-سرورِ verify_purchase بگیره. حالا
|
|
GuestAwareBazaarThrottle برای کاربرهای مهمان کلید رو بر اساسِ IP
|
|
می‌سازه، پس چرخشِ حساب مهمان دیگه bucket تازه نمی‌ده. رفتارِ throttle
|
|
برای کاربرِ غیرمهمان (per-user، مثلِ قبل) دست‌نخورده مونده.
|
|
"""
|
|
|
|
def setUp(self):
|
|
cache.clear()
|
|
self.client = APIClient()
|
|
|
|
def _make_guest_with_order(self, device_id, key):
|
|
guest = User.objects.create(device_id=device_id, is_guest=True, name='')
|
|
order = Order.objects.create(
|
|
user_id=guest.id, amount=10000, provider='bazaar',
|
|
bazaar_product_sku='sku_one_time', idempotency_key=key,
|
|
)
|
|
return guest, order
|
|
|
|
def _verify(self, order_id, purchase_token, token):
|
|
return self.client.post(
|
|
'/api/payment/bazaar/verify',
|
|
{'orderId': order_id, 'purchaseToken': purchase_token},
|
|
format='json',
|
|
HTTP_AUTHORIZATION=f'Bearer {token}',
|
|
)
|
|
|
|
def test_rotating_guest_accounts_share_the_same_throttle_bucket(self):
|
|
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'bazaar_verify': '2/hour'}):
|
|
with patch('payment.views.bazaar.verify_purchase') as mock_verify:
|
|
mock_verify.return_value = BazaarPurchaseResult(is_valid=True, purchase_state=0)
|
|
|
|
guest1, order1 = self._make_guest_with_order('device-bazaar-guest-rot-1', 'order-bazaar-rot-1')
|
|
response1 = self._verify(order1.id, 'token-rot-1', issue_access_token(guest1.id))
|
|
self.assertEqual(response1.status_code, 200)
|
|
|
|
guest2, order2 = self._make_guest_with_order('device-bazaar-guest-rot-2', 'order-bazaar-rot-2')
|
|
response2 = self._verify(order2.id, 'token-rot-2', issue_access_token(guest2.id))
|
|
self.assertEqual(response2.status_code, 200)
|
|
|
|
# سومین حسابِ مهمانِ تازه (pk جدید) باید throttle بشه، چون
|
|
# bucket بر اساسِ IP به‌اشتراک گذاشته شده، نه user.pk.
|
|
guest3, order3 = self._make_guest_with_order('device-bazaar-guest-rot-3', 'order-bazaar-rot-3')
|
|
response3 = self._verify(order3.id, 'token-rot-3', 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, {'bazaar_verify': '1/hour'}):
|
|
with patch('payment.views.bazaar.verify_purchase') as mock_verify:
|
|
mock_verify.return_value = BazaarPurchaseResult(is_valid=True, purchase_state=0)
|
|
|
|
user_a = User.objects.create(is_guest=False, name='', phone='+989120000761')
|
|
order_a = Order.objects.create(
|
|
user_id=user_a.id, amount=10000, provider='bazaar',
|
|
bazaar_product_sku='sku_one_time', idempotency_key='order-bazaar-reg-a',
|
|
)
|
|
response_a = self._verify(order_a.id, 'token-reg-a', issue_access_token(user_a.id))
|
|
self.assertEqual(response_a.status_code, 200)
|
|
|
|
user_b = User.objects.create(is_guest=False, name='', phone='+989120000762')
|
|
order_b = Order.objects.create(
|
|
user_id=user_b.id, amount=10000, provider='bazaar',
|
|
bazaar_product_sku='sku_one_time', idempotency_key='order-bazaar-reg-b',
|
|
)
|
|
response_b = self._verify(order_b.id, 'token-reg-b', issue_access_token(user_b.id))
|
|
|
|
self.assertEqual(response_b.status_code, 200)
|
|
|
|
def test_blocked_guest_request_does_not_call_bazaar_or_mutate_order(self):
|
|
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'bazaar_verify': '1/hour'}):
|
|
with patch('payment.views.bazaar.verify_purchase') as mock_verify:
|
|
mock_verify.return_value = BazaarPurchaseResult(is_valid=True, purchase_state=0)
|
|
|
|
guest1, order1 = self._make_guest_with_order('device-bazaar-guest-rot-4', 'order-bazaar-rot-4')
|
|
first_response = self._verify(order1.id, 'token-rot-4', issue_access_token(guest1.id))
|
|
self.assertEqual(first_response.status_code, 200)
|
|
|
|
mock_verify.reset_mock()
|
|
guest2, order2 = self._make_guest_with_order('device-bazaar-guest-rot-5', 'order-bazaar-rot-5')
|
|
blocked_response = self._verify(order2.id, 'token-rot-5', issue_access_token(guest2.id))
|
|
self.assertEqual(blocked_response.status_code, 429)
|
|
mock_verify.assert_not_called()
|
|
|
|
order2.refresh_from_db()
|
|
self.assertEqual(order2.status, 'pending')
|
|
self.assertEqual(order2.bazaar_purchase_token, '')
|
|
self.assertFalse(Transaction.objects.filter(order=order2).exists())
|