""" حفاظ‌های بازبینیِ امنیتیِ پیش از تحویل (هویت و حساب). هر تست رفتاری را نگه می‌دارد که پیش‌تر غلط بود. """ 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') def _staff_with(*perms, username='staff1'): from django.contrib.auth.models import Permission staff = get_user_model().objects.create_user(username, f'{username}@example.com', 'pw-for-tests-only', is_staff=True) for app_label, codename in perms: staff.user_permissions.add(Permission.objects.get(content_type__app_label=app_label, codename=codename)) return staff class DeviceIdNotGuessableByAdminFilterTests(TestCase): """فیلترِ querystringِ changelist (?device_id__startswith=…) شناسه‌ی دستگاه را برای غیرِ ابرکاربر لو نمی‌دهد.""" def setUp(self): call_command('setup_supervisor_group', verbosity=0) self.guest = User.objects.create(device_id='abc-secret-999', is_guest=True) from accounts.models import Session Session.objects.create(user=self.guest, refresh_token='h' * 64, device_id='abc-secret-999', expires_at=timezone.now() + timezone.timedelta(days=1)) staff = _staff_with(('accounts', 'view_session')) staff.groups.add(Group.objects.get(name='Supervisor')) self.client.force_login(staff) def test_filter_on_device_id_is_refused(self): for lookup in ('device_id__startswith', 'device_id__regex', 'device_id'): r = self.client.get('/admin/accounts/user/', {lookup: 'abc'}) self.assertEqual(r.status_code, 400, lookup) self.assertNotContains(r, f'/admin/accounts/user/{self.guest.pk}/change/', status_code=400) r = self.client.get('/admin/accounts/session/', {'device_id__startswith': 'abc'}) self.assertEqual(r.status_code, 400) def test_superuser_still_filters(self): root = get_user_model().objects.create_superuser('root', 'root@example.com', 'pw-for-tests-only') self.client.force_login(root) r = self.client.get('/admin/accounts/user/', {'device_id__startswith': 'abc'}) self.assertContains(r, f'/admin/accounts/user/{self.guest.pk}/change/') class StaffCannotSwapLoginPhoneTests(TestCase): """کارمندی با change_user شماره‌ی کاربر را عوض نمی‌کند؛ وگرنه با شماره‌ی خودش کد می‌گیرد و واردِ آن حساب می‌شود.""" def setUp(self): cache.clear() self.victim = User.objects.create(phone='09120000077', is_guest=False, name='victim') self.client.force_login(_staff_with(('accounts', 'change_user'), ('accounts', 'view_user'))) def tearDown(self): cache.clear() def test_phone_and_is_guest_are_read_only(self): r = self.client.post(f'/admin/accounts/user/{self.victim.pk}/change/', { 'phone': '09129999999', 'name': 'renamed', 'avatar': '', 'grade_id': '', 'is_guest': 'on'}) self.assertEqual(r.status_code, 302) self.victim.refresh_from_db() self.assertEqual(self.victim.name, 'renamed') self.assertEqual(self.victim.phone, '09120000077') self.assertFalse(self.victim.is_guest) login = APIClient().post('/api/auth/login', {'phone': '09129999999', 'otp': _otp_row('09129999999')}, format='json') self.assertNotEqual(login.json().get('userId'), str(self.victim.pk)) class StaffAccountsAdminSuperuserOnlyTests(TestCase): """auth.User/Group پنل: با change_group یا change_user (auth) کارمند خودش را ابرکاربر یا همه‌کاره نمی‌کند.""" def test_change_group_cannot_grant_all_permissions(self): from django.contrib.auth.models import Permission call_command('setup_supervisor_group', verbosity=0) staff = _staff_with(('auth', 'change_group'), ('auth', 'view_group')) group = Group.objects.get(name='Supervisor') staff.groups.add(group) self.client.force_login(staff) r = self.client.post(f'/admin/auth/group/{group.pk}/change/', {'name': 'Supervisor', 'permissions': list(Permission.objects.values_list('pk', flat=True))}) self.assertEqual(r.status_code, 403) staff = get_user_model().objects.get(pk=staff.pk) self.assertFalse(staff.has_perm('accounts.add_otpcode')) def test_change_user_cannot_make_self_superuser(self): staff = _staff_with(('auth', 'change_user'), ('auth', 'view_user')) self.client.force_login(staff) form = self.client.get(f'/admin/auth/user/{staff.pk}/change/') self.assertEqual(form.status_code, 200) self.assertNotContains(form, 'name="is_superuser"') r = self.client.post(f'/admin/auth/user/{staff.pk}/change/', { 'username': staff.username, 'is_active': 'on', 'is_staff': 'on', 'is_superuser': 'on', 'last_login_0': '', 'last_login_1': '', 'date_joined_0': '2026-01-01', 'date_joined_1': '00:00:00'}) self.assertEqual(r.status_code, 403) self.assertFalse(get_user_model().objects.get(pk=staff.pk).is_superuser) class UnicodeInputsAreNormalisedOrRejectedTests(TestCase): """رقم‌های تمام‌عرض/ریاضی، کدِ غیرِ ASCII، پایه‌ی ناموجود و deviceIdِ null: ۴۰۰ یا نرمال‌شده، نه ۵۰۰ و نه حسابِ جدا.""" def setUp(self): cache.clear() self.addCleanup(cache.clear) def test_fullwidth_and_math_digits_normalise_to_same_phone(self): from accounts.phone import normalize_phone for variant in ('０９１２３４５６７８９', '𝟎𝟗𝟏𝟐𝟑𝟒𝟓𝟔𝟕𝟖𝟗', '۰۹۱۲۳۴۵۶۷۸۹', '+98 912 345 6789'): self.assertEqual(normalize_phone(variant), '09123456789', variant) self.assertIsNone(normalize_phone('0912345678²')) def test_non_ascii_otp_is_400_not_500(self): phone = '09120000099' with override_settings(TEST_LOGIN_PHONES={phone: '123456'}): for code in ('²²²²', '١٢٣٤٥٦٧'[:6] + 'x', 'abcd'): r = APIClient().post('/api/auth/login', {'phone': phone, 'otp': code}, format='json') self.assertEqual(r.status_code, 400, code) def test_login_device_id_null_is_400(self): phone = '09120000098' code = _otp_row(phone) r = APIClient().post('/api/auth/login', {'phone': phone, 'otp': code, 'deviceId': None}, format='json') self.assertEqual(r.status_code, 400) def test_unknown_or_huge_grade_is_400(self): user = User.objects.create(phone='09120000097', is_guest=False) app = _bearer(issue_access_token(user.id)) for gid in (99999, 10 ** 30): self.assertEqual(app.post('/api/user/grade', {'gradeId': gid}, format='json').status_code, 400, gid) def test_birth_date_with_superscript_digit_is_400(self): user = User.objects.create(phone='09120000096', is_guest=False) r = _bearer(issue_access_token(user.id)).patch('/api/user/profile', {'birthDate': '13²5-04-12'}, format='json') self.assertEqual(r.status_code, 400) def test_display_name_nfkc_contacts_removed(self): from accounts.serializers import clean_display_name as clean for sample in ('０９１２３４５６７８９', '𝟎𝟗𝟏𝟐𝟑𝟒𝟓𝟔𝟕𝟖𝟗', '0912*345*6789', '＠kid2014'): self.assertEqual(clean(sample), '', sample) self.assertEqual(clean('علی رضایی'), 'علی رضایی') class DemoGuestNotUpgradableTests(TestCase): """مهمانِ نمایشیِ مشترک با ورودِ یک شماره به آن شماره تبدیل نمی‌شود؛ شماره‌ی تازه حسابِ تازه می‌گیرد.""" def test_demo_guest_keeps_its_identity(self): cache.clear() self.addCleanup(cache.clear) s = _guest('frontend-test-device') demo_id = int(s['userId']) phone = '09120000095' code = _otp_row(phone) r = _bearer(s['token']).post('/api/auth/login', {'phone': phone, 'otp': code}, format='json') self.assertEqual(r.status_code, 200) self.assertNotEqual(int(r.json()['userId']), demo_id) demo = User.objects.get(pk=demo_id) self.assertIsNone(demo.phone) self.assertTrue(demo.is_guest)