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