1from __future__ import absolute_import 2from __future__ import division 3from __future__ import print_function 4 5import email.utils 6import errno 7import httplib2 8import mock 9import os 10import pytest 11from six.moves import http_client, urllib 12import socket 13import tests 14 15 16def _raise_connection_refused_exception(*args, **kwargs): 17 raise socket.error(errno.ECONNREFUSED, "Connection refused.") 18 19 20def test_connection_type(): 21 http = httplib2.Http() 22 http.force_exception_to_status_code = False 23 response, content = http.request( 24 tests.DUMMY_URL, connection_type=tests.MockHTTPConnection 25 ) 26 assert response["content-location"] == tests.DUMMY_URL 27 assert content == b"the body" 28 29 30def test_bad_status_line_retry(): 31 http = httplib2.Http() 32 old_retries = httplib2.RETRIES 33 httplib2.RETRIES = 1 34 http.force_exception_to_status_code = False 35 try: 36 response, content = http.request( 37 tests.DUMMY_URL, connection_type=tests.MockHTTPBadStatusConnection 38 ) 39 except http_client.BadStatusLine: 40 assert tests.MockHTTPBadStatusConnection.num_calls == 2 41 httplib2.RETRIES = old_retries 42 43 44def test_unknown_server(): 45 http = httplib2.Http() 46 http.force_exception_to_status_code = False 47 with tests.assert_raises(httplib2.ServerNotFoundError): 48 with mock.patch("socket.socket.connect", side_effect=socket.gaierror): 49 http.request("http://no-such-hostname./") 50 51 # Now test with exceptions turned off 52 http.force_exception_to_status_code = True 53 response, content = http.request("http://no-such-hostname./") 54 assert response["content-type"] == "text/plain" 55 assert content.startswith(b"Unable to find") 56 assert response.status == 400 57 58 59@pytest.mark.skipif( 60 os.environ.get("TRAVIS_PYTHON_VERSION") in ("2.7", "pypy"), 61 reason="Fails on Travis py27/pypy, works elsewhere. " 62 "See https://travis-ci.org/httplib2/httplib2/jobs/408769880.", 63) 64@mock.patch("socket.socket.connect", spec=True) 65def test_connection_refused_raises_exception(mock_socket_connect): 66 mock_socket_connect.side_effect = _raise_connection_refused_exception 67 http = httplib2.Http() 68 http.force_exception_to_status_code = False 69 with tests.assert_raises(socket.error): 70 http.request(tests.DUMMY_URL) 71 72 73@pytest.mark.skipif( 74 os.environ.get("TRAVIS_PYTHON_VERSION") in ("2.7", "pypy"), 75 reason="Fails on Travis py27/pypy, works elsewhere. " 76 "See https://travis-ci.org/httplib2/httplib2/jobs/408769880.", 77) 78@mock.patch("socket.socket.connect", spec=True) 79def test_connection_refused_returns_response(mock_socket_connect): 80 mock_socket_connect.side_effect = _raise_connection_refused_exception 81 http = httplib2.Http() 82 http.force_exception_to_status_code = True 83 response, content = http.request(tests.DUMMY_URL) 84 content = content.lower() 85 assert response["content-type"] == "text/plain" 86 assert ( 87 b"connection refused" in content 88 or b"actively refused" in content 89 or b"socket is not connected" in content 90 ) 91 assert response.status == 400 92 93 94def test_get_iri(): 95 http = httplib2.Http() 96 query = u"?a=\N{CYRILLIC CAPITAL LETTER DJE}" 97 with tests.server_reflect() as uri: 98 response, content = http.request(uri + query, "GET") 99 assert response.status == 200 100 reflected = tests.HttpRequest.from_bytes(content) 101 assert reflected.uri == "/?a=%D0%82" 102 103 104def test_get_is_default_method(): 105 # Test that GET is the default method 106 http = httplib2.Http() 107 with tests.server_reflect() as uri: 108 response, content = http.request(uri) 109 assert response.status == 200 110 reflected = tests.HttpRequest.from_bytes(content) 111 assert reflected.method == "GET" 112 113 114def test_different_methods(): 115 # Test that all methods can be used 116 http = httplib2.Http() 117 methods = ["GET", "PUT", "DELETE", "POST", "unknown"] 118 with tests.server_reflect(request_count=len(methods)) as uri: 119 for method in methods: 120 response, content = http.request(uri, method, body=b" ") 121 assert response.status == 200 122 reflected = tests.HttpRequest.from_bytes(content) 123 assert reflected.method == method 124 125 126def test_head_read(): 127 # Test that we don't try to read the response of a HEAD request 128 # since httplib blocks response.read() for HEAD requests. 129 http = httplib2.Http() 130 respond_with = b"HTTP/1.0 200 OK\r\ncontent-length: " b"14\r\n\r\nnon-empty-body" 131 with tests.server_const_bytes(respond_with) as uri: 132 response, content = http.request(uri, "HEAD") 133 assert response.status == 200 134 assert content == b"" 135 136 137def test_get_no_cache(): 138 # Test that can do a GET w/o the cache turned on. 139 http = httplib2.Http() 140 with tests.server_const_http() as uri: 141 response, content = http.request(uri, "GET") 142 assert response.status == 200 143 assert response.previous is None 144 145 146def test_user_agent(): 147 # Test that we provide a default user-agent 148 http = httplib2.Http() 149 with tests.server_reflect() as uri: 150 response, content = http.request(uri, "GET") 151 assert response.status == 200 152 reflected = tests.HttpRequest.from_bytes(content) 153 assert reflected.headers.get("user-agent", "").startswith("Python-httplib2/") 154 155 156def test_user_agent_non_default(): 157 # Test that the default user-agent can be over-ridden 158 http = httplib2.Http() 159 with tests.server_reflect() as uri: 160 response, content = http.request(uri, "GET", headers={"User-Agent": "fred/1.0"}) 161 assert response.status == 200 162 reflected = tests.HttpRequest.from_bytes(content) 163 assert reflected.headers.get("user-agent") == "fred/1.0" 164 165 166def test_get_300_with_location(): 167 # Test the we automatically follow 300 redirects if a Location: header is provided 168 http = httplib2.Http() 169 final_content = b"This is the final destination.\n" 170 routes = { 171 "/final": tests.http_response_bytes(body=final_content), 172 "": tests.http_response_bytes( 173 status="300 Multiple Choices", headers={"location": "/final"} 174 ), 175 } 176 with tests.server_route(routes, request_count=2) as uri: 177 response, content = http.request(uri, "GET") 178 assert response.status == 200 179 assert content == final_content 180 assert response.previous.status == 300 181 assert not response.previous.fromcache 182 183 # Confirm that the intermediate 300 is not cached 184 with tests.server_route(routes, request_count=2) as uri: 185 response, content = http.request(uri, "GET") 186 assert response.status == 200 187 assert content == final_content 188 assert response.previous.status == 300 189 assert not response.previous.fromcache 190 191 192def test_get_300_with_location_noredirect(): 193 # Test the we automatically follow 300 redirects if a Location: header is provided 194 http = httplib2.Http() 195 http.follow_redirects = False 196 response = tests.http_response_bytes( 197 status="300 Multiple Choices", 198 headers={"location": "/final"}, 199 body=b"redirect body", 200 ) 201 with tests.server_const_bytes(response) as uri: 202 response, content = http.request(uri, "GET") 203 assert response.status == 300 204 205 206def test_get_300_without_location(): 207 # Not giving a Location: header in a 300 response is acceptable 208 # In which case we just return the 300 response 209 http = httplib2.Http() 210 with tests.server_const_http( 211 status="300 Multiple Choices", body=b"redirect body" 212 ) as uri: 213 response, content = http.request(uri, "GET") 214 assert response.status == 300 215 assert response.previous is None 216 assert content == b"redirect body" 217 218 219def test_get_301(): 220 # Test that we automatically follow 301 redirects 221 # and that we cache the 301 response 222 http = httplib2.Http(cache=tests.get_cache_path()) 223 destination = "" 224 routes = { 225 "/final": tests.http_response_bytes(body=b"This is the final destination.\n"), 226 "": tests.http_response_bytes( 227 status="301 Now where did I leave that URL", 228 headers={"location": "/final"}, 229 body=b"redirect body", 230 ), 231 } 232 with tests.server_route(routes, request_count=3) as uri: 233 destination = urllib.parse.urljoin(uri, "/final") 234 response1, content1 = http.request(uri, "GET") 235 response2, content2 = http.request(uri, "GET") 236 assert response1.status == 200 237 assert "content-location" in response2 238 assert response1["content-location"] == destination 239 assert content1 == b"This is the final destination.\n" 240 assert response1.previous.status == 301 241 assert not response1.previous.fromcache 242 243 assert response2.status == 200 244 assert response2["content-location"] == destination 245 assert content2 == b"This is the final destination.\n" 246 assert response2.previous.status == 301 247 assert response2.previous.fromcache 248 249 250@pytest.mark.skip( 251 not os.environ.get("httplib2_test_still_run_skipped") 252 and os.environ.get("TRAVIS_PYTHON_VERSION") in ("2.7", "pypy"), 253 reason="FIXME: timeout on Travis py27 and pypy, works elsewhere", 254) 255def test_head_301(): 256 # Test that we automatically follow 301 redirects 257 http = httplib2.Http() 258 destination = "" 259 routes = { 260 "/final": tests.http_response_bytes(body=b"This is the final destination.\n"), 261 "": tests.http_response_bytes( 262 status="301 Now where did I leave that URL", 263 headers={"location": "/final"}, 264 body=b"redirect body", 265 ), 266 } 267 with tests.server_route(routes, request_count=2) as uri: 268 destination = urllib.parse.urljoin(uri, "/final") 269 response, content = http.request(uri, "HEAD") 270 assert response.status == 200 271 assert response["content-location"] == destination 272 assert response.previous.status == 301 273 assert not response.previous.fromcache 274 275 276@pytest.mark.xfail( 277 reason=( 278 "FIXME: 301 cache works only with follow_redirects, should work " "regardless" 279 ) 280) 281def test_get_301_no_redirect(): 282 # Test that we cache the 301 response 283 http = httplib2.Http(cache=tests.get_cache_path(), timeout=0.5) 284 http.follow_redirects = False 285 response = tests.http_response_bytes( 286 status="301 Now where did I leave that URL", 287 headers={"location": "/final", "cache-control": "max-age=300"}, 288 body=b"redirect body", 289 add_date=True, 290 ) 291 with tests.server_const_bytes(response) as uri: 292 response, _ = http.request(uri, "GET") 293 assert response.status == 301 294 assert not response.fromcache 295 response, _ = http.request(uri, "GET") 296 assert response.status == 301 297 assert response.fromcache 298 299 300def test_get_302(): 301 # Test that we automatically follow 302 redirects 302 # and that we DO NOT cache the 302 response 303 http = httplib2.Http(cache=tests.get_cache_path()) 304 second_url, final_url = "", "" 305 routes = { 306 "/final": tests.http_response_bytes(body=b"This is the final destination.\n"), 307 "/second": tests.http_response_bytes( 308 status="302 Found", headers={"location": "/final"}, body=b"second redirect" 309 ), 310 "": tests.http_response_bytes( 311 status="302 Found", headers={"location": "/second"}, body=b"redirect body" 312 ), 313 } 314 with tests.server_route(routes, request_count=7) as uri: 315 second_url = urllib.parse.urljoin(uri, "/second") 316 final_url = urllib.parse.urljoin(uri, "/final") 317 response1, content1 = http.request(second_url, "GET") 318 response2, content2 = http.request(second_url, "GET") 319 response3, content3 = http.request(uri, "GET") 320 assert response1.status == 200 321 assert response1["content-location"] == final_url 322 assert content1 == b"This is the final destination.\n" 323 assert response1.previous.status == 302 324 assert not response1.previous.fromcache 325 326 assert response2.status == 200 327 # FIXME: 328 # assert response2.fromcache 329 assert response2["content-location"] == final_url 330 assert content2 == b"This is the final destination.\n" 331 assert response2.previous.status == 302 332 assert not response2.previous.fromcache 333 assert response2.previous["content-location"] == second_url 334 335 assert response3.status == 200 336 # FIXME: 337 # assert response3.fromcache 338 assert content3 == b"This is the final destination.\n" 339 assert response3.previous.status == 302 340 assert not response3.previous.fromcache 341 342 343def test_get_302_redirection_limit(): 344 # Test that we can set a lower redirection limit 345 # and that we raise an exception when we exceed 346 # that limit. 347 http = httplib2.Http() 348 http.force_exception_to_status_code = False 349 routes = { 350 "/second": tests.http_response_bytes( 351 status="302 Found", headers={"location": "/final"}, body=b"second redirect" 352 ), 353 "": tests.http_response_bytes( 354 status="302 Found", headers={"location": "/second"}, body=b"redirect body" 355 ), 356 } 357 with tests.server_route(routes, request_count=4) as uri: 358 try: 359 http.request(uri, "GET", redirections=1) 360 assert False, "This should not happen" 361 except httplib2.RedirectLimit: 362 pass 363 except Exception: 364 assert False, "Threw wrong kind of exception " 365 366 # Re-run the test with out the exceptions 367 http.force_exception_to_status_code = True 368 response, content = http.request(uri, "GET", redirections=1) 369 370 assert response.status == 500 371 assert response.reason.startswith("Redirected more") 372 assert response["status"] == "302" 373 assert content == b"second redirect" 374 assert response.previous is not None 375 376 377def test_get_302_no_location(): 378 # Test that we throw an exception when we get 379 # a 302 with no Location: header. 380 http = httplib2.Http() 381 http.force_exception_to_status_code = False 382 with tests.server_const_http(status="302 Found", request_count=2) as uri: 383 try: 384 http.request(uri, "GET") 385 assert False, "Should never reach here" 386 except httplib2.RedirectMissingLocation: 387 pass 388 except Exception: 389 assert False, "Threw wrong kind of exception " 390 391 # Re-run the test with out the exceptions 392 http.force_exception_to_status_code = True 393 response, content = http.request(uri, "GET") 394 395 assert response.status == 500 396 assert response.reason.startswith("Redirected but") 397 assert "302" == response["status"] 398 assert content == b"" 399 400 401@pytest.mark.skip( 402 not os.environ.get("httplib2_test_still_run_skipped") 403 and os.environ.get("TRAVIS_PYTHON_VERSION") in ("2.7", "pypy"), 404 reason="FIXME: timeout on Travis py27 and pypy, works elsewhere", 405) 406def test_303(): 407 # Do a follow-up GET on a Location: header 408 # returned from a POST that gave a 303. 409 http = httplib2.Http() 410 routes = { 411 "/final": tests.make_http_reflect(), 412 "": tests.make_http_reflect( 413 status="303 See Other", headers={"location": "/final"} 414 ), 415 } 416 with tests.server_route(routes, request_count=2) as uri: 417 response, content = http.request(uri, "POST", " ") 418 assert response.status == 200 419 reflected = tests.HttpRequest.from_bytes(content) 420 assert reflected.uri == "/final" 421 assert response.previous.status == 303 422 423 # Skip follow-up GET 424 http = httplib2.Http() 425 http.follow_redirects = False 426 with tests.server_route(routes, request_count=1) as uri: 427 response, content = http.request(uri, "POST", " ") 428 assert response.status == 303 429 430 # All methods can be used 431 http = httplib2.Http() 432 cases = "DELETE GET HEAD POST PUT EVEN_NEW_ONES".split(" ") 433 with tests.server_route(routes, request_count=len(cases) * 2) as uri: 434 for method in cases: 435 response, content = http.request(uri, method, body=b"q q") 436 assert response.status == 200 437 reflected = tests.HttpRequest.from_bytes(content) 438 assert reflected.method == "GET" 439 440 441def test_etag_used(): 442 # Test that we use ETags properly to validate our cache 443 cache_path = tests.get_cache_path() 444 http = httplib2.Http(cache=cache_path) 445 response_kwargs = dict( 446 add_date=True, 447 add_etag=True, 448 body=b"something", 449 headers={"cache-control": "public,max-age=300"}, 450 ) 451 452 def handler(request): 453 if request.headers.get("range"): 454 return tests.http_response_bytes(status=206, **response_kwargs) 455 return tests.http_response_bytes(**response_kwargs) 456 457 with tests.server_request(handler, request_count=2) as uri: 458 response, _ = http.request(uri, "GET", headers={"accept-encoding": "identity"}) 459 assert response["etag"] == '"437b930db84b8079c2dd804a71936b5f"' 460 461 http.request(uri, "GET", headers={"accept-encoding": "identity"}) 462 response, _ = http.request( 463 uri, 464 "GET", 465 headers={"accept-encoding": "identity", "cache-control": "must-revalidate"}, 466 ) 467 assert response.status == 200 468 assert response.fromcache 469 470 # TODO: API to read cache item, at least internal to tests 471 cache_file_name = os.path.join( 472 cache_path, httplib2.safename(httplib2.urlnorm(uri)[-1]) 473 ) 474 with open(cache_file_name, "r") as f: 475 status_line = f.readline() 476 assert status_line.startswith("status:") 477 478 response, content = http.request( 479 uri, "HEAD", headers={"accept-encoding": "identity"} 480 ) 481 assert response.status == 200 482 assert response.fromcache 483 484 response, content = http.request( 485 uri, "GET", headers={"accept-encoding": "identity", "range": "bytes=0-0"} 486 ) 487 assert response.status == 206 488 assert not response.fromcache 489 490 491def test_etag_ignore(): 492 # Test that we can forcibly ignore ETags 493 http = httplib2.Http(cache=tests.get_cache_path()) 494 response_kwargs = dict(add_date=True, add_etag=True) 495 with tests.server_reflect(request_count=3, **response_kwargs) as uri: 496 response, content = http.request( 497 uri, "GET", headers={"accept-encoding": "identity"} 498 ) 499 assert response.status == 200 500 assert response["etag"] != "" 501 502 response, content = http.request( 503 uri, 504 "GET", 505 headers={"accept-encoding": "identity", "cache-control": "max-age=0"}, 506 ) 507 reflected = tests.HttpRequest.from_bytes(content) 508 assert reflected.headers.get("if-none-match") 509 510 http.ignore_etag = True 511 response, content = http.request( 512 uri, 513 "GET", 514 headers={"accept-encoding": "identity", "cache-control": "max-age=0"}, 515 ) 516 assert not response.fromcache 517 reflected = tests.HttpRequest.from_bytes(content) 518 assert not reflected.headers.get("if-none-match") 519 520 521def test_etag_override(): 522 # Test that we can forcibly ignore ETags 523 http = httplib2.Http(cache=tests.get_cache_path()) 524 response_kwargs = dict(add_date=True, add_etag=True) 525 with tests.server_reflect(request_count=3, **response_kwargs) as uri: 526 response, _ = http.request(uri, "GET", headers={"accept-encoding": "identity"}) 527 assert response.status == 200 528 assert response["etag"] != "" 529 530 response, content = http.request( 531 uri, 532 "GET", 533 headers={"accept-encoding": "identity", "cache-control": "max-age=0"}, 534 ) 535 assert response.status == 200 536 reflected = tests.HttpRequest.from_bytes(content) 537 assert reflected.headers.get("if-none-match") 538 assert reflected.headers.get("if-none-match") != "fred" 539 540 response, content = http.request( 541 uri, 542 "GET", 543 headers={ 544 "accept-encoding": "identity", 545 "cache-control": "max-age=0", 546 "if-none-match": "fred", 547 }, 548 ) 549 assert response.status == 200 550 reflected = tests.HttpRequest.from_bytes(content) 551 assert reflected.headers.get("if-none-match") == "fred" 552 553 554@pytest.mark.skip(reason="was commented in legacy code") 555def test_get_304_end_to_end(): 556 pass 557 # Test that end to end headers get overwritten in the cache 558 # uri = urllib.parse.urljoin(base, "304/end2end.cgi") 559 # response, content = http.request(uri, 'GET') 560 # assertNotEqual(response['etag'], "") 561 # old_date = response['date'] 562 # time.sleep(2) 563 564 # response, content = http.request(uri, 'GET', headers = {'Cache-Control': 'max-age=0'}) 565 # # The response should be from the cache, but the Date: header should be updated. 566 # new_date = response['date'] 567 # assert new_date != old_date 568 # assert response.status == 200 569 # assert response.fromcache == True 570 571 572def test_get_304_last_modified(): 573 # Test that we can still handle a 304 574 # by only using the last-modified cache validator. 575 http = httplib2.Http(cache=tests.get_cache_path()) 576 date = email.utils.formatdate() 577 578 def handler(read): 579 read() 580 yield tests.http_response_bytes( 581 status=200, body=b"something", headers={"date": date, "last-modified": date} 582 ) 583 584 request2 = read() 585 assert request2.headers["if-modified-since"] == date 586 yield tests.http_response_bytes(status=304) 587 588 with tests.server_yield(handler, request_count=2) as uri: 589 response, content = http.request(uri, "GET") 590 assert response.get("last-modified") == date 591 592 response, content = http.request(uri, "GET") 593 assert response.status == 200 594 assert response.fromcache 595 596 597def test_get_307(): 598 # Test that we do follow 307 redirects but 599 # do not cache the 307 600 http = httplib2.Http(cache=tests.get_cache_path(), timeout=1) 601 r307 = tests.http_response_bytes(status=307, headers={"location": "/final"}) 602 r200 = tests.http_response_bytes( 603 status=200, 604 add_date=True, 605 body=b"final content\n", 606 headers={"cache-control": "max-age=300"}, 607 ) 608 609 with tests.server_list_http([r307, r200, r307]) as uri: 610 response, content = http.request(uri, "GET") 611 assert response.previous.status == 307 612 assert not response.previous.fromcache 613 assert response.status == 200 614 assert not response.fromcache 615 assert content == b"final content\n" 616 617 response, content = http.request(uri, "GET") 618 assert response.previous.status == 307 619 assert not response.previous.fromcache 620 assert response.status == 200 621 assert response.fromcache 622 assert content == b"final content\n" 623 624 625def test_post_307(): 626 # 307: follow with same method 627 http = httplib2.Http(cache=tests.get_cache_path(), timeout=1) 628 http.follow_all_redirects = True 629 r307 = tests.http_response_bytes(status=307, headers={"location": "/final"}) 630 r200 = tests.http_response_bytes(status=200, body=b"final content\n") 631 632 with tests.server_list_http([r307, r200, r307, r200]) as uri: 633 response, content = http.request(uri, "POST") 634 assert response.previous.status == 307 635 assert not response.previous.fromcache 636 assert response.status == 200 637 assert not response.fromcache 638 assert content == b"final content\n" 639 640 response, content = http.request(uri, "POST") 641 assert response.previous.status == 307 642 assert not response.previous.fromcache 643 assert response.status == 200 644 assert not response.fromcache 645 assert content == b"final content\n" 646 647 648def test_change_308(): 649 # 308: follow with same method, cache redirect 650 http = httplib2.Http(cache=tests.get_cache_path(), timeout=1) 651 routes = { 652 "/final": tests.make_http_reflect(), 653 "": tests.http_response_bytes( 654 status="308 Permanent Redirect", 655 add_date=True, 656 headers={"cache-control": "max-age=300", "location": "/final"}, 657 ), 658 } 659 660 with tests.server_route(routes, request_count=3) as uri: 661 response, content = http.request(uri, "CHANGE", body=b"hello308") 662 assert response.previous.status == 308 663 assert not response.previous.fromcache 664 assert response.status == 200 665 assert not response.fromcache 666 assert content.startswith(b"CHANGE /final HTTP") 667 668 response, content = http.request(uri, "CHANGE") 669 assert response.previous.status == 308 670 assert response.previous.fromcache 671 assert response.status == 200 672 assert not response.fromcache 673 assert content.startswith(b"CHANGE /final HTTP") 674 675 676def test_get_410(): 677 # Test that we pass 410's through 678 http = httplib2.Http() 679 with tests.server_const_http(status=410) as uri: 680 response, content = http.request(uri, "GET") 681 assert response.status == 410 682 683 684def test_get_duplicate_headers(): 685 # Test that duplicate headers get concatenated via ',' 686 http = httplib2.Http() 687 response = b"""HTTP/1.0 200 OK\r\n\ 688Link: link1\r\n\ 689Content-Length: 7\r\n\ 690Link: link2\r\n\r\n\ 691content""" 692 with tests.server_const_bytes(response) as uri: 693 response, content = http.request(uri, "GET") 694 assert response.status == 200 695 assert content == b"content" 696 assert response["link"], "link1, link2" 697 698 699def test_custom_redirect_codes(): 700 http = httplib2.Http() 701 http.redirect_codes = set([300]) 702 with tests.server_const_http(status=301, request_count=1) as uri: 703 response, content = http.request(uri, "GET") 704 assert response.status == 301 705 assert response.previous is None 706 707 708def test_cwe93_inject_crlf(): 709 # https://cwe.mitre.org/data/definitions/93.html 710 # GET /?q= HTTP/1.1 <- injected "HTTP/1.1" from attacker 711 # injected: attack 712 # ignore-http: HTTP/1.1 <- nominal "HTTP/1.1" from library 713 # Host: localhost:57285 714 http = httplib2.Http() 715 with tests.server_reflect() as uri: 716 danger_url = urllib.parse.urljoin( 717 uri, "?q= HTTP/1.1\r\ninjected: attack\r\nignore-http:" 718 ) 719 response, content = http.request(danger_url, "GET") 720 assert response.status == 200 721 req = tests.HttpRequest.from_bytes(content) 722 assert req.headers.get("injected") is None 723 724 725def test_inject_space(): 726 # Injecting space into request line is precursor to CWE-93 and possibly other injections 727 http = httplib2.Http() 728 with tests.server_reflect() as uri: 729 # "\r\nignore-http:" suffix is nuance for current server implementation 730 # please only pay attention to space after "?q=" 731 danger_url = urllib.parse.urljoin(uri, "?q= HTTP/1.1\r\nignore-http:") 732 response, content = http.request(danger_url, "GET") 733 assert response.status == 200 734 req = tests.HttpRequest.from_bytes(content) 735 assert req.uri == "/?q=%20HTTP/1.1%0D%0Aignore-http:" 736