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": "php", 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": "php", 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": "php", 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