Zareian
8835e01595
POST /api/auth/session is gone (404). Existing is_guest rows keep refresh and upgrade-by-phone until the demo/guest rows are purged; the code that serves them is listed in docs/security-before-production.md §10. Tests that used the endpoint as a fixture now use accounts/testing.guest_session. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
334 lines
17 KiB
Python
334 lines
17 KiB
Python
"""
|
||
حفاظ‌های بازبینیِ امنیتیِ پیش از تحویل (هویت و حساب). هر تست رفتاری را نگه می‌دارد که پیش‌تر غلط بود.
|
||
"""
|
||
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.testing import guest_session
|
||
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 guest_session(device)
|
||
|
||
|
||
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)
|