test script for ftplib module modified by giampaolo rodola to test ftp class ipv6 and tls environment import ftplib import socket import io import errno import os import threading import time import unittest try import ssl except importerror ssl none from unittest import testcase skipunless from test import support from test support import threadinghelper from test support import sockethelper from test support import warningshelper from test support import asynchat from test support import asyncore from test support sockethelper import host hostv6 support requiresworkingsocketmoduletrue timeout support loopbacktimeout defaultencoding utf8 the dummy data returned by server over the data channel when retr list nlst mlsd commands are issued retrdata abcdexb9xb2xb3xa4xa6rn 1000 listdata foornbarrn nonascii char xaern nlstdata foornbarrn nonascii char xaern mlsddata typecdir permel uniquekevo1zf4 testrn typepdir perme uniquekevo1d 3 rn typeos unixslink foobar perm uniquekevo14g4 foobarrn typeos unixchr1329 perm uniquekevo15g4 devicern typeos unixblk11108 perm uniquekevo16g4 blockrn typefile permawr uniquekevo18g4 writablern typedir permcpmel uniquekevo17g4 promiscuousrn typedir perm uniquekevo11t2 noexecrn typefile permr uniquekevo1eg4 two wordsrn typefile permr uniquekevo1ih4 leading spacern typefile permr uniquekevo11g4 file1rn typedir permcpmel uniquekevo17g4 incomingrn typefile permr uniquekevo11g4 file2rn typefile permr uniquekevo11g4 file3rn typefile permr uniquekevo11g4 file4rn typedir permcpmel uniquesgp1 dir xae nonascii charrn typefile permr uniquesgp2 file xae nonascii charrn def defaulterrorhandler bpo44359 silently ignore socket errors such errors occur when a client socket is closed in testftpclass teardown and makepasv tests and the server gets an error on its side pass class dummydtphandlerasynchat asyncchat dtpconnclosed false def initself conn baseclass asynchat asyncchat initself conn self baseclass baseclass self baseclass lastreceiveddata bytearray self encoding baseclass encoding def handlereadself newdata self recv1024 self baseclass lastreceiveddata newdata def handlecloseself xxx this method can be called many times in a row for a single connection including in cleartext nontls mode behaviour witnessed with testdataconnection if not self dtpconnclosed self baseclass push 226 transfer complete self close self dtpconnclosed true def pushself what if self baseclass nextdata is not none what self baseclass nextdata self baseclass nextdata none if not what return self closewhendone superdummydtphandler self pushwhat encodeself encoding def handleerrorself defaulterrorhandler class dummyftphandlerasynchat asyncchat dtphandler dummydtphandler def initself conn encodingdefaultencoding asynchat asyncchat initself conn tells the socket to handle urgent data inline abor command self socket setsockoptsocket solsocket socket sooobinline 1 self setterminatorbrn self inbuffer self dtp none self lastreceivedcmd none self lastreceiveddata bytearray self nextresponse self nextdata none self rest none self nextretrdata retrdata self push 220 welcome self encoding encoding we use this as the string ipv4 address to direct the client to in response to a pasv command to test security behavior https bugs python orgissue43285 self fakepasvserverip 252 253 254 255 def collectincomingdataself data self inbuffer appenddata def foundterminatorself line b joinself inbuffer decodeself encoding self inbuffer if self nextresponse self pushself nextresponse self nextresponse cmd line split 0 lower self lastreceivedcmd cmd space line find if space 1 arg linespace 1 else arg if hasattrself cmd cmd method getattrself cmd cmd methodarg else self push 550 command s not understood cmd def handleerrorself defaulterrorhandler def pushself data asynchat asyncchat pushself data encodeself encoding b rn def cmdportself arg addr listmapint arg split ip d d d d tupleaddr 4 port addr4 256 addr5 s socket createconnectionip port timeouttimeout self dtp self dtphandlers baseclassself self push 200 active data connection established def cmdpasvself arg with socket createserverself socket getsockname0 0 as sock sock settimeouttimeout port sock getsockname1 ip self fakepasvserverip ip ip replace p1 port 256 p2 port 256 self push 227 entering passive mode s d d ip p1 p2 conn addr sock accept self dtp self dtphandlerconn baseclassself def cmdeprtself arg af ip port arg splitarg01 1 port intport s socket createconnectionip port timeouttimeout self dtp self dtphandlers baseclassself self push 200 active data connection established def cmdepsvself arg with socket createserverself socket getsockname0 0 familysocket afinet6 as sock sock settimeouttimeout port sock getsockname1 self push 229 entering extended passive mode d port conn addr sock accept self dtp self dtphandlerconn baseclassself def cmdechoself arg sends back the received string used by the test suite self pusharg def cmdnoopself arg self push 200 noop ok def cmduserself arg self push 331 username ok def cmdpassself arg self push 230 password ok def cmdacctself arg self push 230 acct ok def cmdrnfrself arg self push 350 rnfr ok def cmdrntoself arg self push 250 rnto ok def cmddeleself arg self push 250 dele ok def cmdcwdself arg self push 250 cwd ok def cmdsizeself arg self push 250 1000 def cmdmkdself arg self push 257 s arg def cmdrmdself arg self push 250 rmd ok def cmdpwdself arg self push 257 pwd ok def cmdtypeself arg self push 200 type ok def cmdquitself arg self push 221 quit ok self close def cmdaborself arg self push 226 abor ok def cmdstorself arg self push 125 stor ok def cmdrestself arg self rest arg self push 350 rest ok def cmdretrself arg self push 125 retr ok if self rest is not none offset intself rest else offset 0 self dtp pushself nextretrdataoffset self dtp closewhendone self rest none def cmdlistself arg self push 125 list ok self dtp pushlistdata self dtp closewhendone def cmdnlstself arg self push 125 nlst ok self dtp pushnlstdata self dtp closewhendone def cmdoptsself arg self push 200 opts ok def cmdmlsdself arg self push 125 mlsd ok self dtp pushmlsddata self dtp closewhendone def cmdsetlongretrself arg for testing next retr will return long line self nextretrdata x intarg self push 125 setlongretr ok class dummyftpserverasyncore dispatcher threading thread handler dummyftphandler def initself address afsocket afinet encodingdefaultencoding threading thread initself asyncore dispatcher initself self daemon true self createsocketaf socket sockstream self bindaddress self listen5 self active false self activelock threading lock self host self port self socket getsockname 2 self handlerinstance none self encoding encoding def startself assert not self active self flag threading event threading thread startself self flag wait def runself self active true self flag set while self active and asyncore socketmap self activelock acquire asyncore looptimeout0 1 count1 self activelock release asyncore closeallignorealltrue def stopself assert self active self active false self join def handleacceptedself conn addr self handlerinstance self handlerconn encodingself encoding def handleconnectself self close handleread handleconnect def writableself return 0 def handleerrorself defaulterrorhandler if ssl is not none certfile os path joinos path dirnamefile certdata keycert3 pem cafile os path joinos path dirnamefile certdata pycacert pem class sslconnectionasyncore dispatcher todo sslerror does not expose alert information any socket error corresponds to a sslerrorsyscall return from openssl s sslshutdown corresponding to a closed socket condition see also http www mailarchive comopensslusersopenssl orgmsg60710 html a dummydtphandler subclass supporting tlsssl def initself conn baseclass dummydtphandler initself conn baseclass if self baseclass securedatachannel self secureconnection class dummytlsftphandlersslconnection dummyftphandler set up secure control channel self push 234 auth tls successful self secureconnection def cmdcccself line self push 220 reverting back to cleartext self ccc true self dosslshutdown def cmdpbszself line self push 200 pbsz0 successful def cmdprotself line explicitly clear the attribute to prevent dangling thread passive mode is supposed to be enabled by default ensure the connection gets closed sock attribute should be none test new callback arg test new callback arg storlines expects a binary file not a text file plain in fact value spaces in name in name case sensitiveness no data directory empty ipv4 is in use just make sure sendeprt has not been used ipv4 is in use just make sure sendepsv has not been used test the optin to the old vulnerable behavior self client trustserverpasvipv4address true badhost port self client makepasv self assertequal badhost self server handlerinstance fakepasvserverip opening and closing a connection keeps the dummy server happy instead of timing out on accept socket createconnectionself client sock getpeername0 port timeouttimeout close def testmakepasvissue43285securityenableddefaultself self assertfalseself client trustserverpasvipv4address trustedhost port self client makepasv self assertnotequal trustedhost self server handlerinstance fakepasvserverip opening and closing a connection keeps the dummy server happy instead of timing out on accept socket createconnectiontrustedhost port timeouttimeout close def testwithstatementself self client quit def isclientconnected if self client sock is none return false try self client sendcmd noop except oserror eoferror return false return true base test with ftplib ftptimeouttimeout as self client self client connectself server host self server port self client sendcmd noop self asserttrueisclientconnected self assertequalself server handlerinstance lastreceivedcmd quit self assertfalseisclientconnected quit sent inside the with block with ftplib ftptimeouttimeout as self client self client connectself server host self server port self client sendcmd noop self client quit self assertequalself server handlerinstance lastreceivedcmd quit self assertfalseisclientconnected force a wrong response code to be sent on quit errorperm is expected and the connection is supposed to be closed try with ftplib ftptimeouttimeout as self client self client connectself server host self server port self client sendcmd noop self server handlerinstance nextresponse 550 error on quit except ftplib errorperm as err self assertequalstrerr 550 error on quit else self fail exception not raised needed to give the threaded server some time to set the attribute which otherwise would still be noop time sleep0 1 self assertequalself server handlerinstance lastreceivedcmd quit self assertfalseisclientconnected def testsourceaddressself self client quit port sockethelper findunusedport try self client connectself server host self server port sourceaddresshost port self assertequalself client sock getsockname1 port self client quit except oserror as e if e errno errno eaddrinuse self skiptestcouldn t bind to port d port raise def testsourceaddresspassiveconnectionself port sockethelper findunusedport self client sourceaddress host port try with self client transfercmd list as sock self assertequalsock getsockname1 port except oserror as e if e errno errno eaddrinuse self skiptestcouldn t bind to port d port raise def testparse257self self assertequalftplib parse257 257 foobar foobar self assertequalftplib parse257 257 foobar created foobar self assertequalftplib parse257 257 self assertequalftplib parse257 257 created self assertraisesftplib errorreply ftplib parse257 250 foobar the 257 response is supposed to include the directory name and in case it contains embedded doublequotes they must be doubled see rfc959 chapter 7 appendix 2 self assertequalftplib parse257 257 foobar foobar self assertequalftplib parse257 257 foobar created foobar def testlinetoolongself self assertraisesftplib error self client sendcmd x self client maxline 2 def testretrlinestoolongself self client sendcmd setlongretr d self client maxline 2 received self assertraisesftplib error self client retrlines retr received append def teststorlinestoolongself f io bytesiob x self client maxline 2 self assertraisesftplib error self client storlines stor f def testencodingparamself encodings latin1 utf8 for encoding in encodings with self subtestencodingencoding self teardown self setupencodingencoding self assertequalencoding self client encoding self testretrbinary self teststorbinary self testretrlines newdir self client mkd nonascii dir xae self checkdatanewdir nonascii dir xae check default encoding client ftplib ftptimeouttimeout self assertequaldefaultencoding client encoding skipunlesssockethelper ipv6enabled ipv6 not enabled class testipv6environmenttestcase def setupself self server dummyftpserverhostv6 0 afsocket afinet6 encodingdefaultencoding self server start self client ftplib ftptimeouttimeout encodingdefaultencoding self client connectself server host self server port def teardownself self client close self server stop explicitly clear the attribute to prevent dangling thread self server none asyncore closeallignorealltrue def testafself self assertequalself client af socket afinet6 def testmakeportself with self client makeport self assertequalself server handlerinstance lastreceivedcmd eprt def testmakepasvself host port self client makepasv conn socket createconnectionhost port timeouttimeout conn close self assertequalself server handlerinstance lastreceivedcmd epsv def testtransferself def retr received self client retrbinary retr received append self assertequalb joinreceived retrdata encodeself client encoding self client setpasvtrue retr self client setpasvfalse retr skipunlessssl ssl not available class testtlsftpclassmixintestftpclass def setupself encodingdefaultencoding self server dummytlsftpserverhost 0 encodingencoding self server start self client ftplib ftptlstimeouttimeout encodingencoding self client connectself server host self server port enable tls self client auth self client protp skipunlessssl ssl not available class testtlsftpclasstestcase explicitly clear the attribute to prevent dangling thread clear text secured after prot p consume from ssl socket to finalize handshake and avoid sslerror ssl shutdown while in init prot c is issued the connection must be in cleartext again login is supposed to implicitly secure the control connection make sure that auth tls doesn t get issued again 127 0 0 1 doesn t match san exception quits connection wait for the server to be ready explicitly clear the attribute to prevent dangling thread this method sets the evt 3 times 1 when the connection is ready to be accepted 2 when it is safe for the caller to close the connection 3 when we have closed the socket 1 signal the caller that we are ready to accept the connection 2 signal the caller that it is safe to close the socket default use global socket timeout no timeout do not use global socket timeout a value bpo39259 modified by giampaolo rodola to test ftp class ipv6 and tls environment the dummy data returned by server over the data channel when retr list nlst mlsd commands are issued bpo 44359 silently ignore socket errors such errors occur when a client socket is closed in testftpclass teardown and makepasv tests and the server gets an error on its side xxx this method can be called many times in a row for a single connection including in clear text non tls mode behaviour witnessed with test_data_connection tells the socket to handle urgent data inline abor command we use this as the string ipv4 address to direct the client to in response to a pasv command to test security behavior https bugs python org issue43285 sends back the received string used by the test suite for testing next retr will return long line an asyncore dispatcher subclass supporting tls ssl todo sslerror does not expose alert information any socket error corresponds to a ssl_error_syscall return from openssl s ssl_shutdown corresponding to a closed socket condition see also http www mail archive com openssl users openssl org msg60710 html a dummydtphandler subclass supporting tls ssl a dummyftphandler subclass supporting tls ssl set up secure control channel negotiate size of buffer for secure data transfer for tls ssl the only valid value for the parameter is 0 any other value is accepted but ignored setup un secure data channel explicitly clear the attribute to prevent dangling thread passive mode is supposed to be enabled by default ensure the connection gets closed sock attribute should be none test new callback arg test new callback arg storlines expects a binary file not a text file plain in fact value spaces in name in name case sensitiveness no data directory empty ipv4 is in use just make sure send_eprt has not been used ipv4 is in use just make sure send_epsv has not been used test the opt in to the old vulnerable behavior opening and closing a connection keeps the dummy server happy instead of timing out on accept opening and closing a connection keeps the dummy server happy instead of timing out on accept base test quit sent inside the with block force a wrong response code to be sent on quit error_perm is expected and the connection is supposed to be closed needed to give the threaded server some time to set the attribute which otherwise would still be noop the 257 response is supposed to include the directory name and in case it contains embedded double quotes they must be doubled see rfc 959 chapter 7 appendix 2 check default encoding explicitly clear the attribute to prevent dangling thread repeat testftpclass tests starting the tls layer for both control and data connections first enable tls specific tls_ftp class tests explicitly clear the attribute to prevent dangling thread clear text secured after prot p consume from ssl socket to finalize handshake and avoid sslerror ssl shutdown while in init prot c is issued the connection must be in cleartext again login is supposed to implicitly secure the control connection make sure that auth tls doesn t get issued again 127 0 0 1 doesn t match san exception quits connection wait for the server to be ready explicitly clear the attribute to prevent dangling thread this method sets the evt 3 times 1 when the connection is ready to be accepted 2 when it is safe for the caller to close the connection 3 when we have closed the socket 1 signal the caller that we are ready to accept the connection 2 signal the caller that it is safe to close the socket default use global socket timeout no timeout do not use global socket timeout a value bpo 39259
import ftplib
import socket
import io
import errno
import os
import threading
import time
import unittest
try:
    import ssl
