Zareian
e3bddb3757
nginx on the server also answers on its own IP. A direct request with a fresh X-Forwarded-For each time got a new throttle identity every time and skipped every per-IP limit (OTP request/login, admin login): reproduced on the server 2026-10-13, 65 of 65 health requests with rotating XFF got 200. TrustedProxyMiddleware reads X-Real-IP (set by nginx from the TCP peer) and, when the peer is not in TRUSTED_PROXY_NETWORKS (default: ArvanCloud's published list), replaces XFF with the peer address. Logs a WARNING once an hour per untrusted peer that sent a proxied XFF, so a new Arvan range shows up. Also ignore the parallel test DB clones (.test-db_N.sqlite3). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
74 lines
2.7 KiB
Python
74 lines
2.7 KiB
Python
"""
|
||
X-Forwarded-For فقط از ArvanCloud پذیرفته میشود (config.middleware.TrustedProxyMiddleware).
|
||
|
||
بازتولیدِ ۲۰۲۶-۱۰-۱۳ روی سرور: درخواستِ مستقیم به IPِ سرور با XFFِ تازه در هر بار، همهی سقفهای IP-محور را
|
||
دور میزد. X-Real-IP همان چیزی است که nginx از آدرسِ TCP میگذارد.
|
||
"""
|
||
|
||
from django.core.cache import cache
|
||
from django.test import RequestFactory, TestCase, override_settings
|
||
from rest_framework.throttling import BaseThrottle
|
||
|
||
from config.middleware import TrustedProxyMiddleware
|
||
|
||
ARVAN_EDGE = '37.32.19.8'
|
||
|
||
|
||
def _ident_after_middleware(**meta):
|
||
request = RequestFactory().get('/api/health', **meta)
|
||
seen = {}
|
||
|
||
def view(req):
|
||
seen['ident'] = BaseThrottle().get_ident(req)
|
||
return None
|
||
|
||
TrustedProxyMiddleware(view)(request)
|
||
return seen['ident']
|
||
|
||
|
||
class TrustedProxyIdentTests(TestCase):
|
||
def setUp(self):
|
||
cache.clear()
|
||
|
||
def test_request_through_arvan_uses_the_client_address(self):
|
||
ident = _ident_after_middleware(
|
||
HTTP_X_REAL_IP=ARVAN_EDGE, HTTP_X_FORWARDED_FOR=f'5.5.5.5, {ARVAN_EDGE}',
|
||
)
|
||
self.assertEqual(ident, '5.5.5.5')
|
||
|
||
def test_direct_hit_ignores_a_forged_forwarded_for(self):
|
||
for forged in ('1.2.3.4', '9.9.9.9, 1.2.3.4'):
|
||
ident = _ident_after_middleware(
|
||
HTTP_X_REAL_IP='203.0.113.7', HTTP_X_FORWARDED_FOR=f'{forged}, 203.0.113.7',
|
||
)
|
||
self.assertEqual(ident, '203.0.113.7')
|
||
|
||
def test_direct_hit_is_logged_once_per_peer(self):
|
||
with self.assertLogs('config.middleware', level='WARNING') as logs:
|
||
for _ in range(3):
|
||
_ident_after_middleware(
|
||
HTTP_X_REAL_IP='203.0.113.9', HTTP_X_FORWARDED_FOR='1.1.1.1, 203.0.113.9',
|
||
)
|
||
self.assertEqual(len(logs.records), 1)
|
||
|
||
def test_without_nginx_header_nothing_changes(self):
|
||
# runserverِ محلی: nginx نیست، پس X-Real-IP هم نیست.
|
||
ident = _ident_after_middleware(REMOTE_ADDR='127.0.0.1')
|
||
self.assertEqual(ident, '127.0.0.1')
|
||
|
||
|
||
@override_settings(ADMIN_LOGIN_MAX_ATTEMPTS=2)
|
||
class AdminLoginThrottleCannotBeDodgedTests(TestCase):
|
||
def setUp(self):
|
||
cache.clear()
|
||
|
||
def test_rotating_forwarded_for_on_a_direct_hit_still_hits_the_limit(self):
|
||
codes = [
|
||
self.client.post(
|
||
'/admin/login/', {'username': 'x', 'password': 'y'},
|
||
HTTP_X_REAL_IP='203.0.113.7', HTTP_X_FORWARDED_FOR=f'10.0.{i}.1, 203.0.113.7',
|
||
).status_code
|
||
for i in range(3)
|
||
]
|
||
self.assertEqual(codes[-1], 429)
|