Tadrisino/MathApp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

harden OTP for real SMS and close the second audit round

Browse Source 
Kavenegar is being switched on, so every OTP request now costs credit:
- phone numbers are normalised to 09XXXXXXXXX everywhere (login, OTP,
  family invite, TEST_LOGIN_PHONES keys) and existing rows are migrated;
  non-Iranian and landline numbers are rejected before any SMS
- cooldown and a daily per-phone cap are counted in the database, so the
  three gunicorn workers no longer triple them
- a global daily OTP SMS cap (default 300) stops SMS pumping
- per-IP rates widened for CGNAT / school Wi-Fi (env-overridable)
- provider message id is logged; SMS text warns not to share the code

Second audit round: an offer no longer shortens a longer paid plan,
Zarinpal error bodies no longer 500, island XP reaches the parent report,
a non-numeric questionId no longer 500s after the answer is generated.

596 tests green.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Zareian
2026-09-25 12:28:06 +03:30
parent 74aa9aa8ca
commit 0580aec7b0
24 changed files with 685 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
family/tests.py
View File

@@ -24,10 +24,10 @@ class FamilyLinkAcceptThrottleTests(TestCase):
self.client = APIClient()
self.parent = User.objects.create(device_id='device-family-parent-1', is_guest=True, name='')
self.child = User.objects.create(
device_id='device-family-child-1', is_guest=True, name='', phone='+989120000001',
device_id='device-family-child-1', is_guest=True, name='', phone='09120000001',
)
self.child_token = issue_access_token(self.child.id)
self.registered_user = User.objects.create(is_guest=False, name='', phone='+989120000098')
self.registered_user = User.objects.create(is_guest=False, name='', phone='09120000098')
self.registered_token = issue_access_token(self.registered_user.id)
self.link = FamilyLink.objects.create(
parent_user_id=self.parent.id, child_phone=self.child.phone, invite_code='123456',
@@ -60,7 +60,7 @@ class FamilyLinkAcceptThrottleTests(TestCase):
def test_throttle_is_scoped_per_user_not_shared_globally(self):
# کاربرِ غیرمهمان (نه یک حسابِ مهمانِ دوم، که دیگه bucket جدا نداره).
other_user = User.objects.create(
is_guest=False, name='', phone='+989120000099',
is_guest=False, name='', phone='09120000099',
)
other_token = issue_access_token(other_user.id)
@@ -153,11 +153,11 @@ class FamilyLinkAcceptGuestThrottleBypassTests(TestCase):
# قبل هرکدوم bucket جداگانه‌ی خودشون رو داشته باشن — throttle
# برای اونا هنوز بر اساسِ user.pk است، نه IP.
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'family_invite_accept': '1/hour'}):
user_a = User.objects.create(is_guest=False, name='', phone='+989120000201')
user_a = User.objects.create(is_guest=False, name='', phone='09120000201')
response_a = self._accept('000001', issue_access_token(user_a.id))
self.assertEqual(response_a.status_code, 404)
user_b = User.objects.create(is_guest=False, name='', phone='+989120000202')
user_b = User.objects.create(is_guest=False, name='', phone='09120000202')
response_b = self._accept('000002', issue_access_token(user_b.id))
self.assertEqual(response_b.status_code, 404)
@@ -168,7 +168,7 @@ class FamilyLinkAcceptGuestThrottleBypassTests(TestCase):
# برسه — throttle باید *قبل* از منطقِ accept رد کنه.
parent = User.objects.create(device_id='device-family-guest-rot-parent', is_guest=True, name='')
child = User.objects.create(
device_id='device-family-guest-rot-child', is_guest=True, name='', phone='+989120000203',
device_id='device-family-guest-rot-child', is_guest=True, name='', phone='09120000203',
)
link = FamilyLink.objects.create(
parent_user_id=parent.id, child_phone=child.phone, invite_code='654321',
@@ -217,55 +217,55 @@ class FamilyLinkInviteThrottleTests(TestCase):
)
def test_unauthenticated_request_is_rejected(self):
response = self._invite('+989120001001')
response = self._invite('09120001001')
self.assertEqual(response.status_code, 401)
def test_requests_within_the_limit_still_succeed(self):
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'family_invite_create': '3/hour'}):
for i in range(3):
response = self._invite(f'+98912000110{i}', token=self.token)
response = self._invite(f'0912000110{i}', token=self.token)
self.assertEqual(response.status_code, 200)
self.assertIn('inviteCode', response.data)
def test_excessive_invite_requests_from_same_user_are_blocked(self):
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'family_invite_create': '3/hour'}):
for i in range(3):
response = self._invite(f'+98912000120{i}', token=self.token)
response = self._invite(f'0912000120{i}', token=self.token)
self.assertEqual(response.status_code, 200)
blocked_response = self._invite('+989120001299', token=self.token)
blocked_response = self._invite('09120001299', token=self.token)
self.assertEqual(blocked_response.status_code, 429)
def test_changing_client_ip_does_not_bypass_the_user_limit(self):
# کاربرِ غیرمهمان (نه self.parent که مهمانه)، چون کلیدِ throttleِ
# مهمان‌ها عمداً بر اساسِ IP است؛ این تست فقط رفتارِ per-user رو می‌سنجه.
registered_parent = User.objects.create(is_guest=False, name='', phone='+989120001300')
registered_parent = User.objects.create(is_guest=False, name='', phone='09120001300')
registered_token = issue_access_token(registered_parent.id)
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'family_invite_create': '3/hour'}):
for i in range(3):
response = self._invite(
f'+98912000131{i}', token=registered_token, remote_addr=f'10.5.0.{i + 1}',
f'0912000131{i}', token=registered_token, remote_addr=f'10.5.0.{i + 1}',
)
self.assertEqual(response.status_code, 200)
blocked_response = self._invite('+989120001399', token=registered_token, remote_addr='10.5.0.99')
blocked_response = self._invite('09120001399', token=registered_token, remote_addr='10.5.0.99')
self.assertEqual(blocked_response.status_code, 429)
def test_throttled_request_does_not_create_family_link(self):
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'family_invite_create': '1/hour'}):
first_response = self._invite('+989120001401', token=self.token)
first_response = self._invite('09120001401', token=self.token)
self.assertEqual(first_response.status_code, 200)
link_count_before = FamilyLink.objects.filter(parent_user_id=self.parent.id).count()
blocked_response = self._invite('+989120001402', token=self.token)
blocked_response = self._invite('09120001402', token=self.token)
self.assertEqual(blocked_response.status_code, 429)
self.assertEqual(
FamilyLink.objects.filter(parent_user_id=self.parent.id).count(), link_count_before,
)
self.assertFalse(
FamilyLink.objects.filter(parent_user_id=self.parent.id, child_phone='+989120001402').exists(),
FamilyLink.objects.filter(parent_user_id=self.parent.id, child_phone='09120001402').exists(),
)
@@ -295,39 +295,39 @@ class FamilyLinkInviteGuestThrottleBypassTests(TestCase):
def test_rotating_guest_accounts_share_the_same_throttle_bucket(self):
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'family_invite_create': '2/hour'}):
guest1 = User.objects.create(device_id='device-family-invite-guest-rot-1', is_guest=True, name='')
response1 = self._invite('+989120001501', issue_access_token(guest1.id))
response1 = self._invite('09120001501', issue_access_token(guest1.id))
self.assertEqual(response1.status_code, 200)
guest2 = User.objects.create(device_id='device-family-invite-guest-rot-2', is_guest=True, name='')
response2 = self._invite('+989120001502', issue_access_token(guest2.id))
response2 = self._invite('09120001502', issue_access_token(guest2.id))
self.assertEqual(response2.status_code, 200)
# سومین حسابِ مهمانِ تازه (pk جدید) باید throttle بشه، چون
# bucket بر اساسِ IP به‌اشتراک گذاشته شده، نه user.pk.
guest3 = User.objects.create(device_id='device-family-invite-guest-rot-3', is_guest=True, name='')
response3 = self._invite('+989120001503', issue_access_token(guest3.id))
response3 = self._invite('09120001503', 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, {'family_invite_create': '1/hour'}):
user_a = User.objects.create(is_guest=False, name='', phone='+989120001601')
response_a = self._invite('+989120001701', issue_access_token(user_a.id))
user_a = User.objects.create(is_guest=False, name='', phone='09120001601')
response_a = self._invite('09120001701', issue_access_token(user_a.id))
self.assertEqual(response_a.status_code, 200)
user_b = User.objects.create(is_guest=False, name='', phone='+989120001602')
response_b = self._invite('+989120001702', issue_access_token(user_b.id))
user_b = User.objects.create(is_guest=False, name='', phone='09120001602')
response_b = self._invite('09120001702', issue_access_token(user_b.id))
self.assertEqual(response_b.status_code, 200)
def test_blocked_guest_request_does_not_create_family_link(self):
with patch.dict(ScopedRateThrottle.THROTTLE_RATES, {'family_invite_create': '1/hour'}):
guest1 = User.objects.create(device_id='device-family-invite-guest-rot-4', is_guest=True, name='')
first_response = self._invite('+989120001801', issue_access_token(guest1.id))
first_response = self._invite('09120001801', issue_access_token(guest1.id))
self.assertEqual(first_response.status_code, 200)
guest2 = User.objects.create(device_id='device-family-invite-guest-rot-5', is_guest=True, name='')
blocked_response = self._invite('+989120001802', issue_access_token(guest2.id))
blocked_response = self._invite('09120001802', issue_access_token(guest2.id))
self.assertEqual(blocked_response.status_code, 429)
self.assertFalse(FamilyLink.objects.filter(parent_user_id=guest2.id).exists())