1"""Test script for ftplib module.""" 2 3# Modified by Giampaolo Rodola' to test FTP class, IPv6 and TLS 4# environment 5 6import ftplib 7import socket 8import io 9import errno 10import os 11import threading 12import time 13import unittest 14try: 15 import ssl 16except ImportError: 17 ssl = None 18 19from unittest import TestCase, skipUnless 20from test import support 21from test.support import threading_helper 22from test.support import socket_helper 23from test.support import warnings_helper 24from test.support.socket_helper import HOST, HOSTv6 25 26 27asynchat = warnings_helper.import_deprecated('asynchat') 28asyncore = warnings_helper.import_deprecated('asyncore') 29 30 31support.requires_working_socket(module=True) 32 33TIMEOUT = support.LOOPBACK_TIMEOUT 34DEFAULT_ENCODING = 'utf-8' 35# the dummy data returned by server over the data channel when 36# RETR, LIST, NLST, MLSD commands are issued 37RETR_DATA = 'abcde12345\r\n' * 1000 + 'non-ascii char \xAE\r\n' 38LIST_DATA = 'foo\r\nbar\r\n non-ascii char \xAE\r\n' 39NLST_DATA = 'foo\r\nbar\r\n non-ascii char \xAE\r\n' 40MLSD_DATA = ("type=cdir;perm=el;unique==keVO1+ZF4; test\r\n" 41 "type=pdir;perm=e;unique==keVO1+d?3; ..\r\n" 42 "type=OS.unix=slink:/foobar;perm=;unique==keVO1+4G4; foobar\r\n" 43 "type=OS.unix=chr-13/29;perm=;unique==keVO1+5G4; device\r\n" 44 "type=OS.unix=blk-11/108;perm=;unique==keVO1+6G4; block\r\n" 45 "type=file;perm=awr;unique==keVO1+8G4; writable\r\n" 46 "type=dir;perm=cpmel;unique==keVO1+7G4; promiscuous\r\n" 47 "type=dir;perm=;unique==keVO1+1t2; no-exec\r\n" 48 "type=file;perm=r;unique==keVO1+EG4; two words\r\n" 49 "type=file;perm=r;unique==keVO1+IH4; leading space\r\n" 50 "type=file;perm=r;unique==keVO1+1G4; file1\r\n" 51 "type=dir;perm=cpmel;unique==keVO1+7G4; incoming\r\n" 52 "type=file;perm=r;unique==keVO1+1G4; file2\r\n" 53 "type=file;perm=r;unique==keVO1+1G4; file3\r\n" 54 "type=file;perm=r;unique==keVO1+1G4; file4\r\n" 55 "type=dir;perm=cpmel;unique==SGP1; dir \xAE non-ascii char\r\n" 56 "type=file;perm=r;unique==SGP2; file \xAE non-ascii char\r\n") 57 58 59def default_error_handler(): 60 # bpo-44359: Silently ignore socket errors. Such errors occur when a client 61 # socket is closed, in TestFTPClass.tearDown() and makepasv() tests, and 62 # the server gets an error on its side. 63 pass 64 65 66class DummyDTPHandler(asynchat.async_chat): 67 dtp_conn_closed = False 68 69 def __init__(self, conn, baseclass): 70 asynchat.async_chat.__init__(self, conn) 71 self.baseclass = baseclass 72 self.baseclass.last_received_data = '' 73 self.encoding = baseclass.encoding 74 75 def handle_read(self): 76 new_data = self.recv(1024).decode(self.encoding, 'replace') 77 self.baseclass.last_received_data += new_data 78 79 def handle_close(self): 80 # XXX: this method can be called many times in a row for a single 81 # connection, including in clear-text (non-TLS) mode. 82 # (behaviour witnessed with test_data_connection) 83 if not self.dtp_conn_closed: 84 self.baseclass.push('226 transfer complete') 85 self.close() 86 self.dtp_conn_closed = True 87 88 def push(self, what): 89 if self.baseclass.next_data is not None: 90 what = self.baseclass.next_data 91 self.baseclass.next_data = None 92 if not what: 93 return self.close_when_done() 94 super(DummyDTPHandler, self).push(what.encode(self.encoding)) 95 96 def handle_error(self): 97 default_error_handler() 98 99 100class DummyFTPHandler(asynchat.async_chat): 101 102 dtp_handler = DummyDTPHandler 103 104 def __init__(self, conn, encoding=DEFAULT_ENCODING): 105 asynchat.async_chat.__init__(self, conn) 106 # tells the socket to handle urgent data inline (ABOR command) 107 self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_OOBINLINE, 1) 108 self.set_terminator(b"\r\n") 109 self.in_buffer = [] 110 self.dtp = None 111 self.last_received_cmd = None 112 self.last_received_data = '' 113 self.next_response = '' 114 self.next_data = None 115 self.rest = None 116 self.next_retr_data = RETR_DATA 117 self.push('220 welcome') 118 self.encoding = encoding 119 # We use this as the string IPv4 address to direct the client 120 # to in response to a PASV command. To test security behavior. 121 # https://bugs.python.org/issue43285/. 122 self.fake_pasv_server_ip = '252.253.254.255' 123 124 def collect_incoming_data(self, data): 125 self.in_buffer.append(data) 126 127 def found_terminator(self): 128 line = b''.join(self.in_buffer).decode(self.encoding) 129 self.in_buffer = [] 130 if self.next_response: 131 self.push(self.next_response) 132 self.next_response = '' 133 cmd = line.split(' ')[0].lower() 134 self.last_received_cmd = cmd 135 space = line.find(' ') 136 if space != -1: 137 arg = line[space + 1:] 138 else: 139 arg = "" 140 if hasattr(self, 'cmd_' + cmd): 141 method = getattr(self, 'cmd_' + cmd) 142 method(arg) 143 else: 144 self.push('550 command "%s" not understood.' %cmd) 145 146 def handle_error(self): 147 default_error_handler() 148 149 def push(self, data): 150 asynchat.async_chat.push(self, data.encode(self.encoding) + b'\r\n') 151 152 def cmd_port(self, arg): 153 addr = list(map(int, arg.split(','))) 154 ip = '%d.%d.%d.%d' %tuple(addr[:4]) 155 port = (addr[4] * 256) + addr[5] 156 s = socket.create_connection((ip, port), timeout=TIMEOUT) 157 self.dtp = self.dtp_handler(s, baseclass=self) 158 self.push('200 active data connection established') 159 160 def cmd_pasv(self, arg): 161 with socket.create_server((self.socket.getsockname()[0], 0)) as sock: 162 sock.settimeout(TIMEOUT) 163 port = sock.getsockname()[1] 164 ip = self.fake_pasv_server_ip 165 ip = ip.replace('.', ','); p1 = port / 256; p2 = port % 256 166 self.push('227 entering passive mode (%s,%d,%d)' %(ip, p1, p2)) 167 conn, addr = sock.accept() 168 self.dtp = self.dtp_handler(conn, baseclass=self) 169 170 def cmd_eprt(self, arg): 171 af, ip, port = arg.split(arg[0])[1:-1] 172 port = int(port) 173 s = socket.create_connection((ip, port), timeout=TIMEOUT) 174 self.dtp = self.dtp_handler(s, baseclass=self) 175 self.push('200 active data connection established') 176 177 def cmd_epsv(self, arg): 178 with socket.create_server((self.socket.getsockname()[0], 0), 179 family=socket.AF_INET6) as sock: 180 sock.settimeout(TIMEOUT) 181 port = sock.getsockname()[1] 182 self.push('229 entering extended passive mode (|||%d|)' %port) 183 conn, addr = sock.accept() 184 self.dtp = self.dtp_handler(conn, baseclass=self) 185 186 def cmd_echo(self, arg): 187 # sends back the received string (used by the test suite) 188 self.push(arg) 189 190 def cmd_noop(self, arg): 191 self.push('200 noop ok') 192 193 def cmd_user(self, arg): 194 self.push('331 username ok') 195 196 def cmd_pass(self, arg): 197 self.push('230 password ok') 198 199 def cmd_acct(self, arg): 200 self.push('230 acct ok') 201 202 def cmd_rnfr(self, arg): 203 self.push('350 rnfr ok') 204 205 def cmd_rnto(self, arg): 206 self.push('250 rnto ok') 207 208 def cmd_dele(self, arg): 209 self.push('250 dele ok') 210 211 def cmd_cwd(self, arg): 212 self.push('250 cwd ok') 213 214 def cmd_size(self, arg): 215 self.push('250 1000') 216 217 def cmd_mkd(self, arg): 218 self.push('257 "%s"' %arg) 219 220 def cmd_rmd(self, arg): 221 self.push('250 rmd ok') 222 223 def cmd_pwd(self, arg): 224 self.push('257 "pwd ok"') 225 226 def cmd_type(self, arg): 227 self.push('200 type ok') 228 229 def cmd_quit(self, arg): 230 self.push('221 quit ok') 231 self.close() 232 233 def cmd_abor(self, arg): 234 self.push('226 abor ok') 235 236 def cmd_stor(self, arg): 237 self.push('125 stor ok') 238 239 def cmd_rest(self, arg): 240 self.rest = arg 241 self.push('350 rest ok') 242 243 def cmd_retr(self, arg): 244 self.push('125 retr ok') 245 if self.rest is not None: 246 offset = int(self.rest) 247 else: 248 offset = 0 249 self.dtp.push(self.next_retr_data[offset:]) 250 self.dtp.close_when_done() 251 self.rest = None 252 253 def cmd_list(self, arg): 254 self.push('125 list ok') 255 self.dtp.push(LIST_DATA) 256 self.dtp.close_when_done() 257 258 def cmd_nlst(self, arg): 259 self.push('125 nlst ok') 260 self.dtp.push(NLST_DATA) 261 self.dtp.close_when_done() 262 263 def cmd_opts(self, arg): 264 self.push('200 opts ok') 265 266 def cmd_mlsd(self, arg): 267 self.push('125 mlsd ok') 268 self.dtp.push(MLSD_DATA) 269 self.dtp.close_when_done() 270 271 def cmd_setlongretr(self, arg): 272 # For testing. Next RETR will return long line. 273 self.next_retr_data = 'x' * int(arg) 274 self.push('125 setlongretr ok') 275 276 277class DummyFTPServer(asyncore.dispatcher, threading.Thread): 278 279 handler = DummyFTPHandler 280 281 def __init__(self, address, af=socket.AF_INET, encoding=DEFAULT_ENCODING): 282 threading.Thread.__init__(self) 283 asyncore.dispatcher.__init__(self) 284 self.daemon = True 285 self.create_socket(af, socket.SOCK_STREAM) 286 self.bind(address) 287 self.listen(5) 288 self.active = False 289 self.active_lock = threading.Lock() 290 self.host, self.port = self.socket.getsockname()[:2] 291 self.handler_instance = None 292 self.encoding = encoding 293 294 def start(self): 295 assert not self.active 296 self.__flag = threading.Event() 297 threading.Thread.start(self) 298 self.__flag.wait() 299 300 def run(self): 301 self.active = True 302 self.__flag.set() 303 while self.active and asyncore.socket_map: 304 self.active_lock.acquire() 305 asyncore.loop(timeout=0.1, count=1) 306 self.active_lock.release() 307 asyncore.close_all(ignore_all=True) 308 309 def stop(self): 310 assert self.active 311 self.active = False 312 self.join() 313 314 def handle_accepted(self, conn, addr): 315 self.handler_instance = self.handler(conn, encoding=self.encoding) 316 317 def handle_connect(self): 318 self.close() 319 handle_read = handle_connect 320 321 def writable(self): 322 return 0 323 324 def handle_error(self): 325 default_error_handler() 326 327 328if ssl is not None: 329 330 CERTFILE = os.path.join(os.path.dirname(__file__), "keycert3.pem") 331 CAFILE = os.path.join(os.path.dirname(__file__), "pycacert.pem") 332 333 class SSLConnection(asyncore.dispatcher): 334 """An asyncore.dispatcher subclass supporting TLS/SSL.""" 335 336 _ssl_accepting = False 337 _ssl_closing = False 338 339 def secure_connection(self): 340 context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) 341 context.load_cert_chain(CERTFILE) 342 socket = context.wrap_socket(self.socket, 343 suppress_ragged_eofs=False, 344 server_side=True, 345 do_handshake_on_connect=False) 346 self.del_channel() 347 self.set_socket(socket) 348 self._ssl_accepting = True 349 350 def _do_ssl_handshake(self): 351 try: 352 self.socket.do_handshake() 353 except ssl.SSLError as err: 354 if err.args[0] in (ssl.SSL_ERROR_WANT_READ, 355 ssl.SSL_ERROR_WANT_WRITE): 356 return 357 elif err.args[0] == ssl.SSL_ERROR_EOF: 358 return self.handle_close() 359 # TODO: SSLError does not expose alert information 360 elif "SSLV3_ALERT_BAD_CERTIFICATE" in err.args[1]: 361 return self.handle_close() 362 raise 363 except OSError as err: 364 if err.args[0] == errno.ECONNABORTED: 365 return self.handle_close() 366 else: 367 self._ssl_accepting = False 368 369 def _do_ssl_shutdown(self): 370 self._ssl_closing = True 371 try: 372 self.socket = self.socket.unwrap() 373 except ssl.SSLError as err: 374 if err.args[0] in (ssl.SSL_ERROR_WANT_READ, 375 ssl.SSL_ERROR_WANT_WRITE): 376 return 377 except OSError: 378 # Any "socket error" corresponds to a SSL_ERROR_SYSCALL return 379 # from OpenSSL's SSL_shutdown(), corresponding to a 380 # closed socket condition. See also: 381 # http://www.mail-archive.com/[email protected]/msg60710.html 382 pass 383 self._ssl_closing = False 384 if getattr(self, '_ccc', False) is False: 385 super(SSLConnection, self).close() 386 else: 387 pass 388 389 def handle_read_event(self): 390 if self._ssl_accepting: 391 self._do_ssl_handshake() 392 elif self._ssl_closing: 393 self._do_ssl_shutdown() 394 else: 395 super(SSLConnection, self).handle_read_event() 396 397 def handle_write_event(self): 398 if self._ssl_accepting: 399 self._do_ssl_handshake() 400 elif self._ssl_closing: 401 self._do_ssl_shutdown() 402 else: 403 super(SSLConnection, self).handle_write_event() 404 405 def send(self, data): 406 try: 407 return super(SSLConnection, self).send(data) 408 except ssl.SSLError as err: 409 if err.args[0] in (ssl.SSL_ERROR_EOF, ssl.SSL_ERROR_ZERO_RETURN, 410 ssl.SSL_ERROR_WANT_READ, 411 ssl.SSL_ERROR_WANT_WRITE): 412 return 0 413 raise 414 415 def recv(self, buffer_size): 416 try: 417 return super(SSLConnection, self).recv(buffer_size) 418 except ssl.SSLError as err: 419 if err.args[0] in (ssl.SSL_ERROR_WANT_READ, 420 ssl.SSL_ERROR_WANT_WRITE): 421 return b'' 422 if err.args[0] in (ssl.SSL_ERROR_EOF, ssl.SSL_ERROR_ZERO_RETURN): 423 self.handle_close() 424 return b'' 425 raise 426 427 def handle_error(self): 428 default_error_handler() 429 430 def close(self): 431 if (isinstance(self.socket, ssl.SSLSocket) and 432 self.socket._sslobj is not None): 433 self._do_ssl_shutdown() 434 else: 435 super(SSLConnection, self).close() 436 437 438 class DummyTLS_DTPHandler(SSLConnection, DummyDTPHandler): 439 """A DummyDTPHandler subclass supporting TLS/SSL.""" 440 441 def __init__(self, conn, baseclass): 442 DummyDTPHandler.__init__(self, conn, baseclass) 443 if self.baseclass.secure_data_channel: 444 self.secure_connection() 445 446 447 class DummyTLS_FTPHandler(SSLConnection, DummyFTPHandler): 448 """A DummyFTPHandler subclass supporting TLS/SSL.""" 449 450 dtp_handler = DummyTLS_DTPHandler 451 452 def __init__(self, conn, encoding=DEFAULT_ENCODING): 453 DummyFTPHandler.__init__(self, conn, encoding=encoding) 454 self.secure_data_channel = False 455 self._ccc = False 456 457 def cmd_auth(self, line): 458 """Set up secure control channel.""" 459 self.push('234 AUTH TLS successful') 460 self.secure_connection() 461 462 def cmd_ccc(self, line): 463 self.push('220 Reverting back to clear-text') 464 self._ccc = True 465 self._do_ssl_shutdown() 466 467 def cmd_pbsz(self, line): 468 """Negotiate size of buffer for secure data transfer. 469 For TLS/SSL the only valid value for the parameter is '0'. 470 Any other value is accepted but ignored. 471 """ 472 self.push('200 PBSZ=0 successful.') 473 474 def cmd_prot(self, line): 475 """Setup un/secure data channel.""" 476 arg = line.upper() 477 if arg == 'C': 478 self.push('200 Protection set to Clear') 479 self.secure_data_channel = False 480 elif arg == 'P': 481 self.push('200 Protection set to Private') 482 self.secure_data_channel = True 483 else: 484 self.push("502 Unrecognized PROT type (use C or P).") 485 486 487 class DummyTLS_FTPServer(DummyFTPServer): 488 handler = DummyTLS_FTPHandler 489 490 491class TestFTPClass(TestCase): 492 493 def setUp(self, encoding=DEFAULT_ENCODING): 494 self.server = DummyFTPServer((HOST, 0), encoding=encoding) 495 self.server.start() 496 self.client = ftplib.FTP(timeout=TIMEOUT, encoding=encoding) 497 self.client.connect(self.server.host, self.server.port) 498 499 def tearDown(self): 500 self.client.close() 501 self.server.stop() 502 # Explicitly clear the attribute to prevent dangling thread 503 self.server = None 504 asyncore.close_all(ignore_all=True) 505 506 def check_data(self, received, expected): 507 self.assertEqual(len(received), len(expected)) 508 self.assertEqual(received, expected) 509 510 def test_getwelcome(self): 511 self.assertEqual(self.client.getwelcome(), '220 welcome') 512 513 def test_sanitize(self): 514 self.assertEqual(self.client.sanitize('foo'), repr('foo')) 515 self.assertEqual(self.client.sanitize('pass 12345'), repr('pass *****')) 516 self.assertEqual(self.client.sanitize('PASS 12345'), repr('PASS *****')) 517 518 def test_exceptions(self): 519 self.assertRaises(ValueError, self.client.sendcmd, 'echo 40\r\n0') 520 self.assertRaises(ValueError, self.client.sendcmd, 'echo 40\n0') 521 self.assertRaises(ValueError, self.client.sendcmd, 'echo 40\r0') 522 self.assertRaises(ftplib.error_temp, self.client.sendcmd, 'echo 400') 523 self.assertRaises(ftplib.error_temp, self.client.sendcmd, 'echo 499') 524 self.assertRaises(ftplib.error_perm, self.client.sendcmd, 'echo 500') 525 self.assertRaises(ftplib.error_perm, self.client.sendcmd, 'echo 599') 526 self.assertRaises(ftplib.error_proto, self.client.sendcmd, 'echo 999') 527 528 def test_all_errors(self): 529 exceptions = (ftplib.error_reply, ftplib.error_temp, ftplib.error_perm, 530 ftplib.error_proto, ftplib.Error, OSError, 531 EOFError) 532 for x in exceptions: 533 try: 534 raise x('exception not included in all_errors set') 535 except ftplib.all_errors: 536 pass 537 538 def test_set_pasv(self): 539 # passive mode is supposed to be enabled by default 540 self.assertTrue(self.client.passiveserver) 541 self.client.set_pasv(True) 542 self.assertTrue(self.client.passiveserver) 543 self.client.set_pasv(False) 544 self.assertFalse(self.client.passiveserver) 545 546 def test_voidcmd(self): 547 self.client.voidcmd('echo 200') 548 self.client.voidcmd('echo 299') 549 self.assertRaises(ftplib.error_reply, self.client.voidcmd, 'echo 199') 550 self.assertRaises(ftplib.error_reply, self.client.voidcmd, 'echo 300') 551 552 def test_login(self): 553 self.client.login() 554 555 def test_acct(self): 556 self.client.acct('passwd') 557 558 def test_rename(self): 559 self.client.rename('a', 'b') 560 self.server.handler_instance.next_response = '200' 561 self.assertRaises(ftplib.error_reply, self.client.rename, 'a', 'b') 562 563 def test_delete(self): 564 self.client.delete('foo') 565 self.server.handler_instance.next_response = '199' 566 self.assertRaises(ftplib.error_reply, self.client.delete, 'foo') 567 568 def test_size(self): 569 self.client.size('foo') 570 571 def test_mkd(self): 572 dir = self.client.mkd('/foo') 573 self.assertEqual(dir, '/foo') 574 575 def test_rmd(self): 576 self.client.rmd('foo') 577 578 def test_cwd(self): 579 dir = self.client.cwd('/foo') 580 self.assertEqual(dir, '250 cwd ok') 581 582 def test_pwd(self): 583 dir = self.client.pwd() 584 self.assertEqual(dir, 'pwd ok') 585 586 def test_quit(self): 587 self.assertEqual(self.client.quit(), '221 quit ok') 588 # Ensure the connection gets closed; sock attribute should be None 589 self.assertEqual(self.client.sock, None) 590 591 def test_abort(self): 592 self.client.abort() 593 594 def test_retrbinary(self): 595 def callback(data): 596 received.append(data.decode(self.client.encoding)) 597 received = [] 598 self.client.retrbinary('retr', callback) 599 self.check_data(''.join(received), RETR_DATA) 600 601 def test_retrbinary_rest(self): 602 def callback(data): 603 received.append(data.decode(self.client.encoding)) 604 for rest in (0, 10, 20): 605 received = [] 606 self.client.retrbinary('retr', callback, rest=rest) 607 self.check_data(''.join(received), RETR_DATA[rest:]) 608 609 def test_retrlines(self): 610 received = [] 611 self.client.retrlines('retr', received.append) 612 self.check_data(''.join(received), RETR_DATA.replace('\r\n', '')) 613 614 def test_storbinary(self): 615 f = io.BytesIO(RETR_DATA.encode(self.client.encoding)) 616 self.client.storbinary('stor', f) 617 self.check_data(self.server.handler_instance.last_received_data, RETR_DATA) 618 # test new callback arg 619 flag = [] 620 f.seek(0) 621 self.client.storbinary('stor', f, callback=lambda x: flag.append(None)) 622 self.assertTrue(flag) 623 624 def test_storbinary_rest(self): 625 data = RETR_DATA.replace('\r\n', '\n').encode(self.client.encoding) 626 f = io.BytesIO(data) 627 for r in (30, '30'): 628 f.seek(0) 629 self.client.storbinary('stor', f, rest=r) 630 self.assertEqual(self.server.handler_instance.rest, str(r)) 631 632 def test_storlines(self): 633 data = RETR_DATA.replace('\r\n', '\n').encode(self.client.encoding) 634 f = io.BytesIO(data) 635 self.client.storlines('stor', f) 636 self.check_data(self.server.handler_instance.last_received_data, RETR_DATA) 637 # test new callback arg 638 flag = [] 639 f.seek(0) 640 self.client.storlines('stor foo', f, callback=lambda x: flag.append(None)) 641 self.assertTrue(flag) 642 643 f = io.StringIO(RETR_DATA.replace('\r\n', '\n')) 644 # storlines() expects a binary file, not a text file 645 with warnings_helper.check_warnings(('', BytesWarning), quiet=True): 646 self.assertRaises(TypeError, self.client.storlines, 'stor foo', f) 647 648 def test_nlst(self): 649 self.client.nlst() 650 self.assertEqual(self.client.nlst(), NLST_DATA.split('\r\n')[:-1]) 651 652 def test_dir(self): 653 l = [] 654 self.client.dir(lambda x: l.append(x)) 655 self.assertEqual(''.join(l), LIST_DATA.replace('\r\n', '')) 656 657 def test_mlsd(self): 658 list(self.client.mlsd()) 659 list(self.client.mlsd(path='/')) 660 list(self.client.mlsd(path='/', facts=['size', 'type'])) 661 662 ls = list(self.client.mlsd()) 663 for name, facts in ls: 664 self.assertIsInstance(name, str) 665 self.assertIsInstance(facts, dict) 666 self.assertTrue(name) 667 self.assertIn('type', facts) 668 self.assertIn('perm', facts) 669 self.assertIn('unique', facts) 670 671 def set_data(data): 672 self.server.handler_instance.next_data = data 673 674 def test_entry(line, type=None, perm=None, unique=None, name=None): 675 type = 'type' if type is None else type 676 perm = 'perm' if perm is None else perm 677 unique = 'unique' if unique is None else unique 678 name = 'name' if name is None else name 679 set_data(line) 680 _name, facts = next(self.client.mlsd()) 681 self.assertEqual(_name, name) 682 self.assertEqual(facts['type'], type) 683 self.assertEqual(facts['perm'], perm) 684 self.assertEqual(facts['unique'], unique) 685 686 # plain 687 test_entry('type=type;perm=perm;unique=unique; name\r\n') 688 # "=" in fact value 689 test_entry('type=ty=pe;perm=perm;unique=unique; name\r\n', type="ty=pe") 690 test_entry('type==type;perm=perm;unique=unique; name\r\n', type="=type") 691 test_entry('type=t=y=pe;perm=perm;unique=unique; name\r\n', type="t=y=pe") 692 test_entry('type=====;perm=perm;unique=unique; name\r\n', type="====") 693 # spaces in name 694 test_entry('type=type;perm=perm;unique=unique; na me\r\n', name="na me") 695 test_entry('type=type;perm=perm;unique=unique; name \r\n', name="name ") 696 test_entry('type=type;perm=perm;unique=unique; name\r\n', name=" name") 697 test_entry('type=type;perm=perm;unique=unique; n am e\r\n', name="n am e") 698 # ";" in name 699 test_entry('type=type;perm=perm;unique=unique; na;me\r\n', name="na;me") 700 test_entry('type=type;perm=perm;unique=unique; ;name\r\n', name=";name") 701 test_entry('type=type;perm=perm;unique=unique; ;name;\r\n', name=";name;") 702 test_entry('type=type;perm=perm;unique=unique; ;;;;\r\n', name=";;;;") 703 # case sensitiveness 704 set_data('Type=type;TyPe=perm;UNIQUE=unique; name\r\n') 705 _name, facts = next(self.client.mlsd()) 706 for x in facts: 707 self.assertTrue(x.islower()) 708 # no data (directory empty) 709 set_data('') 710 self.assertRaises(StopIteration, next, self.client.mlsd()) 711 set_data('') 712 for x in self.client.mlsd(): 713 self.fail("unexpected data %s" % x) 714 715 def test_makeport(self): 716 with self.client.makeport(): 717 # IPv4 is in use, just make sure send_eprt has not been used 718 self.assertEqual(self.server.handler_instance.last_received_cmd, 719 'port') 720 721 def test_makepasv(self): 722 host, port = self.client.makepasv() 723 conn = socket.create_connection((host, port), timeout=TIMEOUT) 724 conn.close() 725 # IPv4 is in use, just make sure send_epsv has not been used 726 self.assertEqual(self.server.handler_instance.last_received_cmd, 'pasv') 727 728 def test_makepasv_issue43285_security_disabled(self): 729 """Test the opt-in to the old vulnerable behavior.""" 730 self.client.trust_server_pasv_ipv4_address = True 731 bad_host, port = self.client.makepasv() 732 self.assertEqual( 733 bad_host, self.server.handler_instance.fake_pasv_server_ip) 734 # Opening and closing a connection keeps the dummy server happy 735 # instead of timing out on accept. 736 socket.create_connection((self.client.sock.getpeername()[0], port), 737 timeout=TIMEOUT).close() 738 739 def test_makepasv_issue43285_security_enabled_default(self): 740 self.assertFalse(self.client.trust_server_pasv_ipv4_address) 741 trusted_host, port = self.client.makepasv() 742 self.assertNotEqual( 743 trusted_host, self.server.handler_instance.fake_pasv_server_ip) 744 # Opening and closing a connection keeps the dummy server happy 745 # instead of timing out on accept. 746 socket.create_connection((trusted_host, port), timeout=TIMEOUT).close() 747 748 def test_with_statement(self): 749 self.client.quit() 750 751 def is_client_connected(): 752 if self.client.sock is None: 753 return False 754 try: 755 self.client.sendcmd('noop') 756 except (OSError, EOFError): 757 return False 758 return True 759 760 # base test 761 with ftplib.FTP(timeout=TIMEOUT) as self.client: 762 self.client.connect(self.server.host, self.server.port) 763 self.client.sendcmd('noop') 764 self.assertTrue(is_client_connected()) 765 self.assertEqual(self.server.handler_instance.last_received_cmd, 'quit') 766 self.assertFalse(is_client_connected()) 767 768 # QUIT sent inside the with block 769 with ftplib.FTP(timeout=TIMEOUT) as self.client: 770 self.client.connect(self.server.host, self.server.port) 771 self.client.sendcmd('noop') 772 self.client.quit() 773 self.assertEqual(self.server.handler_instance.last_received_cmd, 'quit') 774 self.assertFalse(is_client_connected()) 775 776 # force a wrong response code to be sent on QUIT: error_perm 777 # is expected and the connection is supposed to be closed 778 try: 779 with ftplib.FTP(timeout=TIMEOUT) as self.client: 780 self.client.connect(self.server.host, self.server.port) 781 self.client.sendcmd('noop') 782 self.server.handler_instance.next_response = '550 error on quit' 783 except ftplib.error_perm as err: 784 self.assertEqual(str(err), '550 error on quit') 785 else: 786 self.fail('Exception not raised') 787 # needed to give the threaded server some time to set the attribute 788 # which otherwise would still be == 'noop' 789 time.sleep(0.1) 790 self.assertEqual(self.server.handler_instance.last_received_cmd, 'quit') 791 self.assertFalse(is_client_connected()) 792 793 def test_source_address(self): 794 self.client.quit() 795 port = socket_helper.find_unused_port() 796 try: 797 self.client.connect(self.server.host, self.server.port, 798 source_address=(HOST, port)) 799 self.assertEqual(self.client.sock.getsockname()[1], port) 800 self.client.quit() 801 except OSError as e: 802 if e.errno == errno.EADDRINUSE: 803 self.skipTest("couldn't bind to port %d" % port) 804 raise 805 806 def test_source_address_passive_connection(self): 807 port = socket_helper.find_unused_port() 808 self.client.source_address = (HOST, port) 809 try: 810 with self.client.transfercmd('list') as sock: 811 self.assertEqual(sock.getsockname()[1], port) 812 except OSError as e: 813 if e.errno == errno.EADDRINUSE: 814 self.skipTest("couldn't bind to port %d" % port) 815 raise 816 817 def test_parse257(self): 818 self.assertEqual(ftplib.parse257('257 "/foo/bar"'), '/foo/bar') 819 self.assertEqual(ftplib.parse257('257 "/foo/bar" created'), '/foo/bar') 820 self.assertEqual(ftplib.parse257('257 ""'), '') 821 self.assertEqual(ftplib.parse257('257 "" created'), '') 822 self.assertRaises(ftplib.error_reply, ftplib.parse257, '250 "/foo/bar"') 823 # The 257 response is supposed to include the directory 824 # name and in case it contains embedded double-quotes 825 # they must be doubled (see RFC-959, chapter 7, appendix 2). 826 self.assertEqual(ftplib.parse257('257 "/foo/b""ar"'), '/foo/b"ar') 827 self.assertEqual(ftplib.parse257('257 "/foo/b""ar" created'), '/foo/b"ar') 828 829 def test_line_too_long(self): 830 self.assertRaises(ftplib.Error, self.client.sendcmd, 831 'x' * self.client.maxline * 2) 832 833 def test_retrlines_too_long(self): 834 self.client.sendcmd('SETLONGRETR %d' % (self.client.maxline * 2)) 835 received = [] 836 self.assertRaises(ftplib.Error, 837 self.client.retrlines, 'retr', received.append) 838 839 def test_storlines_too_long(self): 840 f = io.BytesIO(b'x' * self.client.maxline * 2) 841 self.assertRaises(ftplib.Error, self.client.storlines, 'stor', f) 842 843 def test_encoding_param(self): 844 encodings = ['latin-1', 'utf-8'] 845 for encoding in encodings: 846 with self.subTest(encoding=encoding): 847 self.tearDown() 848 self.setUp(encoding=encoding) 849 self.assertEqual(encoding, self.client.encoding) 850 self.test_retrbinary() 851 self.test_storbinary() 852 self.test_retrlines() 853 new_dir = self.client.mkd('/non-ascii dir \xAE') 854 self.check_data(new_dir, '/non-ascii dir \xAE') 855 # Check default encoding 856 client = ftplib.FTP(timeout=TIMEOUT) 857 self.assertEqual(DEFAULT_ENCODING, client.encoding) 858 859 860@skipUnless(socket_helper.IPV6_ENABLED, "IPv6 not enabled") 861class TestIPv6Environment(TestCase): 862 863 def setUp(self): 864 self.server = DummyFTPServer((HOSTv6, 0), 865 af=socket.AF_INET6, 866 encoding=DEFAULT_ENCODING) 867 self.server.start() 868 self.client = ftplib.FTP(timeout=TIMEOUT, encoding=DEFAULT_ENCODING) 869 self.client.connect(self.server.host, self.server.port) 870 871 def tearDown(self): 872 self.client.close() 873 self.server.stop() 874 # Explicitly clear the attribute to prevent dangling thread 875 self.server = None 876 asyncore.close_all(ignore_all=True) 877 878 def test_af(self): 879 self.assertEqual(self.client.af, socket.AF_INET6) 880 881 def test_makeport(self): 882 with self.client.makeport(): 883 self.assertEqual(self.server.handler_instance.last_received_cmd, 884 'eprt') 885 886 def test_makepasv(self): 887 host, port = self.client.makepasv() 888 conn = socket.create_connection((host, port), timeout=TIMEOUT) 889 conn.close() 890 self.assertEqual(self.server.handler_instance.last_received_cmd, 'epsv') 891 892 def test_transfer(self): 893 def retr(): 894 def callback(data): 895 received.append(data.decode(self.client.encoding)) 896 received = [] 897 self.client.retrbinary('retr', callback) 898 self.assertEqual(len(''.join(received)), len(RETR_DATA)) 899 self.assertEqual(''.join(received), RETR_DATA) 900 self.client.set_pasv(True) 901 retr() 902 self.client.set_pasv(False) 903 retr() 904 905 906@skipUnless(ssl, "SSL not available") 907class TestTLS_FTPClassMixin(TestFTPClass): 908 """Repeat TestFTPClass tests starting the TLS layer for both control 909 and data connections first. 910 """ 911 912 def setUp(self, encoding=DEFAULT_ENCODING): 913 self.server = DummyTLS_FTPServer((HOST, 0), encoding=encoding) 914 self.server.start() 915 self.client = ftplib.FTP_TLS(timeout=TIMEOUT, encoding=encoding) 916 self.client.connect(self.server.host, self.server.port) 917 # enable TLS 918 self.client.auth() 919 self.client.prot_p() 920 921 922@skipUnless(ssl, "SSL not available") 923class TestTLS_FTPClass(TestCase): 924 """Specific TLS_FTP class tests.""" 925 926 def setUp(self, encoding=DEFAULT_ENCODING): 927 self.server = DummyTLS_FTPServer((HOST, 0), encoding=encoding) 928 self.server.start() 929 self.client = ftplib.FTP_TLS(timeout=TIMEOUT) 930 self.client.connect(self.server.host, self.server.port) 931 932 def tearDown(self): 933 self.client.close() 934 self.server.stop() 935 # Explicitly clear the attribute to prevent dangling thread 936 self.server = None 937 asyncore.close_all(ignore_all=True) 938 939 def test_control_connection(self): 940 self.assertNotIsInstance(self.client.sock, ssl.SSLSocket) 941 self.client.auth() 942 self.assertIsInstance(self.client.sock, ssl.SSLSocket) 943 944 def test_data_connection(self): 945 # clear text 946 with self.client.transfercmd('list') as sock: 947 self.assertNotIsInstance(sock, ssl.SSLSocket) 948 self.assertEqual(sock.recv(1024), 949 LIST_DATA.encode(self.client.encoding)) 950 self.assertEqual(self.client.voidresp(), "226 transfer complete") 951 952 # secured, after PROT P 953 self.client.prot_p() 954 with self.client.transfercmd('list') as sock: 955 self.assertIsInstance(sock, ssl.SSLSocket) 956 # consume from SSL socket to finalize handshake and avoid 957 # "SSLError [SSL] shutdown while in init" 958 self.assertEqual(sock.recv(1024), 959 LIST_DATA.encode(self.client.encoding)) 960 self.assertEqual(self.client.voidresp(), "226 transfer complete") 961 962 # PROT C is issued, the connection must be in cleartext again 963 self.client.prot_c() 964 with self.client.transfercmd('list') as sock: 965 self.assertNotIsInstance(sock, ssl.SSLSocket) 966 self.assertEqual(sock.recv(1024), 967 LIST_DATA.encode(self.client.encoding)) 968 self.assertEqual(self.client.voidresp(), "226 transfer complete") 969 970 def test_login(self): 971 # login() is supposed to implicitly secure the control connection 972 self.assertNotIsInstance(self.client.sock, ssl.SSLSocket) 973 self.client.login() 974 self.assertIsInstance(self.client.sock, ssl.SSLSocket) 975 # make sure that AUTH TLS doesn't get issued again 976 self.client.login() 977 978 def test_auth_issued_twice(self): 979 self.client.auth() 980 self.assertRaises(ValueError, self.client.auth) 981 982 def test_context(self): 983 self.client.quit() 984 ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) 985 ctx.check_hostname = False 986 ctx.verify_mode = ssl.CERT_NONE 987 self.assertRaises(ValueError, ftplib.FTP_TLS, keyfile=CERTFILE, 988 context=ctx) 989 self.assertRaises(ValueError, ftplib.FTP_TLS, certfile=CERTFILE, 990 context=ctx) 991 self.assertRaises(ValueError, ftplib.FTP_TLS, certfile=CERTFILE, 992 keyfile=CERTFILE, context=ctx) 993 994 self.client = ftplib.FTP_TLS(context=ctx, timeout=TIMEOUT) 995 self.client.connect(self.server.host, self.server.port) 996 self.assertNotIsInstance(self.client.sock, ssl.SSLSocket) 997 self.client.auth() 998 self.assertIs(self.client.sock.context, ctx) 999 self.assertIsInstance(self.client.sock, ssl.SSLSocket) 1000 1001 self.client.prot_p() 1002 with self.client.transfercmd('list') as sock: 1003 self.assertIs(sock.context, ctx) 1004 self.assertIsInstance(sock, ssl.SSLSocket) 1005 1006 def test_ccc(self): 1007 self.assertRaises(ValueError, self.client.ccc) 1008 self.client.login(secure=True) 1009 self.assertIsInstance(self.client.sock, ssl.SSLSocket) 1010 self.client.ccc() 1011 self.assertRaises(ValueError, self.client.sock.unwrap) 1012 1013 @skipUnless(False, "FIXME: bpo-32706") 1014 def test_check_hostname(self): 1015 self.client.quit() 1016 ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) 1017 self.assertEqual(ctx.verify_mode, ssl.CERT_REQUIRED) 1018 self.assertEqual(ctx.check_hostname, True) 1019 ctx.load_verify_locations(CAFILE) 1020 self.client = ftplib.FTP_TLS(context=ctx, timeout=TIMEOUT) 1021 1022 # 127.0.0.1 doesn't match SAN 1023 self.client.connect(self.server.host, self.server.port) 1024 with self.assertRaises(ssl.CertificateError): 1025 self.client.auth() 1026 # exception quits connection 1027 1028 self.client.connect(self.server.host, self.server.port) 1029 self.client.prot_p() 1030 with self.assertRaises(ssl.CertificateError): 1031 with self.client.transfercmd("list") as sock: 1032 pass 1033 self.client.quit() 1034 1035 self.client.connect("localhost", self.server.port) 1036 self.client.auth() 1037 self.client.quit() 1038 1039 self.client.connect("localhost", self.server.port) 1040 self.client.prot_p() 1041 with self.client.transfercmd("list") as sock: 1042 pass 1043 1044 1045class TestTimeouts(TestCase): 1046 1047 def setUp(self): 1048 self.evt = threading.Event() 1049 self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 1050 self.sock.settimeout(20) 1051 self.port = socket_helper.bind_port(self.sock) 1052 self.server_thread = threading.Thread(target=self.server) 1053 self.server_thread.daemon = True 1054 self.server_thread.start() 1055 # Wait for the server to be ready. 1056 self.evt.wait() 1057 self.evt.clear() 1058 self.old_port = ftplib.FTP.port 1059 ftplib.FTP.port = self.port 1060 1061 def tearDown(self): 1062 ftplib.FTP.port = self.old_port 1063 self.server_thread.join() 1064 # Explicitly clear the attribute to prevent dangling thread 1065 self.server_thread = None 1066 1067 def server(self): 1068 # This method sets the evt 3 times: 1069 # 1) when the connection is ready to be accepted. 1070 # 2) when it is safe for the caller to close the connection 1071 # 3) when we have closed the socket 1072 self.sock.listen() 1073 # (1) Signal the caller that we are ready to accept the connection. 1074 self.evt.set() 1075 try: 1076 conn, addr = self.sock.accept() 1077 except TimeoutError: 1078 pass 1079 else: 1080 conn.sendall(b"1 Hola mundo\n") 1081 conn.shutdown(socket.SHUT_WR) 1082 # (2) Signal the caller that it is safe to close the socket. 1083 self.evt.set() 1084 conn.close() 1085 finally: 1086 self.sock.close() 1087 1088 def testTimeoutDefault(self): 1089 # default -- use global socket timeout 1090 self.assertIsNone(socket.getdefaulttimeout()) 1091 socket.setdefaulttimeout(30) 1092 try: 1093 ftp = ftplib.FTP(HOST) 1094 finally: 1095 socket.setdefaulttimeout(None) 1096 self.assertEqual(ftp.sock.gettimeout(), 30) 1097 self.evt.wait() 1098 ftp.close() 1099 1100 def testTimeoutNone(self): 1101 # no timeout -- do not use global socket timeout 1102 self.assertIsNone(socket.getdefaulttimeout()) 1103 socket.setdefaulttimeout(30) 1104 try: 1105 ftp = ftplib.FTP(HOST, timeout=None) 1106 finally: 1107 socket.setdefaulttimeout(None) 1108 self.assertIsNone(ftp.sock.gettimeout()) 1109 self.evt.wait() 1110 ftp.close() 1111 1112 def testTimeoutValue(self): 1113 # a value 1114 ftp = ftplib.FTP(HOST, timeout=30) 1115 self.assertEqual(ftp.sock.gettimeout(), 30) 1116 self.evt.wait() 1117 ftp.close() 1118 1119 # bpo-39259 1120 with self.assertRaises(ValueError): 1121 ftplib.FTP(HOST, timeout=0) 1122 1123 def testTimeoutConnect(self): 1124 ftp = ftplib.FTP() 1125 ftp.connect(HOST, timeout=30) 1126 self.assertEqual(ftp.sock.gettimeout(), 30) 1127 self.evt.wait() 1128 ftp.close() 1129 1130 def testTimeoutDifferentOrder(self): 1131 ftp = ftplib.FTP(timeout=30) 1132 ftp.connect(HOST) 1133 self.assertEqual(ftp.sock.gettimeout(), 30) 1134 self.evt.wait() 1135 ftp.close() 1136 1137 def testTimeoutDirectAccess(self): 1138 ftp = ftplib.FTP() 1139 ftp.timeout = 30 1140 ftp.connect(HOST) 1141 self.assertEqual(ftp.sock.gettimeout(), 30) 1142 self.evt.wait() 1143 ftp.close() 1144 1145 1146class MiscTestCase(TestCase): 1147 def test__all__(self): 1148 not_exported = { 1149 'MSG_OOB', 'FTP_PORT', 'MAXLINE', 'CRLF', 'B_CRLF', 'Error', 1150 'parse150', 'parse227', 'parse229', 'parse257', 'print_line', 1151 'ftpcp', 'test'} 1152 support.check__all__(self, ftplib, not_exported=not_exported) 1153 1154 1155def setUpModule(): 1156 thread_info = threading_helper.threading_setup() 1157 unittest.addModuleCleanup(threading_helper.threading_cleanup, *thread_info) 1158 1159 1160if __name__ == '__main__': 1161 unittest.main() 1162