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