diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/test_functional.py | 50 | ||||
| -rw-r--r-- | tests/test_parser.py | 22 | ||||
| -rw-r--r-- | tests/test_receiver.py | 51 |
3 files changed, 119 insertions, 4 deletions
diff --git a/tests/test_functional.py b/tests/test_functional.py index f74a252..60eb24a 100644 --- a/tests/test_functional.py +++ b/tests/test_functional.py @@ -322,7 +322,7 @@ class EchoTests: self.assertFalse("transfer-encoding" in headers) def test_chunking_request_with_content(self): - control_line = b"20;\r\n" # 20 hex = 32 dec + control_line = b"20\r\n" # 20 hex = 32 dec s = b"This string has 32 characters.\r\n" expected = s * 12 header = b"GET / HTTP/1.1\r\nTransfer-Encoding: chunked\r\n\r\n" @@ -341,7 +341,7 @@ class EchoTests: self.assertFalse("transfer-encoding" in headers) def test_broken_chunked_encoding(self): - control_line = b"20;\r\n" # 20 hex = 32 dec + control_line = b"20\r\n" # 20 hex = 32 dec s = b"This string has 32 characters.\r\n" to_send = b"GET / HTTP/1.1\r\nTransfer-Encoding: chunked\r\n\r\n" to_send += control_line + s + b"\r\n" @@ -364,8 +364,52 @@ class EchoTests: self.send_check_error(to_send) self.assertRaises(ConnectionClosed, read_http, fp) + def test_broken_chunked_encoding_invalid_hex(self): + control_line = b"0x20\r\n" # 20 hex = 32 dec + s = b"This string has 32 characters.\r\n" + to_send = b"GET / HTTP/1.1\r\nTransfer-Encoding: chunked\r\n\r\n" + to_send += control_line + s + b"\r\n" + self.connect() + self.sock.send(to_send) + with self.sock.makefile("rb", 0) as fp: + line, headers, response_body = read_http(fp) + self.assertline(line, "400", "Bad Request", "HTTP/1.1") + cl = int(headers["content-length"]) + self.assertEqual(cl, len(response_body)) + self.assertIn(b"Invalid chunk size", response_body) + self.assertEqual( + sorted(headers.keys()), + ["connection", "content-length", "content-type", "date", "server"], + ) + self.assertEqual(headers["content-type"], "text/plain") + # connection has been closed + self.send_check_error(to_send) + self.assertRaises(ConnectionClosed, read_http, fp) + + def test_broken_chunked_encoding_invalid_extension(self): + control_line = b"20;invalid=\r\n" # 20 hex = 32 dec + s = b"This string has 32 characters.\r\n" + to_send = b"GET / HTTP/1.1\r\nTransfer-Encoding: chunked\r\n\r\n" + to_send += control_line + s + b"\r\n" + self.connect() + self.sock.send(to_send) + with self.sock.makefile("rb", 0) as fp: + line, headers, response_body = read_http(fp) + self.assertline(line, "400", "Bad Request", "HTTP/1.1") + cl = int(headers["content-length"]) + self.assertEqual(cl, len(response_body)) + self.assertIn(b"Invalid chunk extension", response_body) + self.assertEqual( + sorted(headers.keys()), + ["connection", "content-length", "content-type", "date", "server"], + ) + self.assertEqual(headers["content-type"], "text/plain") + # connection has been closed + self.send_check_error(to_send) + self.assertRaises(ConnectionClosed, read_http, fp) + def test_broken_chunked_encoding_missing_chunk_end(self): - control_line = b"20;\r\n" # 20 hex = 32 dec + control_line = b"20\r\n" # 20 hex = 32 dec s = b"This string has 32 characters.\r\n" to_send = b"GET / HTTP/1.1\r\nTransfer-Encoding: chunked\r\n\r\n" to_send += control_line + s diff --git a/tests/test_parser.py b/tests/test_parser.py index aacef26..4461bde 100644 --- a/tests/test_parser.py +++ b/tests/test_parser.py @@ -155,7 +155,7 @@ class TestHTTPRequestParser(unittest.TestCase): b"Transfer-Encoding: chunked\r\n" b"X-Foo: 1\r\n" b"\r\n" - b"1d;\r\n" + b"1d\r\n" b"This string has 29 characters\r\n" b"0\r\n\r\n" ) @@ -193,6 +193,26 @@ class TestHTTPRequestParser(unittest.TestCase): 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" diff --git a/tests/test_receiver.py b/tests/test_receiver.py index f55aa68..d160cac 100644 --- a/tests/test_receiver.py +++ b/tests/test_receiver.py @@ -1,5 +1,7 @@ import unittest +import pytest + class TestFixedStreamReceiver(unittest.TestCase): def _makeOne(self, cl, buf): @@ -226,6 +228,55 @@ class TestChunkedReceiver(unittest.TestCase): self.assertEqual(inst.error, None) +class TestChunkedReceiverParametrized: + def _makeOne(self, buf): + from waitress.receiver import ChunkedReceiver + + return ChunkedReceiver(buf) + + @pytest.mark.parametrize( + "invalid_extension", [b"\n", b"invalid=", b"\r", b"invalid = true"] + ) + def test_received_invalid_extensions(self, invalid_extension): + from waitress.utilities import BadRequest + + buf = DummyBuffer() + inst = self._makeOne(buf) + data = b"4;" + invalid_extension + b"\r\ntest\r\n" + result = inst.received(data) + assert result == len(data) + assert inst.error.__class__ == BadRequest + assert inst.error.body == "Invalid chunk extension" + + @pytest.mark.parametrize( + "valid_extension", [b"test", b"valid=true", b"valid=true;other=true"] + ) + def test_received_valid_extensions(self, valid_extension): + # While waitress may ignore extensions in Chunked Encoding, we do want + # to make sure that we don't fail when we do encounter one that is + # valid + buf = DummyBuffer() + inst = self._makeOne(buf) + data = b"4;" + valid_extension + b"\r\ntest\r\n" + result = inst.received(data) + assert result == len(data) + assert inst.error == None + + @pytest.mark.parametrize( + "invalid_size", [b"0x04", b"+0x04", b"x04", b"+04", b" 04", b" 0x04"] + ) + def test_received_invalid_size(self, invalid_size): + from waitress.utilities import BadRequest + + buf = DummyBuffer() + inst = self._makeOne(buf) + data = invalid_size + b"\r\ntest\r\n" + result = inst.received(data) + assert result == len(data) + assert inst.error.__class__ == BadRequest + assert inst.error.body == "Invalid chunk size" + + class DummyBuffer: def __init__(self, data=None): if data is None: |