except ImportError:
    ssl = None

from unittest import TestCase, skipUnless
from test import support
from test.support import threading_helper
from test.support import socket_helper
from test.support import warnings_helper
from test.support import asynchat
from test.support import asyncore
from test.support.socket_helper import HOST, HOSTv6


support.requires_working_socket(module=True)

TIMEOUT = support.LOOPBACK_TIMEOUT
DEFAULT_ENCODING = 'utf-8'


RETR_DATA = 'abcde\xB9\xB2\xB3\xA4\xA6\r\n' * 1000
LIST_DATA = 'foo\r\nbar\r\n non-ascii char \xAE\r\n'
NLST_DATA = 'foo\r\nbar\r\n non-ascii char \xAE\r\n'
MLSD_DATA = ("type=cdir;perm=el;unique==keVO1+ZF4; test\r\n"
             "type=pdir;perm=e;unique==keVO1+d?3; ..\r\n"
             "type=OS.unix=slink:/foobar;perm=;unique==keVO1+4G4; foobar\r\n"
             "type=OS.unix=chr-13/29;perm=;unique==keVO1+5G4; device\r\n"
             "type=OS.unix=blk-11/108;perm=;unique==keVO1+6G4; block\r\n"
             "type=file;perm=awr;unique==keVO1+8G4; writable\r\n"
             "type=dir;perm=cpmel;unique==keVO1+7G4; promiscuous\r\n"
             "type=dir;perm=;unique==keVO1+1t2; no-exec\r\n"
             "type=file;perm=r;unique==keVO1+EG4; two words\r\n"
             "type=file;perm=r;unique==keVO1+IH4;  leading space\r\n"
             "type=file;perm=r;unique==keVO1+1G4; file1\r\n"
             "type=dir;perm=cpmel;unique==keVO1+7G4; incoming\r\n"
             "type=file;perm=r;unique==keVO1+1G4; file2\r\n"
             "type=file;perm=r;unique==keVO1+1G4; file3\r\n"
             "type=file;perm=r;unique==keVO1+1G4; file4\r\n"
             "type=dir;perm=cpmel;unique==SGP1; dir \xAE non-ascii char\r\n"
             "type=file;perm=r;unique==SGP2; file \xAE non-ascii char\r\n")


