1import os
2from pathlib import Path
3
4import pytest
5from unit.applications.proto import TestApplicationProto
6
7
8class TestStaticChroot(TestApplicationProto):
9    prerequisites = {'features': ['chroot']}
10
11    @pytest.fixture(autouse=True)
12    def setup_method_fixture(self, temp_dir):
13        os.makedirs(temp_dir + '/assets/dir')
14        Path(temp_dir + '/assets/index.html').write_text('0123456789')
15        Path(temp_dir + '/assets/dir/file').write_text('blah')
16
17        test = Path(__file__)
18        self.test_path = '/' + test.parent.name + '/' + test.name
19
20        self._load_conf(
21            {
22                "listeners": {"*:7080": {"pass": "routes"}},
23                "routes": [{"action": {"share": temp_dir + "/assets$uri"}}],
24            }
25        )
26
27    def update_action(self, share, chroot):
28        return self.conf(
29            {"share": share, "chroot": chroot}, 'routes/0/action',
30        )
31
32    def get_custom(self, uri, host):
33        return self.get(
34            url=uri, headers={'Host': host, 'Connection': 'close'}
35        )['status']
36
37    def test_static_chroot(self, temp_dir):
38        assert self.get(url='/dir/file')['status'] == 200, 'default chroot'
39        assert self.get(url='/index.html')['status'] == 200, 'default chroot 2'
40
41        assert 'success' in self.conf(
42            {
43                "share": temp_dir + "/assets$uri",
44                "chroot": temp_dir + "/assets/dir",
45            },
46            'routes/0/action',
47        ), 'configure chroot'
48
49        assert self.get(url='/dir/file')['status'] == 200, 'chroot'
50        assert self.get(url='/index.html')['status'] == 403, 'chroot 403 2'
51        assert self.get(url='/file')['status'] == 403, 'chroot 403'
52
53    def test_share_chroot_array(self, temp_dir):
54        assert 'success' in self.conf(
55            {
56                "share": ["/blah", temp_dir + "/assets$uri"],
57                "chroot": temp_dir + "/assets/dir",
58            },
59            'routes/0/action',
60        ), 'configure share array'
61        assert self.get(url='/dir/file')['status'] == 200, 'share array'
62
63        assert 'success' in self.update_action(
64            ["/blah", temp_dir + '/assets$uri'], temp_dir + '/assets/$host'
65        )
66        assert self.get_custom('/dir/file', 'dir') == 200, 'array variable'
67
68        assert 'success' in self.conf(
69            {
70                "share": ["/blah", "/blah2"],
71                "chroot": temp_dir + "/assets/dir",
72            },
73            'routes/0/action',
74        ), 'configure share array bad'
75        assert self.get()['status'] != 200, 'share array bad'
76
77    def test_static_chroot_permission(self, is_su, temp_dir):
78        if is_su:
79            pytest.skip('does\'t work under root')
80
81        os.chmod(temp_dir + '/assets/dir', 0o100)
82
83        assert 'success' in self.conf(
84            {
85                "share": temp_dir + "/assets$uri",
86                "chroot": temp_dir + "/assets/dir",
87            },
88            'routes/0/action',
89        ), 'configure chroot'
90
91        assert self.get(url='/dir/file')['status'] == 200, 'chroot'
92
93    def test_static_chroot_empty(self, temp_dir):
94        assert 'success' in self.conf(
95            {"share": temp_dir + "/assets$uri", "chroot": ""},
96            'routes/0/action',
97        ), 'configure chroot empty absolute'
98
99        assert (
100            self.get(url='/dir/file')['status'] == 200
101        ), 'chroot empty absolute'
102
103        assert 'success' in self.conf(
104            {"share": ".$uri", "chroot": ""}, 'routes/0/action',
105        ), 'configure chroot empty relative'
106
107        assert (
108            self.get(url=self.test_path)['status'] == 200
109        ), 'chroot empty relative'
110
111    def test_static_chroot_relative(self, is_su, temp_dir):
112        if is_su:
113            pytest.skip('does\'t work under root')
114
115        assert 'success' in self.conf(
116            {"share": temp_dir + "/assets$uri", "chroot": "."},
117            'routes/0/action',
118        ), 'configure relative chroot'
119
120        assert self.get(url='/dir/file')['status'] == 403, 'relative chroot'
121
122        assert 'success' in self.conf(
123            {"share": ".$uri"}, 'routes/0/action',
124        ), 'configure relative share'
125
126        assert self.get(url=self.test_path)['status'] == 200, 'relative share'
127
128        assert 'success' in self.conf(
129            {"share": ".$uri", "chroot": "."}, 'routes/0/action',
130        ), 'configure relative'
131
132        assert self.get(url=self.test_path)['status'] == 200, 'relative'
133
134    def test_static_chroot_variables(self, temp_dir):
135        assert 'success' in self.update_action(
136            temp_dir + '/assets$uri', temp_dir + '/assets/$host'
137        )
138        assert self.get_custom('/dir/file', 'dir') == 200
139
140        assert 'success' in self.update_action(
141            temp_dir + '/assets$uri', temp_dir + '/assets/${host}'
142        )
143        assert self.get_custom('/dir/file', 'dir') == 200
144
145    def test_static_chroot_variables_buildin_start(self, temp_dir):
146        assert 'success' in self.update_action(
147            temp_dir + '/assets/dir/$host', '$uri/assets/dir'
148        )
149
150        assert self.get_custom(temp_dir, 'file') == 200
151
152    def test_static_chroot_variables_buildin_mid(self, temp_dir):
153        assert 'success' in self.update_action(
154            temp_dir + '/assets$uri', temp_dir + '/$host/dir'
155        )
156
157        assert self.get_custom('/dir/file', 'assets') == 200
158
159    def test_static_chroot_variables_buildin_end(self, temp_dir):
160        assert 'success' in self.update_action(
161            temp_dir + '/assets$uri', temp_dir + '/assets/$host'
162        )
163
164        assert self.get_custom('/dir/file', 'dir') == 200
165
166    def test_static_chroot_slash(self, temp_dir):
167        assert 'success' in self.conf(
168            {
169                "share": temp_dir + "/assets$uri",
170                "chroot": temp_dir + "/assets/dir/",
171            },
172            'routes/0/action',
173        ), 'configure chroot slash end'
174
175        assert self.get(url='/dir/file')['status'] == 200, 'slash end'
176        assert self.get(url='/dirxfile')['status'] == 403, 'slash end bad'
177
178        assert 'success' in self.conf(
179            {
180                "share": temp_dir + "/assets$uri",
181                "chroot": temp_dir + "/assets/dir",
182            },
183            'routes/0/action',
184        ), 'configure chroot no slash end'
185
186        assert self.get(url='/dir/file')['status'] == 200, 'no slash end'
187
188        assert 'success' in self.conf(
189            {
190                "share": temp_dir + "/assets$uri",
191                "chroot": temp_dir + "/assets/dir/",
192            },
193            'routes/0/action',
194        ), 'configure chroot slash end 2'
195
196        assert self.get(url='/dir/file')['status'] == 200, 'slash end 2'
197        assert self.get(url='/dirxfile')['status'] == 403, 'slash end 2 bad'
198
199        assert 'success' in self.conf(
200            {
201                "share": temp_dir + "///assets/////$uri",
202                "chroot": temp_dir + "//assets////dir///",
203            },
204            'routes/0/action',
205        ), 'configure chroot multiple slashes'
206
207        assert self.get(url='/dir/file')['status'] == 200, 'multiple slashes'
208
209    def test_static_chroot_invalid(self, temp_dir):
210        assert 'error' in self.conf(
211            {"share": temp_dir, "chroot": True}, 'routes/0/action',
212        ), 'configure chroot error'
213        assert 'error' in self.conf(
214            {"share": temp_dir, "symlinks": "True"}, 'routes/0/action',
215        ), 'configure symlink error'
216        assert 'error' in self.conf(
217            {"share": temp_dir, "mount": "True"}, 'routes/0/action',
218        ), 'configure mount error'
219
220        assert 'error' in self.update_action(
221            temp_dir + '/assets$uri', temp_dir + '/assets/d$r$uri'
222        )
223        assert 'error' in self.update_action(
224            temp_dir + '/assets$uri', temp_dir + '/assets/$$uri'
225        )
226