xref: /unit/test/test_php_application.py (revision 1770:e4ece1ff4413)
1import os
2import re
3import shutil
4import time
5from subprocess import call
6
7import pytest
8
9from unit.applications.lang.php import TestApplicationPHP
10from unit.option import option
11
12class TestPHPApplication(TestApplicationPHP):
13    prerequisites = {'modules': {'php': 'all'}}
14
15    def before_disable_functions(self):
16        body = self.get()['body']
17
18        assert re.search(r'time: \d+', body), 'disable_functions before time'
19        assert re.search(r'exec: \/\w+', body), 'disable_functions before exec'
20
21    def set_opcache(self, app, val):
22        assert 'success' in self.conf(
23            {"admin": {"opcache.enable": val, "opcache.enable_cli": val}},
24            'applications/' + app + '/options',
25        )
26
27        opcache = self.get()['headers']['X-OPcache']
28
29        if not opcache or opcache == '-1':
30            pytest.skip('opcache is not supported')
31
32        assert opcache == val, 'opcache value'
33
34    def test_php_application_variables(self):
35        self.load('variables')
36
37        body = 'Test body string.'
38
39        resp = self.post(
40            headers={
41                'Host': 'localhost',
42                'Content-Type': 'text/html',
43                'Custom-Header': 'blah',
44                'Connection': 'close',
45            },
46            body=body,
47            url='/index.php/blah?var=val',
48        )
49
50        assert resp['status'] == 200, 'status'
51        headers = resp['headers']
52        header_server = headers.pop('Server')
53        assert re.search(r'Unit/[\d\.]+', header_server), 'server header'
54        assert (
55            headers.pop('Server-Software') == header_server
56        ), 'server software header'
57
58        date = headers.pop('Date')
59        assert date[-4:] == ' GMT', 'date header timezone'
60        assert (
61            abs(self.date_to_sec_epoch(date) - self.sec_epoch()) < 5
62        ), 'date header'
63
64        if 'X-Powered-By' in headers:
65            headers.pop('X-Powered-By')
66
67        headers.pop('Content-type')
68        assert headers == {
69            'Connection': 'close',
70            'Content-Length': str(len(body)),
71            'Request-Method': 'POST',
72            'Path-Info': '/blah',
73            'Request-Uri': '/index.php/blah?var=val',
74            'Http-Host': 'localhost',
75            'Server-Protocol': 'HTTP/1.1',
76            'Custom-Header': 'blah',
77        }, 'headers'
78        assert resp['body'] == body, 'body'
79
80    def test_php_application_query_string(self):
81        self.load('query_string')
82
83        resp = self.get(url='/?var1=val1&var2=val2')
84
85        assert (
86            resp['headers']['Query-String'] == 'var1=val1&var2=val2'
87        ), 'query string'
88
89    def test_php_application_query_string_empty(self):
90        self.load('query_string')
91
92        resp = self.get(url='/?')
93
94        assert resp['status'] == 200, 'query string empty status'
95        assert resp['headers']['Query-String'] == '', 'query string empty'
96
97    def test_php_application_fastcgi_finish_request(self, temp_dir):
98        self.load('fastcgi_finish_request')
99
100        assert self.get()['body'] == '0123'
101
102        with open(temp_dir + '/unit.pid', 'r') as f:
103            pid = f.read().rstrip()
104
105        call(['kill', '-s', 'USR1', pid])
106
107        with open(temp_dir + '/unit.log', 'r', errors='ignore') as f:
108            errs = re.findall(r'Error in fastcgi_finish_request', f.read())
109
110            assert len(errs) == 0, 'no error'
111
112    def test_php_application_fastcgi_finish_request_2(self, temp_dir):
113        self.load('fastcgi_finish_request')
114
115        resp = self.get(url='/?skip')
116        assert resp['status'] == 200
117        assert resp['body'] == ''
118
119        with open(temp_dir + '/unit.pid', 'r') as f:
120            pid = f.read().rstrip()
121
122        call(['kill', '-s', 'USR1', pid])
123
124        with open(temp_dir + '/unit.log', 'r', errors='ignore') as f:
125            errs = re.findall(r'Error in fastcgi_finish_request', f.read())
126
127            assert len(errs) == 0, 'no error'
128
129    def test_php_application_query_string_absent(self):
130        self.load('query_string')
131
132        resp = self.get()
133
134        assert resp['status'] == 200, 'query string absent status'
135        assert resp['headers']['Query-String'] == '', 'query string absent'
136
137    def test_php_application_phpinfo(self):
138        self.load('phpinfo')
139
140        resp = self.get()
141
142        assert resp['status'] == 200, 'status'
143        assert resp['body'] != '', 'body not empty'
144
145    def test_php_application_header_status(self):
146        self.load('header')
147
148        assert (
149            self.get(
150                headers={
151                    'Host': 'localhost',
152                    'Connection': 'close',
153                    'X-Header': 'HTTP/1.1 404 Not Found',
154                }
155            )['status']
156            == 404
157        ), 'status'
158
159        assert (
160            self.get(
161                headers={
162                    'Host': 'localhost',
163                    'Connection': 'close',
164                    'X-Header': 'http/1.1 404 Not Found',
165                }
166            )['status']
167            == 404
168        ), 'status case insensitive'
169
170        assert (
171            self.get(
172                headers={
173                    'Host': 'localhost',
174                    'Connection': 'close',
175                    'X-Header': 'HTTP/ 404 Not Found',
176                }
177            )['status']
178            == 404
179        ), 'status version empty'
180
181    def test_php_application_404(self):
182        self.load('404')
183
184        resp = self.get()
185
186        assert resp['status'] == 404, '404 status'
187        assert re.search(
188            r'<title>404 Not Found</title>', resp['body']
189        ), '404 body'
190
191    def test_php_application_keepalive_body(self):
192        self.load('mirror')
193
194        assert self.get()['status'] == 200, 'init'
195
196        body = '0123456789' * 500
197        (resp, sock) = self.post(
198            headers={
199                'Host': 'localhost',
200                'Connection': 'keep-alive',
201                'Content-Type': 'text/html',
202            },
203            start=True,
204            body=body,
205            read_timeout=1,
206        )
207
208        assert resp['body'] == body, 'keep-alive 1'
209
210        body = '0123456789'
211        resp = self.post(
212            headers={
213                'Host': 'localhost',
214                'Connection': 'close',
215                'Content-Type': 'text/html',
216            },
217            sock=sock,
218            body=body,
219        )
220
221        assert resp['body'] == body, 'keep-alive 2'
222
223    def test_php_application_conditional(self):
224        self.load('conditional')
225
226        assert re.search(r'True', self.get()['body']), 'conditional true'
227        assert re.search(r'False', self.post()['body']), 'conditional false'
228
229    def test_php_application_get_variables(self):
230        self.load('get_variables')
231
232        resp = self.get(url='/?var1=val1&var2=&var3')
233        assert resp['headers']['X-Var-1'] == 'val1', 'GET variables'
234        assert resp['headers']['X-Var-2'] == '', 'GET variables 2'
235        assert resp['headers']['X-Var-3'] == '', 'GET variables 3'
236        assert resp['headers']['X-Var-4'] == 'not set', 'GET variables 4'
237
238    def test_php_application_post_variables(self):
239        self.load('post_variables')
240
241        resp = self.post(
242            headers={
243                'Content-Type': 'application/x-www-form-urlencoded',
244                'Host': 'localhost',
245                'Connection': 'close',
246            },
247            body='var1=val1&var2=',
248        )
249        assert resp['headers']['X-Var-1'] == 'val1', 'POST variables'
250        assert resp['headers']['X-Var-2'] == '', 'POST variables 2'
251        assert resp['headers']['X-Var-3'] == 'not set', 'POST variables 3'
252
253    def test_php_application_cookies(self):
254        self.load('cookies')
255
256        resp = self.get(
257            headers={
258                'Cookie': 'var=val; var2=val2',
259                'Host': 'localhost',
260                'Connection': 'close',
261            }
262        )
263
264        assert resp['headers']['X-Cookie-1'] == 'val', 'cookie'
265        assert resp['headers']['X-Cookie-2'] == 'val2', 'cookie'
266
267    def test_php_application_ini_precision(self):
268        self.load('ini_precision')
269
270        assert self.get()['headers']['X-Precision'] != '4', 'ini value default'
271
272        self.conf(
273            {"file": "ini/php.ini"}, 'applications/ini_precision/options'
274        )
275
276        assert (
277            self.get()['headers']['X-File']
278            == option.test_dir + '/php/ini_precision/ini/php.ini'
279        ), 'ini file'
280        assert self.get()['headers']['X-Precision'] == '4', 'ini value'
281
282    @pytest.mark.skip('not yet')
283    def test_php_application_ini_admin_user(self):
284        self.load('ini_precision')
285
286        assert 'error' in self.conf(
287            {"user": {"precision": "4"}, "admin": {"precision": "5"}},
288            'applications/ini_precision/options',
289        ), 'ini admin user'
290
291    def test_php_application_ini_admin(self):
292        self.load('ini_precision')
293
294        self.conf(
295            {"file": "php.ini", "admin": {"precision": "5"}},
296            'applications/ini_precision/options',
297        )
298
299        assert self.get()['headers']['X-Precision'] == '5', 'ini value admin'
300
301    def test_php_application_ini_user(self):
302        self.load('ini_precision')
303
304        self.conf(
305            {"file": "php.ini", "user": {"precision": "5"}},
306            'applications/ini_precision/options',
307        )
308
309        assert self.get()['headers']['X-Precision'] == '5', 'ini value user'
310
311    def test_php_application_ini_user_2(self):
312        self.load('ini_precision')
313
314        self.conf(
315            {"file": "ini/php.ini"}, 'applications/ini_precision/options'
316        )
317
318        assert self.get()['headers']['X-Precision'] == '4', 'ini user file'
319
320        self.conf(
321            {"precision": "5"}, 'applications/ini_precision/options/user'
322        )
323
324        assert self.get()['headers']['X-Precision'] == '5', 'ini value user'
325
326    def test_php_application_ini_set_admin(self):
327        self.load('ini_precision')
328
329        self.conf(
330            {"admin": {"precision": "5"}}, 'applications/ini_precision/options'
331        )
332
333        assert (
334            self.get(url='/?precision=6')['headers']['X-Precision'] == '5'
335        ), 'ini set admin'
336
337    def test_php_application_ini_set_user(self):
338        self.load('ini_precision')
339
340        self.conf(
341            {"user": {"precision": "5"}}, 'applications/ini_precision/options'
342        )
343
344        assert (
345            self.get(url='/?precision=6')['headers']['X-Precision'] == '6'
346        ), 'ini set user'
347
348    def test_php_application_ini_repeat(self):
349        self.load('ini_precision')
350
351        self.conf(
352            {"user": {"precision": "5"}}, 'applications/ini_precision/options'
353        )
354
355        assert self.get()['headers']['X-Precision'] == '5', 'ini value'
356
357        assert self.get()['headers']['X-Precision'] == '5', 'ini value repeat'
358
359    def test_php_application_disable_functions_exec(self):
360        self.load('time_exec')
361
362        self.before_disable_functions()
363
364        self.conf(
365            {"admin": {"disable_functions": "exec"}},
366            'applications/time_exec/options',
367        )
368
369        body = self.get()['body']
370
371        assert re.search(r'time: \d+', body), 'disable_functions time'
372        assert not re.search(r'exec: \/\w+', body), 'disable_functions exec'
373
374    def test_php_application_disable_functions_comma(self):
375        self.load('time_exec')
376
377        self.before_disable_functions()
378
379        self.conf(
380            {"admin": {"disable_functions": "exec,time"}},
381            'applications/time_exec/options',
382        )
383
384        body = self.get()['body']
385
386        assert not re.search(
387            r'time: \d+', body
388        ), 'disable_functions comma time'
389        assert not re.search(
390            r'exec: \/\w+', body
391        ), 'disable_functions comma exec'
392
393    def test_php_application_auth(self):
394        self.load('auth')
395
396        resp = self.get()
397        assert resp['status'] == 200, 'status'
398        assert resp['headers']['X-Digest'] == 'not set', 'digest'
399        assert resp['headers']['X-User'] == 'not set', 'user'
400        assert resp['headers']['X-Password'] == 'not set', 'password'
401
402        resp = self.get(
403            headers={
404                'Host': 'localhost',
405                'Authorization': 'Basic dXNlcjpwYXNzd29yZA==',
406                'Connection': 'close',
407            }
408        )
409        assert resp['status'] == 200, 'basic status'
410        assert resp['headers']['X-Digest'] == 'not set', 'basic digest'
411        assert resp['headers']['X-User'] == 'user', 'basic user'
412        assert resp['headers']['X-Password'] == 'password', 'basic password'
413
414        resp = self.get(
415            headers={
416                'Host': 'localhost',
417                'Authorization': 'Digest username="blah", realm="", uri="/"',
418                'Connection': 'close',
419            }
420        )
421        assert resp['status'] == 200, 'digest status'
422        assert (
423            resp['headers']['X-Digest'] == 'username="blah", realm="", uri="/"'
424        ), 'digest digest'
425        assert resp['headers']['X-User'] == 'not set', 'digest user'
426        assert resp['headers']['X-Password'] == 'not set', 'digest password'
427
428    def test_php_application_auth_invalid(self):
429        self.load('auth')
430
431        def check_auth(auth):
432            resp = self.get(headers={
433                'Host': 'localhost',
434                'Authorization': auth,
435                'Connection': 'close',
436            })
437
438            assert resp['status'] == 200, 'status'
439            assert resp['headers']['X-Digest'] == 'not set', 'Digest'
440            assert resp['headers']['X-User'] == 'not set', 'User'
441            assert resp['headers']['X-Password'] == 'not set', 'Password'
442
443        check_auth('Basic dXN%cjpwYXNzd29yZA==')
444        check_auth('Basic XNlcjpwYXNzd29yZA==')
445        check_auth('Basic DdXNlcjpwYXNzd29yZA==')
446        check_auth('Basic blah')
447        check_auth('Basic')
448        check_auth('Digest')
449        check_auth('blah')
450
451    def test_php_application_disable_functions_space(self):
452        self.load('time_exec')
453
454        self.before_disable_functions()
455
456        self.conf(
457            {"admin": {"disable_functions": "exec time"}},
458            'applications/time_exec/options',
459        )
460
461        body = self.get()['body']
462
463        assert not re.search(
464            r'time: \d+', body
465        ), 'disable_functions space time'
466        assert not re.search(
467            r'exec: \/\w+', body
468        ), 'disable_functions space exec'
469
470    def test_php_application_disable_functions_user(self):
471        self.load('time_exec')
472
473        self.before_disable_functions()
474
475        self.conf(
476            {"user": {"disable_functions": "exec"}},
477            'applications/time_exec/options',
478        )
479
480        body = self.get()['body']
481
482        assert re.search(r'time: \d+', body), 'disable_functions user time'
483        assert not re.search(
484            r'exec: \/\w+', body
485        ), 'disable_functions user exec'
486
487    def test_php_application_disable_functions_nonexistent(self):
488        self.load('time_exec')
489
490        self.before_disable_functions()
491
492        self.conf(
493            {"admin": {"disable_functions": "blah"}},
494            'applications/time_exec/options',
495        )
496
497        body = self.get()['body']
498
499        assert re.search(
500            r'time: \d+', body
501        ), 'disable_functions nonexistent time'
502        assert re.search(
503            r'exec: \/\w+', body
504        ), 'disable_functions nonexistent exec'
505
506    def test_php_application_disable_classes(self):
507        self.load('date_time')
508
509        assert re.search(
510            r'012345', self.get()['body']
511        ), 'disable_classes before'
512
513        self.conf(
514            {"admin": {"disable_classes": "DateTime"}},
515            'applications/date_time/options',
516        )
517
518        assert not re.search(
519            r'012345', self.get()['body']
520        ), 'disable_classes before'
521
522    def test_php_application_disable_classes_user(self):
523        self.load('date_time')
524
525        assert re.search(
526            r'012345', self.get()['body']
527        ), 'disable_classes before'
528
529        self.conf(
530            {"user": {"disable_classes": "DateTime"}},
531            'applications/date_time/options',
532        )
533
534        assert not re.search(
535            r'012345', self.get()['body']
536        ), 'disable_classes before'
537
538    def test_php_application_error_log(self, temp_dir):
539        self.load('error_log')
540
541        assert self.get()['status'] == 200, 'status'
542
543        time.sleep(1)
544
545        assert self.get()['status'] == 200, 'status 2'
546
547        pattern = r'\d{4}\/\d\d\/\d\d\s\d\d:.+\[notice\].+Error in application'
548
549        assert self.wait_for_record(pattern) is not None, 'errors print'
550
551        with open(temp_dir + '/unit.log', 'r', errors='ignore') as f:
552            errs = re.findall(pattern, f.read())
553
554            assert len(errs) == 2, 'error_log count'
555
556            date = errs[0].split('[')[0]
557            date2 = errs[1].split('[')[0]
558            assert date != date2, 'date diff'
559
560    def test_php_application_script(self):
561        assert 'success' in self.conf(
562            {
563                "listeners": {"*:7080": {"pass": "applications/script"}},
564                "applications": {
565                    "script": {
566                        "type": "php",
567                        "processes": {"spare": 0},
568                        "root": option.test_dir + "/php/script",
569                        "script": "phpinfo.php",
570                    }
571                },
572            }
573        ), 'configure script'
574
575        resp = self.get()
576
577        assert resp['status'] == 200, 'status'
578        assert resp['body'] != '', 'body not empty'
579
580    def test_php_application_index_default(self):
581        assert 'success' in self.conf(
582            {
583                "listeners": {"*:7080": {"pass": "applications/phpinfo"}},
584                "applications": {
585                    "phpinfo": {
586                        "type": "php",
587                        "processes": {"spare": 0},
588                        "root": option.test_dir + "/php/phpinfo",
589                    }
590                },
591            }
592        ), 'configure index default'
593
594        resp = self.get()
595
596        assert resp['status'] == 200, 'status'
597        assert resp['body'] != '', 'body not empty'
598
599    def test_php_application_extension_check(self, temp_dir):
600        self.load('phpinfo')
601
602        assert self.get(url='/index.wrong')['status'] != 200, 'status'
603
604        new_root = temp_dir + "/php"
605        os.mkdir(new_root)
606        shutil.copy(option.test_dir + '/php/phpinfo/index.wrong', new_root)
607
608        assert 'success' in self.conf(
609            {
610                "listeners": {"*:7080": {"pass": "applications/phpinfo"}},
611                "applications": {
612                    "phpinfo": {
613                        "type": "php",
614                        "processes": {"spare": 0},
615                        "root": new_root,
616                        "working_directory": new_root,
617                    }
618                },
619            }
620        ), 'configure new root'
621
622        resp = self.get()
623        assert str(resp['status']) + resp['body'] != '200', 'status new root'
624
625    def run_php_application_cwd_root_tests(self):
626        assert 'success' in self.conf_delete(
627            'applications/cwd/working_directory'
628        )
629
630        script_cwd = option.test_dir + '/php/cwd'
631
632        resp = self.get()
633        assert resp['status'] == 200, 'status ok'
634        assert resp['body'] == script_cwd, 'default cwd'
635
636        assert 'success' in self.conf(
637            '"' + option.test_dir + '"', 'applications/cwd/working_directory',
638        )
639
640        resp = self.get()
641        assert resp['status'] == 200, 'status ok'
642        assert resp['body'] == script_cwd, 'wdir cwd'
643
644        resp = self.get(url='/?chdir=/')
645        assert resp['status'] == 200, 'status ok'
646        assert resp['body'] == '/', 'cwd after chdir'
647
648        # cwd must be restored
649
650        resp = self.get()
651        assert resp['status'] == 200, 'status ok'
652        assert resp['body'] == script_cwd, 'cwd restored'
653
654        resp = self.get(url='/subdir/')
655        assert resp['body'] == script_cwd + '/subdir', 'cwd subdir'
656
657    def test_php_application_cwd_root(self):
658        self.load('cwd')
659        self.run_php_application_cwd_root_tests()
660
661    def test_php_application_cwd_opcache_disabled(self):
662        self.load('cwd')
663        self.set_opcache('cwd', '0')
664        self.run_php_application_cwd_root_tests()
665
666    def test_php_application_cwd_opcache_enabled(self):
667        self.load('cwd')
668        self.set_opcache('cwd', '1')
669        self.run_php_application_cwd_root_tests()
670
671    def run_php_application_cwd_script_tests(self):
672        self.load('cwd')
673
674        script_cwd = option.test_dir + '/php/cwd'
675
676        assert 'success' in self.conf_delete(
677            'applications/cwd/working_directory'
678        )
679
680        assert 'success' in self.conf('"index.php"', 'applications/cwd/script')
681
682        assert self.get()['body'] == script_cwd, 'default cwd'
683
684        assert self.get(url='/?chdir=/')['body'] == '/', 'cwd after chdir'
685
686        # cwd must be restored
687        assert self.get()['body'] == script_cwd, 'cwd restored'
688
689    def test_php_application_cwd_script(self):
690        self.load('cwd')
691        self.run_php_application_cwd_script_tests()
692
693    def test_php_application_cwd_script_opcache_disabled(self):
694        self.load('cwd')
695        self.set_opcache('cwd', '0')
696        self.run_php_application_cwd_script_tests()
697
698    def test_php_application_cwd_script_opcache_enabled(self):
699        self.load('cwd')
700        self.set_opcache('cwd', '1')
701        self.run_php_application_cwd_script_tests()
702
703    def test_php_application_path_relative(self):
704        self.load('open')
705
706        assert self.get()['body'] == 'test', 'relative path'
707
708        assert (
709            self.get(url='/?chdir=/')['body'] != 'test'
710        ), 'relative path w/ chdir'
711
712        assert self.get()['body'] == 'test', 'relative path 2'
713