diff -r 936621d33c38 Lib/poplib.py --- a/Lib/poplib.py Wed Feb 20 18:19:55 2013 -0500 +++ b/Lib/poplib.py Mon Sep 30 19:32:35 2013 +0200 @@ -32,6 +32,12 @@ LF = '\n' CRLF = CR+LF +# maximal line length when calling readline(). This is to prevent +# reading arbitrary lenght lines. RFC 1939 limits POP3 line length to +# 512 characters, including CRLF. We have selected 2048 just to be on +# the safe side. +_MAXLINE = 2048 + class POP3: @@ -103,7 +109,10 @@ # Raise error_proto('-ERR EOF') if the connection is closed. def _getline(self): - line = self.file.readline() + line = self.file.readline(_MAXLINE + 1) + if len(line) > _MAXLINE: + raise error_proto('line too long') + if self._debugging > 1: print '*get*', repr(line) if not line: raise error_proto('-ERR EOF') octets = len(line) @@ -363,7 +372,10 @@ line = "" renewline = re.compile(r'.*?\n') match = renewline.match(self.buffer) + while not match: + if len(self.buffer) > _MAXLINE: + raise error_proto('line too long') self._fillBuffer() match = renewline.match(self.buffer) line = match.group(0) diff -r 936621d33c38 Lib/test/test_poplib.py --- a/Lib/test/test_poplib.py Wed Feb 20 18:19:55 2013 -0500 +++ b/Lib/test/test_poplib.py Mon Sep 30 19:32:35 2013 +0200 @@ -1,3 +1,4 @@ +import os import socket import threading import poplib @@ -21,6 +22,34 @@ serv.close() evt.set() + +def evil_server(evt, serv, use_ssl=False): + serv.listen(5) + try: + conn, addr = serv.accept() + if use_ssl: + conn = ssl.wrap_socket( + conn, + server_side=True, + certfile=CERTFILE, + ) + except socket.timeout: + pass + else: + if use_ssl: + try: + conn.do_handshake() + except ssl.SSLError, err: + if err.args[0] not in (ssl.SSL_ERROR_WANT_READ, + ssl.SSL_ERROR_WANT_WRITE): + raise + conn.send("+ Hola mundo" * 1000 + "\n") + conn.close() + finally: + serv.close() + evt.set() + + class GeneralTests(TestCase): def setUp(self): @@ -65,8 +94,49 @@ pop.sock.close() +class EvilServerTests(TestCase): + use_ssl = False + + def setUp(self): + self.evt = threading.Event() + self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.sock.settimeout(3) + self.port = test_support.bind_port(self.sock) + threading.Thread( + target=evil_server, + args=(self.evt, self.sock, self.use_ssl)).start() + time.sleep(.1) + + def tearDown(self): + self.evt.wait() + + def testTooLongLines(self): + self.assertRaises(poplib.error_proto, poplib.POP3, + 'localhost', self.port, timeout=30) + + +SUPPORTS_SSL = False + +if hasattr(poplib, 'POP3_SSL'): + import ssl + + SUPPORTS_SSL = True + CERTFILE = os.path.join(os.path.dirname(__file__) or os.curdir, "keycert.pem") + + class EvilSSLServerTests(EvilServerTests): + use_ssl = True + + def testTooLongLines(self): + self.assertRaises(poplib.error_proto, poplib.POP3_SSL, + 'localhost', self.port) + + def test_main(verbose=None): test_support.run_unittest(GeneralTests) + test_support.run_unittest(EvilServerTests) + + if SUPPORTS_SSL: + test_support.run_unittest(EvilSSLServerTests) if __name__ == '__main__': test_main()