def default_error_handler():
    
    
    
    pass


class DummyDTPHandler(asynchat.async_chat):
    dtp_conn_closed = False

    def __init__(self, conn, baseclass):
        asynchat.async_chat.__init__(self, conn)
        self.baseclass = baseclass
        self.baseclass.last_received_data = bytearray()
        self.encoding = baseclass.encoding

    def handle_read(self):
        new_data = self.recv(1024)
        self.baseclass.last_received_data += new_data

    def handle_close(self):
        
        
        
        if not self.dtp_conn_closed:
            self.baseclass.push('226 transfer complete')
            self.close()
            self.dtp_conn_closed = True

    def push(self, what):
        if self.baseclass.next_data is not None:
            what = self.baseclass.next_data
            self.baseclass.next_data = None
        if not what:
            return self.close_when_done()
        super(DummyDTPHandler, self).push(what.encode(self.encoding))

    def handle_error(self):
        default_error_handler()


class DummyFTPHandler(asynchat.async_chat):

    dtp_handler = DummyDTPHandler

    def __init__(self, conn, encoding=DEFAULT_ENCODING):
        asynchat.async_chat.__init__(self, conn)
        
        self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_OOBINLINE, 1)
        self.set_terminator(b"\r\n")
        self.in_buffer = []
        self.dtp = None
        self.last_received_cmd = None
        self.last_received_data = bytearray()
        self.next_response = ''
        self.next_data = None
        self.rest = None
        self.next_retr_data = RETR_DATA
        self.push('220 welcome')
        self.encoding = encoding
        
        
        
        self.fake_pasv_server_ip = '252.253.254.255'

    def collect_incoming_data(self, data):
        self.in_buffer.append(data)

    def found_terminator(self):
        line = b''.join(self.in_buffer).decode(self.encoding)
        self.in_buffer = []
        if self.next_response:
            self.push(self.next_response)
            self.next_response = ''
        cmd = line.split(' ')[0].lower()
        self.last_received_cmd = cmd
        space = line.find(' ')
        if space != -1:
            arg = line[space + 1:]
        else:
            arg = ""
        if hasattr(self, 'cmd_' + cmd):
            method = getattr(self, 'cmd_' + cmd)
            method(arg)
        else:
            self.push('550 command "%s" not understood.' %cmd)

    def handle_error(self):
        default_error_handler()

    def push(self, data):
        asynchat.async_chat.push(self, data.encode(self.encoding) + b'\r\n')

    def cmd_port(self, arg):
        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=TIMEOUT)
        self.dtp = self.dtp_handler(s, baseclass=self)
        self.push('200 active data connection established')

    def cmd_pasv(self, arg):
        with socket.create_server((self.socket.getsockname()[0], 0)) as sock:
            sock.settimeout(TIMEOUT)
            port = sock.getsockname()[1]
            ip = self.fake_pasv_server_ip
            ip = ip.replace('.', ','); p1 = port / 256; p2 = port % 256
            self.push('227 entering passive mode (%s,%d,%d)' %(ip, p1, p2))
            conn, addr = sock.accept()
            self.dtp = self.dtp_handler(conn, baseclass=self)

    def cmd_eprt(self, arg):
        af, ip, port = arg.split(arg[0])[1:-1]
        port = int(port)
        s = socket.create_connection((ip, port), timeout=TIMEOUT)
        self.dtp = self.dtp_handler(s, baseclass=self)
        self.push('200 active data connection established')

    def cmd_epsv(self, arg):
        with socket.create_server((self.socket.getsockname()[0], 0),
                                  family=socket.AF_INET6) as sock:
            sock.settimeout(TIMEOUT)
            port = sock.getsockname()[1]
            self.push('229 entering extended passive mode (|||%d|)' %port)
            conn, addr = sock.accept()
            self.dtp = self.dtp_handler(conn, baseclass=self)

    def cmd_echo(self, arg):
        
        self.push(arg)

    def cmd_noop(self, arg):
        self.push('200 noop ok')

    def cmd_user(self, arg):
        self.push('331 username ok')

    def cmd_pass(self, arg):
        self.push('230 password ok')

    def cmd_acct(self, arg):
        self.push('230 acct ok')

    def cmd_rnfr(self, arg):
        self.push('350 rnfr ok')

    def cmd_rnto(self, arg):
        self.push('250 rnto ok')

    def cmd_dele(self, arg):
        self.push('250 dele ok')

    def cmd_cwd(self, arg):
        self.push('250 cwd ok')

    def cmd_size(self, arg):
        self.push('250 1000')

    def cmd_mkd(self, arg):
        self.push('257 "%s"' %arg)

    def cmd_rmd(self, arg):
        self.push('250 rmd ok')

    def cmd_pwd(self, arg):
        self.push('257 "pwd ok"')

    def cmd_type(self, arg):
        self.push('200 type ok')

    def cmd_quit(self, arg):
        self.push('221 quit ok')
        self.close()

    def cmd_abor(self, arg):
        self.push('226 abor ok')

    def cmd_stor(self, arg):
        self.push('125 stor ok')

    def cmd_rest(self, arg):
        self.rest = arg
        self.push('350 rest ok')

    def cmd_retr(self, arg):
        self.push('125 retr ok')
        if self.rest is not None:
            offset = int(self.rest)
        else:
            offset = 0
        self.dtp.push(self.next_retr_data[offset:])
        self.dtp.close_when_done()
        self.rest = None

    def cmd_list(self, arg):
        self.push('125 list ok')
        self.dtp.push(LIST_DATA)
        self.dtp.close_when_done()

    def cmd_nlst(self, arg):
        self.push('125 nlst ok')
        self.dtp.push(NLST_DATA)
        self.dtp.close_when_done()

    def cmd_opts(self, arg):
        self.push('200 opts ok')

    def cmd_mlsd(self, arg):
        self.push('125 mlsd ok')
        self.dtp.push(MLSD_DATA)
        self.dtp.close_when_done()

    def cmd_setlongretr(self, arg):
        
        self.next_retr_data = 'x' * int(arg)
        self.push('125 setlongretr ok')


