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