| ############################################################################## |
| # |
| # 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_bad_transfer_encoding(self): |
| data = ( |
| b"GET /foobar HTTP/1.1\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, 48) |
| 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"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, 62) |
| 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"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"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, 62) |
| 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.assertIn("Content-Length is invalid", e.args[0]) |
| 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\ntransfer-encoding: ChUnKed\r\n" |
| self.parser.parse_header(data) |
| self.assertEqual(self.parser.body_rcv.__class__.__name__, "ChunkedReceiver") |
| |
| def test_parse_header_transfer_encoding_invalid(self): |
| data = b"GET /foobar HTTP/1.1\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\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_whitespace(self): |
| data = b"GET /foobar HTTP/1.1\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\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\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\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\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\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\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\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\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\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\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\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\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\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\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\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) |
| self.assertEqual(parser.headers, {"CONTENT_LENGTH": "6"}) |
| 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 |