class DummyFTPServer(asyncore.dispatcher, threading.Thread):

    handler = DummyFTPHandler

    def __init__(self, address, af=socket.AF_INET, encoding=DEFAULT_ENCODING):
        threading.Thread.__init__(self)
        asyncore.dispatcher.__init__(self)
        self.daemon = True
        self.create_socket(af, socket.SOCK_STREAM)
        self.bind(address)
        self.listen(5)
        self.active = False
        self.active_lock = threading.Lock()
        self.host, self.port = self.socket.getsockname()[:2]
        self.handler_instance = None
        self.encoding = encoding

    def start(self):
        assert not self.active
        self.__flag = threading.Event()
        threading.Thread.start(self)
        self.__flag.wait()

    def run(self):
        self.active = True
        self.__flag.set()
        while self.active and asyncore.socket_map:
            self.active_lock.acquire()
            asyncore.loop(timeout=0.1, count=1)
            self.active_lock.release()
        asyncore.close_all(ignore_all=True)

    def stop(self):
        assert self.active
        self.active = False
        self.join()

    def handle_accepted(self, conn, addr):
        self.handler_instance = self.handler(conn, encoding=self.encoding)

    def handle_connect(self):
        self.close()
    handle_read = handle_connect

    def writable(self):
        return 0

    def handle_error(self):
        default_error_handler()


