1import os
2import shutil
3import socket
4
5import pytest
6from conftest import unit_run, unit_stop
7from unit.applications.proto import TestApplicationProto
8from unit.option import option
9from unit.utils import waitforfiles
10
11
12class TestStatic(TestApplicationProto):
13    prerequisites = {}
14
15    def setup_method(self):
16        os.makedirs(option.temp_dir + '/assets/dir')
17        with open(option.temp_dir + '/assets/index.html', 'w') as index, open(
18            option.temp_dir + '/assets/README', 'w'
19        ) as readme, open(
20            option.temp_dir + '/assets/log.log', 'w'
21        ) as log, open(
22            option.temp_dir + '/assets/dir/file', 'w'
23        ) as file:
24            index.write('0123456789')
25            readme.write('readme')
26            log.write('[debug]')
27            file.write('blah')
28
29        self._load_conf(
30            {
31                "listeners": {"*:7080": {"pass": "routes"}},
32                "routes": [
33                    {"action": {"share": option.temp_dir + "/assets$uri"}}
34                ],
35                "settings": {
36                    "http": {
37                        "static": {
38                            "mime_types": {"text/plain": [".log", "README"]}
39                        }
40                    }
41                },
42            }
43        )
44
45    def test_static_migration(self, skip_fds_check, temp_dir):
46        skip_fds_check(True, True, True)
47
48        def set_conf_version(path, version):
49            with open(path, 'w+') as f:
50                f.write(str(version))
51
52        with open(temp_dir + '/state/version', 'r') as f:
53            assert int(f.read().rstrip()) > 12500, 'current version'
54
55        assert 'success' in self.conf(
56            {"share": temp_dir + "/assets"}, 'routes/0/action'
57        ), 'configure migration 12500'
58
59        shutil.copytree(temp_dir + '/state', temp_dir + '/state_copy_12500')
60        set_conf_version(temp_dir + '/state_copy_12500/version', 12500)
61
62        assert 'success' in self.conf(
63            {"share": temp_dir + "/assets$uri"}, 'routes/0/action'
64        ), 'configure migration 12600'
65        shutil.copytree(temp_dir + '/state', temp_dir + '/state_copy_12600')
66        set_conf_version(temp_dir + '/state_copy_12600/version', 12600)
67
68        assert 'success' in self.conf(
69            {"share": temp_dir + "/assets"}, 'routes/0/action'
70        ), 'configure migration no version'
71        shutil.copytree(
72            temp_dir + '/state', temp_dir + '/state_copy_no_version'
73        )
74        os.remove(temp_dir + '/state_copy_no_version/version')
75
76        unit_stop()
77        unit_run(temp_dir + '/state_copy_12500')
78        assert self.get(url='/')['body'] == '0123456789', 'before 1.26.0'
79
80        unit_stop()
81        unit_run(temp_dir + '/state_copy_12600')
82        assert self.get(url='/')['body'] == '0123456789', 'after 1.26.0'
83
84        unit_stop()
85        unit_run(temp_dir + '/state_copy_no_version')
86        assert self.get(url='/')['body'] == '0123456789', 'before 1.26.0 2'
87
88    def test_static_index(self):
89        assert self.get(url='/index.html')['body'] == '0123456789', 'index'
90        assert self.get(url='/')['body'] == '0123456789', 'index 2'
91        assert self.get(url='//')['body'] == '0123456789', 'index 3'
92        assert self.get(url='/.')['body'] == '0123456789', 'index 4'
93        assert self.get(url='/./')['body'] == '0123456789', 'index 5'
94        assert self.get(url='/?blah')['body'] == '0123456789', 'index vars'
95        assert self.get(url='/#blah')['body'] == '0123456789', 'index anchor'
96        assert self.get(url='/dir/')['status'] == 404, 'index not found'
97
98        resp = self.get(url='/index.html/')
99        assert resp['status'] == 404, 'index not found 2 status'
100        assert (
101            resp['headers']['Content-Type'] == 'text/html'
102        ), 'index not found 2 Content-Type'
103
104    def test_static_large_file(self, temp_dir):
105        file_size = 32 * 1024 * 1024
106        with open(temp_dir + '/assets/large', 'wb') as f:
107            f.seek(file_size - 1)
108            f.write(b'\0')
109
110        assert (
111            len(self.get(url='/large', read_buffer_size=1024 * 1024)['body'])
112            == file_size
113        ), 'large file'
114
115    def test_static_etag(self, temp_dir):
116        etag = self.get(url='/')['headers']['ETag']
117        etag_2 = self.get(url='/README')['headers']['ETag']
118
119        assert etag != etag_2, 'different ETag'
120        assert etag == self.get(url='/')['headers']['ETag'], 'same ETag'
121
122        with open(temp_dir + '/assets/index.html', 'w') as f:
123            f.write('blah')
124
125        assert etag != self.get(url='/')['headers']['ETag'], 'new ETag'
126
127    def test_static_redirect(self):
128        resp = self.get(url='/dir')
129        assert resp['status'] == 301, 'redirect status'
130        assert resp['headers']['Location'] == '/dir/', 'redirect Location'
131        assert 'Content-Type' not in resp['headers'], 'redirect Content-Type'
132
133    def test_static_space_in_name(self, temp_dir):
134        os.rename(
135            temp_dir + '/assets/dir/file', temp_dir + '/assets/dir/fi le',
136        )
137        assert waitforfiles(temp_dir + '/assets/dir/fi le')
138        assert self.get(url='/dir/fi le')['body'] == 'blah', 'file name'
139
140        os.rename(temp_dir + '/assets/dir', temp_dir + '/assets/di r')
141        assert waitforfiles(temp_dir + '/assets/di r/fi le')
142        assert self.get(url='/di r/fi le')['body'] == 'blah', 'dir name'
143
144        os.rename(temp_dir + '/assets/di r', temp_dir + '/assets/ di r ')
145        assert waitforfiles(temp_dir + '/assets/ di r /fi le')
146        assert (
147            self.get(url='/ di r /fi le')['body'] == 'blah'
148        ), 'dir name enclosing'
149
150        assert (
151            self.get(url='/%20di%20r%20/fi le')['body'] == 'blah'
152        ), 'dir encoded'
153        assert (
154            self.get(url='/ di r %2Ffi le')['body'] == 'blah'
155        ), 'slash encoded'
156        assert (
157            self.get(url='/ di r /fi%20le')['body'] == 'blah'
158        ), 'file encoded'
159        assert (
160            self.get(url='/%20di%20r%20%2Ffi%20le')['body'] == 'blah'
161        ), 'encoded'
162        assert (
163            self.get(url='/%20%64%69%20%72%20%2F%66%69%20%6C%65')['body']
164            == 'blah'
165        ), 'encoded 2'
166
167        os.rename(
168            temp_dir + '/assets/ di r /fi le',
169            temp_dir + '/assets/ di r / fi le ',
170        )
171        assert waitforfiles(temp_dir + '/assets/ di r / fi le ')
172        assert (
173            self.get(url='/%20di%20r%20/%20fi%20le%20')['body'] == 'blah'
174        ), 'file name enclosing'
175
176        try:
177            open(temp_dir + '/ф а', 'a').close()
178            utf8 = True
179
180        except KeyboardInterrupt:
181            raise
182
183        except:
184            utf8 = False
185
186        if utf8:
187            os.rename(
188                temp_dir + '/assets/ di r / fi le ',
189                temp_dir + '/assets/ di r /фа йл',
190            )
191            assert waitforfiles(temp_dir + '/assets/ di r /фа йл')
192            assert (
193                self.get(url='/ di r /фа йл')['body'] == 'blah'
194            ), 'file name 2'
195
196            os.rename(
197                temp_dir + '/assets/ di r ', temp_dir + '/assets/ди ректория',
198            )
199            assert waitforfiles(temp_dir + '/assets/ди ректория/фа йл')
200            assert (
201                self.get(url='/ди ректория/фа йл')['body'] == 'blah'
202            ), 'dir name 2'
203
204    def test_static_unix_socket(self, temp_dir):
205        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
206        sock.bind(temp_dir + '/assets/unix_socket')
207
208        assert self.get(url='/unix_socket')['status'] == 404, 'socket'
209
210        sock.close()
211
212    def test_static_unix_fifo(self, temp_dir):
213        os.mkfifo(temp_dir + '/assets/fifo')
214
215        assert self.get(url='/fifo')['status'] == 404, 'fifo'
216
217    def test_static_method(self):
218        resp = self.head()
219        assert resp['status'] == 200, 'HEAD status'
220        assert resp['body'] == '', 'HEAD empty body'
221
222        assert self.delete()['status'] == 405, 'DELETE'
223        assert self.post()['status'] == 405, 'POST'
224        assert self.put()['status'] == 405, 'PUT'
225
226    def test_static_path(self):
227        assert self.get(url='/dir/../dir/file')['status'] == 200, 'relative'
228
229        assert self.get(url='./')['status'] == 400, 'path invalid'
230        assert self.get(url='../')['status'] == 400, 'path invalid 2'
231        assert self.get(url='/..')['status'] == 400, 'path invalid 3'
232        assert self.get(url='../assets/')['status'] == 400, 'path invalid 4'
233        assert self.get(url='/../assets/')['status'] == 400, 'path invalid 5'
234
235    def test_static_two_clients(self):
236        _, sock = self.get(url='/', start=True, no_recv=True)
237        _, sock2 = self.get(url='/', start=True, no_recv=True)
238
239        assert sock.recv(1) == b'H', 'client 1'
240        assert sock2.recv(1) == b'H', 'client 2'
241        assert sock.recv(1) == b'T', 'client 1 again'
242        assert sock2.recv(1) == b'T', 'client 2 again'
243
244        sock.close()
245        sock2.close()
246
247    def test_static_mime_types(self):
248        assert 'success' in self.conf(
249            {
250                "text/x-code/x-blah/x-blah": "readme",
251                "text/plain": [".html", ".log", "file"],
252            },
253            'settings/http/static/mime_types',
254        ), 'configure mime_types'
255
256        assert (
257            self.get(url='/README')['headers']['Content-Type']
258            == 'text/x-code/x-blah/x-blah'
259        ), 'mime_types string case insensitive'
260        assert (
261            self.get(url='/index.html')['headers']['Content-Type']
262            == 'text/plain'
263        ), 'mime_types html'
264        assert (
265            self.get(url='/')['headers']['Content-Type'] == 'text/plain'
266        ), 'mime_types index default'
267        assert (
268            self.get(url='/dir/file')['headers']['Content-Type']
269            == 'text/plain'
270        ), 'mime_types file in dir'
271
272    def test_static_mime_types_partial_match(self):
273        assert 'success' in self.conf(
274            {"text/x-blah": ["ile", "fil", "f", "e", ".file"],},
275            'settings/http/static/mime_types',
276        ), 'configure mime_types'
277        assert 'Content-Type' not in self.get(url='/dir/file'), 'partial match'
278
279    def test_static_mime_types_reconfigure(self):
280        assert 'success' in self.conf(
281            {
282                "text/x-code": "readme",
283                "text/plain": [".html", ".log", "file"],
284            },
285            'settings/http/static/mime_types',
286        ), 'configure mime_types'
287
288        assert self.conf_get('settings/http/static/mime_types') == {
289            'text/x-code': 'readme',
290            'text/plain': ['.html', '.log', 'file'],
291        }, 'mime_types get'
292        assert (
293            self.conf_get('settings/http/static/mime_types/text%2Fx-code')
294            == 'readme'
295        ), 'mime_types get string'
296        assert self.conf_get(
297            'settings/http/static/mime_types/text%2Fplain'
298        ) == ['.html', '.log', 'file'], 'mime_types get array'
299        assert (
300            self.conf_get('settings/http/static/mime_types/text%2Fplain/1')
301            == '.log'
302        ), 'mime_types get array element'
303
304        assert 'success' in self.conf_delete(
305            'settings/http/static/mime_types/text%2Fplain/2'
306        ), 'mime_types remove array element'
307        assert (
308            'Content-Type' not in self.get(url='/dir/file')['headers']
309        ), 'mime_types removed'
310
311        assert 'success' in self.conf_post(
312            '"file"', 'settings/http/static/mime_types/text%2Fplain'
313        ), 'mime_types add array element'
314        assert (
315            self.get(url='/dir/file')['headers']['Content-Type']
316            == 'text/plain'
317        ), 'mime_types reverted'
318
319        assert 'success' in self.conf(
320            '"file"', 'settings/http/static/mime_types/text%2Fplain'
321        ), 'configure mime_types update'
322        assert (
323            self.get(url='/dir/file')['headers']['Content-Type']
324            == 'text/plain'
325        ), 'mime_types updated'
326        assert (
327            'Content-Type' not in self.get(url='/log.log')['headers']
328        ), 'mime_types updated 2'
329
330        assert 'success' in self.conf(
331            '".log"', 'settings/http/static/mime_types/text%2Fblahblahblah'
332        ), 'configure mime_types create'
333        assert (
334            self.get(url='/log.log')['headers']['Content-Type']
335            == 'text/blahblahblah'
336        ), 'mime_types create'
337
338    def test_static_mime_types_correct(self):
339        assert 'error' in self.conf(
340            {"text/x-code": "readme", "text/plain": "readme"},
341            'settings/http/static/mime_types',
342        ), 'mime_types same extensions'
343        assert 'error' in self.conf(
344            {"text/x-code": [".h", ".c"], "text/plain": ".c"},
345            'settings/http/static/mime_types',
346        ), 'mime_types same extensions array'
347        assert 'error' in self.conf(
348            {"text/x-code": [".h", ".c", "readme"], "text/plain": "README",},
349            'settings/http/static/mime_types',
350        ), 'mime_types same extensions case insensitive'
351
352    @pytest.mark.skip('not yet')
353    def test_static_mime_types_invalid(self, temp_dir):
354        assert 'error' in self.http(
355            b"""PUT /config/settings/http/static/mime_types/%0%00% HTTP/1.1\r
356Host: localhost\r
357Connection: close\r
358Content-Length: 6\r
359\r
360\"blah\"""",
361            raw_resp=True,
362            raw=True,
363            sock_type='unix',
364            addr=temp_dir + '/control.unit.sock',
365        ), 'mime_types invalid'
366