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