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(r'time: \d+', body), 'disable_functions comma time' 423 assert not re.search( 424 r'exec: \/\w+', body 425 ), 'disable_functions comma exec' 426 427 def test_php_application_auth(self): 428 self.load('auth') 429 430 resp = self.get() 431 assert resp['status'] == 200, 'status' 432 assert resp['headers']['X-Digest'] == 'not set', 'digest' 433 assert resp['headers']['X-User'] == 'not set', 'user' 434 assert resp['headers']['X-Password'] == 'not set', 'password' 435 436 resp = self.get( 437 headers={ 438 'Host': 'localhost', 439 'Authorization': 'Basic dXNlcjpwYXNzd29yZA==', 440 'Connection': 'close', 441 } 442 ) 443 assert resp['status'] == 200, 'basic status' 444 assert resp['headers']['X-Digest'] == 'not set', 'basic digest' 445 assert resp['headers']['X-User'] == 'user', 'basic user' 446 assert resp['headers']['X-Password'] == 'password', 'basic password' 447 448 resp = self.get( 449 headers={ 450 'Host': 'localhost', 451 'Authorization': 'Digest username="blah", realm="", uri="/"', 452 'Connection': 'close', 453 } 454 ) 455 assert resp['status'] == 200, 'digest status' 456 assert ( 457 resp['headers']['X-Digest'] == 'username="blah", realm="", uri="/"' 458 ), 'digest digest' 459 assert resp['headers']['X-User'] == 'not set', 'digest user' 460 assert resp['headers']['X-Password'] == 'not set', 'digest password' 461 462 def test_php_application_auth_invalid(self): 463 self.load('auth') 464 465 def check_auth(auth): 466 resp = self.get( 467 headers={ 468 'Host': 'localhost', 469 'Authorization': auth, 470 'Connection': 'close', 471 } 472 ) 473 474 assert resp['status'] == 200, 'status' 475 assert resp['headers']['X-Digest'] == 'not set', 'Digest' 476 assert resp['headers']['X-User'] == 'not set', 'User' 477 assert resp['headers']['X-Password'] == 'not set', 'Password' 478 479 check_auth('Basic dXN%cjpwYXNzd29yZA==') 480 check_auth('Basic XNlcjpwYXNzd29yZA==') 481 check_auth('Basic DdXNlcjpwYXNzd29yZA==') 482 check_auth('Basic blah') 483 check_auth('Basic') 484 check_auth('Digest') 485 check_auth('blah') 486 487 def test_php_application_disable_functions_space(self): 488 self.load('time_exec') 489 490 self.before_disable_functions() 491 492 assert 'success' in self.conf( 493 {"admin": {"disable_functions": "exec time"}}, 494 'applications/time_exec/options', 495 ) 496 497 body = self.get()['body'] 498 499 assert not re.search(r'time: \d+', body), 'disable_functions space time' 500 assert not re.search( 501 r'exec: \/\w+', body 502 ), 'disable_functions space exec' 503 504 def test_php_application_disable_functions_user(self): 505 self.load('time_exec') 506 507 self.before_disable_functions() 508 509 assert 'success' in self.conf( 510 {"user": {"disable_functions": "exec"}}, 511 'applications/time_exec/options', 512 ) 513 514 body = self.get()['body'] 515 516 assert re.search(r'time: \d+', body), 'disable_functions user time' 517 assert not re.search( 518 r'exec: \/\w+', body 519 ), 'disable_functions user exec' 520 521 def test_php_application_disable_functions_nonexistent(self): 522 self.load('time_exec') 523 524 self.before_disable_functions() 525 526 assert 'success' in self.conf( 527 {"admin": {"disable_functions": "blah"}}, 528 'applications/time_exec/options', 529 ) 530 531 body = self.get()['body'] 532 533 assert re.search( 534 r'time: \d+', body 535 ), 'disable_functions nonexistent time' 536 assert re.search( 537 r'exec: \/\w+', body 538 ), 'disable_functions nonexistent exec' 539 540 def test_php_application_disable_classes(self): 541 self.load('date_time') 542 543 assert re.search( 544 r'012345', self.get()['body'] 545 ), 'disable_classes before' 546 547 assert 'success' in self.conf( 548 {"admin": {"disable_classes": "DateTime"}}, 549 'applications/date_time/options', 550 ) 551 552 assert not re.search( 553 r'012345', self.get()['body'] 554 ), 'disable_classes before' 555 556 def test_php_application_disable_classes_user(self): 557 self.load('date_time') 558 559 assert re.search( 560 r'012345', self.get()['body'] 561 ), 'disable_classes before' 562 563 assert 'success' in self.conf( 564 {"user": {"disable_classes": "DateTime"}}, 565 'applications/date_time/options', 566 ) 567 568 assert not re.search( 569 r'012345', self.get()['body'] 570 ), 'disable_classes before' 571 572 def test_php_application_error_log(self): 573 self.load('error_log') 574 575 assert self.get()['status'] == 200, 'status' 576 577 time.sleep(1) 578 579 assert self.get()['status'] == 200, 'status 2' 580 581 pattern = r'\d{4}\/\d\d\/\d\d\s\d\d:.+\[notice\].+Error in application' 582 583 assert self.wait_for_record(pattern) is not None, 'errors print' 584 585 errs = self.findall(pattern) 586 587 assert len(errs) == 2, 'error_log count' 588 589 date = errs[0].split('[')[0] 590 date2 = errs[1].split('[')[0] 591 assert date != date2, 'date diff' 592 593 def test_php_application_script(self): 594 assert 'success' in self.conf( 595 { 596 "listeners": {"*:7080": {"pass": "applications/script"}}, 597 "applications": { 598 "script": { 599 "type": self.get_application_type(), 600 "processes": {"spare": 0}, 601 "root": option.test_dir + "/php/script", 602 "script": "phpinfo.php", 603 } 604 }, 605 } 606 ), 'configure script' 607 608 resp = self.get() 609 610 assert resp['status'] == 200, 'status' 611 assert resp['body'] != '', 'body not empty' 612 613 def test_php_application_index_default(self): 614 assert 'success' in self.conf( 615 { 616 "listeners": {"*:7080": {"pass": "applications/phpinfo"}}, 617 "applications": { 618 "phpinfo": { 619 "type": self.get_application_type(), 620 "processes": {"spare": 0}, 621 "root": option.test_dir + "/php/phpinfo", 622 } 623 }, 624 } 625 ), 'configure index default' 626 627 resp = self.get() 628 629 assert resp['status'] == 200, 'status' 630 assert resp['body'] != '', 'body not empty' 631 632 def test_php_application_extension_check(self, temp_dir): 633 self.load('phpinfo') 634 635 assert self.get(url='/index.wrong')['status'] != 200, 'status' 636 637 new_root = temp_dir + "/php" 638 os.mkdir(new_root) 639 shutil.copy(option.test_dir + '/php/phpinfo/index.wrong', new_root) 640 641 assert 'success' in self.conf( 642 { 643 "listeners": {"*:7080": {"pass": "applications/phpinfo"}}, 644 "applications": { 645 "phpinfo": { 646 "type": self.get_application_type(), 647 "processes": {"spare": 0}, 648 "root": new_root, 649 "working_directory": new_root, 650 } 651 }, 652 } 653 ), 'configure new root' 654 655 resp = self.get() 656 assert str(resp['status']) + resp['body'] != '200', 'status new root' 657 658 def run_php_application_cwd_root_tests(self): 659 assert 'success' in self.conf_delete( 660 'applications/cwd/working_directory' 661 ) 662 663 script_cwd = option.test_dir + '/php/cwd' 664 665 resp = self.get() 666 assert resp['status'] == 200, 'status ok' 667 assert resp['body'] == script_cwd, 'default cwd' 668 669 assert 'success' in self.conf( 670 '"' + option.test_dir + '"', 671 'applications/cwd/working_directory', 672 ) 673 674 resp = self.get() 675 assert resp['status'] == 200, 'status ok' 676 assert resp['body'] == script_cwd, 'wdir cwd' 677 678 resp = self.get(url='/?chdir=/') 679 assert resp['status'] == 200, 'status ok' 680 assert resp['body'] == '/', 'cwd after chdir' 681 682 # cwd must be restored 683 684 resp = self.get() 685 assert resp['status'] == 200, 'status ok' 686 assert resp['body'] == script_cwd, 'cwd restored' 687 688 resp = self.get(url='/subdir/') 689 assert resp['body'] == script_cwd + '/subdir', 'cwd subdir' 690 691 def test_php_application_cwd_root(self): 692 self.load('cwd') 693 self.run_php_application_cwd_root_tests() 694 695 def test_php_application_cwd_opcache_disabled(self): 696 self.load('cwd') 697 self.set_opcache('cwd', '0') 698 self.run_php_application_cwd_root_tests() 699 700 def test_php_application_cwd_opcache_enabled(self): 701 self.load('cwd') 702 self.set_opcache('cwd', '1') 703 self.run_php_application_cwd_root_tests() 704 705 def run_php_application_cwd_script_tests(self): 706 self.load('cwd') 707 708 script_cwd = option.test_dir + '/php/cwd' 709 710 assert 'success' in self.conf_delete( 711 'applications/cwd/working_directory' 712 ) 713 714 assert 'success' in self.conf('"index.php"', 'applications/cwd/script') 715 716 assert self.get()['body'] == script_cwd, 'default cwd' 717 718 assert self.get(url='/?chdir=/')['body'] == '/', 'cwd after chdir' 719 720 # cwd must be restored 721 assert self.get()['body'] == script_cwd, 'cwd restored' 722 723 def test_php_application_cwd_script(self): 724 self.load('cwd') 725 self.run_php_application_cwd_script_tests() 726 727 def test_php_application_cwd_script_opcache_disabled(self): 728 self.load('cwd') 729 self.set_opcache('cwd', '0') 730 self.run_php_application_cwd_script_tests() 731 732 def test_php_application_cwd_script_opcache_enabled(self): 733 self.load('cwd') 734 self.set_opcache('cwd', '1') 735 self.run_php_application_cwd_script_tests() 736 737 def test_php_application_path_relative(self): 738 self.load('open') 739 740 assert self.get()['body'] == 'test', 'relative path' 741 742 assert ( 743 self.get(url='/?chdir=/')['body'] != 'test' 744 ), 'relative path w/ chdir' 745 746 assert self.get()['body'] == 'test', 'relative path 2' 747 748 def test_php_application_shared_opcache(self): 749 self.load('opcache', limits={'requests': 1}) 750 751 r = self.check_opcache() 752 pid = r['headers']['X-Pid'] 753 assert r['headers']['X-Cached'] == '0', 'not cached' 754 755 r = self.get() 756 757 assert r['headers']['X-Pid'] != pid, 'new instance' 758 assert r['headers']['X-Cached'] == '1', 'cached' 759 760 def test_php_application_opcache_preload_chdir(self, temp_dir): 761 self.load('opcache') 762 763 self.check_opcache() 764 765 self.set_preload('chdir.php') 766 767 assert self.get()['headers']['X-Cached'] == '0', 'not cached' 768 assert self.get()['headers']['X-Cached'] == '1', 'cached' 769 770 def test_php_application_opcache_preload_ffr(self, temp_dir): 771 self.load('opcache') 772 773 self.check_opcache() 774 775 self.set_preload('fastcgi_finish_request.php') 776 777 assert self.get()['headers']['X-Cached'] == '0', 'not cached' 778 assert self.get()['headers']['X-Cached'] == '1', 'cached' 779