xref: /unit/test/test_php_application.py (revision 2000:c42fcbf93789)
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": "php.ini", "admin": {"precision": "5"}},
298            'applications/ini_precision/options',
299        )
300
301        assert self.get()['headers']['X-Precision'] == '5', 'ini value admin'
302
303    def test_php_application_ini_user(self):
304        self.load('ini_precision')
305
306        assert 'success' in self.conf(
307            {"file": "php.ini", "user": {"precision": "5"}},
308            'applications/ini_precision/options',
309        )
310
311        assert self.get()['headers']['X-Precision'] == '5', 'ini value user'
312
313    def test_php_application_ini_user_2(self):
314        self.load('ini_precision')
315
316        assert 'success' in self.conf(
317            {"file": "ini/php.ini"}, 'applications/ini_precision/options'
318        )
319
320        assert self.get()['headers']['X-Precision'] == '4', 'ini user file'
321
322        assert 'success' in self.conf(
323            {"precision": "5"}, 'applications/ini_precision/options/user'
324        )
325
326        assert self.get()['headers']['X-Precision'] == '5', 'ini value user'
327
328    def test_php_application_ini_set_admin(self):
329        self.load('ini_precision')
330
331        assert 'success' in self.conf(
332            {"admin": {"precision": "5"}}, 'applications/ini_precision/options'
333        )
334
335        assert (
336            self.get(url='/?precision=6')['headers']['X-Precision'] == '5'
337        ), 'ini set admin'
338
339    def test_php_application_ini_set_user(self):
340        self.load('ini_precision')
341
342        assert 'success' in self.conf(
343            {"user": {"precision": "5"}}, 'applications/ini_precision/options'
344        )
345
346        assert (
347            self.get(url='/?precision=6')['headers']['X-Precision'] == '6'
348        ), 'ini set user'
349
350    def test_php_application_ini_repeat(self):
351        self.load('ini_precision')
352
353        assert 'success' in self.conf(
354            {"user": {"precision": "5"}}, 'applications/ini_precision/options'
355        )
356
357        assert self.get()['headers']['X-Precision'] == '5', 'ini value'
358
359        assert self.get()['headers']['X-Precision'] == '5', 'ini value repeat'
360
361    def test_php_application_disable_functions_exec(self):
362        self.load('time_exec')
363
364        self.before_disable_functions()
365
366        assert 'success' in self.conf(
367            {"admin": {"disable_functions": "exec"}},
368            'applications/time_exec/options',
369        )
370
371        body = self.get()['body']
372
373        assert re.search(r'time: \d+', body), 'disable_functions time'
374        assert not re.search(r'exec: \/\w+', body), 'disable_functions exec'
375
376    def test_php_application_disable_functions_comma(self):
377        self.load('time_exec')
378
379        self.before_disable_functions()
380
381        assert 'success' in self.conf(
382            {"admin": {"disable_functions": "exec,time"}},
383            'applications/time_exec/options',
384        )
385
386        body = self.get()['body']
387
388        assert not re.search(
389            r'time: \d+', body
390        ), 'disable_functions comma time'
391        assert not re.search(
392            r'exec: \/\w+', body
393        ), 'disable_functions comma exec'
394
395    def test_php_application_auth(self):
396        self.load('auth')
397
398        resp = self.get()
399        assert resp['status'] == 200, 'status'
400        assert resp['headers']['X-Digest'] == 'not set', 'digest'
401        assert resp['headers']['X-User'] == 'not set', 'user'
402        assert resp['headers']['X-Password'] == 'not set', 'password'
403
404        resp = self.get(
405            headers={
406                'Host': 'localhost',
407                'Authorization': 'Basic dXNlcjpwYXNzd29yZA==',
408                'Connection': 'close',
409            }
410        )
411        assert resp['status'] == 200, 'basic status'
412        assert resp['headers']['X-Digest'] == 'not set', 'basic digest'
413        assert resp['headers']['X-User'] == 'user', 'basic user'
414        assert resp['headers']['X-Password'] == 'password', 'basic password'
415
416        resp = self.get(
417            headers={
418                'Host': 'localhost',
419                'Authorization': 'Digest username="blah", realm="", uri="/"',
420                'Connection': 'close',
421            }
422        )
423        assert resp['status'] == 200, 'digest status'
424        assert (
425            resp['headers']['X-Digest'] == 'username="blah", realm="", uri="/"'
426        ), 'digest digest'
427        assert resp['headers']['X-User'] == 'not set', 'digest user'
428        assert resp['headers']['X-Password'] == 'not set', 'digest password'
429
430    def test_php_application_auth_invalid(self):
431        self.load('auth')
432
433        def check_auth(auth):
434            resp = self.get(
435                headers={
436                    'Host': 'localhost',
437                    'Authorization': auth,
438                    'Connection': 'close',
439                }
440            )
441
442            assert resp['status'] == 200, 'status'
443            assert resp['headers']['X-Digest'] == 'not set', 'Digest'
444            assert resp['headers']['X-User'] == 'not set', 'User'
445            assert resp['headers']['X-Password'] == 'not set', 'Password'
446
447        check_auth('Basic dXN%cjpwYXNzd29yZA==')
448        check_auth('Basic XNlcjpwYXNzd29yZA==')
449        check_auth('Basic DdXNlcjpwYXNzd29yZA==')
450        check_auth('Basic blah')
451        check_auth('Basic')
452        check_auth('Digest')
453        check_auth('blah')
454
455    def test_php_application_disable_functions_space(self):
456        self.load('time_exec')
457
458        self.before_disable_functions()
459
460        assert 'success' in self.conf(
461            {"admin": {"disable_functions": "exec time"}},
462            'applications/time_exec/options',
463        )
464
465        body = self.get()['body']
466
467        assert not re.search(
468            r'time: \d+', body
469        ), 'disable_functions space time'
470        assert not re.search(
471            r'exec: \/\w+', body
472        ), 'disable_functions space exec'
473
474    def test_php_application_disable_functions_user(self):
475        self.load('time_exec')
476
477        self.before_disable_functions()
478
479        assert 'success' in self.conf(
480            {"user": {"disable_functions": "exec"}},
481            'applications/time_exec/options',
482        )
483
484        body = self.get()['body']
485
486        assert re.search(r'time: \d+', body), 'disable_functions user time'
487        assert not re.search(
488            r'exec: \/\w+', body
489        ), 'disable_functions user exec'
490
491    def test_php_application_disable_functions_nonexistent(self):
492        self.load('time_exec')
493
494        self.before_disable_functions()
495
496        assert 'success' in self.conf(
497            {"admin": {"disable_functions": "blah"}},
498            'applications/time_exec/options',
499        )
500
501        body = self.get()['body']
502
503        assert re.search(
504            r'time: \d+', body
505        ), 'disable_functions nonexistent time'
506        assert re.search(
507            r'exec: \/\w+', body
508        ), 'disable_functions nonexistent exec'
509
510    def test_php_application_disable_classes(self):
511        self.load('date_time')
512
513        assert re.search(
514            r'012345', self.get()['body']
515        ), 'disable_classes before'
516
517        assert 'success' in self.conf(
518            {"admin": {"disable_classes": "DateTime"}},
519            'applications/date_time/options',
520        )
521
522        assert not re.search(
523            r'012345', self.get()['body']
524        ), 'disable_classes before'
525
526    def test_php_application_disable_classes_user(self):
527        self.load('date_time')
528
529        assert re.search(
530            r'012345', self.get()['body']
531        ), 'disable_classes before'
532
533        assert 'success' in self.conf(
534            {"user": {"disable_classes": "DateTime"}},
535            'applications/date_time/options',
536        )
537
538        assert not re.search(
539            r'012345', self.get()['body']
540        ), 'disable_classes before'
541
542    def test_php_application_error_log(self):
543        self.load('error_log')
544
545        assert self.get()['status'] == 200, 'status'
546
547        time.sleep(1)
548
549        assert self.get()['status'] == 200, 'status 2'
550
551        pattern = r'\d{4}\/\d\d\/\d\d\s\d\d:.+\[notice\].+Error in application'
552
553        assert self.wait_for_record(pattern) is not None, 'errors print'
554
555        errs = self.findall(pattern)
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
717    def test_php_application_shared_opcache(self):
718        self.load('opcache', limits={'requests': 1})
719
720        r = self.get()
721        cached = r['headers']['X-Cached']
722        if cached == '-1':
723            pytest.skip('opcache is not supported')
724
725        pid = r['headers']['X-Pid']
726
727        assert cached == '0', 'not cached'
728
729        r = self.get()
730
731        assert r['headers']['X-Pid'] != pid, 'new instance'
732        assert r['headers']['X-Cached'] == '1', 'cached'
733