""" حفاظ‌های بازبینیِ امنیتیِ پیش از تحویل (هویت و حساب). هر تست رفتاری را نگه می‌دارد که پیش‌تر غلط بود. """ import logging import socket from unittest import mock from django.contrib.auth import get_user_model from django.contrib.auth.models import Group from django.core.cache import cache from django.core.management import call_command from django.test import TestCase, override_settings from django.utils import timezone from rest_framework.test import APIClient from accounts.authentication import issue_access_token from accounts.models import OTPCode, User from config.log_formatters import JSONLogFormatter def _bearer(token): c = APIClient() c.credentials(HTTP_AUTHORIZATION=f'Bearer {token}') return c def _otp_row(phone, code='482913'): OTPCode.objects.create(phone=phone, code=code, expires_at=timezone.now() + timezone.timedelta(minutes=5)) return code def _guest(device): return APIClient().post('/api/auth/session', {'type': 'guest', 'deviceId': device}, format='json').json() class GuestUpgradeRevokesEarlierSessionsTests(TestCase): """کسی که پیش از ثبتِ شماره با شناسه‌ی دستگاه نشست گرفته بود، بعد از آن به حسابِ شماره‌دار نمی‌رسید.""" def setUp(self): cache.clear() def test_pre_upgrade_tokens_stop_working(self): attacker = _guest('shared-device') victim = _guest('shared-device') login = _bearer(victim['token']).post('/api/auth/login', {'phone': '09121230001', 'otp': _otp_row('09121230001')}, format='json') self.assertEqual(login.status_code, 200) self.assertEqual(login.json()['userId'], victim['userId']) refresh = APIClient().post('/api/auth/refresh', {'refreshToken': attacker['refreshToken']}, format='json') self.assertEqual(refresh.status_code, 401) # توکنِ دسترسیِ پیش از ثبت هم (اگر همان ثانیه صادر نشده باشد) رد می‌شود؛ نشستِ تازه کار می‌کند. self.assertEqual(_bearer(login.json()['token']).get('/api/user/profile').status_code, 200) fresh = APIClient().post('/api/auth/refresh', {'refreshToken': login.json()['refreshToken']}, format='json') self.assertEqual(fresh.status_code, 200) class AccountDeletionKeepsLimitsTests(TestCase): def setUp(self): cache.clear() from discount.models import DiscountCode from subscription.models import Plan self.plan = Plan.objects.create(title='pack20', price=45000, kind='pack', question_credits=20, hearts_days=7, duration_days=0) DiscountCode.objects.create(code='FREE1', type='percent', value=100, max_uses=1) def _guest_redeems(self, device): s = _guest(device) c = _bearer(s['token']) p = c.post('/api/subscription/purchase', {'planId': str(self.plan.id)}, format='json') return c, c.post('/api/discount/redeem', {'code': 'FREE1', 'orderId': p.json()['orderId']}, format='json') def test_single_use_code_stays_used_after_delete(self): c1, first = self._guest_redeems('g-1') self.assertEqual(first.status_code, 200) self.assertEqual(c1.post('/api/user/delete', {'confirm': 'حذف'}, format='json').status_code, 200) _, again = self._guest_redeems('g-2') self.assertEqual(again.status_code, 400) @override_settings(DEBUG=False, OTP_REQUEST_PHONE_COOLDOWN_SECONDS=0) def test_per_phone_sms_cap_survives_delete(self): phone = '09121230009' user = User.objects.create(phone=phone, is_guest=False) for i in range(10): OTPCode.objects.create(phone=phone, code=f'{i:06d}', is_used=True, expires_at=timezone.now() - timezone.timedelta(minutes=i + 1)) with mock.patch('accounts.views.send_otp_sms'): OTPCode.objects.create(phone=phone, code='777777', expires_at=timezone.now() + timezone.timedelta(minutes=5)) d = _bearer(issue_access_token(user.id)).post('/api/user/delete', {'otp': '777777'}, format='json') self.assertEqual(d.status_code, 200) cache.clear() r = APIClient().post('/api/auth/otp/request', {'phone': phone}, format='json') self.assertEqual(r.status_code, 429) def test_public_demo_guest_cannot_be_deleted(self): demo = User.objects.create(device_id='frontend-test-device', is_guest=True, name='demo') s = _guest('frontend-test-device') self.assertEqual(int(s['userId']), demo.id) r = _bearer(s['token']).post('/api/user/delete', {'confirm': 'حذف'}, format='json') self.assertEqual(r.status_code, 403) self.assertTrue(User.objects.filter(pk=demo.id).exists()) class DisplayNameContactFilterTests(TestCase): def test_contacts_are_removed_and_names_kept(self): from accounts.serializers import clean_display_name as clean for sample in ('0912/123/4567', '۰۹۱۲/۱۲۳/۴۵۶۷', '0912 _123_ 4567', 'ig: ali_kid_2014', 'آیدی: kid2014'): self.assertEqual(clean(sample), '', sample) for name in ('علی رضایی', 'آیدین', 'Sara 2014', 'ایتالیا', 'تلما'): self.assertEqual(clean(name), name) class _Capture(logging.Handler): def __init__(self): super().__init__() self.lines = [] self.setFormatter(JSONLogFormatter()) def emit(self, record): self.lines.append(self.format(record)) FAKE_KEY = '4E6F74415265616C4B6176656E656761724B6579414243444546' @override_settings(KAVENEGAR_API_KEY=FAKE_KEY, KAVENEGAR_OTP_TEMPLATE='SMSVerify') class KavenegarKeyNotLoggedTests(TestCase): """کلیدِ کاوه‌نگار بخشی از آدرس است و متنِ خطای requests آدرس را دارد.""" def test_connection_error_log_has_no_key(self): from accounts import services cap = _Capture() log = logging.getLogger('accounts.services') log.addHandler(cap) def no_dns(*args, **kwargs): raise socket.gaierror(socket.EAI_NONAME, 'network disabled in tests') try: with mock.patch('socket.getaddrinfo', no_dns), self.assertRaises(services.SMSDeliveryError): services.send_verify_lookup(phone='09120000000', template='SMSVerify', token='123456') finally: log.removeHandler(cap) out = '\n'.join(cap.lines) self.assertIn('SMS request to provider failed', out) self.assertNotIn(FAKE_KEY, out) class SupervisorDoesNotSeeDeviceIdsTests(TestCase): """شناسه‌ی دستگاه برای مهمان رمزِ ورود است؛ کارمندِ فقط‌خواندنی آن را نمی‌بیند.""" def setUp(self): call_command('setup_supervisor_group', verbosity=0) self.guest = User.objects.create(device_id='secret-device-123', is_guest=True) from accounts.models import Session Session.objects.create(user=self.guest, refresh_token='h' * 64, device_id='secret-device-123', expires_at=timezone.now() + timezone.timedelta(days=1)) staff = get_user_model().objects.create_user('sup', 'sup@example.com', 'pw-for-tests-only', is_staff=True) staff.groups.add(Group.objects.get(name='Supervisor')) # Supervisor نشست‌ها را نمی‌بیند؛ این‌جا اجازه‌ی دیدن داده می‌شود تا ستونِ آن صفحه هم سنجیده شود. from django.contrib.auth.models import Permission staff.user_permissions.add(Permission.objects.get(codename='view_session', content_type__app_label='accounts')) self.client.force_login(staff) def test_hidden_in_list_search_and_form(self): listing = self.client.get('/admin/accounts/user/') self.assertEqual(listing.status_code, 200) self.assertNotContains(listing, 'secret-device-123') # متنِ جستجو در خودِ صفحه تکرار می‌شود؛ نتیجه نباید همان مهمان باشد. search = self.client.get('/admin/accounts/user/', {'q': 'secret-device-123'}) self.assertNotContains(search, f'/admin/accounts/user/{self.guest.pk}/change/') form = self.client.get(f'/admin/accounts/user/{self.guest.pk}/change/') self.assertNotContains(form, 'secret-device-123') sessions = self.client.get('/admin/accounts/session/') self.assertNotContains(sessions, 'secret-device-123')