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 }, 230 start=True, 231 body=body, 232 read_timeout=1, 233 ) 234 235 assert resp['body'] == body, 'keep-alive 1' 236 237 body = '0123456789' 238 resp = self.post(sock=sock, body=body) 239 240 assert resp['body'] == body, 'keep-alive 2' 241 242 def test_php_application_conditional(self): 243 self.load('conditional') 244 245 assert re.search(r'True', self.get()['body']), 'conditional true' 246 assert re.search(r'False', self.post()['body']), 'conditional false' 247 248 def test_php_application_get_variables(self): 249 self.load('get_variables') 250 251 resp = self.get(url='/?var1=val1&var2=&var3') 252 assert resp['headers']['X-Var-1'] == 'val1', 'GET variables' 253 assert resp['headers']['X-Var-2'] == '', 'GET variables 2' 254 assert resp['headers']['X-Var-3'] == '', 'GET variables 3' 255 assert resp['headers']['X-Var-4'] == 'not set', 'GET variables 4' 256 257 def test_php_application_post_variables(self): 258 self.load('post_variables') 259 260 resp = self.post( 261 headers={ 262 'Content-Type': 'application/x-www-form-urlencoded', 263 'Host': 'localhost', 264 'Connection': 'close', 265 }, 266 body='var1=val1&var2=', 267 ) 268 assert resp['headers']['X-Var-1'] == 'val1', 'POST variables' 269 assert resp['headers']['X-Var-2'] == '', 'POST variables 2' 270 assert resp['headers']['X-Var-3'] == 'not set', 'POST variables 3' 271 272 def test_php_application_cookies(self): 273 self.load('cookies') 274 275 resp = self.get( 276 headers={ 277 'Cookie': 'var=val; var2=val2', 278 'Host': 'localhost', 279 'Connection': 'close', 280 } 281 ) 282 283 assert resp['headers']['X-Cookie-1'] == 'val', 'cookie' 284 assert resp['headers']['X-Cookie-2'] == 'val2', 'cookie' 285 286 def test_php_application_ini_precision(self): 287 self.load('ini_precision') 288 289 assert self.get()['headers']['X-Precision'] != '4', 'ini value default' 290 291 assert 'success' in self.conf( 292 {"file": "ini/php.ini"}, 'applications/ini_precision/options' 293 ) 294 295 assert ( 296 self.get()['headers']['X-File'] 297 == option.test_dir + '/php/ini_precision/ini/php.ini' 298 ), 'ini file' 299 assert self.get()['headers']['X-Precision'] == '4', 'ini value' 300 301 @pytest.mark.skip('not yet') 302 def test_php_application_ini_admin_user(self): 303 self.load('ini_precision') 304 305 assert 'error' in self.conf( 306 {"user": {"precision": "4"}, "admin": {"precision": "5"}}, 307 'applications/ini_precision/options', 308 ), 'ini admin user' 309 310 def test_php_application_ini_admin(self): 311 self.load('ini_precision') 312 313 assert 'success' in self.conf( 314 {"file": "ini/php.ini", "admin": {"precision": "5"}}, 315 'applications/ini_precision/options', 316 ) 317 318 assert ( 319 self.get()['headers']['X-File'] 320 == option.test_dir + '/php/ini_precision/ini/php.ini' 321 ), 'ini file' 322 assert self.get()['headers']['X-Precision'] == '5', 'ini value admin' 323 324 def test_php_application_ini_user(self): 325 self.load('ini_precision') 326 327 assert 'success' in self.conf( 328 {"file": "ini/php.ini", "user": {"precision": "5"}}, 329 'applications/ini_precision/options', 330 ) 331 332 assert ( 333 self.get()['headers']['X-File'] 334 == option.test_dir + '/php/ini_precision/ini/php.ini' 335 ), 'ini file' 336 assert self.get()['headers']['X-Precision'] == '5', 'ini value user' 337 338 def test_php_application_ini_user_2(self): 339 self.load('ini_precision') 340 341 assert 'success' in self.conf( 342 {"file": "ini/php.ini"}, 'applications/ini_precision/options' 343 ) 344 345 assert self.get()['headers']['X-Precision'] == '4', 'ini user file' 346 347 assert 'success' in self.conf( 348 {"precision": "5"}, 'applications/ini_precision/options/user' 349 ) 350 351 assert self.get()['headers']['X-Precision'] == '5', 'ini value user' 352 353 def test_php_application_ini_set_admin(self): 354 self.load('ini_precision') 355 356 assert 'success' in self.conf( 357 {"admin": {"precision": "5"}}, 'applications/ini_precision/options' 358 ) 359 360 assert ( 361 self.get(url='/?precision=6')['headers']['X-Precision'] == '5' 362 ), 'ini set admin' 363 364 def test_php_application_ini_set_user(self): 365 self.load('ini_precision') 366 367 assert 'success' in self.conf( 368 {"user": {"precision": "5"}}, 'applications/ini_precision/options' 369 ) 370 371 assert ( 372 self.get(url='/?precision=6')['headers']['X-Precision'] == '6' 373 ), 'ini set user' 374 375 def test_php_application_ini_repeat(self): 376 self.load('ini_precision') 377 378 assert 'success' in self.conf( 379 {"user": {"precision": "5"}}, 'applications/ini_precision/options' 380 ) 381 382 assert self.get()['headers']['X-Precision'] == '5', 'ini value' 383 384 assert self.get()['headers']['X-Precision'] == '5', 'ini value repeat' 385 386 def test_php_application_disable_functions_exec(self): 387 self.load('time_exec') 388 389 self.before_disable_functions() 390 391 assert 'success' in self.conf( 392 {"admin": {"disable_functions": "exec"}}, 393 'applications/time_exec/options', 394 ) 395 396 body = self.get()['body'] 397 398 assert re.search(r'time: \d+', body), 'disable_functions time' 399 assert not re.search(r'exec: \/\w+', body), 'disable_functions exec' 400 401 def test_php_application_disable_functions_comma(self): 402 self.load('time_exec') 403 404 self.before_disable_functions() 405 406 assert 'success' in self.conf( 407 {"admin": {"disable_functions": "exec,time"}}, 408 'applications/time_exec/options', 409 ) 410 411 body = self.get()['body'] 412 413 assert not re.search(r'time: \d+', body), 'disable_functions comma time' 414 assert not re.search( 415 r'exec: \/\w+', body 416 ), 'disable_functions comma exec' 417 418 def test_php_application_auth(self): 419 self.load('auth') 420 421 resp = self.get() 422 assert resp['status'] == 200, 'status' 423 assert resp['headers']['X-Digest'] == 'not set', 'digest' 424 assert resp['headers']['X-User'] == 'not set', 'user' 425 assert resp['headers']['X-Password'] == 'not set', 'password' 426 427 resp = self.get( 428 headers={ 429 'Host': 'localhost', 430 'Authorization': 'Basic dXNlcjpwYXNzd29yZA==', 431 'Connection': 'close', 432 } 433 ) 434 assert resp['status'] == 200, 'basic status' 435 assert resp['headers']['X-Digest'] == 'not set', 'basic digest' 436 assert resp['headers']['X-User'] == 'user', 'basic user' 437 assert resp['headers']['X-Password'] == 'password', 'basic password' 438 439 resp = self.get( 440 headers={ 441 'Host': 'localhost', 442 'Authorization': 'Digest username="blah", realm="", uri="/"', 443 'Connection': 'close', 444 } 445 ) 446 assert resp['status'] == 200, 'digest status' 447 assert ( 448 resp['headers']['X-Digest'] == 'username="blah", realm="", uri="/"' 449 ), 'digest digest' 450 assert resp['headers']['X-User'] == 'not set', 'digest user' 451 assert resp['headers']['X-Password'] == 'not set', 'digest password' 452 453 def test_php_application_auth_invalid(self): 454 self.load('auth') 455 456 def check_auth(auth): 457 resp = self.get( 458 headers={ 459 'Host': 'localhost', 460 'Authorization': auth, 461 'Connection': 'close', 462 } 463 ) 464 465 assert resp['status'] == 200, 'status' 466 assert resp['headers']['X-Digest'] == 'not set', 'Digest' 467 assert resp['headers']['X-User'] == 'not set', 'User' 468 assert resp['headers']['X-Password'] == 'not set', 'Password' 469 470 check_auth('Basic dXN%cjpwYXNzd29yZA==') 471 check_auth('Basic XNlcjpwYXNzd29yZA==') 472 check_auth('Basic DdXNlcjpwYXNzd29yZA==') 473 check_auth('Basic blah') 474 check_auth('Basic') 475 check_auth('Digest') 476 check_auth('blah') 477 478 def test_php_application_disable_functions_space(self): 479 self.load('time_exec') 480 481 self.before_disable_functions() 482 483 assert 'success' in self.conf( 484 {"admin": {"disable_functions": "exec time"}}, 485 'applications/time_exec/options', 486 ) 487 488 body = self.get()['body'] 489 490 assert not re.search(r'time: \d+', body), 'disable_functions space time' 491 assert not re.search( 492 r'exec: \/\w+', body 493 ), 'disable_functions space exec' 494 495 def test_php_application_disable_functions_user(self): 496 self.load('time_exec') 497 498 self.before_disable_functions() 499 500 assert 'success' in self.conf( 501 {"user": {"disable_functions": "exec"}}, 502 'applications/time_exec/options', 503 ) 504 505 body = self.get()['body'] 506 507 assert re.search(r'time: \d+', body), 'disable_functions user time' 508 assert not re.search( 509 r'exec: \/\w+', body 510 ), 'disable_functions user exec' 511 512 def test_php_application_disable_functions_nonexistent(self): 513 self.load('time_exec') 514 515 self.before_disable_functions() 516 517 assert 'success' in self.conf( 518 {"admin": {"disable_functions": "blah"}}, 519 'applications/time_exec/options', 520 ) 521 522 body = self.get()['body'] 523 524 assert re.search( 525 r'time: \d+', body 526 ), 'disable_functions nonexistent time' 527 assert re.search( 528 r'exec: \/\w+', body 529 ), 'disable_functions nonexistent exec' 530 531 def test_php_application_disable_classes(self): 532 self.load('date_time') 533 534 assert re.search( 535 r'012345', self.get()['body'] 536 ), 'disable_classes before' 537 538 assert 'success' in self.conf( 539 {"admin": {"disable_classes": "DateTime"}}, 540 'applications/date_time/options', 541 ) 542 543 assert not re.search( 544 r'012345', self.get()['body'] 545 ), 'disable_classes before' 546 547 def test_php_application_disable_classes_user(self): 548 self.load('date_time') 549 550 assert re.search( 551 r'012345', self.get()['body'] 552 ), 'disable_classes before' 553 554 assert 'success' in self.conf( 555 {"user": {"disable_classes": "DateTime"}}, 556 'applications/date_time/options', 557 ) 558 559 assert not re.search( 560 r'012345', self.get()['body'] 561 ), 'disable_classes before' 562 563 def test_php_application_error_log(self): 564 self.load('error_log') 565 566 assert self.get()['status'] == 200, 'status' 567 568 time.sleep(1) 569 570 assert self.get()['status'] == 200, 'status 2' 571 572 pattern = r'\d{4}\/\d\d\/\d\d\s\d\d:.+\[notice\].+Error in application' 573 574 assert self.wait_for_record(pattern) is not None, 'errors print' 575 576 errs = self.findall(pattern) 577 578 assert len(errs) == 2, 'error_log count' 579 580 date = errs[0].split('[')[0] 581 date2 = errs[1].split('[')[0] 582 assert date != date2, 'date diff' 583 584 def test_php_application_script(self): 585 assert 'success' in self.conf( 586 { 587 "listeners": {"*:7080": {"pass": "applications/script"}}, 588 "applications": { 589 "script": { 590 "type": self.get_application_type(), 591 "processes": {"spare": 0}, 592 "root": option.test_dir + "/php/script", 593 "script": "phpinfo.php", 594 } 595 }, 596 } 597 ), 'configure script' 598 599 resp = self.get() 600 601 assert resp['status'] == 200, 'status' 602 assert resp['body'] != '', 'body not empty' 603 604 def test_php_application_index_default(self): 605 assert 'success' in self.conf( 606 { 607 "listeners": {"*:7080": {"pass": "applications/phpinfo"}}, 608 "applications": { 609 "phpinfo": { 610 "type": self.get_application_type(), 611 "processes": {"spare": 0}, 612 "root": option.test_dir + "/php/phpinfo", 613 } 614 }, 615 } 616 ), 'configure index default' 617 618 resp = self.get() 619 620 assert resp['status'] == 200, 'status' 621 assert resp['body'] != '', 'body not empty' 622 623 def test_php_application_extension_check(self, temp_dir): 624 self.load('phpinfo') 625 626 assert self.get(url='/index.wrong')['status'] != 200, 'status' 627 628 new_root = temp_dir + "/php" 629 os.mkdir(new_root) 630 shutil.copy(option.test_dir + '/php/phpinfo/index.wrong', new_root) 631 632 assert 'success' in self.conf( 633 { 634 "listeners": {"*:7080": {"pass": "applications/phpinfo"}}, 635 "applications": { 636 "phpinfo": { 637 "type": self.get_application_type(), 638 "processes": {"spare": 0}, 639 "root": new_root, 640 "working_directory": new_root, 641 } 642 }, 643 } 644 ), 'configure new root' 645 646 resp = self.get() 647 assert str(resp['status']) + resp['body'] != '200', 'status new root' 648 649 def run_php_application_cwd_root_tests(self): 650 assert 'success' in self.conf_delete( 651 'applications/cwd/working_directory' 652 ) 653 654 script_cwd = option.test_dir + '/php/cwd' 655 656 resp = self.get() 657 assert resp['status'] == 200, 'status ok' 658 assert resp['body'] == script_cwd, 'default cwd' 659 660 assert 'success' in self.conf( 661 '"' + option.test_dir + '"', 662 'applications/cwd/working_directory', 663 ) 664 665 resp = self.get() 666 assert resp['status'] == 200, 'status ok' 667 assert resp['body'] == script_cwd, 'wdir cwd' 668 669 resp = self.get(url='/?chdir=/') 670 assert resp['status'] == 200, 'status ok' 671 assert resp['body'] == '/', 'cwd after chdir' 672 673 # cwd must be restored 674 675 resp = self.get() 676 assert resp['status'] == 200, 'status ok' 677 assert resp['body'] == script_cwd, 'cwd restored' 678 679 resp = self.get(url='/subdir/') 680 assert resp['body'] == script_cwd + '/subdir', 'cwd subdir' 681 682 def test_php_application_cwd_root(self): 683 self.load('cwd') 684 self.run_php_application_cwd_root_tests() 685 686 def test_php_application_cwd_opcache_disabled(self): 687 self.load('cwd') 688 self.set_opcache('cwd', '0') 689 self.run_php_application_cwd_root_tests() 690 691 def test_php_application_cwd_opcache_enabled(self): 692 self.load('cwd') 693 self.set_opcache('cwd', '1') 694 self.run_php_application_cwd_root_tests() 695 696 def run_php_application_cwd_script_tests(self): 697 self.load('cwd') 698 699 script_cwd = option.test_dir + '/php/cwd' 700 701 assert 'success' in self.conf_delete( 702 'applications/cwd/working_directory' 703 ) 704 705 assert 'success' in self.conf('"index.php"', 'applications/cwd/script') 706 707 assert self.get()['body'] == script_cwd, 'default cwd' 708 709 assert self.get(url='/?chdir=/')['body'] == '/', 'cwd after chdir' 710 711 # cwd must be restored 712 assert self.get()['body'] == script_cwd, 'cwd restored' 713 714 def test_php_application_cwd_script(self): 715 self.load('cwd') 716 self.run_php_application_cwd_script_tests() 717 718 def test_php_application_cwd_script_opcache_disabled(self): 719 self.load('cwd') 720 self.set_opcache('cwd', '0') 721 self.run_php_application_cwd_script_tests() 722 723 def test_php_application_cwd_script_opcache_enabled(self): 724 self.load('cwd') 725 self.set_opcache('cwd', '1') 726 self.run_php_application_cwd_script_tests() 727 728 def test_php_application_path_relative(self): 729 self.load('open') 730 731 assert self.get()['body'] == 'test', 'relative path' 732 733 assert ( 734 self.get(url='/?chdir=/')['body'] != 'test' 735 ), 'relative path w/ chdir' 736 737 assert self.get()['body'] == 'test', 'relative path 2' 738 739 def test_php_application_shared_opcache(self): 740 self.load('opcache', limits={'requests': 1}) 741 742 r = self.check_opcache() 743 pid = r['headers']['X-Pid'] 744 assert r['headers']['X-Cached'] == '0', 'not cached' 745 746 r = self.get() 747 748 assert r['headers']['X-Pid'] != pid, 'new instance' 749 assert r['headers']['X-Cached'] == '1', 'cached' 750 751 def test_php_application_opcache_preload_chdir(self, temp_dir): 752 self.load('opcache') 753 754 self.check_opcache() 755 756 self.set_preload('chdir.php') 757 758 assert self.get()['headers']['X-Cached'] == '0', 'not cached' 759 assert self.get()['headers']['X-Cached'] == '1', 'cached' 760 761 def test_php_application_opcache_preload_ffr(self, temp_dir): 762 self.load('opcache') 763 764 self.check_opcache() 765 766 self.set_preload('fastcgi_finish_request.php') 767 768 assert self.get()['headers']['X-Cached'] == '0', 'not cached' 769 assert self.get()['headers']['X-Cached'] == '1', 'cached' 770