if ssl is not None:

    CERTFILE = os.path.join(os.path.dirname(__file__), "certdata", "keycert3.pem")
    CAFILE = os.path.join(os.path.dirname(__file__), "certdata", "pycacert.pem")

    class SSLConnection(asyncore.dispatcher):
        

        _ssl_accepting = False
        _ssl_closing = False

        def secure_connection(self):
            context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
            context.load_cert_chain(CERTFILE)
            socket = context.wrap_socket(self.socket,
                                         suppress_ragged_eofs=False,
                                         server_side=True,
                                         do_handshake_on_connect=False)
            self.del_channel()
            self.set_socket(socket)
            self._ssl_accepting = True

        def _do_ssl_handshake(self):
            try:
                self.socket.do_handshake()
            except ssl.SSLError as err:
                if err.args[0] in (ssl.SSL_ERROR_WANT_READ,
                                   ssl.SSL_ERROR_WANT_WRITE):
                    return
                elif err.args[0] == ssl.SSL_ERROR_EOF:
                    return self.handle_close()
                
                elif "SSLV3_ALERT_BAD_CERTIFICATE" in err.args[1]:
                    return self.handle_close()
                raise
            except OSError as err:
                if err.args[0] == errno.ECONNABORTED:
                    return self.handle_close()
            else:
                self._ssl_accepting = False

        def _do_ssl_shutdown(self):
            self._ssl_closing = True
            try:
                self.socket = self.socket.unwrap()
            except ssl.SSLError as err:
                if err.args[0] in (ssl.SSL_ERROR_WANT_READ,
                                   ssl.SSL_ERROR_WANT_WRITE):
                    return
            except OSError:
                
                
                
                
                pass
            self._ssl_closing = False
            if getattr(self, '_ccc', False) is False:
                super(SSLConnection, self).close()
            else:
                pass

        def handle_read_event(self):
            if self._ssl_accepting:
                self._do_ssl_handshake()
            elif self._ssl_closing:
                self._do_ssl_shutdown()
            else:
                super(SSLConnection, self).handle_read_event()

        def handle_write_event(self):
            if self._ssl_accepting:
                self._do_ssl_handshake()
            elif self._ssl_closing:
                self._do_ssl_shutdown()
            else:
                super(SSLConnection, self).handle_write_event()

        def send(self, data):
            try:
                return super(SSLConnection, self).send(data)
            except ssl.SSLError as err:
                if err.args[0] in (ssl.SSL_ERROR_EOF, ssl.SSL_ERROR_ZERO_RETURN,
                                   ssl.SSL_ERROR_WANT_READ,
                                   ssl.SSL_ERROR_WANT_WRITE):
                    return 0
                raise

        def recv(self, buffer_size):
            try:
                return super(SSLConnection, self).recv(buffer_size)
            except ssl.SSLError as err:
                if err.args[0] in (ssl.SSL_ERROR_WANT_READ,
                                   ssl.SSL_ERROR_WANT_WRITE):
                    return b''
                if err.args[0] in (ssl.SSL_ERROR_EOF, ssl.SSL_ERROR_ZERO_RETURN):
                    self.handle_close()
                    return b''
                raise

        def handle_error(self):
            default_error_handler()

        def close(self):
            if (isinstance(self.socket, ssl.SSLSocket) and
                    self.socket._sslobj is not None):
                self._do_ssl_shutdown()
            else:
                super(SSLConnection, self).close()


    class DummyTLS_DTPHandler(SSLConnection, DummyDTPHandler):
        

        def __init__(self, conn, baseclass):
            DummyDTPHandler.__init__(self, conn, baseclass)
            if self.baseclass.secure_data_channel:
                self.secure_connection()


    class DummyTLS_FTPHandler(SSLConnection, DummyFTPHandler):
        

        dtp_handler = DummyTLS_DTPHandler

        def __init__(self, conn, encoding=DEFAULT_ENCODING):
            DummyFTPHandler.__init__(self, conn, encoding=encoding)
            self.secure_data_channel = False
            self._ccc = False

        def cmd_auth(self, line):
            
            self.push('234 AUTH TLS successful')
            self.secure_connection()

        def cmd_ccc(self, line):
            self.push('220 Reverting back to clear-text')
            self._ccc = True
            self._do_ssl_shutdown()

        def cmd_pbsz(self, line):
            
            self.push('200 PBSZ=0 successful.')

        def cmd_prot(self, line):
            
            arg = line.upper()
            if arg == 'C':
                self.push('200 Protection set to Clear')
                self.secure_data_channel = False
            elif arg == 'P':
                self.push('200 Protection set to Private')
                self.secure_data_channel = True
            else:
                self.push("502 Unrecognized PROT type (use C or P).")


    class DummyTLS_FTPServer(DummyFTPServer):
        handler = DummyTLS_FTPHandler


