diff options
Diffstat (limited to 'Lib/test/test_ftplib.py')
| -rw-r--r-- | Lib/test/test_ftplib.py | 91 |
1 files changed, 69 insertions, 22 deletions
diff --git a/Lib/test/test_ftplib.py b/Lib/test/test_ftplib.py index 64cd80bc..d0fd92e 100644 --- a/Lib/test/test_ftplib.py +++ b/Lib/test/test_ftplib.py @@ -15,12 +15,16 @@ try: import ssl except ImportError: ssl = None + HAS_SNI = False +else: + from ssl import HAS_SNI from unittest import TestCase, skipUnless from test import support from test.support import HOST, HOSTv6 threading = support.import_module('threading') +TIMEOUT = 3 # the dummy data returned by server over the data channel when # RETR, LIST, NLST, MLSD commands are issued RETR_DATA = 'abcde12345\r\n' * 1000 @@ -126,7 +130,7 @@ class DummyFTPHandler(asynchat.async_chat): addr = list(map(int, arg.split(','))) ip = '%d.%d.%d.%d' %tuple(addr[:4]) port = (addr[4] * 256) + addr[5] - s = socket.create_connection((ip, port), timeout=2) + s = socket.create_connection((ip, port), timeout=TIMEOUT) self.dtp = self.dtp_handler(s, baseclass=self) self.push('200 active data connection established') @@ -134,7 +138,7 @@ class DummyFTPHandler(asynchat.async_chat): with socket.socket() as sock: sock.bind((self.socket.getsockname()[0], 0)) sock.listen(5) - sock.settimeout(10) + sock.settimeout(TIMEOUT) ip, port = sock.getsockname()[:2] ip = ip.replace('.', ','); p1 = port / 256; p2 = port % 256 self.push('227 entering passive mode (%s,%d,%d)' %(ip, p1, p2)) @@ -144,7 +148,7 @@ class DummyFTPHandler(asynchat.async_chat): def cmd_eprt(self, arg): af, ip, port = arg.split(arg[0])[1:-1] port = int(port) - s = socket.create_connection((ip, port), timeout=2) + s = socket.create_connection((ip, port), timeout=TIMEOUT) self.dtp = self.dtp_handler(s, baseclass=self) self.push('200 active data connection established') @@ -152,7 +156,7 @@ class DummyFTPHandler(asynchat.async_chat): with socket.socket(socket.AF_INET6) as sock: sock.bind((self.socket.getsockname()[0], 0)) sock.listen(5) - sock.settimeout(10) + sock.settimeout(TIMEOUT) port = sock.getsockname()[1] self.push('229 entering extended passive mode (|||%d|)' %port) conn, addr = sock.accept() @@ -300,7 +304,8 @@ class DummyFTPServer(asyncore.dispatcher, threading.Thread): if ssl is not None: - CERTFILE = os.path.join(os.path.dirname(__file__), "keycert.pem") + CERTFILE = os.path.join(os.path.dirname(__file__), "keycert3.pem") + CAFILE = os.path.join(os.path.dirname(__file__), "pycacert.pem") class SSLConnection(asyncore.dispatcher): """An asyncore.dispatcher subclass supporting TLS/SSL.""" @@ -327,7 +332,7 @@ if ssl is not None: elif err.args[0] == ssl.SSL_ERROR_EOF: return self.handle_close() raise - except socket.error as err: + except OSError as err: if err.args[0] == errno.ECONNABORTED: return self.handle_close() else: @@ -341,7 +346,7 @@ if ssl is not None: if err.args[0] in (ssl.SSL_ERROR_WANT_READ, ssl.SSL_ERROR_WANT_WRITE): return - except socket.error as err: + except OSError as err: # Any "socket error" corresponds to a SSL_ERROR_SYSCALL return # from OpenSSL's SSL_shutdown(), corresponding to a # closed socket condition. See also: @@ -460,7 +465,7 @@ class TestFTPClass(TestCase): def setUp(self): self.server = DummyFTPServer((HOST, 0)) self.server.start() - self.client = ftplib.FTP(timeout=2) + self.client = ftplib.FTP(timeout=TIMEOUT) self.client.connect(self.server.host, self.server.port) def tearDown(self): @@ -488,7 +493,7 @@ class TestFTPClass(TestCase): def test_all_errors(self): exceptions = (ftplib.error_reply, ftplib.error_temp, ftplib.error_perm, - ftplib.error_proto, ftplib.Error, IOError, EOFError) + ftplib.error_proto, ftplib.Error, OSError, EOFError) for x in exceptions: try: raise x('exception not included in all_errors set') @@ -678,7 +683,7 @@ class TestFTPClass(TestCase): def test_makepasv(self): host, port = self.client.makepasv() - conn = socket.create_connection((host, port), 10) + conn = socket.create_connection((host, port), timeout=TIMEOUT) conn.close() # IPv4 is in use, just make sure send_epsv has not been used self.assertEqual(self.server.handler_instance.last_received_cmd, 'pasv') @@ -691,12 +696,12 @@ class TestFTPClass(TestCase): return False try: self.client.sendcmd('noop') - except (socket.error, EOFError): + except (OSError, EOFError): return False return True # base test - with ftplib.FTP(timeout=2) as self.client: + with ftplib.FTP(timeout=TIMEOUT) as self.client: self.client.connect(self.server.host, self.server.port) self.client.sendcmd('noop') self.assertTrue(is_client_connected()) @@ -704,7 +709,7 @@ class TestFTPClass(TestCase): self.assertFalse(is_client_connected()) # QUIT sent inside the with block - with ftplib.FTP(timeout=2) as self.client: + with ftplib.FTP(timeout=TIMEOUT) as self.client: self.client.connect(self.server.host, self.server.port) self.client.sendcmd('noop') self.client.quit() @@ -714,7 +719,7 @@ class TestFTPClass(TestCase): # force a wrong response code to be sent on QUIT: error_perm # is expected and the connection is supposed to be closed try: - with ftplib.FTP(timeout=2) as self.client: + with ftplib.FTP(timeout=TIMEOUT) as self.client: self.client.connect(self.server.host, self.server.port) self.client.sendcmd('noop') self.server.handler_instance.next_response = '550 error on quit' @@ -736,7 +741,7 @@ class TestFTPClass(TestCase): source_address=(HOST, port)) self.assertEqual(self.client.sock.getsockname()[1], port) self.client.quit() - except IOError as e: + except OSError as e: if e.errno == errno.EADDRINUSE: self.skipTest("couldn't bind to port %d" % port) raise @@ -747,7 +752,7 @@ class TestFTPClass(TestCase): try: with self.client.transfercmd('list') as sock: self.assertEqual(sock.getsockname()[1], port) - except IOError as e: + except OSError as e: if e.errno == errno.EADDRINUSE: self.skipTest("couldn't bind to port %d" % port) raise @@ -785,7 +790,7 @@ class TestIPv6Environment(TestCase): def setUp(self): self.server = DummyFTPServer((HOSTv6, 0), af=socket.AF_INET6) self.server.start() - self.client = ftplib.FTP() + self.client = ftplib.FTP(timeout=TIMEOUT) self.client.connect(self.server.host, self.server.port) def tearDown(self): @@ -802,7 +807,7 @@ class TestIPv6Environment(TestCase): def test_makepasv(self): host, port = self.client.makepasv() - conn = socket.create_connection((host, port), 10) + conn = socket.create_connection((host, port), timeout=TIMEOUT) conn.close() self.assertEqual(self.server.handler_instance.last_received_cmd, 'epsv') @@ -829,7 +834,7 @@ class TestTLS_FTPClassMixin(TestFTPClass): def setUp(self): self.server = DummyTLS_FTPServer((HOST, 0)) self.server.start() - self.client = ftplib.FTP_TLS(timeout=2) + self.client = ftplib.FTP_TLS(timeout=TIMEOUT) self.client.connect(self.server.host, self.server.port) # enable TLS self.client.auth() @@ -843,7 +848,7 @@ class TestTLS_FTPClass(TestCase): def setUp(self): self.server = DummyTLS_FTPServer((HOST, 0)) self.server.start() - self.client = ftplib.FTP_TLS(timeout=2) + self.client = ftplib.FTP_TLS(timeout=TIMEOUT) self.client.connect(self.server.host, self.server.port) def tearDown(self): @@ -903,7 +908,7 @@ class TestTLS_FTPClass(TestCase): self.assertRaises(ValueError, ftplib.FTP_TLS, certfile=CERTFILE, keyfile=CERTFILE, context=ctx) - self.client = ftplib.FTP_TLS(context=ctx, timeout=2) + self.client = ftplib.FTP_TLS(context=ctx, timeout=TIMEOUT) self.client.connect(self.server.host, self.server.port) self.assertNotIsInstance(self.client.sock, ssl.SSLSocket) self.client.auth() @@ -922,6 +927,37 @@ class TestTLS_FTPClass(TestCase): self.client.ccc() self.assertRaises(ValueError, self.client.sock.unwrap) + @skipUnless(HAS_SNI, 'No SNI support in ssl module') + def test_check_hostname(self): + self.client.quit() + ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1) + ctx.verify_mode = ssl.CERT_REQUIRED + ctx.check_hostname = True + ctx.load_verify_locations(CAFILE) + self.client = ftplib.FTP_TLS(context=ctx, timeout=TIMEOUT) + + # 127.0.0.1 doesn't match SAN + self.client.connect(self.server.host, self.server.port) + with self.assertRaises(ssl.CertificateError): + self.client.auth() + # exception quits connection + + self.client.connect(self.server.host, self.server.port) + self.client.prot_p() + with self.assertRaises(ssl.CertificateError): + with self.client.transfercmd("list") as sock: + pass + self.client.quit() + + self.client.connect("localhost", self.server.port) + self.client.auth() + self.client.quit() + + self.client.connect("localhost", self.server.port) + self.client.prot_p() + with self.client.transfercmd("list") as sock: + pass + class TestTimeouts(TestCase): @@ -1017,8 +1053,19 @@ class TestTimeouts(TestCase): ftp.close() +class TestNetrcDeprecation(TestCase): + + def test_deprecation(self): + with support.temp_cwd(), support.EnvironmentVarGuard() as env: + env['HOME'] = os.getcwd() + open('.netrc', 'w').close() + with self.assertWarns(DeprecationWarning): + ftplib.Netrc() + + + def test_main(): - tests = [TestFTPClass, TestTimeouts, + tests = [TestFTPClass, TestTimeouts, TestNetrcDeprecation, TestIPv6Environment, TestTLS_FTPClassMixin, TestTLS_FTPClass] |