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