import base64 import binascii import contextlib from contextlib import closing import copy import functools import gzip import logging import threading import datetime from io import BytesIO import subprocess import sys import time import typing # noqa: F401 import unicodedata import unittest import zlib from tornado.escape import utf8, native_str, to_unicode, json_encode, json_decode from tornado import gen from tornado.httpclient import ( HTTPRequest, HTTPResponse, _RequestProxy, HTTPError, HTTPClient, ) from tornado.httpserver import HTTPServer from tornado.ioloop import IOLoop from tornado.iostream import IOStream from tornado.log import gen_log, app_log from tornado import netutil from tornado.testing import AsyncHTTPTestCase, bind_unused_port, gen_test, ExpectLog from tornado.test.util import abstract_base_test, ignore_deprecation from tornado.web import Application, RequestHandler, url from tornado.httputil import format_timestamp, HTTPHeaders try: import tracemalloc except ImportError: # Not available on pypy. tracemalloc = None # type: ignore skipIfNoTracemalloc = unittest.skipIf( tracemalloc is None, "tracemalloc module not present" ) class HelloWorldHandler(RequestHandler): def get(self): name = self.get_argument("name", "world") self.set_header("Content-Type", "text/plain") self.finish("Hello %s!" % name) class PostHandler(RequestHandler): def post(self): self.finish( "Post arg1: %s, arg2: %s" % (self.get_argument("arg1"), self.get_argument("arg2")) ) class PutHandler(RequestHandler): def put(self): self.write("Put body: ") self.write(self.request.body) class RedirectHandler(RequestHandler): def prepare(self): self.write("redirects can have bodies too") self.redirect( self.get_argument("url"), status=int(self.get_argument("status", "302")) ) class RedirectWithoutLocationHandler(RequestHandler): def prepare(self): # For testing error handling of a redirect with no location header. self.set_status(301) self.finish() class ChunkHandler(RequestHandler): @gen.coroutine def get(self): self.write("asdf") self.flush() # Wait a bit to ensure the chunks are sent and received separately. yield gen.sleep(0.01) self.write("qwer") class AuthHandler(RequestHandler): def get(self): self.finish(self.request.headers["Authorization"]) class CountdownHandler(RequestHandler): def get(self, count): count = int(count) if count > 0: self.redirect(self.reverse_url("countdown", count - 1)) else: self.write("Zero") class EchoPostHandler(RequestHandler): def post(self): self.write(self.request.body) class UserAgentHandler(RequestHandler): def get(self): self.write(self.request.headers.get("User-Agent", "User agent not set")) class ContentLength304Handler(RequestHandler): def get(self): self.set_status(304) self.set_header("Content-Length", 42) def _clear_representation_headers(self): # Tornado strips content-length from 304 responses, but here we # want to simulate servers that include the headers anyway. pass class PatchHandler(RequestHandler): def patch(self): "Return the request payload - so we can check it is being kept" self.write(self.request.body) class AllMethodsHandler(RequestHandler): SUPPORTED_METHODS = RequestHandler.SUPPORTED_METHODS + ("OTHER",) # type: ignore def method(self): assert self.request.method is not None self.write(self.request.method) get = head = post = put = delete = options = patch = other = method # type: ignore class SetHeaderHandler(RequestHandler): def get(self): # Use get_arguments for keys to get strings, but # request.arguments for values to get bytes. for k, v in zip(self.get_arguments("k"), self.request.arguments["v"]): self.set_header(k, v) class InvalidGzipHandler(RequestHandler): def get(self) -> None: # set Content-Encoding manually to avoid automatic gzip encoding self.set_header("Content-Type", "text/plain") self.set_header("Content-Encoding", "gzip") # Triggering the potential bug seems to depend on input length. # This length is taken from the bad-response example reported in # https://github.com/tornadoweb/tornado/pull/2875 (uncompressed). text = "".join(f"Hello World {i}\n" for i in range(9000))[:149051] body = gzip.compress(text.encode(), compresslevel=6) + b"\00" self.write(body) class HeaderEncodingHandler(RequestHandler): def get(self): self.finish(self.request.headers["Foo"].encode("ISO8859-1")) class EchoHeadersHandler(RequestHandler): def get(self): self.write(json_encode(dict(self.request.headers.get_all()))) # Decompressed size of the response served by `GzipBombHandler`. Kept below # every implementation's default ``max_body_size`` of 100MB so that a # streaming_callback can retrieve the whole thing; raising it past that limit # would turn this into a test of the limit instead (see # `HTTPClientMaxBodySizeTestCase`). GZIP_BOMB_SIZE = 64 * 1024 * 1024 # Decompressed size of the bomb used for the buffered (no # streaming_callback) case. Larger than any implementation is willing to # hold in memory, so such a request cannot complete and can only be # refused. UNBUFFERABLE_BOMB_SIZE = 256 * 1024 * 1024 @functools.lru_cache(maxsize=None) def gzip_bomb(size: int = GZIP_BOMB_SIZE) -> bytes: """Returns a small gzip stream that expands to ``size`` bytes. Compressed incrementally so that building it does not itself allocate the decompressed size. """ compressor = zlib.compressobj(9, zlib.DEFLATED, 16 + zlib.MAX_WBITS) block = b"\0" * (1024 * 1024) pieces = [compressor.compress(block) for _ in range(size // len(block))] pieces.append(compressor.flush()) return b"".join(pieces) # A 421-byte brotli stream that expands to BROTLI_BOMB_SIZE bytes of zeros, # a ratio of ~637000:1 (deflate cannot exceed ~1032:1). Embedded as a # constant so that the test suite does not need a brotli module for Python. # Regenerate with: # import base64, brotli # base64.b64encode(brotli.compress(b"\0" * BROTLI_BOMB_SIZE, quality=11)) BROTLI_BOMB_SIZE = 256 * 1024 * 1024 BROTLI_BOMB = base64.b64decode( "y///P/gnAOKxQCD3/o///3/wTwDEYRGA7v0f////4J8AiMMCAN37P/7//8E/ARCHBQC6" "93/8//+DfwIgDgsAdO//+P//B/8EQBwWAOje//H//w/+CYA4LADQvf/j//8f/BMAcVgA" "oHv/x///P/gnAOKwAED3/o///3/wTwDEYQGA7v0f////4J8AiMMCAN37P/7//8E/ARCH" "BQC693/8//+DfwIgDgsAdO//+P//B/8EQBwWAOje//H//w/+CYA4LADQvf/j//8f/BMA" "cVgAoHv/x///P/gnAOKwAED3/o///3/wTwDEYQGA7v0f////4J8AiMMCAN37P/7//8E/" "ARCHBQC693/8//+DfwIgDgsAdO//+P//B/8EQBwWAOje//H//w/+CYA4LADQvf/j//8f" "/BMAcVgAoHv/x///P/gnAOKwAED3/o///3/wTwDEYQGA7v0f////4J8AiMMCAN37P/7/" "/8E/ARCHBQC693/8//+DfwIgDgsAdO//+P//B/8EQBwWAOje//H//w/+CYA4LADQvf/j" "//8f/BMAcVgAoHv/Nw==" ) @functools.lru_cache(maxsize=None) def large_body() -> bytes: """Returns a body big enough to be delivered in several chunks. The contents are non-repeating so that a chunk which was dropped, duplicated or delivered out of order can be detected. """ return b"".join(b"%09d\n" % i for i in range(100000)) class GzipBombHandler(RequestHandler): """Sends a small response that decompresses to a very large one.""" def get(self): size = int(self.get_argument("size", str(GZIP_BOMB_SIZE))) # Set Content-Encoding manually to avoid automatic gzip encoding. self.set_header("Content-Encoding", "gzip") self.write(gzip_bomb(size)) class BrotliBombHandler(RequestHandler): """Sends a brotli-encoded bomb that the client never asked for. No client implementation requests brotli, but libcurl builds its decoder from the ``Content-Encoding`` of the response without checking it against what was requested, so ``curl_httpclient`` decodes this anyway. """ def get(self): self.set_header("Content-Encoding", "br") self.write(BROTLI_BOMB) class LargeBodyHandler(RequestHandler): def get(self): # An incompressible content type, so that the server's gzip # transform leaves the body alone and this exercises the plain # streaming path. self.set_header("Content-Type", "application/octet-stream") self.write(large_body()) # These tests end up getting run redundantly: once here with the default # HTTPClient implementation, and then again in each implementation's own # test suite. class HTTPClientCommonTestCase(AsyncHTTPTestCase): # Set by subclasses whose client decodes ``Content-Encoding: br``. No # implementation asks for brotli, so this is only true where the client # decodes an encoding it did not request (see BrotliBombHandler). decompresses_brotli = False def get_app(self): return Application( [ url("/hello", HelloWorldHandler), url("/post", PostHandler), url("/put", PutHandler), url("/redirect", RedirectHandler), url("/redirect_without_location", RedirectWithoutLocationHandler), url("/chunk", ChunkHandler), url("/auth", AuthHandler), url("/countdown/([0-9]+)", CountdownHandler, name="countdown"), url("/echopost", EchoPostHandler), url("/user_agent", UserAgentHandler), url("/304_with_content_length", ContentLength304Handler), url("/all_methods", AllMethodsHandler), url("/patch", PatchHandler), url("/set_header", SetHeaderHandler), url("/invalid_gzip", InvalidGzipHandler), url("/header-encoding", HeaderEncodingHandler), url("/echo_headers", EchoHeadersHandler), url("/gzip_bomb", GzipBombHandler), url("/brotli_bomb", BrotliBombHandler), url("/large_body", LargeBodyHandler), ], gzip=True, ) def setUp(self): super().setUp() # Add a second port (serving the same app) to the HTTP server, so we can test the effects # of redirects that span different origins. sock, port = bind_unused_port() self.http_server.add_socket(sock) self.__port2 = port def get_url2(self, path: str) -> str: return f"{self.get_protocol()}://127.0.0.1:{self.__port2}{path}" def test_patch_receives_payload(self): body = b"some patch data" response = self.fetch("/patch", method="PATCH", body=body) self.assertEqual(response.code, 200) self.assertEqual(response.body, body) def test_hello_world(self): response = self.fetch("/hello") self.assertEqual(response.code, 200) self.assertEqual(response.headers["Content-Type"], "text/plain") self.assertEqual(response.body, b"Hello world!") assert response.request_time is not None self.assertEqual(int(response.request_time), 0) response = self.fetch("/hello?name=Ben") self.assertEqual(response.body, b"Hello Ben!") def test_streaming_callback(self): # streaming_callback is also tested in test_chunked chunks = [] # type: typing.List[bytes] response = self.fetch("/hello", streaming_callback=chunks.append) # with streaming_callback, data goes to the callback and not response.body self.assertEqual(chunks, [b"Hello world!"]) self.assertFalse(response.body) def test_streaming_callback_large_body(self): # A body large enough to arrive in several chunks must reach the # callback exactly once each, in order. chunks: list[bytes] = [] response = self.fetch("/large_body", streaming_callback=chunks.append) self.assertEqual(response.code, 200) self.assertEqual(b"".join(chunks), large_body()) self.assertFalse(response.body) @skipIfNoTracemalloc def test_streaming_decompression_bomb(self): # A malicious server can turn a small compressed response into an # arbitrarily large decompressed one. With a streaming_callback it # must be possible to consume the whole thing without the client # buffering more than a bounded amount of it at a time. gzip_bomb() # Precompute so it isn't counted in the measurement. received = 0 def streaming_callback(chunk): nonlocal received received += len(chunk) tracemalloc.start() try: tracemalloc.reset_peak() response = self.fetch("/gzip_bomb", streaming_callback=streaming_callback) peak = tracemalloc.get_traced_memory()[1] finally: tracemalloc.stop() self.assertEqual(response.code, 200) self.assertEqual(received, GZIP_BOMB_SIZE) self.assertFalse(response.body) self.assertLess(peak, 16 * 1024 * 1024) @skipIfNoTracemalloc def test_decompression_bomb_without_streaming_callback(self): # Without a streaming_callback the whole body has to be buffered, so # a bomb this size cannot be retrieved at all. It has to be refused # before it exhausts memory rather than buffered in full. gzip_bomb(UNBUFFERABLE_BOMB_SIZE) # Precompute. tracemalloc.start() try: tracemalloc.reset_peak() with ( self.assertRaises(HTTPError), ExpectLog(gen_log, ".*decompressed body too large"), ): self.fetch( "/gzip_bomb?size=%d" % UNBUFFERABLE_BOMB_SIZE, raise_error=True ) peak = tracemalloc.get_traced_memory()[1] finally: tracemalloc.stop() # Well short of the decompressed size: the point is that the body # was refused, not accumulated. self.assertLess(peak, UNBUFFERABLE_BOMB_SIZE // 2) @skipIfNoTracemalloc def test_streaming_unsolicited_brotli_bomb(self): # A client that decodes an encoding it did not request can be # reached by a codec whose expansion ratio is hundreds of times # deflate's. The transfer need not survive -- libcurl aborts it once # its own pause buffer cannot hold the expansion -- but either way # the expansion must not be buffered. if not self.decompresses_brotli: self.skipTest("client does not decompress Content-Encoding: br") received = 0 def streaming_callback(chunk): nonlocal received received += len(chunk) tracemalloc.start() try: tracemalloc.reset_peak() try: self.fetch("/brotli_bomb", streaming_callback=streaming_callback) except HTTPError: pass peak = tracemalloc.get_traced_memory()[1] finally: tracemalloc.stop() # More than was sent on the wire, i.e. the response really was # decoded with an encoding we never asked for. self.assertGreater(received, len(BROTLI_BOMB)) self.assertLess(peak, 16 * 1024 * 1024) def test_post(self): response = self.fetch("/post", method="POST", body="arg1=foo&arg2=bar") self.assertEqual(response.code, 200) self.assertEqual(response.body, b"Post arg1: foo, arg2: bar") def test_chunked(self): response = self.fetch("/chunk") self.assertEqual(response.body, b"asdfqwer") chunks = [] # type: typing.List[bytes] response = self.fetch("/chunk", streaming_callback=chunks.append) self.assertEqual(chunks, [b"asdf", b"qwer"]) self.assertFalse(response.body) def test_chunked_close(self): # test case in which chunks spread read-callback processing # over several ioloop iterations, but the connection is already closed. sock, port = bind_unused_port() with closing(sock): @gen.coroutine def accept_callback(conn, address): # fake an HTTP server using chunked encoding where the final chunks # and connection close all happen at once stream = IOStream(conn) request_data = yield stream.read_until(b"\r\n\r\n") if b"HTTP/1." not in request_data: self.skipTest("requires HTTP/1.x") yield stream.write( b"""\ HTTP/1.1 200 OK Transfer-Encoding: chunked 1 1 1 2 0 """.replace( b"\n", b"\r\n" ) ) stream.close() netutil.add_accept_handler(sock, accept_callback) # type: ignore resp = self.fetch("http://127.0.0.1:%d/" % port) resp.rethrow() self.assertEqual(resp.body, b"12") self.io_loop.remove_handler(sock.fileno()) def test_basic_auth(self): # This test data appears in section 2 of RFC 7617. self.assertEqual( self.fetch( "/auth", auth_username="Aladdin", auth_password="open sesame" ).body, b"Basic QWxhZGRpbjpvcGVuIHNlc2FtZQ==", ) def test_basic_auth_explicit_mode(self): self.assertEqual( self.fetch( "/auth", auth_username="Aladdin", auth_password="open sesame", auth_mode="basic", ).body, b"Basic QWxhZGRpbjpvcGVuIHNlc2FtZQ==", ) def test_basic_auth_unicode(self): # This test data appears in section 2.1 of RFC 7617. self.assertEqual( self.fetch("/auth", auth_username="test", auth_password="123£").body, b"Basic dGVzdDoxMjPCow==", ) # The standard mandates NFC. Give it a decomposed username # and ensure it is normalized to composed form. username = unicodedata.normalize("NFD", "josé") self.assertEqual( self.fetch("/auth", auth_username=username, auth_password="səcrət").body, b"Basic am9zw6k6c8mZY3LJmXQ=", ) def test_unsupported_auth_mode(self): # curl and simple clients handle errors a bit differently; the # important thing is that they don't fall back to basic auth # on an unknown mode. with ExpectLog(gen_log, "uncaught exception", required=False): with self.assertRaises((ValueError, HTTPError)): # type: ignore self.fetch( "/auth", auth_username="Aladdin", auth_password="open sesame", auth_mode="asdf", raise_error=True, ) def test_follow_redirect(self): response = self.fetch("/countdown/2", follow_redirects=False) self.assertEqual(302, response.code) self.assertTrue(response.headers["Location"].endswith("/countdown/1")) response = self.fetch("/countdown/2") self.assertEqual(200, response.code) self.assertTrue(response.effective_url.endswith("/countdown/0")) self.assertEqual(b"Zero", response.body) def test_redirect_without_location(self): response = self.fetch("/redirect_without_location", follow_redirects=True) # If there is no location header, the redirect response should # just be returned as-is. (This should arguably raise an # error, but libcurl doesn't treat this as an error, so we # don't either). self.assertEqual(301, response.code) def test_redirect_put_with_body(self): response = self.fetch( "/redirect?url=/put&status=307", method="PUT", body="hello" ) self.assertEqual(response.body, b"Put body: hello") def test_redirect_put_without_body(self): # This "without body" edge case is similar to what happens with body_producer. response = self.fetch( "/redirect?url=/put&status=307", method="PUT", allow_nonstandard_methods=True, ) self.assertEqual(response.body, b"Put body: ") def test_method_after_redirect(self): # Legacy redirect codes (301, 302) convert POST requests to GET. for status in [301, 302, 303]: url = "/redirect?url=/all_methods&status=%d" % status resp = self.fetch(url, method="POST", body=b"") self.assertEqual(b"GET", resp.body) # Other methods are left alone, except for 303 redirect, depending on client for method in ["GET", "OPTIONS", "PUT", "DELETE"]: resp = self.fetch(url, method=method, allow_nonstandard_methods=True) if status in [301, 302]: self.assertEqual(utf8(method), resp.body) else: self.assertIn(resp.body, [utf8(method), b"GET"]) # HEAD is different so check it separately. resp = self.fetch(url, method="HEAD") self.assertEqual(200, resp.code) self.assertEqual(b"", resp.body) # Newer redirects always preserve the original method. for status in [307, 308]: url = "/redirect?url=/all_methods&status=307" for method in ["GET", "OPTIONS", "POST", "PUT", "DELETE"]: resp = self.fetch(url, method=method, allow_nonstandard_methods=True) self.assertEqual(method, to_unicode(resp.body)) resp = self.fetch(url, method="HEAD") self.assertEqual(200, resp.code) self.assertEqual(b"", resp.body) def test_credentials_in_url(self): url = self.get_url("/auth").replace("http://", "http://me:secret@") response = self.fetch(url) self.assertEqual(b"Basic " + base64.b64encode(b"me:secret"), response.body) def test_body_encoding(self): unicode_body = "\xe9" byte_body = binascii.a2b_hex(b"e9") # unicode string in body gets converted to utf8 response = self.fetch( "/echopost", method="POST", body=unicode_body, headers={"Content-Type": "application/blah"}, ) self.assertEqual(response.headers["Content-Length"], "2") self.assertEqual(response.body, utf8(unicode_body)) # byte strings pass through directly response = self.fetch( "/echopost", method="POST", body=byte_body, headers={"Content-Type": "application/blah"}, ) self.assertEqual(response.headers["Content-Length"], "1") self.assertEqual(response.body, byte_body) # Mixing unicode in headers and byte string bodies shouldn't # break anything response = self.fetch( "/echopost", method="POST", body=byte_body, headers={"Content-Type": "application/blah"}, user_agent="foo", ) self.assertEqual(response.headers["Content-Length"], "1") self.assertEqual(response.body, byte_body) def test_types(self): response = self.fetch("/hello") self.assertEqual(type(response.body), bytes) self.assertEqual(type(response.headers["Content-Type"]), str) self.assertEqual(type(response.code), int) self.assertEqual(type(response.effective_url), str) def test_gzip(self): # All the tests in this file should be using gzip, but this test # ensures that it is in fact getting compressed, and also tests # the httpclient's decompress=False option. # Setting Accept-Encoding manually bypasses the client's # decompression so we can see the raw data. response = self.fetch( "/chunk", decompress_response=False, headers={"Accept-Encoding": "gzip"} ) self.assertEqual(response.headers["Content-Encoding"], "gzip") self.assertNotEqual(response.body, b"asdfqwer") # Our test data gets bigger when gzipped. Oops. :) # Chunked encoding bypasses the MIN_LENGTH check. self.assertEqual(len(response.body), 34) f = gzip.GzipFile(mode="r", fileobj=response.buffer) self.assertEqual(f.read(), b"asdfqwer") def test_invalid_gzip(self): # test if client hangs on tricky invalid gzip # curl/simple httpclient have different behavior (exception, logging) with ExpectLog( gen_log, ".*Malformed HTTP message.*unconsumed gzip data", required=False ): try: response = self.fetch("/invalid_gzip") self.assertEqual(response.code, 200) self.assertEqual(response.body[:14], b"Hello World 0\n") except HTTPError: pass # acceptable def test_header_callback(self): first_line = [] headers = {} chunks = [] def header_callback(header_line): if header_line.startswith("HTTP/1.1 101"): # Upgrading to HTTP/2 pass elif header_line.startswith("HTTP/"): first_line.append(header_line) elif header_line != "\r\n": k, v = header_line.split(":", 1) headers[k.lower()] = v.strip() def streaming_callback(chunk): # All header callbacks are run before any streaming callbacks, # so the header data is available to process the data as it # comes in. self.assertEqual(headers["content-type"], "text/html; charset=UTF-8") chunks.append(chunk) self.fetch( "/chunk", header_callback=header_callback, streaming_callback=streaming_callback, ) self.assertEqual(len(first_line), 1, first_line) self.assertRegex(first_line[0], "HTTP/[0-9]\\.[0-9] 200.*\r\n") self.assertEqual(chunks, [b"asdf", b"qwer"]) def test_header_callback_to_parse_line(self): # Make a request with header_callback and feed the headers to HTTPHeaders.parse_line. # (Instead of HTTPHeaders.parse which is used in normal cases). Ensure that the resulting # headers are as expected, and in particular do not have trailing whitespace added # due to the final CRLF line. headers = HTTPHeaders() def header_callback(line): if line.startswith("HTTP/"): # Ignore the first status line return headers.parse_line(line) self.fetch("/hello", header_callback=header_callback) for k, v in headers.get_all(): self.assertTrue(v == v.strip(), (k, v)) @gen_test def test_configure_defaults(self): defaults = dict(user_agent="TestDefaultUserAgent", allow_ipv6=False) # Construct a new instance of the configured client class client = self.http_client.__class__(force_instance=True, defaults=defaults) try: response = yield client.fetch(self.get_url("/user_agent")) self.assertEqual(response.body, b"TestDefaultUserAgent") finally: client.close() def test_header_types(self): # Header values may be passed as character or utf8 byte strings, # in a plain dictionary or an HTTPHeaders object. # Keys must always be the native str type. # All combinations should have the same results on the wire. for value in ["MyUserAgent", b"MyUserAgent"]: for container in [dict, HTTPHeaders]: headers = container() headers["User-Agent"] = value resp = self.fetch("/user_agent", headers=headers) self.assertEqual( resp.body, b"MyUserAgent", "response=%r, value=%r, container=%r" % (resp.body, value, container), ) def test_multi_line_headers(self): # Multi-line http headers are rare but rfc-allowed # http://www.w3.org/Protocols/rfc2616/rfc2616-sec4.html#sec4.2 sock, port = bind_unused_port() with closing(sock): @gen.coroutine def accept_callback(conn, address): stream = IOStream(conn) request_data = yield stream.read_until(b"\r\n\r\n") if b"HTTP/1." not in request_data: self.skipTest("requires HTTP/1.x") yield stream.write( b"""\ HTTP/1.1 200 OK X-XSS-Protection: 1; \tmode=block """.replace( b"\n", b"\r\n" ) ) stream.close() netutil.add_accept_handler(sock, accept_callback) # type: ignore try: resp = self.fetch("http://127.0.0.1:%d/" % port) resp.rethrow() self.assertEqual(resp.headers["X-XSS-Protection"], "1; mode=block") finally: self.io_loop.remove_handler(sock.fileno()) @gen_test def test_header_encoding(self): response = yield self.http_client.fetch( self.get_url("/header-encoding"), headers={ "Foo": "b\xe4r", }, ) self.assertEqual(response.body, "b\xe4r".encode("ISO8859-1")) def test_304_with_content_length(self): # According to the spec 304 responses SHOULD NOT include # Content-Length or other entity headers, but some servers do it # anyway. # http://www.w3.org/Protocols/rfc2616/rfc2616-sec10.html#sec10.3.5 response = self.fetch("/304_with_content_length") self.assertEqual(response.code, 304) self.assertEqual(response.headers["Content-Length"], "42") @gen_test def test_future_interface(self): response = yield self.http_client.fetch(self.get_url("/hello")) self.assertEqual(response.body, b"Hello world!") @gen_test def test_future_http_error(self): with self.assertRaises(HTTPError) as context: yield self.http_client.fetch(self.get_url("/notfound")) assert context.exception is not None assert context.exception.response is not None self.assertEqual(context.exception.code, 404) self.assertEqual(context.exception.response.code, 404) @gen_test def test_future_http_error_no_raise(self): response = yield self.http_client.fetch( self.get_url("/notfound"), raise_error=False ) self.assertEqual(response.code, 404) @gen_test def test_reuse_request_from_response(self): # The response.request attribute should be an HTTPRequest, not # a _RequestProxy. # This test uses self.http_client.fetch because self.fetch calls # self.get_url on the input unconditionally. url = self.get_url("/hello") response = yield self.http_client.fetch(url) self.assertEqual(response.request.url, url) self.assertTrue(isinstance(response.request, HTTPRequest)) response2 = yield self.http_client.fetch(response.request) self.assertEqual(response2.body, b"Hello world!") @gen_test def test_bind_source_ip(self): url = self.get_url("/hello") request = HTTPRequest(url, network_interface="127.0.0.1") response = yield self.http_client.fetch(request) self.assertEqual(response.code, 200) with self.assertRaises((ValueError, HTTPError)) as context: # type: ignore request = HTTPRequest(url, network_interface="not-interface-or-ip") yield self.http_client.fetch(request) self.assertTrue( "not-interface-or-ip" in str(context.exception) or "Failed binding local connection end" in str(context.exception) ) def test_all_methods(self): for method in ["GET", "DELETE", "OPTIONS"]: response = self.fetch("/all_methods", method=method) self.assertEqual(response.body, utf8(method)) for method in ["POST", "PUT", "PATCH"]: response = self.fetch("/all_methods", method=method, body=b"") self.assertEqual(response.body, utf8(method)) response = self.fetch("/all_methods", method="HEAD") self.assertEqual(response.body, b"") response = self.fetch( "/all_methods", method="OTHER", allow_nonstandard_methods=True ) self.assertEqual(response.body, b"OTHER") def test_body_sanity_checks(self): # These methods require a body. for method in ("POST", "PUT", "PATCH"): with self.assertRaises(ValueError) as context: self.fetch("/all_methods", method=method, raise_error=True) self.assertIn("must not be None", str(context.exception)) resp = self.fetch( "/all_methods", method=method, allow_nonstandard_methods=True ) self.assertEqual(resp.code, 200) # These methods don't allow a body. for method in ("GET", "DELETE", "OPTIONS"): with self.assertRaises(ValueError) as context: self.fetch( "/all_methods", method=method, body=b"asdf", raise_error=True ) self.assertIn("must be None", str(context.exception)) # In most cases this can be overridden, but curl_httpclient # does not allow body with a GET at all. if method != "GET": self.fetch( "/all_methods", method=method, body=b"asdf", allow_nonstandard_methods=True, raise_error=True, ) self.assertEqual(resp.code, 200) # This test causes odd failures with the combination of # curl_httpclient (at least with the version of libcurl available # on ubuntu 12.04), TwistedIOLoop, and epoll. For POST (but not PUT), # curl decides the response came back too soon and closes the connection # to start again. It does this *before* telling the socket callback to # unregister the FD. Some IOLoop implementations have special kernel # integration to discover this immediately. Tornado's IOLoops # ignore errors on remove_handler to accommodate this behavior, but # Twisted's reactor does not. The removeReader call fails and so # do all future removeAll calls (which our tests do at cleanup). # # def test_post_307(self): # response = self.fetch("/redirect?status=307&url=/post", # method="POST", body=b"arg1=foo&arg2=bar") # self.assertEqual(response.body, b"Post arg1: foo, arg2: bar") def test_put_307(self): response = self.fetch( "/redirect?status=307&url=/put", method="PUT", body=b"hello" ) response.rethrow() self.assertEqual(response.body, b"Put body: hello") def test_non_ascii_header(self): # Non-ascii headers are sent as latin1. response = self.fetch("/set_header?k=foo&v=%E9") response.rethrow() self.assertEqual(response.headers["Foo"], native_str("\u00e9")) def test_response_times(self): # A few simple sanity checks of the response time fields to # make sure they're using the right basis (between the # wall-time and monotonic clocks). start_time = time.time() response = self.fetch("/hello") response.rethrow() self.assertIsNotNone(response.request_time) assert response.request_time is not None # for mypy self.assertGreaterEqual(response.request_time, 0) self.assertLess(response.request_time, 1.0) # A very crude check to make sure that start_time is based on # wall time and not the monotonic clock. self.assertIsNotNone(response.start_time) assert response.start_time is not None # for mypy self.assertLess(abs(response.start_time - start_time), 1.0) for k, v in response.time_info.items(): self.assertTrue(0 <= v < 1.0, f"time_info[{k}] out of bounds: {v}") def test_zero_timeout(self): response = self.fetch("/hello", connect_timeout=0) self.assertEqual(response.code, 200) response = self.fetch("/hello", request_timeout=0) self.assertEqual(response.code, 200) response = self.fetch("/hello", connect_timeout=0, request_timeout=0) self.assertEqual(response.code, 200) @gen_test def test_error_after_cancel(self): fut = self.http_client.fetch(self.get_url("/404")) self.assertTrue(fut.cancel()) with ExpectLog(app_log, "Exception after Future was cancelled") as el: # We can't wait on the cancelled Future any more, so just # let the IOLoop run until the exception gets logged (or # not, in which case we exit the loop and ExpectLog will # raise). for i in range(100): yield gen.sleep(0.01) if el.logged_stack: break def test_header_crlf(self): # Ensure that the client doesn't allow CRLF injection in headers. RFC 9112 section 2.2 # prohibits a bare CR specifically and "a recipient MAY recognize a single LF as a line # terminator" so we check each character separately as well as the (redundant) CRLF pair. for header, name in [ ("foo\rbar:", "cr"), ("foo\nbar:", "lf"), ("foo\r\nbar:", "crlf"), ]: with self.subTest(name=name, position="value"): with self.assertRaises(ValueError): self.fetch("/hello", headers={"foo": header}) with self.subTest(name=name, position="key"): with self.assertRaises(ValueError): self.fetch("/hello", headers={header: "foo"}) def test_strip_headers_on_redirect(self): # Ensure that headers that should be stripped on cross-origin redirects # are stripped, even if the redirect is to a different port on localhost. test_cases: list[tuple[str, dict, str]] = [ ("manual auth header", dict(headers={"Authorization": "secret"}), ""), ("credentials in URL", dict(), "me:secret"), ("auth parameters", dict(auth_username="me", auth_password="secret"), ""), ("manual cookie header", dict(headers={"Cookie": "secret"}), ""), ] for name, kwargs, url_creds in test_cases: with self.subTest(name=name, origin="different"): url = self.get_url( "/redirect?url=%s&status=302" % self.get_url2("/echo_headers") ) if url_creds: # Only add credentials to the outer URL being fetched, not to the # "url" query parameter (the redirect target), which also starts # with "http://". Otherwise the redirect's Location header would # carry its own explicit credentials for the new origin, which # libcurl legitimately honors instead of stripping. url = url.replace("http://", "http://%s@" % url_creds, 1) response = self.fetch(**dict(path=url) | kwargs) response.rethrow() echoed_headers = json_decode(response.body) # Confirm that non-auth headers are getting through self.assertIn("User-Agent", echoed_headers) # Auth headers are stripped, however they were set. self.assertNotIn("Authorization", echoed_headers) self.assertNotIn("Cookie", echoed_headers) with self.subTest(name=name, origin="same"): url = self.get_url( "/redirect?url=%s&status=302" % self.get_url("/echo_headers") ) if url_creds: url = url.replace("http://", "http://%s@" % url_creds, 1) response = self.fetch(**dict(path=url) | kwargs) response.rethrow() echoed_headers = json_decode(response.body) # Confirm that non-auth headers are getting through self.assertIn("User-Agent", echoed_headers) if name == "credentials in URL": # Some libcurl versions (known regression as of 8.20/8.21, # still present as of curl's git master) drop credentials # embedded in the URL across a same-origin redirect whose # Location header is an absolute URL, even though they # should be preserved. This isn't a security concern # (nothing is leaked to another origin), so just don't # assert on it either way here. pass else: # Auth headers are not stripped when the redirect is same-origin. # Each of our tests uses one of these headers, but not both. self.assertTrue( "Authorization" in echoed_headers or "Cookie" in echoed_headers ) class RequestProxyTest(unittest.TestCase): def test_request_set(self): proxy = _RequestProxy( HTTPRequest("http://example.com/", user_agent="foo"), dict() ) self.assertEqual(proxy.user_agent, "foo") def test_default_set(self): proxy = _RequestProxy( HTTPRequest("http://example.com/"), dict(network_interface="foo") ) self.assertEqual(proxy.network_interface, "foo") def test_both_set(self): proxy = _RequestProxy( HTTPRequest("http://example.com/", proxy_host="foo"), dict(proxy_host="bar") ) self.assertEqual(proxy.proxy_host, "foo") def test_neither_set(self): proxy = _RequestProxy(HTTPRequest("http://example.com/"), dict()) self.assertIsNone(proxy.auth_username) def test_bad_attribute(self): proxy = _RequestProxy(HTTPRequest("http://example.com/"), dict()) with self.assertRaises(AttributeError): proxy.foo def test_defaults_none(self): proxy = _RequestProxy(HTTPRequest("http://example.com/"), None) self.assertIsNone(proxy.auth_username) class HTTPResponseTestCase(unittest.TestCase): def test_str(self): response = HTTPResponse( # type: ignore HTTPRequest("http://example.com"), 200, buffer=BytesIO() ) s = str(response) self.assertTrue(s.startswith("HTTPResponse(")) self.assertIn("code=200", s) class SyncHTTPClientTest(unittest.TestCase): def setUp(self): self.server_ioloop = IOLoop(make_current=False) event = threading.Event() @gen.coroutine def init_server(): sock, self.port = bind_unused_port() app = Application([("/", HelloWorldHandler)]) self.server = HTTPServer(app) self.server.add_socket(sock) event.set() def start(): self.server_ioloop.run_sync(init_server) self.server_ioloop.start() self.server_thread = threading.Thread(target=start) self.server_thread.start() event.wait() self.http_client = HTTPClient() def tearDown(self): def stop_server(): self.server.stop() # Delay the shutdown of the IOLoop by several iterations because # the server may still have some cleanup work left when # the client finishes with the response (this is noticeable # with http/2, which leaves a Future with an unexamined # StreamClosedError on the loop). @gen.coroutine def slow_stop(): yield self.server.close_all_connections() # The number of iterations is difficult to predict. Typically, # one is sufficient, although sometimes it needs more. for i in range(5): yield self.server_ioloop.stop() self.server_ioloop.add_callback(slow_stop) self.server_ioloop.add_callback(stop_server) self.server_thread.join() self.http_client.close() self.server_ioloop.close(all_fds=True) def get_url(self, path): return "http://127.0.0.1:%d%s" % (self.port, path) def test_sync_client(self): response = self.http_client.fetch(self.get_url("/")) self.assertEqual(b"Hello world!", response.body) def test_sync_client_error(self): # Synchronous HTTPClient raises errors directly; no need for # response.rethrow() with self.assertRaises(HTTPError) as assertion: self.http_client.fetch(self.get_url("/notfound")) self.assertEqual(assertion.exception.code, 404) class SyncHTTPClientSubprocessTest(unittest.TestCase): def test_destructor_log(self): # Regression test for # https://github.com/tornadoweb/tornado/issues/2539 # # In the past, the following program would log an # "inconsistent AsyncHTTPClient cache" error from a destructor # when the process is shutting down. The shutdown process is # subtle and I don't fully understand it; the failure does not # manifest if that lambda isn't there or is a simpler object # like an int (nor does it manifest in the tornado test suite # as a whole, which is why we use this subprocess). proc = subprocess.run( [ sys.executable, "-c", "from tornado.httpclient import HTTPClient; f = lambda: None; c = HTTPClient()", ], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=True, timeout=15, ) if proc.stdout: print("STDOUT:") print(to_unicode(proc.stdout)) if proc.stdout: self.fail("subprocess produced unexpected output") class HTTPRequestTestCase(unittest.TestCase): def test_headers(self): request = HTTPRequest("http://example.com", headers={"foo": "bar"}) self.assertEqual(request.headers, {"foo": "bar"}) def test_headers_setter(self): request = HTTPRequest("http://example.com") request.headers = {"bar": "baz"} # type: ignore self.assertEqual(request.headers, {"bar": "baz"}) def test_null_headers_setter(self): request = HTTPRequest("http://example.com") request.headers = None # type: ignore self.assertEqual(request.headers, {}) def test_body(self): request = HTTPRequest("http://example.com", body="foo") self.assertEqual(request.body, utf8("foo")) def test_body_setter(self): request = HTTPRequest("http://example.com") request.body = "foo" # type: ignore self.assertEqual(request.body, utf8("foo")) def test_if_modified_since(self): http_date = datetime.datetime.now(datetime.timezone.utc) request = HTTPRequest("http://example.com", if_modified_since=http_date) self.assertEqual( request.headers, {"If-Modified-Since": format_timestamp(http_date)} ) def test_if_modified_since_naive_deprecated(self): with ignore_deprecation(): http_date = datetime.datetime.utcnow() request = HTTPRequest("http://example.com", if_modified_since=http_date) self.assertEqual( request.headers, {"If-Modified-Since": format_timestamp(http_date)} ) class HTTPErrorTestCase(unittest.TestCase): def test_copy(self): e = HTTPError(403) e2 = copy.copy(e) self.assertIsNot(e, e2) self.assertEqual(e.code, e2.code) def test_plain_error(self): e = HTTPError(403) self.assertEqual(str(e), "HTTP 403: Forbidden") self.assertEqual(repr(e), "HTTP 403: Forbidden") def test_error_with_response(self): resp = HTTPResponse(HTTPRequest("http://example.com/"), 403) with self.assertRaises(HTTPError) as cm: resp.rethrow() e = cm.exception self.assertEqual(str(e), "HTTP 403: Forbidden") self.assertEqual(repr(e), "HTTP 403: Forbidden") # Limit used by `HTTPClientMaxBodySizeTestCase`. Subclasses configure their # client with it. MAX_BODY_SIZE = 1024 * 64 class MaxBodySizeSmallHandler(RequestHandler): def get(self): self.set_header("Content-Type", "application/octet-stream") self.write(b"a" * MAX_BODY_SIZE) class MaxBodySizeLargeHandler(RequestHandler): def get(self): self.set_header("Content-Type", "application/octet-stream") self.write(b"a" * (MAX_BODY_SIZE + 1)) class BodyBearingRedirectHandler(RequestHandler): """A redirect carrying a body of its own, larger than the limit. Implementations differ here: libcurl discards the body of a redirect it follows, while simple_httpclient applies ``max_body_size`` to every response including the redirect, and so refuses this one before following it. """ def prepare(self): self.write("x" * (MAX_BODY_SIZE * 2)) self.redirect("/small") @abstract_base_test class HTTPClientMaxBodySizeTestCase(AsyncHTTPTestCase): """Tests for the limit on a response body held in memory. Without a ``streaming_callback`` the whole body is buffered, so a response larger than the client is prepared to hold cannot be retrieved at all -- it can only be refused. Subclasses supply a client configured with `MAX_BODY_SIZE`. """ # Format of the message the implementation logs to `.gen_log` when it # refuses a body, or None if it does not log one. ``{reason}`` is filled # in with why the body was refused. refusal_log_format = None # type: typing.Optional[str] def get_app(self): return Application( [ url("/small", MaxBodySizeSmallHandler), url("/large", MaxBodySizeLargeHandler), url("/bomb", GzipBombHandler), url("/body_bearing_redirect", BodyBearingRedirectHandler), ] ) def assert_refused(self, path, reason, **fetch_kwargs): with contextlib.ExitStack() as stack: if self.refusal_log_format is not None: stack.enter_context( ExpectLog( gen_log, self.refusal_log_format.format(reason=reason), level=logging.INFO, ) ) with self.assertRaises(HTTPError) as cm: self.fetch(path, raise_error=True, **fetch_kwargs) self.assertEqual(cm.exception.code, 599) def assert_streaming_refused(self, path, reason): received = 0 def streaming_callback(chunk): nonlocal received received += len(chunk) self.assert_refused(path, reason=reason, streaming_callback=streaming_callback) # Nothing past the limit reached the callback. self.assertLessEqual(received, MAX_BODY_SIZE) def test_small_body(self): # A body of exactly max_body_size is accepted. response = self.fetch("/small") response.rethrow() self.assertEqual(response.body, b"a" * MAX_BODY_SIZE) def test_large_body(self): # One byte more is not. self.assert_refused("/large", reason="Content-Length too long") def test_decompressed_body_size(self): # The limit is measured after decompression, so a compressed # response cannot expand past it. self.assert_refused("/bomb", reason="decompressed body too large") def test_streaming_callback_over_limit(self): # max_body_size applies to a streaming_callback as well as to a # buffered response. It is a safeguard for applications that do not # impose a limit of their own on what they do with the chunks # (writing them to disk, say), so retrieving a response larger than # this means raising the limit deliberately. self.assert_streaming_refused("/large", reason="Content-Length too long") def test_streaming_callback_decompressed_over_limit(self): # As above, measured after decompression. self.assert_streaming_refused("/bomb", reason="decompressed body too large") def test_reuse_after_refusal(self): # Refusing a body must not leave the client unable to make the next # request. This matters most for curl_httpclient, which reuses a # pool of curl handles. self.assert_refused("/large", reason="Content-Length too long") response = self.fetch("/small") response.rethrow() self.assertEqual(response.body, b"a" * MAX_BODY_SIZE)