class TestFTPClass(TestCase):

    def setUp(self, encoding=DEFAULT_ENCODING):
        self.server = DummyFTPServer((HOST, 0), encoding=encoding)
        self.server.start()
        self.client = ftplib.FTP(timeout=TIMEOUT, encoding=encoding)
        self.client.connect(self.server.host, self.server.port)

    def tearDown(self):
        self.client.close()
        self.server.stop()
        
        self.server = None
        asyncore.close_all(ignore_all=True)

    def check_data(self, received, expected):
        self.assertEqual(len(received), len(expected))
        self.assertEqual(received, expected)

    def test_getwelcome(self):
        self.assertEqual(self.client.getwelcome(), '220 welcome')

    def test_sanitize(self):
        self.assertEqual(self.client.sanitize('foo'), repr('foo'))
        self.assertEqual(self.client.sanitize('pass 12345'), repr('pass *****'))
        self.assertEqual(self.client.sanitize('PASS 12345'), repr('PASS *****'))

    def test_exceptions(self):
        self.assertRaises(ValueError, self.client.sendcmd, 'echo 40\r\n0')
        self.assertRaises(ValueError, self.client.sendcmd, 'echo 40\n0')
        self.assertRaises(ValueError, self.client.sendcmd, 'echo 40\r0')
        self.assertRaises(ftplib.error_temp, self.client.sendcmd, 'echo 400')
        self.assertRaises(ftplib.error_temp, self.client.sendcmd, 'echo 499')
        self.assertRaises(ftplib.error_perm, self.client.sendcmd, 'echo 500')
        self.assertRaises(ftplib.error_perm, self.client.sendcmd, 'echo 599')
        self.assertRaises(ftplib.error_proto, self.client.sendcmd, 'echo 999')

    def test_all_errors(self):
        exceptions = (ftplib.error_reply, ftplib.error_temp, ftplib.error_perm,
                      ftplib.error_proto, ftplib.Error, OSError,
                      EOFError)
        for x in exceptions:
            try:
                raise x('exception not included in all_errors set')
            except ftplib.all_errors:
                pass

    def test_set_pasv(self):
        
        self.assertTrue(self.client.passiveserver)
        self.client.set_pasv(True)
        self.assertTrue(self.client.passiveserver)
        self.client.set_pasv(False)
        self.assertFalse(self.client.passiveserver)

    def test_voidcmd(self):
        self.client.voidcmd('echo 200')
        self.client.voidcmd('echo 299')
        self.assertRaises(ftplib.error_reply, self.client.voidcmd, 'echo 199')
        self.assertRaises(ftplib.error_reply, self.client.voidcmd, 'echo 300')

    def test_login(self):
        self.client.login()

    def test_acct(self):
        self.client.acct('passwd')

    def test_rename(self):
        self.client.rename('a', 'b')
        self.server.handler_instance.next_response = '200'
        self.assertRaises(ftplib.error_reply, self.client.rename, 'a', 'b')

    def test_delete(self):
        self.client.delete('foo')
        self.server.handler_instance.next_response = '199'
        self.assertRaises(ftplib.error_reply, self.client.delete, 'foo')

    def test_size(self):
        self.client.size('foo')

    def test_mkd(self):
        dir = self.client.mkd('/foo')
        self.assertEqual(dir, '/foo')

    def test_rmd(self):
        self.client.rmd('foo')

    def test_cwd(self):
        dir = self.client.cwd('/foo')
        self.assertEqual(dir, '250 cwd ok')

    def test_pwd(self):
        dir = self.client.pwd()
        self.assertEqual(dir, 'pwd ok')

    def test_quit(self):
        self.assertEqual(self.client.quit(), '221 quit ok')
        
        self.assertEqual(self.client.sock, None)

    def test_abort(self):
        self.client.abort()

    def test_retrbinary(self):
        received = []
        self.client.retrbinary('retr', received.append)
        self.check_data(b''.join(received),
                        RETR_DATA.encode(self.client.encoding))

    def test_retrbinary_rest(self):
        for rest in (0, 10, 20):
            received = []
            self.client.retrbinary('retr', received.append, rest=rest)
            self.check_data(b''.join(received),
                            RETR_DATA[rest:].encode(self.client.encoding))

    def test_retrlines(self):
        received = []
        self.client.retrlines('retr', received.append)
        self.check_data(''.join(received), RETR_DATA.replace('\r\n', ''))

    def test_storbinary(self):
        f = io.BytesIO(RETR_DATA.encode(self.client.encoding))
        self.client.storbinary('stor', f)
        self.check_data(self.server.handler_instance.last_received_data,
                        RETR_DATA.encode(self.server.encoding))
        
        flag = []
        f.seek(0)
        self.client.storbinary('stor', f, callback=lambda x: flag.append(None))
        self.assertTrue(flag)

    def test_storbinary_rest(self):
        data = RETR_DATA.replace('\r\n', '\n').encode(self.client.encoding)
        f = io.BytesIO(data)
        for r in (30, '30'):
            f.seek(0)
            self.client.storbinary('stor', f, rest=r)
            self.assertEqual(self.server.handler_instance.rest, str(r))

    def test_storlines(self):
        data = RETR_DATA.replace('\r\n', '\n').encode(self.client.encoding)
        f = io.BytesIO(data)
        self.client.storlines('stor', f)
        self.check_data(self.server.handler_instance.last_received_data,
                        RETR_DATA.encode(self.server.encoding))
        
        flag = []
        f.seek(0)
        self.client.storlines('stor foo', f, callback=lambda x: flag.append(None))
        self.assertTrue(flag)

        f = io.StringIO(RETR_DATA.replace('\r\n', '\n'))
        
        with warnings_helper.check_warnings(('', BytesWarning), quiet=True):
            self.assertRaises(TypeError, self.client.storlines, 'stor foo', f)

    def test_nlst(self):
        self.client.nlst()
        self.assertEqual(self.client.nlst(), NLST_DATA.split('\r\n')[:-1])

    def test_dir(self):
        l = []
        self.client.dir(l.append)
        self.assertEqual(''.join(l), LIST_DATA.replace('\r\n', ''))

    def test_mlsd(self):
        list(self.client.mlsd())
        list(self.client.mlsd(path='/'))
        list(self.client.mlsd(path='/', facts=['size', 'type']))

        ls = list(self.client.mlsd())
        for name, facts in ls:
            self.assertIsInstance(name, str)
            self.assertIsInstance(facts, dict)
            self.assertTrue(name)
            self.assertIn('type', facts)
            self.assertIn('perm', facts)
            self.assertIn('unique', facts)

        def set_data(data):
            self.server.handler_instance.next_data = data

        def test_entry(line, type=None, perm=None, unique=None, name=None):
            type = 'type' if type is None else type
            perm = 'perm' if perm is None else perm
            unique = 'unique' if unique is None else unique
            name = 'name' if name is None else name
            set_data(line)
            _name, facts = next(self.client.mlsd())
            self.assertEqual(_name, name)
            self.assertEqual(facts['type'], type)
            self.assertEqual(facts['perm'], perm)
            self.assertEqual(facts['unique'], unique)

        
        test_entry('type=type;perm=perm;unique=unique; name\r\n')
        
        test_entry('type=ty=pe;perm=perm;unique=unique; name\r\n', type="ty=pe")
        test_entry('type==type;perm=perm;unique=unique; name\r\n', type="=type")
        test_entry('type=t=y=pe;perm=perm;unique=unique; name\r\n', type="t=y=pe")
        test_entry('type=====;perm=perm;unique=unique; name\r\n', type="====")
        
        test_entry('type=type;perm=perm;unique=unique; na me\r\n', name="na me")
        test_entry('type=type;perm=perm;unique=unique; name \r\n', name="name ")
        test_entry('type=type;perm=perm;unique=unique;  name\r\n', name=" name")
        test_entry('type=type;perm=perm;unique=unique; n am  e\r\n', name="n am  e")
        
        test_entry('type=type;perm=perm;unique=unique; na;me\r\n', name="na;me")
        test_entry('type=type;perm=perm;unique=unique; ;name\r\n', name=";name")
        test_entry('type=type;perm=perm;unique=unique; ;name;\r\n', name=";name;")
        test_entry('type=type;perm=perm;unique=unique; ;;;;\r\n', name=";;;;")
        
        set_data('Type=type;TyPe=perm;UNIQUE=unique; name\r\n')
        _name, facts = next(self.client.mlsd())
        for x in facts:
            self.assertTrue(x.islower())
        
        set_data('')
        self.assertRaises(StopIteration, next, self.client.mlsd())
        set_data('')
        for x in self.client.mlsd():
            self.fail("unexpected data %s" % x)

    def test_makeport(self):
        with self.client.makeport():
            
            self.assertEqual(self.server.handler_instance.last_received_cmd,
                                'port')

    def test_makepasv(self):
        host, port = self.client.makepasv()
        conn = socket.create_connection((host, port), timeout=TIMEOUT)
        conn.close()
        
        self.assertEqual(self.server.handler_instance.last_received_cmd, 'pasv')

    def test_makepasv_issue43285_security_disabled(self):
        
        self.client.trust_server_pasv_ipv4_address = True
        bad_host, port = self.client.makepasv()
        self.assertEqual(
                bad_host, self.server.handler_instance.fake_pasv_server_ip)
        
        
        socket.create_connection((self.client.sock.getpeername()[0], port),
                                 timeout=TIMEOUT).close()

    def test_makepasv_issue43285_security_enabled_default(self):
        self.assertFalse(self.client.trust_server_pasv_ipv4_address)
        trusted_host, port = self.client.makepasv()
        self.assertNotEqual(
                trusted_host, self.server.handler_instance.fake_pasv_server_ip)
        
        
        socket.create_connection((trusted_host, port), timeout=TIMEOUT).close()

    def test_with_statement(self):
        self.client.quit()

        def is_client_connected():
            if self.client.sock is None:
                return False
            try:
                self.client.sendcmd('noop')
            except (OSError, EOFError):
                return False
            return True

        
        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())
        self.assertEqual(self.server.handler_instance.last_received_cmd, 'quit')
        self.assertFalse(is_client_connected())

        
        with ftplib.FTP(timeout=TIMEOUT) as self.client:
            self.client.connect(self.server.host, self.server.port)
            self.client.sendcmd('noop')
            self.client.quit()
        self.assertEqual(self.server.handler_instance.last_received_cmd, 'quit')
        self.assertFalse(is_client_connected())

        
        
        try:
            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'
        except ftplib.error_perm as err:
            self.assertEqual(str(err), '550 error on quit')
        else:
            self.fail('Exception not raised')
        
        
        time.sleep(0.1)
        self.assertEqual(self.server.handler_instance.last_received_cmd, 'quit')
        self.assertFalse(is_client_connected())

    def test_source_address(self):
        self.client.quit()
        port = socket_helper.find_unused_port()
        try:
            self.client.connect(self.server.host, self.server.port,
                                source_address=(HOST, port))
            self.assertEqual(self.client.sock.getsockname()[1], port)
            self.client.quit()
        except OSError as e:
            if e.errno == errno.EADDRINUSE:
                self.skipTest("couldn't bind to port %d" % port)
            raise

    def test_source_address_passive_connection(self):
        port = socket_helper.find_unused_port()
        self.client.source_address = (HOST, port)
        try:
            with self.client.transfercmd('list') as sock:
                self.assertEqual(sock.getsockname()[1], port)
        except OSError as e:
            if e.errno == errno.EADDRINUSE:
                self.skipTest("couldn't bind to port %d" % port)
            raise

    def test_parse257(self):
        self.assertEqual(ftplib.parse257('257 "/foo/bar"'), '/foo/bar')
        self.assertEqual(ftplib.parse257('257 "/foo/bar" created'), '/foo/bar')
        self.assertEqual(ftplib.parse257('257 ""'), '')
        self.assertEqual(ftplib.parse257('257 "" created'), '')
        self.assertRaises(ftplib.error_reply, ftplib.parse257, '250 "/foo/bar"')
        
        
        
        self.assertEqual(ftplib.parse257('257 "/foo/b""ar"'), '/foo/b"ar')
        self.assertEqual(ftplib.parse257('257 "/foo/b""ar" created'), '/foo/b"ar')

    def test_line_too_long(self):
        self.assertRaises(ftplib.Error, self.client.sendcmd,
                          'x' * self.client.maxline * 2)

    def test_retrlines_too_long(self):
        self.client.sendcmd('SETLONGRETR %d' % (self.client.maxline * 2))
        received = []
        self.assertRaises(ftplib.Error,
                          self.client.retrlines, 'retr', received.append)

    def test_storlines_too_long(self):
        f = io.BytesIO(b'x' * self.client.maxline * 2)
        self.assertRaises(ftplib.Error, self.client.storlines, 'stor', f)

    def test_encoding_param(self):
        encodings = ['latin-1', 'utf-8']
        for encoding in encodings:
            with self.subTest(encoding=encoding):
                self.tearDown()
                self.setUp(encoding=encoding)
                self.assertEqual(encoding, self.client.encoding)
                self.test_retrbinary()
                self.test_storbinary()
                self.test_retrlines()
                new_dir = self.client.mkd('/non-ascii dir \xAE')
                self.check_data(new_dir, '/non-ascii dir \xAE')
        
        client = ftplib.FTP(timeout=TIMEOUT)
        self.assertEqual(DEFAULT_ENCODING, client.encoding)


