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