# maximal amount of data to read at one time in _safe_read
MAXAMOUNT = 1048576
+# maximal line length when calling readline().
+_MAXLINE = 65536
+
class HTTPMessage(mimetools.Message):
def addheader(self, key, value):
except IOError:
startofline = tell = None
self.seekable = 0
- line = self.fp.readline()
+ line = self.fp.readline(_MAXLINE + 1)
+ if len(line) > _MAXLINE:
+ raise LineTooLong("header line")
if not line:
self.status = 'EOF in headers'
break
break
# skip the header from the 100 response
while True:
- skip = self.fp.readline().strip()
+ skip = self.fp.readline(_MAXLINE + 1)
+ if len(skip) > _MAXLINE:
+ raise LineTooLong("header line")
+ skip = skip.strip()
if not skip:
break
if self.debuglevel > 0:
value = []
while True:
if chunk_left is None:
- line = self.fp.readline()
+ line = self.fp.readline(_MAXLINE + 1)
+ if len(line) > _MAXLINE:
+ raise LineTooLong("chunk size")
i = line.find(';')
if i >= 0:
line = line[:i] # strip chunk-extensions
# read and discard trailer up to the CRLF terminator
### note: we shouldn't have any trailers!
while True:
- line = self.fp.readline()
+ line = self.fp.readline(_MAXLINE + 1)
+ if len(line) > _MAXLINE:
+ raise LineTooLong("trailer line")
if not line:
# a vanishingly small number of sites EOF without
# sending the trailer
raise socket.error("Tunnel connection failed: %d %s" % (code,
message.strip()))
while True:
- line = response.fp.readline()
+ line = response.fp.readline(_MAXLINE + 1)
+ if len(line) > _MAXLINE:
+ raise LineTooLong("header line")
if line == '\r\n': break
self.args = line,
self.line = line
+class LineTooLong(HTTPException):
+ def __init__(self, line_type):
+ HTTPException.__init__(self, "got more than %d bytes when reading %s"
+ % (_MAXLINE, line_type))
+
# for backwards compatibility
error = HTTPException
self.assertTrue(hasattr(resp,'fileno'),
'HTTPResponse should expose a fileno attribute')
+ # Test lines overflowing the max line size (_MAXLINE in http.client)
+
+ def test_overflowing_status_line(self):
+ self.skipTest("disabled for HTTP 0.9 support")
+ body = "HTTP/1.1 200 Ok" + "k" * 65536 + "\r\n"
+ resp = httplib.HTTPResponse(FakeSocket(body))
+ self.assertRaises((httplib.LineTooLong, httplib.BadStatusLine), resp.begin)
+
+ def test_overflowing_header_line(self):
+ body = (
+ 'HTTP/1.1 200 OK\r\n'
+ 'X-Foo: bar' + 'r' * 65536 + '\r\n\r\n'
+ )
+ resp = httplib.HTTPResponse(FakeSocket(body))
+ self.assertRaises(httplib.LineTooLong, resp.begin)
+
+ def test_overflowing_chunked_line(self):
+ body = (
+ 'HTTP/1.1 200 OK\r\n'
+ 'Transfer-Encoding: chunked\r\n\r\n'
+ + '0' * 65536 + 'a\r\n'
+ 'hello world\r\n'
+ '0\r\n'
+ )
+ resp = httplib.HTTPResponse(FakeSocket(body))
+ resp.begin()
+ self.assertRaises(httplib.LineTooLong, resp.read)
+
+
class OfflineTest(TestCase):
def test_responses(self):
self.assertEqual(httplib.responses[httplib.NOT_FOUND], "Not Found")