@skipUnless(socket_helper.IPV6_ENABLED, "IPv6 not enabled")
class TestIPv6Environment(TestCase):

    def setUp(self):
        self.server = DummyFTPServer((HOSTv6, 0),
                                     af=socket.AF_INET6,
                                     encoding=DEFAULT_ENCODING)
        self.server.start()
        self.client = ftplib.FTP(timeout=TIMEOUT, encoding=DEFAULT_ENCODING)
        self.client.connect(self.server.host, self.server.port)

    def tearDown(self):
        self.client.close()
        self.server.stop()
        
        self.server = None
        asyncore.close_all(ignore_all=True)

    def test_af(self):
        self.assertEqual(self.client.af, socket.AF_INET6)

    def test_makeport(self):
        with self.client.makeport():
            self.assertEqual(self.server.handler_instance.last_received_cmd,
                                'eprt')

    def test_makepasv(self):
        host, port = self.client.makepasv()
        conn = socket.create_connection((host, port), timeout=TIMEOUT)
        conn.close()
        self.assertEqual(self.server.handler_instance.last_received_cmd, 'epsv')

    def test_transfer(self):
        def retr():
            received = []
            self.client.retrbinary('retr', received.append)
            self.assertEqual(b''.join(received),
                             RETR_DATA.encode(self.client.encoding))
        self.client.set_pasv(True)
        retr()
        self.client.set_pasv(False)
        retr()


@skipUnless(ssl, "SSL not available")
class TestTLS_FTPClassMixin(TestFTPClass):
    

    def setUp(self, encoding=DEFAULT_ENCODING):
        self.server = DummyTLS_FTPServer((HOST, 0), encoding=encoding)
        self.server.start()
        self.client = ftplib.FTP_TLS(timeout=TIMEOUT, encoding=encoding)
        self.client.connect(self.server.host, self.server.port)
        
        self.client.auth()
        self.client.prot_p()


