import collections from urllib import parse import http_strategies from http_base import DaphneTestCase from hypothesis import assume, given, settings from hypothesis.strategies import integers class TestHTTPRequest(DaphneTestCase): """ Tests the HTTP request handling. """ def assert_valid_http_scope( self, scope, method, path, params=None, headers=None, scheme=None ): """ Checks that the passed scope is a valid ASGI HTTP scope regarding types and some urlencoding things. """ # Check overall keys self.assert_key_sets( required_keys={ "asgi", "type", "http_version", "method", "path", "raw_path", "query_string", "headers", }, optional_keys={"scheme", "root_path", "client", "server"}, actual_keys=scope.keys(), ) self.assertEqual(scope["asgi"]["version"], "3.0") # Check that it is the right type self.assertEqual(scope["type"], "http") # Method (uppercased unicode string) self.assertIsInstance(scope["method"], str) self.assertEqual(scope["method"], method.upper()) # Path self.assert_valid_path(scope["path"]) # HTTP version self.assertIn(scope["http_version"], ["1.0", "1.1", "1.2"]) # Scheme self.assertIn(scope["scheme"], ["http", "https"]) if scheme: self.assertEqual(scheme, scope["scheme"]) # Query string (byte string and still url encoded) query_string = scope["query_string"] self.assertIsInstance(query_string, bytes) if params: self.assertEqual( query_string, parse.urlencode(params or []).encode("ascii") ) # Ordering of header names is not important, but the order of values for a header # name is. To assert whether that order is kept, we transform both the request # headers and the channel message headers into a dictionary # {name: [value1, value2, ...]} and check if they're equal. transformed_scope_headers = collections.defaultdict(list) for name, value in scope["headers"]: transformed_scope_headers[name].append(value) transformed_request_headers = collections.defaultdict(list) for name, value in headers or []: expected_name = name.lower().strip() expected_value = value.strip() transformed_request_headers[expected_name].append(expected_value) for name, value in transformed_request_headers.items(): self.assertIn(name, transformed_scope_headers) self.assertEqual(value, transformed_scope_headers[name]) # Root path self.assertIsInstance(scope.get("root_path", ""), str) # Client and server addresses client = scope.get("client") if client is not None: self.assert_valid_address_and_port(client) server = scope.get("server") if server is not None: self.assert_valid_address_and_port(server) def assert_valid_http_request_message(self, message, body=None): """ Asserts that a message is a valid http.request message """ # Check overall keys self.assert_key_sets( required_keys={"type"}, optional_keys={"body", "more_body"}, actual_keys=message.keys(), ) # Check that it is the right type self.assertEqual(message["type"], "http.request") # If there's a body present, check its type self.assertIsInstance(message.get("body", b""), bytes) if body is not None: self.assertEqual(body, message.get("body", b"")) def test_minimal_request(self): """ Smallest viable example. Mostly verifies that our request building works. """ scope, messages = self.run_daphne_request("GET", "/") self.assert_valid_http_scope(scope, "GET", "/") self.assert_valid_http_request_message(messages[0], body=b"") @given( request_path=http_strategies.http_path(), request_params=http_strategies.query_params(), ) @settings(max_examples=5, deadline=5000) def test_get_request(self, request_path, request_params): """ Tests a typical HTTP GET request, with a path and query parameters """ scope, messages = self.run_daphne_request( "GET", request_path, params=request_params ) self.assert_valid_http_scope(scope, "GET", request_path, params=request_params) self.assert_valid_http_request_message(messages[0], body=b"") @given(request_path=http_strategies.http_path(), chunk_size=integers(min_value=1)) @settings(max_examples=5, deadline=5000) def test_request_body_chunking(self, request_path, chunk_size): """ Tests request body chunking logic. """ body = b"The quick brown fox jumps over the lazy dog" _, messages = self.run_daphne_request( "POST", request_path, body=body, request_buffer_size=chunk_size, ) # Avoid running those asserts when there's a single "http.disconnect" if len(messages) > 1: assert messages[0]["body"].decode() == body.decode()[:chunk_size] assert not messages[-2]["more_body"] assert messages[-1] == {"type": "http.disconnect"} @given( request_path=http_strategies.http_path(), request_body=http_strategies.http_body(), ) @settings(max_examples=5, deadline=5000) def test_post_request(self, request_path, request_body): """ Tests a typical HTTP POST request, with a path and body. """ scope, messages = self.run_daphne_request( "POST", request_path, body=request_body ) self.assert_valid_http_scope(scope, "POST", request_path) self.assert_valid_http_request_message(messages[0], body=request_body) def test_raw_path(self): """ Tests that /foo%2Fbar produces raw_path and a decoded path """ scope, _ = self.run_daphne_request("GET", "/foo%2Fbar") self.assertEqual(scope["path"], "/foo/bar") self.assertEqual(scope["raw_path"], b"/foo%2Fbar") @given(request_headers=http_strategies.headers()) @settings(max_examples=5, deadline=5000) def test_headers(self, request_headers): """ Tests that HTTP header fields are handled as specified """ request_path = parse.quote("/te st-à/") scope, messages = self.run_daphne_request( "OPTIONS", request_path, headers=request_headers ) self.assert_valid_http_scope( scope, "OPTIONS", request_path, headers=request_headers ) self.assert_valid_http_request_message(messages[0], body=b"") @given(request_headers=http_strategies.headers()) @settings(max_examples=5, deadline=5000) def test_duplicate_headers(self, request_headers): """ Tests that duplicate header values are preserved """ # Make sure there's duplicate headers assume(len(request_headers) >= 2) header_name = request_headers[0][0] duplicated_headers = [(header_name, header[1]) for header in request_headers] # Run the request request_path = parse.quote("/te st-à/") scope, messages = self.run_daphne_request( "OPTIONS", request_path, headers=duplicated_headers ) self.assert_valid_http_scope( scope, "OPTIONS", request_path, headers=duplicated_headers ) self.assert_valid_http_request_message(messages[0], body=b"") @given( request_method=http_strategies.http_method(), request_path=http_strategies.http_path(), request_params=http_strategies.query_params(), request_headers=http_strategies.headers(), request_body=http_strategies.http_body(), ) @settings(max_examples=2, deadline=5000) def test_kitchen_sink( self, request_method, request_path, request_params, request_headers, request_body, ): """ Throw everything at Daphne that we dare. The idea is that if a combination of method/path/headers/body would break the spec, hypothesis will eventually find it. """ scope, messages = self.run_daphne_request( request_method, request_path, params=request_params, headers=request_headers, body=request_body, ) self.assert_valid_http_scope( scope, request_method, request_path, params=request_params, headers=request_headers, ) self.assert_valid_http_request_message(messages[0], body=request_body) def test_headers_are_lowercased_and_stripped(self): """ Make sure headers are normalized as the spec says they are. """ headers = [(b"MYCUSTOMHEADER", b" foobar ")] scope, messages = self.run_daphne_request("GET", "/", headers=headers) self.assert_valid_http_scope(scope, "GET", "/", headers=headers) self.assert_valid_http_request_message(messages[0], body=b"") # Note that Daphne returns a list of tuples here, which is fine, because the spec # asks to treat them interchangeably. assert [list(x) for x in scope["headers"]] == [[b"mycustomheader", b"foobar"]] @given(daphne_path=http_strategies.http_path()) @settings(max_examples=5, deadline=5000) def test_root_path_header(self, daphne_path): """ Tests root_path handling. """ # Daphne-Root-Path must be URL encoded when submitting as HTTP header field headers = [("Daphne-Root-Path", parse.quote(daphne_path.encode("utf8")))] scope, messages = self.run_daphne_request("GET", "/", headers=headers) # Daphne-Root-Path is not included in the returned 'headers' section. So we expect # empty headers. self.assert_valid_http_scope(scope, "GET", "/", headers=[]) self.assert_valid_http_request_message(messages[0], body=b"") # And what we're looking for, root_path being set. assert scope["root_path"] == daphne_path def test_x_forwarded_for_ignored(self): """ Make sure that, by default, X-Forwarded-For is ignored. """ headers = [[b"X-Forwarded-For", b"10.1.2.3"], [b"X-Forwarded-Port", b"80"]] scope, messages = self.run_daphne_request("GET", "/", headers=headers) self.assert_valid_http_scope(scope, "GET", "/", headers=headers) self.assert_valid_http_request_message(messages[0], body=b"") # It should NOT appear in the client scope item self.assertNotEqual(scope["client"], ["10.1.2.3", 80]) def test_x_forwarded_for_parsed(self): """ When X-Forwarded-For is enabled, make sure it is respected. """ headers = [[b"X-Forwarded-For", b"10.1.2.3"], [b"X-Forwarded-Port", b"80"]] scope, messages = self.run_daphne_request("GET", "/", headers=headers, xff=True) self.assert_valid_http_scope(scope, "GET", "/", headers=headers) self.assert_valid_http_request_message(messages[0], body=b"") # It should now appear in the client scope item self.assertEqual(scope["client"], ["10.1.2.3", 80]) def test_x_forwarded_for_no_port(self): """ When X-Forwarded-For is enabled but only the host is passed, make sure that at least makes it through. """ headers = [[b"X-Forwarded-For", b"10.1.2.3"]] scope, messages = self.run_daphne_request("GET", "/", headers=headers, xff=True) self.assert_valid_http_scope(scope, "GET", "/", headers=headers) self.assert_valid_http_request_message(messages[0], body=b"") # It should now appear in the client scope item self.assertEqual(scope["client"], ["10.1.2.3", 0]) def test_bad_requests(self): """ Tests that requests with invalid (non-ASCII) characters fail. """ # Bad path response = self.run_daphne_raw( b"GET /\xc3\xa4\xc3\xb6\xc3\xbc HTTP/1.0\r\n\r\n" ) self.assertTrue(b"400 Bad Request" in response) # Bad querystring response = self.run_daphne_raw( b"GET /?\xc3\xa4\xc3\xb6\xc3\xbc HTTP/1.0\r\n\r\n" ) self.assertTrue(b"400 Bad Request" in response) def test_invalid_header_name(self): """ Tests that requests with invalid header names fail. """ # Test cases follow those used by h11 # https://github.com/python-hyper/h11/blob/a2c68948accadc3876dffcf979d98002e4a4ed27/h11/tests/test_headers.py#L24-L35 for header_name in [b"foo bar", b"foo\x00bar", b"foo\xffbar", b"foo\x01bar"]: response = self.run_daphne_raw( f"GET / HTTP/1.0\r\n{header_name}: baz\r\n\r\n".encode("ascii") ) self.assertTrue(b"400 Bad Request" in response)