diff -r 936621d33c38 Lib/poplib.py --- a/Lib/poplib.py Wed Feb 20 18:19:55 2013 -0500 +++ b/Lib/poplib.py Sun Sep 29 21:56:02 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 Sun Sep 29 21:56:02 2013 +0200 @@ -21,6 +21,21 @@ serv.close() evt.set() + +def evil_server(evt, serv): + serv.listen(5) + try: + conn, addr = serv.accept() + except socket.timeout: + pass + else: + conn.send("+ Hola mundo" * 1000 + "\n") + conn.close() + finally: + serv.close() + evt.set() + + class GeneralTests(TestCase): def setUp(self): @@ -65,8 +80,27 @@ pop.sock.close() +class EvilServerTests(TestCase): + 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)).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) + + def test_main(verbose=None): test_support.run_unittest(GeneralTests) + test_support.run_unittest(EvilServerTests) if __name__ == '__main__': test_main()