@skipUnless(ssl, "SSL not available")
class TestTLS_FTPClass(TestCase):
    

    def setUp(self, encoding=DEFAULT_ENCODING):
        self.server = DummyTLS_FTPServer((HOST, 0), encoding=encoding)
        self.server.start()
        self.client = ftplib.FTP_TLS(timeout=TIMEOUT)
        self.client.connect(self.server.host, self.server.port)

    def tearDown(self):
        self.client.close()
        self.server.stop()
        
        self.server = None
        asyncore.close_all(ignore_all=True)

    def test_control_connection(self):
        self.assertNotIsInstance(self.client.sock, ssl.SSLSocket)
        self.client.auth()
        self.assertIsInstance(self.client.sock, ssl.SSLSocket)

    def test_data_connection(self):
        
        with self.client.transfercmd('list') as sock:
            self.assertNotIsInstance(sock, ssl.SSLSocket)
            self.assertEqual(sock.recv(1024),
                             LIST_DATA.encode(self.client.encoding))
        self.assertEqual(self.client.voidresp(), "226 transfer complete")

        
        self.client.prot_p()
        with self.client.transfercmd('list') as sock:
            self.assertIsInstance(sock, ssl.SSLSocket)
            
            
            self.assertEqual(sock.recv(1024),
                             LIST_DATA.encode(self.client.encoding))
        self.assertEqual(self.client.voidresp(), "226 transfer complete")

        
        self.client.prot_c()
        with self.client.transfercmd('list') as sock:
            self.assertNotIsInstance(sock, ssl.SSLSocket)
            self.assertEqual(sock.recv(1024),
                             LIST_DATA.encode(self.client.encoding))
        self.assertEqual(self.client.voidresp(), "226 transfer complete")

    def test_login(self):
        
        self.assertNotIsInstance(self.client.sock, ssl.SSLSocket)
        self.client.login()
        self.assertIsInstance(self.client.sock, ssl.SSLSocket)
        
        self.client.login()

    def test_auth_issued_twice(self):
        self.client.auth()
        self.assertRaises(ValueError, self.client.auth)

    def test_context(self):
        self.client.quit()
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        self.assertRaises(TypeError, ftplib.FTP_TLS, keyfile=CERTFILE,
                          context=ctx)
        self.assertRaises(TypeError, ftplib.FTP_TLS, certfile=CERTFILE,
                          context=ctx)
        self.assertRaises(TypeError, ftplib.FTP_TLS, certfile=CERTFILE,
                          keyfile=CERTFILE, context=ctx)

        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()
        self.assertIs(self.client.sock.context, ctx)
        self.assertIsInstance(self.client.sock, ssl.SSLSocket)

        self.client.prot_p()
        with self.client.transfercmd('list') as sock:
            self.assertIs(sock.context, ctx)
            self.assertIsInstance(sock, ssl.SSLSocket)

    def test_ccc(self):
        self.assertRaises(ValueError, self.client.ccc)
        self.client.login(secure=True)
        self.assertIsInstance(self.client.sock, ssl.SSLSocket)
        self.client.ccc()
        self.assertRaises(ValueError, self.client.sock.unwrap)

    @skipUnless(False, "FIXME: bpo-32706")
    def test_check_hostname(self):
        self.client.quit()
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        self.assertEqual(ctx.verify_mode, ssl.CERT_REQUIRED)
        self.assertEqual(ctx.check_hostname, True)
        ctx.load_verify_locations(CAFILE)
        self.client = ftplib.FTP_TLS(context=ctx, timeout=TIMEOUT)

        
        self.client.connect(self.server.host, self.server.port)
        with self.assertRaises(ssl.CertificateError):
            self.client.auth()
        

        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):

    def setUp(self):
        self.evt = threading.Event()
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.settimeout(20)
        self.port = socket_helper.bind_port(self.sock)
        self.server_thread = threading.Thread(target=self.server)
        self.server_thread.daemon = True
        self.server_thread.start()
        
        self.evt.wait()
        self.evt.clear()
        self.old_port = ftplib.FTP.port
        ftplib.FTP.port = self.port

    def tearDown(self):
        ftplib.FTP.port = self.old_port
        self.server_thread.join()
        
        self.server_thread = None

    def server(self):
        
        
        
        
        self.sock.listen()
        
        self.evt.set()
        try:
            conn, addr = self.sock.accept()
        except TimeoutError:
            pass
        else:
            conn.sendall(b"1 Hola mundo\n")
            conn.shutdown(socket.SHUT_WR)
            
            self.evt.set()
            conn.close()
        finally:
            self.sock.close()

    def testTimeoutDefault(self):
        
        self.assertIsNone(socket.getdefaulttimeout())
        socket.setdefaulttimeout(30)
        try:
            ftp = ftplib.FTP(HOST)
        finally:
            socket.setdefaulttimeout(None)
        self.assertEqual(ftp.sock.gettimeout(), 30)
        self.evt.wait()
        ftp.close()

    def testTimeoutNone(self):
        
        self.assertIsNone(socket.getdefaulttimeout())
        socket.setdefaulttimeout(30)
        try:
            ftp = ftplib.FTP(HOST, timeout=None)
        finally:
            socket.setdefaulttimeout(None)
        self.assertIsNone(ftp.sock.gettimeout())
        self.evt.wait()
        ftp.close()

    def testTimeoutValue(self):
        
        ftp = ftplib.FTP(HOST, timeout=30)
        self.assertEqual(ftp.sock.gettimeout(), 30)
        self.evt.wait()
        ftp.close()

        
        with self.assertRaises(ValueError):
            ftplib.FTP(HOST, timeout=0)

    def testTimeoutConnect(self):
        ftp = ftplib.FTP()
        ftp.connect(HOST, timeout=30)
        self.assertEqual(ftp.sock.gettimeout(), 30)
        self.evt.wait()
        ftp.close()

    def testTimeoutDifferentOrder(self):
        ftp = ftplib.FTP(timeout=30)
        ftp.connect(HOST)
        self.assertEqual(ftp.sock.gettimeout(), 30)
        self.evt.wait()
        ftp.close()

    def testTimeoutDirectAccess(self):
        ftp = ftplib.FTP()
        ftp.timeout = 30
        ftp.connect(HOST)
        self.assertEqual(ftp.sock.gettimeout(), 30)
        self.evt.wait()
        ftp.close()


class MiscTestCase(TestCase):
    def test__all__(self):
        not_exported = {
            'MSG_OOB', 'FTP_PORT', 'MAXLINE', 'CRLF', 'B_CRLF', 'Error',
            'parse150', 'parse227', 'parse229', 'parse257', 'print_line',
            'ftpcp', 'test'}
        support.check__all__(self, ftplib, not_exported=not_exported)


def setUpModule():
    thread_info = threading_helper.threading_setup()
    unittest.addModuleCleanup(threading_helper.threading_cleanup, *thread_info)


if __name__ == '__main__':
    unittest.main()