"""回归测试: 限流与登录防爆破在不可信 X-Forwarded-For 下的 IP 伪造问题。 验证: 1. TRUST_PROXY_HEADERS=False(默认)时忽略伪造的 X-Forwarded-For 2. TRUST_PROXY_HEADERS=True 时解析 X-Forwarded-For 3. 限流与登录共用同一套 IP 提取逻辑 """ from __future__ import annotations from unittest.mock import MagicMock, patch import pytest from src.api.deps import get_client_ip from src.core.config import Settings class TestGetClientIP: """get_client_ip 防伪造逻辑。""" def _make_request(self, client_host: str | None, xff: str | None = None): req = MagicMock() req.client = MagicMock(host=client_host) if client_host else None req.headers = {} if xff is not None: req.headers["X-Forwarded-For"] = xff return req @patch("src.api.deps.settings", Settings(TRUST_PROXY_HEADERS=False)) def test_untrusted_proxy_ignores_xff(self): """TRUST_PROXY_HEADERS=False 时忽略伪造的 X-Forwarded-For。""" # 客户端伪造 X-Forwarded-For,但 TRUST_PROXY_HEADERS=False req = self._make_request("1.2.3.4", xff="10.0.0.1, 192.168.1.1") ip = get_client_ip(req) assert ip == "1.2.3.4", f"应使用连接层 IP,实际 {ip}" @patch("src.api.deps.settings", Settings(TRUST_PROXY_HEADERS=True)) def test_trusted_proxy_parses_xff(self): """TRUST_PROXY_HEADERS=True 时解析 X-Forwarded-For 第一个 IP。""" req = self._make_request("127.0.0.1", xff="10.0.0.1, 192.168.1.1") ip = get_client_ip(req) assert ip == "10.0.0.1", f"应使用 XFF 第一个 IP,实际 {ip}" @patch("src.api.deps.settings", Settings(TRUST_PROXY_HEADERS=True)) def test_trusted_proxy_without_xff(self): """TRUST_PROXY_HEADERS=True 但无 XFF 头时回退到 client.host。""" req = self._make_request("1.2.3.4", xff=None) ip = get_client_ip(req) assert ip == "1.2.3.4", f"应回退到连接层 IP,实际 {ip}" @patch("src.api.deps.settings", Settings(TRUST_PROXY_HEADERS=False)) def test_untrusted_proxy_no_client(self): """TRUST_PROXY_HEADERS=False 且无 client 时返回 unknown。""" req = self._make_request(None, xff="10.0.0.1") ip = get_client_ip(req) assert ip == "unknown", f"应返回 unknown,实际 {ip}" @patch("src.api.deps.settings", Settings(TRUST_PROXY_HEADERS=True)) def test_trusted_proxy_single_ip(self): """TRUST_PROXY_HEADERS=True 且 XFF 只有一个 IP。""" req = self._make_request("127.0.0.1", xff="10.0.0.1") ip = get_client_ip(req) assert ip == "10.0.0.1", f"应返回 10.0.0.1,实际 {ip}" class TestRateLimitIPSpoofing: """验证限流使用 get_client_ip 防伪造。""" @patch("src.api.deps.settings", Settings(TRUST_PROXY_HEADERS=False)) def test_rate_limit_ignores_spoofed_xff(self): """限流在 TRUST_PROXY_HEADERS=False 时不受 XFF 伪造影响。""" from src.api.deps import _RateLimiter, get_client_ip limiter = _RateLimiter(max_requests=10, window_seconds=60) # 模拟不同伪造 XFF,但真实 IP 相同 def make_request(spoofed_xff): req = MagicMock() req.client = MagicMock(host="1.2.3.4") req.headers = {"X-Forwarded-For": spoofed_xff} return req # 伪造不同 XFF,但真实 IP 都是 1.2.3.4 for i in range(10): req = make_request(f"10.0.0.{i}") ip = get_client_ip(req) assert ip == "1.2.3.4", f"迭代 {i}: 应返回 1.2.3.4,实际 {ip}" assert limiter.is_allowed(ip), f"迭代 {i}: 应允许"