|
| 1 | +# -*- coding: utf-8 -*- |
| 2 | +"""Tests for CWE-345 fix: X-Forwarded-For IP spoofing prevention in get_client_ip.""" |
| 3 | + |
| 4 | +import os |
| 5 | +import unittest |
| 6 | +from types import SimpleNamespace |
| 7 | +from unittest.mock import patch |
| 8 | + |
| 9 | +from src.auth import get_client_ip |
| 10 | + |
| 11 | + |
| 12 | +def _make_request(xff_value=None, client_host=None): |
| 13 | + """Build a minimal request-like object.""" |
| 14 | + headers = {} |
| 15 | + if xff_value is not None: |
| 16 | + headers["X-Forwarded-For"] = xff_value |
| 17 | + client = SimpleNamespace(host=client_host) if client_host else None |
| 18 | + return SimpleNamespace(headers=headers, client=client) |
| 19 | + |
| 20 | + |
| 21 | +class TestGetClientIpXffFix(unittest.TestCase): |
| 22 | + """Verify get_client_ip uses rightmost XFF entry (proxy-appended).""" |
| 23 | + |
| 24 | + # --- TRUST_X_FORWARDED_FOR enabled --- |
| 25 | + |
| 26 | + @patch.dict(os.environ, {"TRUST_X_FORWARDED_FOR": "true"}) |
| 27 | + def test_single_ip_returns_that_ip(self): |
| 28 | + """Single-entry XFF should return that entry.""" |
| 29 | + req = _make_request(xff_value="1.2.3.4") |
| 30 | + self.assertEqual(get_client_ip(req), "1.2.3.4") |
| 31 | + |
| 32 | + @patch.dict(os.environ, {"TRUST_X_FORWARDED_FOR": "true"}) |
| 33 | + def test_multiple_ips_returns_rightmost(self): |
| 34 | + """Rightmost entry is the one appended by the trusted proxy.""" |
| 35 | + req = _make_request(xff_value="spoofed.ip, 10.0.0.1, 192.168.1.1") |
| 36 | + self.assertEqual(get_client_ip(req), "192.168.1.1") |
| 37 | + |
| 38 | + @patch.dict(os.environ, {"TRUST_X_FORWARDED_FOR": "true"}) |
| 39 | + def test_attacker_cannot_control_rate_limit_bucket(self): |
| 40 | + """Attacker-injected leftmost IP must NOT be selected (the old [0] bug).""" |
| 41 | + req = _make_request(xff_value="evil-rotated-ip, real-client-ip") |
| 42 | + ip = get_client_ip(req) |
| 43 | + self.assertNotEqual(ip, "evil-rotated-ip", |
| 44 | + "Leftmost (attacker-controlled) IP must not be used") |
| 45 | + self.assertEqual(ip, "real-client-ip") |
| 46 | + |
| 47 | + @patch.dict(os.environ, {"TRUST_X_FORWARDED_FOR": "true"}) |
| 48 | + def test_whitespace_is_stripped(self): |
| 49 | + req = _make_request(xff_value="10.0.0.1, 192.168.1.1 ") |
| 50 | + self.assertEqual(get_client_ip(req), "192.168.1.1") |
| 51 | + |
| 52 | + @patch.dict(os.environ, {"TRUST_X_FORWARDED_FOR": "true"}) |
| 53 | + def test_no_xff_header_falls_back_to_client(self): |
| 54 | + req = _make_request(client_host="172.16.0.1") |
| 55 | + self.assertEqual(get_client_ip(req), "172.16.0.1") |
| 56 | + |
| 57 | + @patch.dict(os.environ, {"TRUST_X_FORWARDED_FOR": "true"}) |
| 58 | + def test_no_xff_no_client_returns_localhost(self): |
| 59 | + req = _make_request() |
| 60 | + self.assertEqual(get_client_ip(req), "127.0.0.1") |
| 61 | + |
| 62 | + # --- TRUST_X_FORWARDED_FOR disabled (default) --- |
| 63 | + |
| 64 | + @patch.dict(os.environ, {"TRUST_X_FORWARDED_FOR": "false"}) |
| 65 | + def test_xff_ignored_when_trust_disabled(self): |
| 66 | + """XFF header should be completely ignored when trust is off.""" |
| 67 | + req = _make_request(xff_value="1.2.3.4", client_host="10.0.0.5") |
| 68 | + self.assertEqual(get_client_ip(req), "10.0.0.5") |
| 69 | + |
| 70 | + @patch.dict(os.environ, {}, clear=False) |
| 71 | + def test_xff_ignored_when_env_unset(self): |
| 72 | + """If TRUST_X_FORWARDED_FOR is not set, default to not trusting.""" |
| 73 | + env = os.environ.copy() |
| 74 | + env.pop("TRUST_X_FORWARDED_FOR", None) |
| 75 | + with patch.dict(os.environ, env, clear=True): |
| 76 | + req = _make_request(xff_value="1.2.3.4", client_host="10.0.0.5") |
| 77 | + self.assertEqual(get_client_ip(req), "10.0.0.5") |
| 78 | + |
| 79 | + # --- Edge cases --- |
| 80 | + |
| 81 | + @patch.dict(os.environ, {"TRUST_X_FORWARDED_FOR": "true"}) |
| 82 | + def test_empty_xff_header(self): |
| 83 | + """Empty XFF string should fall back to client.""" |
| 84 | + req = SimpleNamespace(headers={"X-Forwarded-For": ""}, client=SimpleNamespace(host="10.0.0.1")) |
| 85 | + self.assertEqual(get_client_ip(req), "10.0.0.1") |
| 86 | + |
| 87 | + @patch.dict(os.environ, {"TRUST_X_FORWARDED_FOR": "TRUE"}) |
| 88 | + def test_case_insensitive_trust_flag(self): |
| 89 | + """TRUST_X_FORWARDED_FOR=TRUE (uppercase) should still work.""" |
| 90 | + req = _make_request(xff_value="1.1.1.1, 2.2.2.2") |
| 91 | + self.assertEqual(get_client_ip(req), "2.2.2.2") |
| 92 | + |
| 93 | + |
| 94 | +if __name__ == "__main__": |
| 95 | + unittest.main() |
0 commit comments