xref: /unit/test/test_client_ip.py (revision 1984:06514cd08a35)
1from unit.applications.lang.python import TestApplicationPython
2
3
4class TestClientIP(TestApplicationPython):
5    prerequisites = {'modules': {'python': 'any'}}
6
7    def client_ip(self, options):
8        assert 'success' in self.conf(
9            {
10                "127.0.0.1:7081":
11                    {"client_ip": options, "pass": "applications/client_ip"},
12                "[::1]:7082":
13                    {"client_ip": options, "pass": "applications/client_ip"},
14            },
15            'listeners',
16        ), 'listeners configure'
17
18    def get_xff(self, xff, sock_type='ipv4'):
19        port = 7081 if sock_type == 'ipv4' else 7082
20
21        return self.get(
22            sock_type=sock_type,
23            port=port,
24            headers={'Connection': 'close', 'X-Forwarded-For': xff},
25        )['body']
26
27    def setup_method(self):
28        self.load('client_ip')
29
30    def test_settings_client_ip_single_ip(self):
31        self.client_ip(
32            {'header': 'X-Forwarded-For', 'source': '123.123.123.123'}
33        )
34
35        assert self.get(port=7081)['body'] == '127.0.0.1', 'ipv4 default'
36        assert (
37            self.get(sock_type='ipv6', port=7082)['body'] == '::1'
38        ), 'ipv6 default'
39        assert self.get_xff('1.1.1.1') == '127.0.0.1', 'bad source'
40        assert self.get_xff('blah') == '127.0.0.1', 'bad header'
41        assert self.get_xff('1.1.1.1', 'ipv6') == '::1', 'bad source ipv6'
42
43        self.client_ip({'header': 'X-Forwarded-For', 'source': '127.0.0.1'})
44
45        assert self.get(port=7081)['body'] == '127.0.0.1', 'ipv4 default 2'
46        assert (
47            self.get(sock_type='ipv6', port=7082)['body'] == '::1'
48        ), 'ipv6 default 2'
49        assert self.get_xff('1.1.1.1') == '1.1.1.1', 'replace'
50        assert self.get_xff('blah') == '127.0.0.1', 'bad header 2'
51        assert (
52            self.get_xff('1.1.1.1', 'ipv6') == '::1'
53        ), 'bad source ipv6 2'
54
55        self.client_ip({'header': 'X-Forwarded-For', 'source': '!127.0.0.1'})
56
57        assert self.get_xff('1.1.1.1') == '127.0.0.1', 'bad source 3'
58        assert self.get_xff('1.1.1.1', 'ipv6') == '1.1.1.1', 'replace 2'
59
60    def test_settings_client_ip_ipv4(self):
61        self.client_ip({'header': 'X-Forwarded-For', 'source': '127.0.0.1'})
62
63        assert (
64            self.get_xff('8.8.8.8, 84.23.23.11') == '84.23.23.11'
65        ), 'xff replace'
66        assert (
67            self.get_xff('8.8.8.8, 84.23.23.11, 127.0.0.1') == '127.0.0.1'
68        ), 'xff replace 2'
69        assert (
70            self.get_xff(['8.8.8.8', '127.0.0.1, 10.0.1.1']) == '10.0.1.1'
71        ), 'xff replace multi'
72
73    def test_settings_client_ip_ipv6(self):
74        self.client_ip({'header': 'X-Forwarded-For', 'source': '::1'})
75
76        assert self.get_xff('1.1.1.1') == '127.0.0.1', 'bad source ipv4'
77
78        for ip in [
79            'f607:7403:1e4b:6c66:33b2:843f:2517:da27',
80            '2001:db8:3c4d:15::1a2f:1a2b',
81            '2001::3c4d:15:1a2f:1a2b',
82            '::11.22.33.44',
83        ]:
84            assert self.get_xff(ip, 'ipv6') == ip, 'replace'
85
86    def test_settings_client_ip_recursive(self):
87        self.client_ip(
88            {
89                'header': 'X-Forwarded-For',
90                'recursive': True,
91                'source': ['127.0.0.1', '10.50.0.17', '10.5.2.1'],
92            }
93        )
94
95        assert self.get_xff('1.1.1.1') == '1.1.1.1', 'xff chain'
96        assert self.get_xff('1.1.1.1, 10.5.2.1') == '1.1.1.1', 'xff chain 2'
97        assert (
98            self.get_xff('8.8.8.8, 1.1.1.1, 10.5.2.1') == '1.1.1.1'
99        ), 'xff chain 3'
100        assert (
101            self.get_xff('10.50.0.17, 10.5.2.1, 10.5.2.1') == '10.50.0.17'
102        ), 'xff chain 4'
103        assert (
104            self.get_xff(['8.8.8.8', '1.1.1.1, 127.0.0.1']) == '1.1.1.1'
105        ), 'xff replace multi'
106        assert (
107            self.get_xff(['8.8.8.8', '1.1.1.1, 127.0.0.1', '10.5.2.1'])
108            == '1.1.1.1'
109        ), 'xff replace multi 2'
110        assert (
111            self.get_xff(['10.5.2.1', '10.50.0.17, 1.1.1.1', '10.5.2.1'])
112            == '1.1.1.1'
113        ), 'xff replace multi 3'
114        assert (
115            self.get_xff('8.8.8.8, 2001:db8:3c4d:15::1a2f:1a2b, 127.0.0.1')
116            == '2001:db8:3c4d:15::1a2f:1a2b'
117        ), 'xff chain ipv6'
118
119    def test_settings_client_ip_invalid(self):
120        assert 'error' in self.conf(
121            {"http": {"client_ip": {'header': 'X-Forwarded-For', 'source': []}}},
122            'settings',
123        ), 'empty array source'
124        assert 'error' in self.conf(
125            {"http":{"client_ip": {'header': 'X-Forwarded-For', 'source': 'a'}}},
126            'settings',
127        ), 'empty source invalid'
128