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