Zareian
7ecb337263
- content: POST nodes/{id}/quiz/complete and lessons/{id}/complete refuse a locked island
(403 node_locked, same rule as /api/home) and a user who has not answered every question
(409 answers_missing). Before, only a user with zero hearts and no answers was refused, so
anyone else finished any island, locked or not, and finishing a locked island's lessons
unlocked it. A node or lesson with no questions still passes when unlocked, with no XP.
The client can no longer understate totalQuestions to buy stars.
- throttle_scope on 15 routes that had none (question bank without login, intro, hearts,
notifications, tickets, plans, subscription status, orders, offers, affiliate, assistant
history and feedback, logout).
- admin: OTPCode and Session can only be added/changed by a superuser.
- deploy.sh: integrity-checked DB backup before migrate (keeps the last 2); no migrate without it.
- docs and CLAUDE.md updated to say what is verified and what is not.
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
243 lines
11 KiB
Python
243 lines
11 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.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
|
||
|
||
|
||
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 _user_redeems(self, phone):
|
||
user = User.objects.create(phone=phone)
|
||
c = _bearer(issue_access_token(user.id))
|
||
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._user_redeems('09121230101')
|
||
self.assertEqual(first.status_code, 200)
|
||
OTPCode.objects.create(phone='09121230101', code='777777',
|
||
expires_at=timezone.now() + timezone.timedelta(minutes=5))
|
||
self.assertEqual(c1.post('/api/user/delete', {'otp': '777777'}, format='json').status_code, 200)
|
||
_, again = self._user_redeems('09121230102')
|
||
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)
|
||
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)
|
||
|
||
|
||
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)
|
||
|
||
|
||
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 StaffCannotSwapLoginPhoneTests(TestCase):
|
||
"""کارمندی با change_user شمارهی کاربر را عوض نمیکند؛ وگرنه با شمارهی خودش کد میگیرد و واردِ آن حساب میشود."""
|
||
|
||
def setUp(self):
|
||
cache.clear()
|
||
self.victim = User.objects.create(phone='09120000077', name='victim')
|
||
self.client.force_login(_staff_with(('accounts', 'change_user'), ('accounts', 'view_user')))
|
||
|
||
def tearDown(self):
|
||
cache.clear()
|
||
|
||
def test_phone_is_read_only(self):
|
||
r = self.client.post(f'/admin/accounts/user/{self.victim.pk}/change/', {
|
||
'phone': '09129999999', 'name': 'renamed', 'avatar': '', 'grade_id': ''})
|
||
self.assertEqual(r.status_code, 302)
|
||
self.victim.refresh_from_db()
|
||
self.assertEqual(self.victim.name, 'renamed')
|
||
self.assertEqual(self.victim.phone, '09120000077')
|
||
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و پایهی ناموجود: ۴۰۰ یا نرمالشده، نه ۵۰۰ و نه حسابِ جدا."""
|
||
|
||
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_unknown_or_huge_grade_is_400(self):
|
||
user = User.objects.create(phone='09120000097')
|
||
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')
|
||
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 StaffCannotMintLoginsTests(TestCase):
|
||
"""کارمندِ غیرابرکاربر با اجازهی add نمیتواند OTPCode یا Session دستی بسازد (= ورود به هر شماره)."""
|
||
|
||
def setUp(self):
|
||
from django.contrib.auth.models import Permission, User as AuthUser
|
||
self.staff = AuthUser.objects.create_user('staff1', 's@x.c', 'pw12345!', is_staff=True)
|
||
self.staff.user_permissions.set(Permission.objects.filter(
|
||
codename__in=['add_otpcode', 'change_otpcode', 'view_otpcode',
|
||
'add_session', 'change_session', 'view_session']))
|
||
self.root = AuthUser.objects.create_superuser('root1', 'r@x.c', 'pw12345!')
|
||
|
||
def test_staff_gets_403_on_add_pages(self):
|
||
self.client.force_login(self.staff)
|
||
for model in ('otpcode', 'session'):
|
||
self.assertEqual(self.client.get(f'/admin/accounts/{model}/add/').status_code, 403, model)
|
||
|
||
def test_superuser_can_still_add(self):
|
||
self.client.force_login(self.root)
|
||
for model in ('otpcode', 'session'):
|
||
self.assertEqual(self.client.get(f'/admin/accounts/{model}/add/').status_code, 200, model)
|