From f07031f32c909836038a79c122456da9b9efcf5c Mon Sep 17 00:00:00 2001 From: BADtochka Date: Sat, 6 Jun 2026 00:45:21 +0300 Subject: [PATCH] fix(security): resolve forwarded client ip chain --- backend/bot/utils/request_security.py | 31 ++++++++++++++++++++------- tests/test_security.py | 21 +++++++++++++----- 2 files changed, 39 insertions(+), 13 deletions(-) diff --git a/backend/bot/utils/request_security.py b/backend/bot/utils/request_security.py index bba0bc5..b1b579e 100644 --- a/backend/bot/utils/request_security.py +++ b/backend/bot/utils/request_security.py @@ -34,12 +34,27 @@ def _parse_ip(value: Optional[str]) -> Optional[ipaddress._BaseAddress]: return None -def _last_forwarded_ip(header_value: str) -> Optional[str]: - candidates = [item.strip() for item in header_value.split(",") if item.strip()] - if not candidates: +def _forwarded_ips(header_value: str) -> list[ipaddress._BaseAddress]: + parsed: list[ipaddress._BaseAddress] = [] + for item in header_value.split(","): + ip = _parse_ip(item.strip()) + if ip is not None: + parsed.append(ip) + return parsed + + +def _trusted_forwarded_ip( + header_value: str, + trusted_networks: Sequence[ipaddress._BaseNetwork], +) -> Optional[str]: + forwarded_ips = _forwarded_ips(header_value) + if not forwarded_ips: return None - candidate = candidates[-1] - return candidate if _parse_ip(candidate) is not None else None + + for ip in reversed(forwarded_ips): + if not any(ip in network for network in trusted_networks): + return str(ip) + return str(forwarded_ips[0]) def request_client_ip( @@ -53,15 +68,15 @@ def request_client_ip( if remote_ip and forwarded_for: trusted_networks = parse_ip_entries(trusted_proxies) if any(remote_ip in network for network in trusted_networks): - forwarded_ip = _last_forwarded_ip(forwarded_for) + forwarded_ip = _trusted_forwarded_ip(forwarded_for, trusted_networks) if forwarded_ip: return forwarded_ip if remote_ip: return str(remote_ip) - forwarded_ip = _last_forwarded_ip(forwarded_for) - return forwarded_ip + forwarded_ips = _forwarded_ips(forwarded_for) + return str(forwarded_ips[-1]) if forwarded_ips else None def ip_in_allowlist( diff --git a/tests/test_security.py b/tests/test_security.py index e8b3f58..22a56bf 100644 --- a/tests/test_security.py +++ b/tests/test_security.py @@ -48,15 +48,26 @@ class RequestSecurityTests(unittest.IsolatedAsyncioTestCase): "198.51.100.7", ) + async def test_request_client_ip_skips_trusted_forwarded_proxy_chain(self): + request = SimpleNamespace( + remote="172.19.0.6", + headers={"X-Forwarded-For": "203.0.113.10, 172.19.0.7"}, + ) + + self.assertEqual( + request_client_ip(request, trusted_proxies=["172.19.0.0/16"]), + "203.0.113.10", + ) + async def test_access_logger_uses_forwarded_ip_only_for_trusted_proxy(self): trusted_request = SimpleNamespace( - remote="172.19.0.7", - headers={"X-Forwarded-For": "203.0.113.10"}, + remote="172.19.0.6", + headers={"X-Forwarded-For": "203.0.113.10, 172.19.0.7"}, app={"settings": SimpleNamespace(trusted_proxies=["172.19.0.0/16"])}, ) untrusted_request = SimpleNamespace( - remote="172.19.0.7", - headers={"X-Forwarded-For": "203.0.113.10"}, + remote="172.19.0.6", + headers={"X-Forwarded-For": "203.0.113.10, 172.19.0.7"}, app={"settings": SimpleNamespace(trusted_proxies=["127.0.0.1"])}, ) @@ -66,7 +77,7 @@ class RequestSecurityTests(unittest.IsolatedAsyncioTestCase): ) self.assertEqual( TrustedProxyAccessLogger._format_a(untrusted_request, object(), 0), - "172.19.0.7", + "172.19.0.6", ) async def test_yookassa_webhook_rejects_untrusted_ip_before_reading_body(self):