| ############################################################################## |
| # |
| # Copyright (c) 2002 Zope Foundation and Contributors. |
| # All Rights Reserved. |
| # |
| # This software is subject to the provisions of the Zope Public License, |
| # Version 2.1 (ZPL). A copy of the ZPL should accompany this distribution. |
| # THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED |
| # WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED |
| # WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS |
| # FOR A PARTICULAR PURPOSE. |
| # |
| ############################################################################## |
| """HTTP Request Parser tests""" |
| |
| import unittest |
| |
| from waitress.adjustments import Adjustments |
| from waitress.parser import ( |
| HTTPRequestParser, |
| ParsingError, |
| TransferEncodingNotImplemented, |
| crack_first_line, |
| get_header_lines, |
| split_uri, |
| unquote_bytes_to_wsgi, |
| ) |
| from waitress.utilities import ( |
| BadRequest, |
| RequestEntityTooLarge, |
| RequestHeaderFieldsTooLarge, |
| ServerNotImplemented, |
| ) |
| |
| |
| class TestHTTPRequestParser(unittest.TestCase): |
| def setUp(self): |
| my_adj = Adjustments() |
| self.parser = HTTPRequestParser(my_adj) |
| |
| def test_get_body_stream_None(self): |
| self.parser.body_recv = None |
| result = self.parser.get_body_stream() |
| self.assertEqual(result.getvalue(), b"") |
| |
| def test_get_body_stream_nonNone(self): |
| body_rcv = DummyBodyStream() |
| self.parser.body_rcv = body_rcv |
| result = self.parser.get_body_stream() |
| self.assertEqual(result, body_rcv) |
| |
| def test_received_get_no_headers(self): |
| data = b"HTTP/1.0 GET /foobar\r\n\r\n" |
| result = self.parser.received(data) |
| self.assertEqual(result, 24) |
| self.assertTrue(self.parser.completed) |
| self.assertDictEqual(self.parser.headers, {}) |
| |
| def test_received_bad_host_header(self): |
| data = b"HTTP/1.0 GET /foobar\r\n Host: foo\r\n\r\n" |
| result = self.parser.received(data) |
| self.assertEqual(result, 36) |
| self.assertTrue(self.parser.completed) |
| self.assertIsInstance(self.parser.error, BadRequest) |
| |
| def test_received_duplicate_host_header(self): |
| # RFC 9112: MUST reject HTTP/1.1 requests with more than one Host header |
| data = b"GET / HTTP/1.1\r\nHOST: test1.com\r\nHost: test2.com\r\n\r\n" |
| result = self.parser.received(data) |
| self.assertEqual(result, len(data)) |
| self.assertTrue(self.parser.completed) |
| self.assertIsInstance(self.parser.error, BadRequest) |
| self.assertTrue(self.parser.error.body.startswith("Duplicate header:")) |
| |
| def test_received_duplicate_content_length_header(self): |
| # RFC 7230: MUST reject requests with duplicate Content-Length headers |
| data = b"GET / HTTP/1.1\r\nHost: example.com\r\nContent-Length: 10\r\nContent-Length: 20\r\n\r\n" |
| result = self.parser.received(data) |
| self.assertEqual(result, len(data)) |
| self.assertTrue(self.parser.completed) |
| self.assertIsInstance(self.parser.error, BadRequest) |
| self.assertTrue(self.parser.error.body.startswith("Duplicate header:")) |
| |
| def test_received_duplicate_content_type_header(self): |
| data = b"GET / HTTP/1.1\r\nHost: example.com\r\nContent-Type: text/plain\r\nContent-Type: text/html\r\n\r\n" |
| result = self.parser.received(data) |
| self.assertEqual(result, len(data)) |
| self.assertTrue(self.parser.completed) |
| self.assertIsInstance(self.parser.error, BadRequest) |
| self.assertTrue(self.parser.error.body.startswith("Duplicate header:")) |
| |
| def test_received_missing_host_header_11(self): |
| # RFC 9112 section 3.2: MUST reject an HTTP/1.1 request that lacks a |
| # Host header field |
| data = b"POST / HTTP/1.1\r\nContent-Length: 0\r\n\r\n" |
| result = self.parser.received(data) |
| self.assertEqual(result, len(data)) |
| self.assertTrue(self.parser.completed) |
| self.assertIsInstance(self.parser.error, BadRequest) |
| self.assertIn("does not contain a Host header", self.parser.error.body) |
| |
| def test_received_missing_host_header_10(self): |
| # the Host header is only required from HTTP/1.1 onwards |
| data = b"POST / HTTP/1.0\r\nContent-Length: 0\r\n\r\n" |
| result = self.parser.received(data) |
| self.assertEqual(result, len(data)) |
| self.assertTrue(self.parser.completed) |
| self.assertIsNone(self.parser.error) |
| |
| def test_received_valid_host_headers(self): |
| for host in ( |
| b"example.com", |
| b"example.com:8080", |
| b"example.com.", |
| b"under_score.example.com", |
| b"127.0.0.1:80", |
| b"[::1]", |
| b"[::1]:8080", |
| b"[v7.host:in-the:future]", |
| b"xn--n3h.example", |
| b"", |
| ): |
| with self.subTest(host=host): |
| parser = HTTPRequestParser(Adjustments()) |
| data = b"GET / HTTP/1.1\r\nHost: " + host + b"\r\n\r\n" |
| parser.received(data) |
| self.assertIsNone(parser.error) |
| self.assertEqual(parser.headers["HOST"], host.decode("latin-1")) |
| |
| def test_received_invalid_host_headers(self): |
| # RFC 9112 section 3.2: MUST reject a Host header field with an invalid |
| # field value |
| for host in ( |
| b"exa mple.com", # whitespace is not allowed in a uri-host |
| b"example.com:80x", # the port must be numeric |
| b"example.com:80:80", |
| b"user@example.com", # a userinfo subcomponent is not an uri-host |
| b"example.com/path", |
| b"example.com?query", |
| b"[::1", # unterminated IP-literal |
| b"exampl\xc3\xa9.com", # IDNs need to be punycoded by the client |
| ): |
| with self.subTest(host=host): |
| parser = HTTPRequestParser(Adjustments()) |
| data = b"GET / HTTP/1.1\r\nHost: " + host + b"\r\n\r\n" |
| parser.received(data) |
| self.assertTrue(parser.completed) |
| self.assertIsInstance(parser.error, BadRequest) |
| self.assertIn("Invalid Host header", parser.error.body) |
| |
| def test_received_absolute_form_overrides_host_header(self): |
| # RFC 9112 section 3.2.2: an origin server MUST ignore the Host header |
| # field of a request with an absolute-form request-target and use the |
| # host information from the request-target instead |
| data = b"GET http://evil.com/page HTTP/1.1\r\nHost: victim.com\r\n\r\n" |
| result = self.parser.received(data) |
| self.assertEqual(result, len(data)) |
| self.assertTrue(self.parser.completed) |
| self.assertIsNone(self.parser.error) |
| self.assertEqual(self.parser.headers["HOST"], "evil.com") |
| self.assertEqual(self.parser.path, "/page") |
| |
| def test_received_absolute_form_without_host_header(self): |
| data = b"GET http://evil.com:8080/page HTTP/1.1\r\n\r\n" |
| self.parser.received(data) |
| self.assertTrue(self.parser.completed) |
| self.assertIsNone(self.parser.error) |
| self.assertEqual(self.parser.headers["HOST"], "evil.com:8080") |
| |
| def test_received_absolute_form_invalid_authority(self): |
| for uri in ( |
| b"http:///page", # RFC 9110 4.2.1: MUST reject an empty host |
| b"http://user@evil.com/page", # userinfo is not allowed |
| b"http://evil.com:80x/page", |
| ): |
| with self.subTest(uri=uri): |
| parser = HTTPRequestParser(Adjustments()) |
| data = b"GET " + uri + b" HTTP/1.1\r\nHost: victim.com\r\n\r\n" |
| parser.received(data) |
| self.assertTrue(parser.completed) |
| self.assertIsInstance(parser.error, BadRequest) |
| |
| def test_received_network_path_reference_is_a_path(self): |
| # a request-target that starts with "//" has no scheme, so it stays a |
| # path and the Host header is left alone. See |
| # https://github.com/Pylons/waitress/issues/260 |
| data = b"GET //testing/whatever HTTP/1.1\r\nHost: victim.com\r\n\r\n" |
| self.parser.received(data) |
| self.assertIsNone(self.parser.error) |
| self.assertEqual(self.parser.headers["HOST"], "victim.com") |
| self.assertEqual(self.parser.path, "//testing/whatever") |
| |
| def test_received_asterisk_form_options(self): |
| # RFC 9112 section 3.2.4: the asterisk-form is used for a server-wide |
| # OPTIONS request |
| data = b"OPTIONS * HTTP/1.1\r\nHost: localhost\r\n\r\n" |
| self.parser.received(data) |
| self.assertTrue(self.parser.completed) |
| self.assertIsNone(self.parser.error) |
| self.assertEqual(self.parser.command, "OPTIONS") |
| self.assertEqual(self.parser.path, "*") |
| |
| def test_received_asterisk_form_rejected_for_other_methods(self): |
| # the asterisk-form is only a request-target for OPTIONS; for anything |
| # else it names nothing that could be routed |
| for command in (b"GET", b"POST", b"HEAD", b"DELETE"): |
| with self.subTest(command=command): |
| parser = HTTPRequestParser(Adjustments()) |
| data = command + b" * HTTP/1.1\r\nHost: localhost\r\n\r\n" |
| parser.received(data) |
| self.assertTrue(parser.completed) |
| self.assertIsInstance(parser.error, BadRequest) |
| |
| def test_received_origin_form_must_be_an_absolute_path(self): |
| # RFC 9112 section 3.2.1: an origin-form is an absolute-path, which |
| # begins with a "/". PEP 3333 wants the same of PATH_INFO. |
| for uri in ( |
| b"foo/bar", |
| b"1.2.3.4:80/x", |
| b"%2Ffoo", # a percent encoded "/" does not make an absolute-path |
| b".", |
| ): |
| with self.subTest(uri=uri): |
| parser = HTTPRequestParser(Adjustments()) |
| data = b"GET " + uri + b" HTTP/1.1\r\nHost: localhost\r\n\r\n" |
| parser.received(data) |
| self.assertTrue(parser.completed) |
| self.assertIsInstance(parser.error, BadRequest) |
| |
| def test_received_origin_form_accepted(self): |
| for uri, path in ( |
| (b"/", "/"), |
| (b"/foo", "/foo"), |
| (b"/foo?a=b", "/foo"), |
| (b"/foo%2Fbar", "/foo/bar"), |
| ): |
| with self.subTest(uri=uri): |
| parser = HTTPRequestParser(Adjustments()) |
| data = b"GET " + uri + b" HTTP/1.1\r\nHost: localhost\r\n\r\n" |
| parser.received(data) |
| self.assertIsNone(parser.error) |
| self.assertEqual(parser.path, path) |
| |
| def test_received_connect_is_not_an_absolute_form(self): |
| # a CONNECT uses the authority-form of request-target, which urlsplit() |
| # reads as a scheme followed by a path. Waitress passes a CONNECT |
| # through for the WSGI application to deal with, so the request-target |
| # must not be mistaken for an absolute-form and rejected. |
| for uri, path in ( |
| (b"example.com:443", "443"), |
| (b"[::1]:443", "[::1]:443"), |
| ): |
| with self.subTest(uri=uri): |
| parser = HTTPRequestParser(Adjustments()) |
| data = b"CONNECT " + uri + b" HTTP/1.1\r\nHost: example.com\r\n\r\n" |
| parser.received(data) |
| self.assertIsNone(parser.error) |
| self.assertEqual(parser.command, "CONNECT") |
| self.assertEqual(parser.request_uri, uri.decode("latin-1")) |
| self.assertEqual(parser.path, path) |
| # the Host header field is left exactly as the client sent it |
| self.assertEqual(parser.headers["HOST"], "example.com") |
| |
| def test_received_connect_invalid_authority_form(self): |
| # RFC 9112 section 3.2.3: MUST reject a CONNECT request that targets an |
| # empty or invalid port number |
| for uri in ( |
| b"example.com", # no port at all |
| b"example.com:", # empty port |
| b"example.com:0", # port zero is not a valid port number |
| b"example.com:65536", # out of range |
| b"example.com:999999", |
| b"example.com:https", # the port must be numeric |
| b":443", # no host |
| b"http://example.com:443", |
| b"/", |
| b"*", |
| ): |
| with self.subTest(uri=uri): |
| parser = HTTPRequestParser(Adjustments()) |
| data = b"CONNECT " + uri + b" HTTP/1.1\r\nHost: example.com\r\n\r\n" |
| parser.received(data) |
| self.assertTrue(parser.completed) |
| self.assertIsInstance(parser.error, BadRequest) |
| self.assertIn("Request-target", parser.error.body) |
| |
| def test_received_bad_transfer_encoding(self): |
| data = ( |
| b"GET /foobar HTTP/1.1\r\n" |
| b"Host: example.com\r\n" |
| b"Transfer-Encoding: foo\r\n" |
| b"\r\n" |
| b"1d;\r\n" |
| b"This string has 29 characters\r\n" |
| b"0\r\n\r\n" |
| ) |
| result = self.parser.received(data) |
| self.assertEqual(result, 67) |
| self.assertTrue(self.parser.completed) |
| self.assertIsInstance(self.parser.error, ServerNotImplemented) |
| |
| def test_received_nonsense_nothing(self): |
| data = b"\r\n\r\n" |
| result = self.parser.received(data) |
| self.assertEqual(result, 4) |
| self.assertTrue(self.parser.completed) |
| self.assertDictEqual(self.parser.headers, {}) |
| |
| def test_received_no_doublecr(self): |
| data = b"GET /foobar HTTP/8.4\r\n" |
| result = self.parser.received(data) |
| self.assertEqual(result, 22) |
| self.assertFalse(self.parser.completed) |
| self.assertDictEqual(self.parser.headers, {}) |
| |
| def test_received_already_completed(self): |
| self.parser.completed = True |
| result = self.parser.received(b"a") |
| self.assertEqual(result, 0) |
| |
| def test_received_cl_too_large(self): |
| self.parser.adj.max_request_body_size = 2 |
| data = b"GET /foobar HTTP/8.4\r\nContent-Length: 10\r\n\r\n" |
| result = self.parser.received(data) |
| self.assertEqual(result, 44) |
| self.assertTrue(self.parser.completed) |
| self.assertIsInstance(self.parser.error, RequestEntityTooLarge) |
| |
| def test_received_headers_not_too_large_multiple_chunks(self): |
| data = b"GET /foobar HTTP/8.4\r\nX-Foo: 1\r\n" |
| data2 = b"X-Foo-Other: 3\r\n\r\n" |
| self.parser.adj.max_request_header_size = len(data) + len(data2) + 1 |
| result = self.parser.received(data) |
| self.assertEqual(result, 32) |
| result = self.parser.received(data2) |
| self.assertEqual(result, 18) |
| self.assertTrue(self.parser.completed) |
| self.assertFalse(self.parser.error) |
| |
| def test_received_headers_too_large(self): |
| self.parser.adj.max_request_header_size = 2 |
| data = b"GET /foobar HTTP/8.4\r\nX-Foo: 1\r\n\r\n" |
| result = self.parser.received(data) |
| self.assertEqual(result, 34) |
| self.assertTrue(self.parser.completed) |
| self.assertIsInstance(self.parser.error, RequestHeaderFieldsTooLarge) |
| |
| def test_received_body_too_large(self): |
| self.parser.adj.max_request_body_size = 2 |
| data = ( |
| b"GET /foobar HTTP/1.1\r\n" |
| b"Host: example.com\r\n" |
| b"Transfer-Encoding: chunked\r\n" |
| b"X-Foo: 1\r\n" |
| b"\r\n" |
| b"1d;\r\n" |
| b"This string has 29 characters\r\n" |
| b"0\r\n\r\n" |
| ) |
| |
| result = self.parser.received(data) |
| self.assertEqual(result, 81) |
| self.parser.received(data[result:]) |
| self.assertTrue(self.parser.completed) |
| self.assertIsInstance(self.parser.error, RequestEntityTooLarge) |
| |
| def test_received_error_from_parser(self): |
| data = ( |
| b"GET /foobar HTTP/1.1\r\n" |
| b"Host: example.com\r\n" |
| b"Transfer-Encoding: chunked\r\n" |
| b"X-Foo: 1\r\n" |
| b"\r\n" |
| b"garbage\r\n" |
| ) |
| # header |
| result = self.parser.received(data) |
| # body |
| result = self.parser.received(data[result:]) |
| self.assertEqual(result, 9) |
| self.assertTrue(self.parser.completed) |
| self.assertIsInstance(self.parser.error, BadRequest) |
| |
| def test_received_chunked_completed_sets_content_length(self): |
| data = ( |
| b"GET /foobar HTTP/1.1\r\n" |
| b"Host: example.com\r\n" |
| b"Transfer-Encoding: chunked\r\n" |
| b"X-Foo: 1\r\n" |
| b"\r\n" |
| b"1d\r\n" |
| b"This string has 29 characters\r\n" |
| b"0\r\n\r\n" |
| ) |
| result = self.parser.received(data) |
| self.assertEqual(result, 81) |
| data = data[result:] |
| result = self.parser.received(data) |
| self.assertTrue(self.parser.completed) |
| self.assertIsNone(self.parser.error) |
| self.assertEqual(self.parser.headers["CONTENT_LENGTH"], "29") |
| |
| def test_parse_header_gardenpath(self): |
| data = b"GET /foobar HTTP/8.4\r\nfoo: bar\r\n" |
| self.parser.parse_header(data) |
| self.assertEqual(self.parser.first_line, b"GET /foobar HTTP/8.4") |
| self.assertEqual(self.parser.headers["FOO"], "bar") |
| |
| def test_parse_header_no_cr_in_headerplus(self): |
| data = b"GET /foobar HTTP/8.4" |
| |
| try: |
| self.parser.parse_header(data) |
| except ParsingError: |
| pass |
| else: # pragma: nocover |
| self.assertTrue(False) |
| |
| def test_parse_header_bad_content_length(self): |
| data = b"GET /foobar HTTP/8.4\r\ncontent-length: abc\r\n" |
| |
| try: |
| self.parser.parse_header(data) |
| except ParsingError as e: |
| self.assertIn("Content-Length is invalid", e.args[0]) |
| else: # pragma: nocover |
| self.assertTrue(False) |
| |
| def test_parse_header_bad_content_length_plus(self): |
| data = b"GET /foobar HTTP/8.4\r\ncontent-length: +10\r\n" |
| |
| try: |
| self.parser.parse_header(data) |
| except ParsingError as e: |
| self.assertIn("Content-Length is invalid", e.args[0]) |
| else: # pragma: nocover |
| self.assertTrue(False) |
| |
| def test_parse_header_bad_content_length_minus(self): |
| data = b"GET /foobar HTTP/8.4\r\ncontent-length: -10\r\n" |
| |
| try: |
| self.parser.parse_header(data) |
| except ParsingError as e: |
| self.assertIn("Content-Length is invalid", e.args[0]) |
| else: # pragma: nocover |
| self.assertTrue(False) |
| |
| def test_parse_header_multiple_content_length(self): |
| data = b"GET /foobar HTTP/8.4\r\ncontent-length: 10\r\ncontent-length: 20\r\n" |
| |
| try: |
| self.parser.parse_header(data) |
| except ParsingError as e: |
| self.assertTrue(e.args[0].startswith("Duplicate header:")) |
| else: # pragma: nocover |
| self.assertTrue(False) |
| |
| def test_parse_header_11_te_chunked(self): |
| # NB: test that capitalization of header value is unimportant |
| data = b"GET /foobar HTTP/1.1\r\nHost: example.com\r\ntransfer-encoding: ChUnKed\r\n" |
| self.parser.parse_header(data) |
| self.assertEqual(self.parser.body_rcv.__class__.__name__, "ChunkedReceiver") |
| |
| def test_parse_header_11_te_chunked_with_cl_close_connection(self): |
| # NB: test that capitalization of header value is unimportant |
| data = b"GET /foobar HTTP/1.1\r\nHost: example.com\r\ntransfer-encoding: chunked\r\ncontent-length: 10\r\n" |
| self.parser.parse_header(data) |
| self.assertEqual(self.parser.body_rcv.__class__.__name__, "ChunkedReceiver") |
| self.assertEqual(self.parser.connection_close, True) |
| |
| def test_parse_header_transfer_encoding_invalid(self): |
| data = ( |
| b"GET /foobar HTTP/1.1\r\nHost: example.com\r\ntransfer-encoding: gzip\r\n" |
| ) |
| |
| try: |
| self.parser.parse_header(data) |
| except TransferEncodingNotImplemented as e: |
| self.assertIn("Transfer-Encoding requested is not supported.", e.args[0]) |
| else: # pragma: nocover |
| self.assertTrue(False) |
| |
| def test_parse_header_transfer_encoding_invalid_multiple(self): |
| data = b"GET /foobar HTTP/1.1\r\nHost: example.com\r\ntransfer-encoding: gzip\r\ntransfer-encoding: chunked\r\n" |
| |
| try: |
| self.parser.parse_header(data) |
| except TransferEncodingNotImplemented as e: |
| self.assertIn("Transfer-Encoding requested is not supported.", e.args[0]) |
| else: # pragma: nocover |
| self.assertTrue(False) |
| |
| def test_parse_header_transfer_encoding_invalid_multiple_chunked(self): |
| data = b"GET /foobar HTTP/1.1\r\nHost: example.com\r\ntransfer-encoding: chunked\r\ntransfer-encoding: chunked\r\n" |
| |
| try: |
| self.parser.parse_header(data) |
| except TransferEncodingNotImplemented as e: |
| self.assertIn( |
| "Transfer-Encoding is invalid. Multiple chunked encodings requested.", |
| e.args[0], |
| ) |
| else: # pragma: nocover |
| self.assertTrue(False) |
| |
| def test_parse_header_transfer_encoding_invalid_whitespace(self): |
| data = b"GET /foobar HTTP/1.1\r\nHost: example.com\r\nTransfer-Encoding:\x85chunked\r\n" |
| |
| try: |
| self.parser.parse_header(data) |
| except TransferEncodingNotImplemented as e: |
| self.assertIn("Transfer-Encoding requested is not supported.", e.args[0]) |
| else: # pragma: nocover |
| self.assertTrue(False) |
| |
| def test_parse_header_transfer_encoding_invalid_unicode(self): |
| # This is the binary encoding for the UTF-8 character |
| # https://www.compart.com/en/unicode/U+212A "unicode character "K"" |
| # which if waitress were to accidentally do the wrong thing get |
| # lowercased to just the ascii "k" due to unicode collisions during |
| # transformation |
| data = b"GET /foobar HTTP/1.1\r\nHost: example.com\r\nTransfer-Encoding: chun\xe2\x84\xaaed\r\n" |
| |
| try: |
| self.parser.parse_header(data) |
| except TransferEncodingNotImplemented as e: |
| self.assertIn("Transfer-Encoding requested is not supported.", e.args[0]) |
| else: # pragma: nocover |
| self.assertTrue(False) |
| |
| def test_parse_header_11_expect_continue(self): |
| data = b"GET /foobar HTTP/1.1\r\nHost: example.com\r\nexpect: 100-continue\r\n" |
| self.parser.parse_header(data) |
| self.assertTrue(self.parser.expect_continue) |
| |
| def test_parse_header_connection_close(self): |
| data = b"GET /foobar HTTP/1.1\r\nHost: example.com\r\nConnection: close\r\n" |
| self.parser.parse_header(data) |
| self.assertTrue(self.parser.connection_close) |
| |
| def test_close_with_body_rcv(self): |
| body_rcv = DummyBodyStream() |
| self.parser.body_rcv = body_rcv |
| self.parser.close() |
| self.assertTrue(body_rcv.closed) |
| |
| def test_close_with_no_body_rcv(self): |
| self.parser.body_rcv = None |
| self.parser.close() # doesn't raise |
| |
| def test_parse_header_lf_only(self): |
| data = b"GET /foobar HTTP/8.4\nfoo: bar" |
| |
| try: |
| self.parser.parse_header(data) |
| except ParsingError: |
| pass |
| else: # pragma: nocover |
| self.assertTrue(False) |
| |
| def test_parse_header_cr_only(self): |
| data = b"GET /foobar HTTP/8.4\rfoo: bar" |
| try: |
| self.parser.parse_header(data) |
| except ParsingError: |
| pass |
| else: # pragma: nocover |
| self.assertTrue(False) |
| |
| def test_parse_header_extra_lf_in_header(self): |
| data = b"GET /foobar HTTP/8.4\r\nfoo: \nbar\r\n" |
| try: |
| self.parser.parse_header(data) |
| except ParsingError as e: |
| self.assertIn("Bare CR or LF found in header line", e.args[0]) |
| else: # pragma: nocover |
| self.assertTrue(False) |
| |
| def test_parse_header_extra_lf_in_first_line(self): |
| data = b"GET /foobar\n HTTP/8.4\r\n" |
| try: |
| self.parser.parse_header(data) |
| except ParsingError as e: |
| self.assertIn("Bare CR or LF found in HTTP message", e.args[0]) |
| else: # pragma: nocover |
| self.assertTrue(False) |
| |
| def test_parse_header_invalid_whitespace(self): |
| data = b"GET /foobar HTTP/8.4\r\nfoo : bar\r\n" |
| try: |
| self.parser.parse_header(data) |
| except ParsingError as e: |
| self.assertIn("Invalid header", e.args[0]) |
| else: # pragma: nocover |
| self.assertTrue(False) |
| |
| def test_parse_header_invalid_whitespace_vtab(self): |
| data = b"GET /foobar HTTP/1.1\r\nHost: example.com\r\nfoo:\x0bbar\r\n" |
| try: |
| self.parser.parse_header(data) |
| except ParsingError as e: |
| self.assertIn("Invalid header", e.args[0]) |
| else: # pragma: nocover |
| self.assertTrue(False) |
| |
| def test_parse_header_invalid_no_colon(self): |
| data = b"GET /foobar HTTP/1.1\r\nHost: example.com\r\nfoo: bar\r\nnotvalid\r\n" |
| try: |
| self.parser.parse_header(data) |
| except ParsingError as e: |
| self.assertIn("Invalid header", e.args[0]) |
| else: # pragma: nocover |
| self.assertTrue(False) |
| |
| def test_parse_header_invalid_folding_spacing(self): |
| data = b"GET /foobar HTTP/1.1\r\nHost: example.com\r\nfoo: bar\r\n\t\x0bbaz\r\n" |
| try: |
| self.parser.parse_header(data) |
| except ParsingError as e: |
| self.assertIn("Invalid header", e.args[0]) |
| else: # pragma: nocover |
| self.assertTrue(False) |
| |
| def test_parse_header_invalid_chars(self): |
| data = ( |
| b"GET /foobar HTTP/1.1\r\nHost: example.com\r\nfoo: bar\r\nfoo: \x0bbaz\r\n" |
| ) |
| try: |
| self.parser.parse_header(data) |
| except ParsingError as e: |
| self.assertIn("Invalid header", e.args[0]) |
| else: # pragma: nocover |
| self.assertTrue(False) |
| |
| def test_parse_header_other_whitespace(self): |
| data = ( |
| b"GET /foobar HTTP/1.1\r\nHost: example.com\r\nfoo: \xa0something\x85\r\n" |
| ) |
| self.parser.parse_header(data) |
| self.assertEqual(self.parser.headers["FOO"], "\xa0something\x85") |
| |
| def test_parse_header_empty(self): |
| data = b"GET /foobar HTTP/1.1\r\nHost: example.com\r\nfoo: bar\r\nempty:\r\n" |
| self.parser.parse_header(data) |
| |
| self.assertIn("EMPTY", self.parser.headers) |
| self.assertIn("FOO", self.parser.headers) |
| self.assertEqual(self.parser.headers["EMPTY"], "") |
| self.assertEqual(self.parser.headers["FOO"], "bar") |
| |
| def test_parse_header_multiple_values(self): |
| data = b"GET /foobar HTTP/1.1\r\nHost: example.com\r\nfoo: bar, whatever, more, please, yes\r\n" |
| self.parser.parse_header(data) |
| |
| self.assertIn("FOO", self.parser.headers) |
| self.assertEqual(self.parser.headers["FOO"], "bar, whatever, more, please, yes") |
| |
| def test_parse_header_multiple_values_header_folded(self): |
| data = b"GET /foobar HTTP/1.1\r\nHost: example.com\r\nfoo: bar, whatever,\r\n more, please, yes\r\n" |
| self.parser.parse_header(data) |
| |
| self.assertIn("FOO", self.parser.headers) |
| self.assertEqual(self.parser.headers["FOO"], "bar, whatever, more, please, yes") |
| |
| def test_parse_header_multiple_values_header_folded_multiple(self): |
| data = b"GET /foobar HTTP/1.1\r\nHost: example.com\r\nfoo: bar, whatever,\r\n more\r\nfoo: please, yes\r\n" |
| self.parser.parse_header(data) |
| |
| self.assertIn("FOO", self.parser.headers) |
| self.assertEqual(self.parser.headers["FOO"], "bar, whatever, more, please, yes") |
| |
| def test_parse_header_multiple_values_extra_space(self): |
| # Tests errata from: https://www.rfc-editor.org/errata_search.php?rfc=7230&eid=4189 |
| data = b"GET /foobar HTTP/1.1\r\nHost: example.com\r\nfoo: abrowser/0.001 (C O M M E N T)\r\n" |
| self.parser.parse_header(data) |
| |
| self.assertIn("FOO", self.parser.headers) |
| self.assertEqual(self.parser.headers["FOO"], "abrowser/0.001 (C O M M E N T)") |
| |
| def test_parse_header_invalid_backtrack_bad(self): |
| data = b"GET /foobar HTTP/1.1\r\nHost: example.com\r\nfoo: bar\r\nfoo: xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\x10\r\n" |
| try: |
| self.parser.parse_header(data) |
| except ParsingError as e: |
| self.assertIn("Invalid header", e.args[0]) |
| else: # pragma: nocover |
| self.assertTrue(False) |
| |
| def test_parse_header_short_values(self): |
| data = b"GET /foobar HTTP/1.1\r\nHost: example.com\r\none: 1\r\ntwo: 22\r\n" |
| self.parser.parse_header(data) |
| |
| self.assertIn("ONE", self.parser.headers) |
| self.assertIn("TWO", self.parser.headers) |
| self.assertEqual(self.parser.headers["ONE"], "1") |
| self.assertEqual(self.parser.headers["TWO"], "22") |
| |
| |
| class Test_split_uri(unittest.TestCase): |
| def _callFUT(self, uri): |
| ( |
| self.proxy_scheme, |
| self.proxy_netloc, |
| self.path, |
| self.query, |
| self.fragment, |
| ) = split_uri(uri) |
| |
| def test_split_uri_unquoting_unneeded(self): |
| self._callFUT(b"http://localhost:8080/abc def") |
| self.assertEqual(self.path, "/abc def") |
| |
| def test_split_uri_unquoting_needed(self): |
| self._callFUT(b"http://localhost:8080/abc%20def") |
| self.assertEqual(self.path, "/abc def") |
| |
| def test_split_url_with_query(self): |
| self._callFUT(b"http://localhost:8080/abc?a=1&b=2") |
| self.assertEqual(self.path, "/abc") |
| self.assertEqual(self.query, "a=1&b=2") |
| |
| def test_split_url_with_query_empty(self): |
| self._callFUT(b"http://localhost:8080/abc?") |
| self.assertEqual(self.path, "/abc") |
| self.assertEqual(self.query, "") |
| |
| def test_split_url_with_fragment(self): |
| self._callFUT(b"http://localhost:8080/#foo") |
| self.assertEqual(self.path, "/") |
| self.assertEqual(self.fragment, "foo") |
| |
| def test_split_url_https(self): |
| self._callFUT(b"https://localhost:8080/") |
| self.assertEqual(self.path, "/") |
| self.assertEqual(self.proxy_scheme, "https") |
| self.assertEqual(self.proxy_netloc, "localhost:8080") |
| |
| def test_split_uri_unicode_error_raises_parsing_error(self): |
| # See https://github.com/Pylons/waitress/issues/64 |
| |
| # Either pass or throw a ParsingError, just don't throw another type of |
| # exception as that will cause the connection to close badly: |
| try: |
| self._callFUT(b"/\xd0") |
| except ParsingError: |
| pass |
| |
| def test_split_uri_path(self): |
| self._callFUT(b"//testing/whatever") |
| self.assertEqual(self.path, "//testing/whatever") |
| self.assertEqual(self.proxy_scheme, "") |
| self.assertEqual(self.proxy_netloc, "") |
| self.assertEqual(self.query, "") |
| self.assertEqual(self.fragment, "") |
| |
| def test_split_uri_path_query(self): |
| self._callFUT(b"//testing/whatever?a=1&b=2") |
| self.assertEqual(self.path, "//testing/whatever") |
| self.assertEqual(self.proxy_scheme, "") |
| self.assertEqual(self.proxy_netloc, "") |
| self.assertEqual(self.query, "a=1&b=2") |
| self.assertEqual(self.fragment, "") |
| |
| def test_split_uri_path_query_fragment(self): |
| self._callFUT(b"//testing/whatever?a=1&b=2#fragment") |
| self.assertEqual(self.path, "//testing/whatever") |
| self.assertEqual(self.proxy_scheme, "") |
| self.assertEqual(self.proxy_netloc, "") |
| self.assertEqual(self.query, "a=1&b=2") |
| self.assertEqual(self.fragment, "fragment") |
| |
| |
| class Test_get_header_lines(unittest.TestCase): |
| def _callFUT(self, data): |
| return get_header_lines(data) |
| |
| def test_get_header_lines(self): |
| result = self._callFUT(b"slam\r\nslim") |
| self.assertListEqual(result, [b"slam", b"slim"]) |
| |
| def test_get_header_lines_folded(self): |
| # From RFC2616: |
| # HTTP/1.1 header field values can be folded onto multiple lines if the |
| # continuation line begins with a space or horizontal tab. All linear |
| # white space, including folding, has the same semantics as SP. A |
| # recipient MAY replace any linear white space with a single SP before |
| # interpreting the field value or forwarding the message downstream. |
| |
| # We are just preserving the whitespace that indicates folding. |
| result = self._callFUT(b"slim\r\n slam") |
| self.assertListEqual(result, [b"slim slam"]) |
| |
| def test_get_header_lines_tabbed(self): |
| result = self._callFUT(b"slam\r\n\tslim") |
| self.assertListEqual(result, [b"slam\tslim"]) |
| |
| def test_get_header_lines_malformed(self): |
| # https://corte.si/posts/code/pathod/pythonservers/index.html |
| self.assertRaises(ParsingError, self._callFUT, b" Host: localhost\r\n\r\n") |
| |
| |
| class Test_crack_first_line(unittest.TestCase): |
| def _callFUT(self, line): |
| return crack_first_line(line) |
| |
| def test_crack_first_line_matchok(self): |
| result = self._callFUT(b"GET / HTTP/1.0") |
| self.assertTupleEqual(result, (b"GET", b"/", b"1.0")) |
| |
| def test_crack_first_line_lowercase_method(self): |
| self.assertRaises(ParsingError, self._callFUT, b"get / HTTP/1.0") |
| |
| def test_crack_first_line_nomatch(self): |
| result = self._callFUT(b"GET / bleh") |
| self.assertTupleEqual(result, (b"", b"", b"")) |
| |
| result = self._callFUT(b"GET /info?txtAirPlay&txtRAOP RTSP/1.0") |
| self.assertTupleEqual(result, (b"", b"", b"")) |
| |
| def test_crack_first_line_missing_version(self): |
| result = self._callFUT(b"GET /") |
| self.assertTupleEqual(result, (b"GET", b"/", b"")) |
| |
| def test_crack_first_line_bad_method(self): |
| result = self._callFUT(b"GE\x00 /foobar HTTP/8.4") |
| self.assertTupleEqual(result, (b"", b"", b"")) |
| |
| def test_crack_first_line_bad_version(self): |
| result = self._callFUT(b"GET /foobar HTTP/.1.") |
| self.assertTupleEqual(result, (b"", b"", b"")) |
| |
| |
| class TestHTTPRequestParserIntegration(unittest.TestCase): |
| def setUp(self): |
| my_adj = Adjustments() |
| self.parser = HTTPRequestParser(my_adj) |
| |
| def feed(self, data): |
| parser = self.parser |
| |
| for n in range(100): # make sure we never loop forever |
| consumed = parser.received(data) |
| data = data[consumed:] |
| |
| if parser.completed: |
| return |
| raise ValueError("Looping") # pragma: no cover |
| |
| def testSimpleGET(self): |
| data = ( |
| b"GET /foobar HTTP/8.4\r\n" |
| b"FirstName: mickey\r\n" |
| b"lastname: Mouse\r\n" |
| b"content-length: 6\r\n" |
| b"\r\n" |
| b"Hello." |
| ) |
| parser = self.parser |
| self.feed(data) |
| self.assertTrue(parser.completed) |
| self.assertEqual(parser.version, "8.4") |
| self.assertFalse(parser.empty) |
| self.assertDictEqual( |
| parser.headers, |
| { |
| "FIRSTNAME": "mickey", |
| "LASTNAME": "Mouse", |
| "CONTENT_LENGTH": "6", |
| }, |
| ) |
| self.assertEqual(parser.path, "/foobar") |
| self.assertEqual(parser.command, "GET") |
| self.assertEqual(parser.query, "") |
| self.assertEqual(parser.proxy_scheme, "") |
| self.assertEqual(parser.proxy_netloc, "") |
| self.assertEqual(parser.get_body_stream().getvalue(), b"Hello.") |
| |
| def testComplexGET(self): |
| data = ( |
| b"GET /foo/a+%2B%2F%C3%A4%3D%26a%3Aint?d=b+%2B%2F%3D%26b%3Aint&c+%2B%2F%3D%26c%3Aint=6 HTTP/8.4\r\n" |
| b"FirstName: mickey\r\n" |
| b"lastname: Mouse\r\n" |
| b"content-length: 10\r\n" |
| b"\r\n" |
| b"Hello mickey." |
| ) |
| parser = self.parser |
| self.feed(data) |
| self.assertEqual(parser.command, "GET") |
| self.assertEqual(parser.version, "8.4") |
| self.assertFalse(parser.empty) |
| self.assertDictEqual( |
| parser.headers, |
| {"FIRSTNAME": "mickey", "LASTNAME": "Mouse", "CONTENT_LENGTH": "10"}, |
| ) |
| # path should be utf-8 encoded |
| self.assertEqual( |
| parser.path.encode("latin-1").decode("utf-8"), |
| b"/foo/a++/\xc3\xa4=&a:int".decode("utf-8"), |
| ) |
| # parser.request_uri should preserve the % escape sequences and the query string. |
| self.assertEqual( |
| parser.request_uri, |
| "/foo/a+%2B%2F%C3%A4%3D%26a%3Aint?d=b+%2B%2F%3D%26b%3Aint&c+%2B%2F%3D%26c%3Aint=6", |
| ) |
| self.assertEqual( |
| parser.query, "d=b+%2B%2F%3D%26b%3Aint&c+%2B%2F%3D%26c%3Aint=6" |
| ) |
| self.assertEqual(parser.get_body_stream().getvalue(), b"Hello mick") |
| |
| def testProxyGET(self): |
| data = ( |
| b"GET https://example.com:8080/foobar HTTP/8.4\r\n" |
| b"content-length: 6\r\n" |
| b"\r\n" |
| b"Hello." |
| ) |
| parser = self.parser |
| self.feed(data) |
| self.assertTrue(parser.completed) |
| self.assertEqual(parser.version, "8.4") |
| self.assertFalse(parser.empty) |
| # the authority of the absolute-form request-target is used as the Host |
| self.assertEqual( |
| parser.headers, {"CONTENT_LENGTH": "6", "HOST": "example.com:8080"} |
| ) |
| self.assertEqual(parser.path, "/foobar") |
| self.assertEqual(parser.command, "GET") |
| self.assertEqual(parser.proxy_scheme, "https") |
| self.assertEqual(parser.proxy_netloc, "example.com:8080") |
| self.assertEqual(parser.command, "GET") |
| self.assertEqual(parser.query, "") |
| self.assertEqual(parser.get_body_stream().getvalue(), b"Hello.") |
| |
| def testDuplicateHeaders(self): |
| # Ensure that headers with the same key get concatenated as per |
| # RFC2616. |
| data = ( |
| b"GET /foobar HTTP/8.4\r\n" |
| b"x-forwarded-for: 10.11.12.13\r\n" |
| b"x-forwarded-for: unknown,127.0.0.1\r\n" |
| b"X-Forwarded_for: 255.255.255.255\r\n" |
| b"content-length: 6\r\n" |
| b"\r\n" |
| b"Hello." |
| ) |
| self.feed(data) |
| self.assertTrue(self.parser.completed) |
| self.assertDictEqual( |
| self.parser.headers, |
| { |
| "CONTENT_LENGTH": "6", |
| "X_FORWARDED_FOR": "10.11.12.13, unknown,127.0.0.1", |
| }, |
| ) |
| |
| def testSpoofedHeadersDropped(self): |
| data = ( |
| b"GET /foobar HTTP/8.4\r\n" |
| b"x-auth_user: bob\r\n" |
| b"content-length: 6\r\n" |
| b"\r\n" |
| b"Hello." |
| ) |
| self.feed(data) |
| self.assertTrue(self.parser.completed) |
| self.assertDictEqual( |
| self.parser.headers, |
| { |
| "CONTENT_LENGTH": "6", |
| }, |
| ) |
| |
| |
| class Test_unquote_bytes_to_wsgi(unittest.TestCase): |
| def _callFUT(self, v): |
| return unquote_bytes_to_wsgi(v) |
| |
| def test_highorder(self): |
| val = b"/a%C5%9B" |
| result = self._callFUT(val) |
| # PEP 3333 urlunquoted-latin1-decoded-bytes |
| self.assertEqual(result, "/aÅ\x9b") |
| |
| |
| class DummyBodyStream: |
| def getfile(self): |
| return self |
| |
| def getbuf(self): |
| return self |
| |
| def close(self): |
| self.closed = True |
