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