1import os 2import re 3import shutil 4import time 5from subprocess import call 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, temp_dir): 99 self.load('fastcgi_finish_request') 100 101 assert self.get()['body'] == '0123' 102 103 with open(temp_dir + '/unit.pid', 'r') as f: 104 pid = f.read().rstrip() 105 106 call(['kill', '-s', 'USR1', pid]) 107 108 with open(temp_dir + '/unit.log', 'r', errors='ignore') as f: 109 errs = re.findall(r'Error in fastcgi_finish_request', f.read()) 110 111 assert len(errs) == 0, 'no error' 112 113 def test_php_application_fastcgi_finish_request_2(self, temp_dir): 114 self.load('fastcgi_finish_request') 115 116 resp = self.get(url='/?skip') 117 assert resp['status'] == 200 118 assert resp['body'] == '' 119 120 with open(temp_dir + '/unit.pid', 'r') as f: 121 pid = f.read().rstrip() 122 123 call(['kill', '-s', 'USR1', pid]) 124 125 with open(temp_dir + '/unit.log', 'r', errors='ignore') as f: 126 errs = re.findall(r'Error in fastcgi_finish_request', f.read()) 127 128 assert len(errs) == 0, 'no error' 129 130 def test_php_application_query_string_absent(self): 131 self.load('query_string') 132 133 resp = self.get() 134 135 assert resp['status'] == 200, 'query string absent status' 136 assert resp['headers']['Query-String'] == '', 'query string absent' 137 138 def test_php_application_phpinfo(self): 139 self.load('phpinfo') 140 141 resp = self.get() 142 143 assert resp['status'] == 200, 'status' 144 assert resp['body'] != '', 'body not empty' 145 146 def test_php_application_header_status(self): 147 self.load('header') 148 149 assert ( 150 self.get( 151 headers={ 152 'Host': 'localhost', 153 'Connection': 'close', 154 'X-Header': 'HTTP/1.1 404 Not Found', 155 } 156 )['status'] 157 == 404 158 ), 'status' 159 160 assert ( 161 self.get( 162 headers={ 163 'Host': 'localhost', 164 'Connection': 'close', 165 'X-Header': 'http/1.1 404 Not Found', 166 } 167 )['status'] 168 == 404 169 ), 'status case insensitive' 170 171 assert ( 172 self.get( 173 headers={ 174 'Host': 'localhost', 175 'Connection': 'close', 176 'X-Header': 'HTTP/ 404 Not Found', 177 } 178 )['status'] 179 == 404 180 ), 'status version empty' 181 182 def test_php_application_404(self): 183 self.load('404') 184 185 resp = self.get() 186 187 assert resp['status'] == 404, '404 status' 188 assert re.search( 189 r'<title>404 Not Found</title>', resp['body'] 190 ), '404 body' 191 192 def test_php_application_keepalive_body(self): 193 self.load('mirror') 194 195 assert self.get()['status'] == 200, 'init' 196 197 body = '0123456789' * 500 198 (resp, sock) = self.post( 199 headers={ 200 'Host': 'localhost', 201 'Connection': 'keep-alive', 202 'Content-Type': 'text/html', 203 }, 204 start=True, 205 body=body, 206 read_timeout=1, 207 ) 208 209 assert resp['body'] == body, 'keep-alive 1' 210 211 body = '0123456789' 212 resp = self.post( 213 headers={ 214 'Host': 'localhost', 215 'Connection': 'close', 216 'Content-Type': 'text/html', 217 }, 218 sock=sock, 219 body=body, 220 ) 221 222 assert resp['body'] == body, 'keep-alive 2' 223 224 def test_php_application_conditional(self): 225 self.load('conditional') 226 227 assert re.search(r'True', self.get()['body']), 'conditional true' 228 assert re.search(r'False', self.post()['body']), 'conditional false' 229 230 def test_php_application_get_variables(self): 231 self.load('get_variables') 232 233 resp = self.get(url='/?var1=val1&var2=&var3') 234 assert resp['headers']['X-Var-1'] == 'val1', 'GET variables' 235 assert resp['headers']['X-Var-2'] == '', 'GET variables 2' 236 assert resp['headers']['X-Var-3'] == '', 'GET variables 3' 237 assert resp['headers']['X-Var-4'] == 'not set', 'GET variables 4' 238 239 def test_php_application_post_variables(self): 240 self.load('post_variables') 241 242 resp = self.post( 243 headers={ 244 'Content-Type': 'application/x-www-form-urlencoded', 245 'Host': 'localhost', 246 'Connection': 'close', 247 }, 248 body='var1=val1&var2=', 249 ) 250 assert resp['headers']['X-Var-1'] == 'val1', 'POST variables' 251 assert resp['headers']['X-Var-2'] == '', 'POST variables 2' 252 assert resp['headers']['X-Var-3'] == 'not set', 'POST variables 3' 253 254 def test_php_application_cookies(self): 255 self.load('cookies') 256 257 resp = self.get( 258 headers={ 259 'Cookie': 'var=val; var2=val2', 260 'Host': 'localhost', 261 'Connection': 'close', 262 } 263 ) 264 265 assert resp['headers']['X-Cookie-1'] == 'val', 'cookie' 266 assert resp['headers']['X-Cookie-2'] == 'val2', 'cookie' 267 268 def test_php_application_ini_precision(self): 269 self.load('ini_precision') 270 271 assert self.get()['headers']['X-Precision'] != '4', 'ini value default' 272 273 assert 'success' in self.conf( 274 {"file": "ini/php.ini"}, 'applications/ini_precision/options' 275 ) 276 277 assert ( 278 self.get()['headers']['X-File'] 279 == option.test_dir + '/php/ini_precision/ini/php.ini' 280 ), 'ini file' 281 assert self.get()['headers']['X-Precision'] == '4', 'ini value' 282 283 @pytest.mark.skip('not yet') 284 def test_php_application_ini_admin_user(self): 285 self.load('ini_precision') 286 287 assert 'error' in self.conf( 288 {"user": {"precision": "4"}, "admin": {"precision": "5"}}, 289 'applications/ini_precision/options', 290 ), 'ini admin user' 291 292 def test_php_application_ini_admin(self): 293 self.load('ini_precision') 294 295 assert 'success' in self.conf( 296 {"file": "php.ini", "admin": {"precision": "5"}}, 297 'applications/ini_precision/options', 298 ) 299 300 assert self.get()['headers']['X-Precision'] == '5', 'ini value admin' 301 302 def test_php_application_ini_user(self): 303 self.load('ini_precision') 304 305 assert 'success' in self.conf( 306 {"file": "php.ini", "user": {"precision": "5"}}, 307 'applications/ini_precision/options', 308 ) 309 310 assert self.get()['headers']['X-Precision'] == '5', 'ini value user' 311 312 def test_php_application_ini_user_2(self): 313 self.load('ini_precision') 314 315 assert 'success' in self.conf( 316 {"file": "ini/php.ini"}, 'applications/ini_precision/options' 317 ) 318 319 assert self.get()['headers']['X-Precision'] == '4', 'ini user file' 320 321 assert 'success' in self.conf( 322 {"precision": "5"}, 'applications/ini_precision/options/user' 323 ) 324 325 assert self.get()['headers']['X-Precision'] == '5', 'ini value user' 326 327 def test_php_application_ini_set_admin(self): 328 self.load('ini_precision') 329 330 assert 'success' in self.conf( 331 {"admin": {"precision": "5"}}, 'applications/ini_precision/options' 332 ) 333 334 assert ( 335 self.get(url='/?precision=6')['headers']['X-Precision'] == '5' 336 ), 'ini set admin' 337 338 def test_php_application_ini_set_user(self): 339 self.load('ini_precision') 340 341 assert 'success' in self.conf( 342 {"user": {"precision": "5"}}, 'applications/ini_precision/options' 343 ) 344 345 assert ( 346 self.get(url='/?precision=6')['headers']['X-Precision'] == '6' 347 ), 'ini set user' 348 349 def test_php_application_ini_repeat(self): 350 self.load('ini_precision') 351 352 assert 'success' in self.conf( 353 {"user": {"precision": "5"}}, 'applications/ini_precision/options' 354 ) 355 356 assert self.get()['headers']['X-Precision'] == '5', 'ini value' 357 358 assert self.get()['headers']['X-Precision'] == '5', 'ini value repeat' 359 360 def test_php_application_disable_functions_exec(self): 361 self.load('time_exec') 362 363 self.before_disable_functions() 364 365 assert 'success' in self.conf( 366 {"admin": {"disable_functions": "exec"}}, 367 'applications/time_exec/options', 368 ) 369 370 body = self.get()['body'] 371 372 assert re.search(r'time: \d+', body), 'disable_functions time' 373 assert not re.search(r'exec: \/\w+', body), 'disable_functions exec' 374 375 def test_php_application_disable_functions_comma(self): 376 self.load('time_exec') 377 378 self.before_disable_functions() 379 380 assert 'success' in self.conf( 381 {"admin": {"disable_functions": "exec,time"}}, 382 'applications/time_exec/options', 383 ) 384 385 body = self.get()['body'] 386 387 assert not re.search( 388 r'time: \d+', body 389 ), 'disable_functions comma time' 390 assert not re.search( 391 r'exec: \/\w+', body 392 ), 'disable_functions comma exec' 393 394 def test_php_application_auth(self): 395 self.load('auth') 396 397 resp = self.get() 398 assert resp['status'] == 200, 'status' 399 assert resp['headers']['X-Digest'] == 'not set', 'digest' 400 assert resp['headers']['X-User'] == 'not set', 'user' 401 assert resp['headers']['X-Password'] == 'not set', 'password' 402 403 resp = self.get( 404 headers={ 405 'Host': 'localhost', 406 'Authorization': 'Basic dXNlcjpwYXNzd29yZA==', 407 'Connection': 'close', 408 } 409 ) 410 assert resp['status'] == 200, 'basic status' 411 assert resp['headers']['X-Digest'] == 'not set', 'basic digest' 412 assert resp['headers']['X-User'] == 'user', 'basic user' 413 assert resp['headers']['X-Password'] == 'password', 'basic password' 414 415 resp = self.get( 416 headers={ 417 'Host': 'localhost', 418 'Authorization': 'Digest username="blah", realm="", uri="/"', 419 'Connection': 'close', 420 } 421 ) 422 assert resp['status'] == 200, 'digest status' 423 assert ( 424 resp['headers']['X-Digest'] == 'username="blah", realm="", uri="/"' 425 ), 'digest digest' 426 assert resp['headers']['X-User'] == 'not set', 'digest user' 427 assert resp['headers']['X-Password'] == 'not set', 'digest password' 428 429 def test_php_application_auth_invalid(self): 430 self.load('auth') 431 432 def check_auth(auth): 433 resp = self.get( 434 headers={ 435 'Host': 'localhost', 436 'Authorization': auth, 437 'Connection': 'close', 438 } 439 ) 440 441 assert resp['status'] == 200, 'status' 442 assert resp['headers']['X-Digest'] == 'not set', 'Digest' 443 assert resp['headers']['X-User'] == 'not set', 'User' 444 assert resp['headers']['X-Password'] == 'not set', 'Password' 445 446 check_auth('Basic dXN%cjpwYXNzd29yZA==') 447 check_auth('Basic XNlcjpwYXNzd29yZA==') 448 check_auth('Basic DdXNlcjpwYXNzd29yZA==') 449 check_auth('Basic blah') 450 check_auth('Basic') 451 check_auth('Digest') 452 check_auth('blah') 453 454 def test_php_application_disable_functions_space(self): 455 self.load('time_exec') 456 457 self.before_disable_functions() 458 459 assert 'success' in self.conf( 460 {"admin": {"disable_functions": "exec time"}}, 461 'applications/time_exec/options', 462 ) 463 464 body = self.get()['body'] 465 466 assert not re.search( 467 r'time: \d+', body 468 ), 'disable_functions space time' 469 assert not re.search( 470 r'exec: \/\w+', body 471 ), 'disable_functions space exec' 472 473 def test_php_application_disable_functions_user(self): 474 self.load('time_exec') 475 476 self.before_disable_functions() 477 478 assert 'success' in self.conf( 479 {"user": {"disable_functions": "exec"}}, 480 'applications/time_exec/options', 481 ) 482 483 body = self.get()['body'] 484 485 assert re.search(r'time: \d+', body), 'disable_functions user time' 486 assert not re.search( 487 r'exec: \/\w+', body 488 ), 'disable_functions user exec' 489 490 def test_php_application_disable_functions_nonexistent(self): 491 self.load('time_exec') 492 493 self.before_disable_functions() 494 495 assert 'success' in self.conf( 496 {"admin": {"disable_functions": "blah"}}, 497 'applications/time_exec/options', 498 ) 499 500 body = self.get()['body'] 501 502 assert re.search( 503 r'time: \d+', body 504 ), 'disable_functions nonexistent time' 505 assert re.search( 506 r'exec: \/\w+', body 507 ), 'disable_functions nonexistent exec' 508 509 def test_php_application_disable_classes(self): 510 self.load('date_time') 511 512 assert re.search( 513 r'012345', self.get()['body'] 514 ), 'disable_classes before' 515 516 assert 'success' in self.conf( 517 {"admin": {"disable_classes": "DateTime"}}, 518 'applications/date_time/options', 519 ) 520 521 assert not re.search( 522 r'012345', self.get()['body'] 523 ), 'disable_classes before' 524 525 def test_php_application_disable_classes_user(self): 526 self.load('date_time') 527 528 assert re.search( 529 r'012345', self.get()['body'] 530 ), 'disable_classes before' 531 532 assert 'success' in self.conf( 533 {"user": {"disable_classes": "DateTime"}}, 534 'applications/date_time/options', 535 ) 536 537 assert not re.search( 538 r'012345', self.get()['body'] 539 ), 'disable_classes before' 540 541 def test_php_application_error_log(self, temp_dir): 542 self.load('error_log') 543 544 assert self.get()['status'] == 200, 'status' 545 546 time.sleep(1) 547 548 assert self.get()['status'] == 200, 'status 2' 549 550 pattern = r'\d{4}\/\d\d\/\d\d\s\d\d:.+\[notice\].+Error in application' 551 552 assert self.wait_for_record(pattern) is not None, 'errors print' 553 554 with open(temp_dir + '/unit.log', 'r', errors='ignore') as f: 555 errs = re.findall(pattern, f.read()) 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