<s> """<STR_LIT>""" <EOL> import os . path <EOL> from struct import pack <EOL> from errno import ENOSYS <EOL> from zope . interface . verify import verifyObject , verifyClass <EOL> from zope . interface import implementer <EOL> from twisted . python . log import msg <EOL> from twisted . python . filepath import FilePath <EOL> from twisted . python . failure import Failure <EOL> from twisted . python . reflect import requireModule <EOL> from twisted . internet . interfaces import IAddress , IStreamClientEndpoint <EOL> from twisted . internet . protocol import Factory , Protocol <EOL> from twisted . internet . defer import CancelledError , Deferred , succeed , fail <EOL> from twisted . internet . error import ConnectionDone , ConnectionRefusedError <EOL> from twisted . internet . address import IPv4Address <EOL> from twisted . trial . unittest import TestCase <EOL> from twisted . test . proto_helpers import MemoryReactorClock <EOL> from twisted . internet . error import ProcessTerminated , ConnectingCancelledError <EOL> from twisted . cred . portal import Portal <EOL> from twisted . cred . checkers import InMemoryUsernamePasswordDatabaseDontUse <EOL> from twisted . conch . interfaces import IConchUser <EOL> from twisted . conch . error import ConchError , UserRejectedKey , HostKeyChanged <EOL> if requireModule ( '<STR_LIT>' ) and requireModule ( '<STR_LIT>' ) : <EOL> from twisted . conch . ssh . factory import SSHFactory <EOL> from twisted . conch . ssh . userauth import SSHUserAuthServer <EOL> from twisted . conch . ssh . connection import SSHConnection <EOL> from twisted . conch . ssh . keys import Key <EOL> from twisted . conch . ssh . channel import SSHChannel <EOL> from twisted . conch . ssh . agent import SSHAgentServer <EOL> from twisted . conch . client . knownhosts import KnownHostsFile , ConsoleUI <EOL> from twisted . conch . checkers import SSHPublicKeyChecker , InMemorySSHKeyDB <EOL> from twisted . conch . avatar import ConchUser <EOL> from twisted . conch . test . keydata import ( <EOL> publicRSA_openssh , privateRSA_openssh , privateDSA_openssh ) <EOL> from twisted . conch . endpoints import ( <EOL> _ISSHConnectionCreator , AuthenticationFailed , SSHCommandAddress , <EOL> SSHCommandClientEndpoint , _ReadFile , _NewConnectionHelper , <EOL> _ExistingConnectionHelper ) <EOL> from twisted . conch . ssh . transport import SSHClientTransport <EOL> else : <EOL> skip = "<STR_LIT>" <EOL> SSHFactory = SSHUserAuthServer = SSHConnection = Key = SSHChannel = SSHAgentServer = KnownHostsFile = SSHPublicKeyChecker = ConchUser = object <EOL> from twisted . test . proto_helpers import StringTransport <EOL> from twisted . test . iosim import FakeTransport , connect <EOL> class AbortableFakeTransport ( FakeTransport ) : <EOL> """<STR_LIT>""" <EOL> aborted = False <EOL> def abortConnection ( self ) : <EOL> """<STR_LIT>""" <EOL> self . aborted = True <EOL> class BrokenExecSession ( SSHChannel ) : <EOL> """<STR_LIT>""" <EOL> def request_exec ( self , data ) : <EOL> """<STR_LIT>""" <EOL> return <NUM_LIT:0> <EOL> class WorkingExecSession ( SSHChannel ) : <EOL> """<STR_LIT>""" <EOL> def request_exec ( self , data ) : <EOL> """<STR_LIT>""" <EOL> return <NUM_LIT:1> <EOL> class UnsatisfiedExecSession ( SSHChannel ) : <EOL> """<STR_LIT>""" <EOL> def request_exec ( self , data ) : <EOL> """<STR_LIT>""" <EOL> return Deferred ( ) <EOL> class TrivialRealm ( object ) : <EOL> def __init__ ( self ) : <EOL> self . channelLookup = { } <EOL> def requestAvatar ( self , avatarId , mind , * interfaces ) : <EOL> avatar = ConchUser ( ) <EOL> avatar . channelLookup = self . channelLookup <EOL> return ( IConchUser , avatar , lambda : None ) <EOL> class AddressSpyFactory ( Factory ) : <EOL> address = None <EOL> def buildProtocol ( self , address ) : <EOL> self . address = address <EOL> return Factory . buildProtocol ( self , address ) <EOL> class FixedResponseUI ( object ) : <EOL> def __init__ ( self , result ) : <EOL> self . result = result <EOL> def prompt ( self , text ) : <EOL> return succeed ( self . result ) <EOL> def warn ( self , text ) : <EOL> pass <EOL> class FakeClockSSHUserAuthServer ( SSHUserAuthServer ) : <EOL> @ property <EOL> def attemptsBeforeDisconnect ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . transport . factory . attemptsBeforeDisconnect <EOL> @ property <EOL> def clock ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . transport . factory . reactor <EOL> class CommandFactory ( SSHFactory ) : <EOL> @ property <EOL> def publicKeys ( self ) : <EOL> return { <EOL> '<STR_LIT>' : Key . fromString ( data = publicRSA_openssh ) <EOL> } <EOL> @ property <EOL> def privateKeys ( self ) : <EOL> return { <EOL> '<STR_LIT>' : Key . fromString ( data = privateRSA_openssh ) <EOL> } <EOL> services = { <EOL> '<STR_LIT>' : FakeClockSSHUserAuthServer , <EOL> '<STR_LIT>' : SSHConnection <EOL> } <EOL> attemptsBeforeDisconnect = <NUM_LIT:0> <EOL> @ implementer ( IAddress ) <EOL> class MemoryAddress ( object ) : <EOL> pass <EOL> @ implementer ( IStreamClientEndpoint ) <EOL> class SingleUseMemoryEndpoint ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , server ) : <EOL> """<STR_LIT>""" <EOL> self . pump = None <EOL> self . _server = server <EOL> def connect ( self , factory ) : <EOL> if self . pump is not None : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> try : <EOL> protocol = factory . buildProtocol ( MemoryAddress ( ) ) <EOL> except : <EOL> return fail ( ) <EOL> else : <EOL> self . pump = connect ( <EOL> self . _server , AbortableFakeTransport ( <EOL> self . _server , isServer = True ) , <EOL> protocol , AbortableFakeTransport ( protocol , isServer = False ) ) <EOL> return succeed ( protocol ) <EOL> class SSHCommandClientEndpointTestsMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> self . hostname = b"<STR_LIT>" <EOL> self . port = <NUM_LIT> <EOL> self . user = b"<STR_LIT:user>" <EOL> self . password = b"<STR_LIT:password>" <EOL> self . reactor = MemoryReactorClock ( ) <EOL> self . realm = TrivialRealm ( ) <EOL> self . portal = Portal ( self . realm ) <EOL> self . passwdDB = InMemoryUsernamePasswordDatabaseDontUse ( ) <EOL> self . passwdDB . addUser ( self . user , self . password ) <EOL> self . portal . registerChecker ( self . passwdDB ) <EOL> self . factory = CommandFactory ( ) <EOL> self . factory . reactor = self . reactor <EOL> self . factory . portal = self . portal <EOL> self . factory . doStart ( ) <EOL> self . addCleanup ( self . factory . doStop ) <EOL> self . clientAddress = IPv4Address ( "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT> ) <EOL> self . serverAddress = IPv4Address ( "<STR_LIT>" , "<STR_LIT>" , <NUM_LIT> ) <EOL> def create ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( <EOL> "<STR_LIT>" % ( self . __class__ . __name__ , ) ) <EOL> def assertClientTransportState ( self , client , immediateClose ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( <EOL> "<STR_LIT>" % ( <EOL> self . __class__ . __name__ , ) ) <EOL> def finishConnection ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( <EOL> "<STR_LIT>" % ( <EOL> self . __class__ . __name__ , ) ) <EOL> def connectedServerAndClient ( self , serverFactory , clientFactory ) : <EOL> """<STR_LIT>""" <EOL> clientProtocol = clientFactory . buildProtocol ( None ) <EOL> serverProtocol = serverFactory . buildProtocol ( None ) <EOL> clientTransport = AbortableFakeTransport ( <EOL> clientProtocol , isServer = False , hostAddress = self . clientAddress , <EOL> peerAddress = self . serverAddress ) <EOL> serverTransport = AbortableFakeTransport ( <EOL> serverProtocol , isServer = True , hostAddress = self . serverAddress , <EOL> peerAddress = self . clientAddress ) <EOL> pump = connect ( <EOL> serverProtocol , serverTransport , clientProtocol , clientTransport ) <EOL> return serverProtocol , clientProtocol , pump <EOL> def test_channelOpenFailure ( self ) : <EOL> """<STR_LIT>""" <EOL> endpoint = self . create ( ) <EOL> factory = Factory ( ) <EOL> factory . protocol = Protocol <EOL> connected = endpoint . connect ( factory ) <EOL> server , client , pump = self . finishConnection ( ) <EOL> errors = self . flushLoggedErrors ( ConchError ) <EOL> self . assertIn ( <EOL> '<STR_LIT>' , ( errors [ <NUM_LIT:0> ] . value . data , errors [ <NUM_LIT:0> ] . value . value ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( errors ) ) <EOL> f = self . failureResultOf ( connected ) <EOL> f . trap ( ConchError ) <EOL> self . assertEqual ( '<STR_LIT>' , f . value . value ) <EOL> self . assertClientTransportState ( client , False ) <EOL> def test_execFailure ( self ) : <EOL> """<STR_LIT>""" <EOL> self . realm . channelLookup [ b'<STR_LIT>' ] = BrokenExecSession <EOL> endpoint = self . create ( ) <EOL> factory = Factory ( ) <EOL> factory . protocol = Protocol <EOL> connected = endpoint . connect ( factory ) <EOL> server , client , pump = self . finishConnection ( ) <EOL> f = self . failureResultOf ( connected ) <EOL> f . trap ( ConchError ) <EOL> self . assertEqual ( '<STR_LIT>' , f . value . value ) <EOL> self . assertClientTransportState ( client , False ) <EOL> def test_execCancelled ( self ) : <EOL> """<STR_LIT>""" <EOL> self . realm . channelLookup [ b'<STR_LIT>' ] = UnsatisfiedExecSession <EOL> endpoint = self . create ( ) <EOL> factory = Factory ( ) <EOL> factory . protocol = Protocol <EOL> connected = endpoint . connect ( factory ) <EOL> server , client , pump = self . finishConnection ( ) <EOL> connected . cancel ( ) <EOL> f = self . failureResultOf ( connected ) <EOL> f . trap ( CancelledError ) <EOL> self . assertClientTransportState ( client , True ) <EOL> def test_buildProtocol ( self ) : <EOL> """<STR_LIT>""" <EOL> self . realm . channelLookup [ b'<STR_LIT>' ] = WorkingExecSession <EOL> endpoint = self . create ( ) <EOL> factory = AddressSpyFactory ( ) <EOL> factory . protocol = Protocol <EOL> endpoint . connect ( factory ) <EOL> server , client , pump = self . finishConnection ( ) <EOL> self . assertIsInstance ( factory . address , SSHCommandAddress ) <EOL> self . assertEqual ( server . transport . getHost ( ) , factory . address . server ) <EOL> self . assertEqual ( self . user , factory . address . username ) <EOL> self . assertEqual ( b"<STR_LIT>" , factory . address . command ) <EOL> def test_makeConnection ( self ) : <EOL> """<STR_LIT>""" <EOL> self . realm . channelLookup [ b'<STR_LIT>' ] = WorkingExecSession <EOL> endpoint = self . create ( ) <EOL> factory = Factory ( ) <EOL> factory . protocol = Protocol <EOL> connected = endpoint . connect ( factory ) <EOL> server , client , pump = self . finishConnection ( ) <EOL> protocol = self . successResultOf ( connected ) <EOL> self . assertIsNot ( None , protocol . transport ) <EOL> def test_dataReceived ( self ) : <EOL> """<STR_LIT>""" <EOL> self . realm . channelLookup [ b'<STR_LIT>' ] = WorkingExecSession <EOL> endpoint = self . create ( ) <EOL> factory = Factory ( ) <EOL> factory . protocol = Protocol <EOL> connected = endpoint . connect ( factory ) <EOL> server , client , pump = self . finishConnection ( ) <EOL> protocol = self . successResultOf ( connected ) <EOL> dataReceived = [ ] <EOL> protocol . dataReceived = dataReceived . append <EOL> channelId = protocol . transport . id <EOL> server . service . channels [ channelId ] . write ( b"<STR_LIT>" ) <EOL> pump . pump ( ) <EOL> self . assertEqual ( b"<STR_LIT>" , b"<STR_LIT>" . join ( dataReceived ) ) <EOL> def test_connectionLost ( self ) : <EOL> """<STR_LIT>""" <EOL> self . realm . channelLookup [ b'<STR_LIT>' ] = WorkingExecSession <EOL> endpoint = self . create ( ) <EOL> factory = Factory ( ) <EOL> factory . protocol = Protocol <EOL> connected = endpoint . connect ( factory ) <EOL> server , client , pump = self . finishConnection ( ) <EOL> protocol = self . successResultOf ( connected ) <EOL> connectionLost = [ ] <EOL> protocol . connectionLost = connectionLost . append <EOL> channelId = protocol . transport . id <EOL> server . service . channels [ channelId ] . loseConnection ( ) <EOL> pump . pump ( ) <EOL> connectionLost [ <NUM_LIT:0> ] . trap ( ConnectionDone ) <EOL> self . assertClientTransportState ( client , False ) <EOL> def _exitStatusTest ( self , request , requestArg ) : <EOL> """<STR_LIT>""" <EOL> self . realm . channelLookup [ b'<STR_LIT>' ] = WorkingExecSession <EOL> endpoint = self . create ( ) <EOL> factory = Factory ( ) <EOL> factory . protocol = Protocol <EOL> connected = endpoint . connect ( factory ) <EOL> server , client , pump = self . finishConnection ( ) <EOL> protocol = self . successResultOf ( connected ) <EOL> connectionLost = [ ] <EOL> protocol . connectionLost = connectionLost . append <EOL> channelId = protocol . transport . id <EOL> channel = server . service . channels [ channelId ] <EOL> server . service . sendRequest ( channel , request , requestArg ) <EOL> channel . loseConnection ( ) <EOL> pump . pump ( ) <EOL> self . assertClientTransportState ( client , False ) <EOL> return connectionLost [ <NUM_LIT:0> ] <EOL> def test_zeroExitCode ( self ) : <EOL> """<STR_LIT>""" <EOL> exitCode = <NUM_LIT:0> <EOL> exc = self . _exitStatusTest ( '<STR_LIT>' , pack ( '<STR_LIT>' , exitCode ) ) <EOL> exc . trap ( ConnectionDone ) <EOL> def test_nonZeroExitStatus ( self ) : <EOL> """<STR_LIT>""" <EOL> exitCode = <NUM_LIT> <EOL> signal = None <EOL> exc = self . _exitStatusTest ( '<STR_LIT>' , pack ( '<STR_LIT>' , exitCode ) ) <EOL> exc . trap ( ProcessTerminated ) <EOL> self . assertEqual ( exitCode , exc . value . exitCode ) <EOL> self . assertEqual ( signal , exc . value . signal ) <EOL> def test_nonZeroExitSignal ( self ) : <EOL> """<STR_LIT>""" <EOL> exitCode = None <EOL> signal = <NUM_LIT> <EOL> exc = self . _exitStatusTest ( '<STR_LIT>' , pack ( '<STR_LIT>' , signal ) ) <EOL> exc . trap ( ProcessTerminated ) <EOL> self . assertEqual ( exitCode , exc . value . exitCode ) <EOL> self . assertEqual ( signal , exc . value . signal ) <EOL> def record ( self , server , protocol , event , noArgs = False ) : <EOL> """<STR_LIT>""" <EOL> channelId = protocol . transport . id <EOL> recorder = [ ] <EOL> if noArgs : <EOL> f = lambda : recorder . append ( None ) <EOL> else : <EOL> f = recorder . append <EOL> setattr ( server . service . channels [ channelId ] , event , f ) <EOL> return recorder <EOL> def test_write ( self ) : <EOL> """<STR_LIT>""" <EOL> self . realm . channelLookup [ b'<STR_LIT>' ] = WorkingExecSession <EOL> endpoint = self . create ( ) <EOL> factory = Factory ( ) <EOL> factory . protocol = Protocol <EOL> connected = endpoint . connect ( factory ) <EOL> server , client , pump = self . finishConnection ( ) <EOL> protocol = self . successResultOf ( connected ) <EOL> dataReceived = self . record ( server , protocol , '<STR_LIT>' ) <EOL> protocol . transport . write ( b"<STR_LIT>" ) <EOL> pump . pump ( ) <EOL> self . assertEqual ( b"<STR_LIT>" , b"<STR_LIT>" . join ( dataReceived ) ) <EOL> def test_writeSequence ( self ) : <EOL> """<STR_LIT>""" <EOL> self . realm . channelLookup [ b'<STR_LIT>' ] = WorkingExecSession <EOL> endpoint = self . create ( ) <EOL> factory = Factory ( ) <EOL> factory . protocol = Protocol <EOL> connected = endpoint . connect ( factory ) <EOL> server , client , pump = self . finishConnection ( ) <EOL> protocol = self . successResultOf ( connected ) <EOL> dataReceived = self . record ( server , protocol , '<STR_LIT>' ) <EOL> protocol . transport . writeSequence ( list ( b"<STR_LIT>" ) ) <EOL> pump . pump ( ) <EOL> self . assertEqual ( b"<STR_LIT>" , b"<STR_LIT>" . join ( dataReceived ) ) <EOL> class NewConnectionTests ( TestCase , SSHCommandClientEndpointTestsMixin ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> SSHCommandClientEndpointTestsMixin . setUp ( self ) <EOL> self . hostKeyPath = FilePath ( self . mktemp ( ) ) <EOL> self . knownHosts = KnownHostsFile ( self . hostKeyPath ) <EOL> self . knownHosts . addHostKey ( <EOL> self . hostname , self . factory . publicKeys [ '<STR_LIT>' ] ) <EOL> self . knownHosts . addHostKey ( <EOL> self . serverAddress . host , self . factory . publicKeys [ '<STR_LIT>' ] ) <EOL> self . knownHosts . save ( ) <EOL> def create ( self ) : <EOL> """<STR_LIT>""" <EOL> return SSHCommandClientEndpoint . newConnection ( <EOL> self . reactor , b"<STR_LIT>" , self . user , self . hostname , self . port , <EOL> password = self . password , knownHosts = self . knownHosts , <EOL> ui = FixedResponseUI ( False ) ) <EOL> def finishConnection ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . connectedServerAndClient ( <EOL> self . factory , self . reactor . tcpClients [ <NUM_LIT:0> ] [ <NUM_LIT:2> ] ) <EOL> def assertClientTransportState ( self , client , immediateClose ) : <EOL> """<STR_LIT>""" <EOL> if immediateClose : <EOL> self . assertTrue ( client . transport . aborted ) <EOL> else : <EOL> self . assertTrue ( client . transport . disconnecting ) <EOL> def test_interface ( self ) : <EOL> """<STR_LIT>""" <EOL> endpoint = SSHCommandClientEndpoint . newConnection ( <EOL> self . reactor , b"<STR_LIT>" , b"<STR_LIT>" , <EOL> self . hostname , self . port ) <EOL> self . assertTrue ( verifyObject ( IStreamClientEndpoint , endpoint ) ) <EOL> def test_defaultPort ( self ) : <EOL> """<STR_LIT>""" <EOL> endpoint = SSHCommandClientEndpoint . newConnection ( <EOL> self . reactor , b"<STR_LIT>" , b"<STR_LIT>" , self . hostname ) <EOL> self . assertEqual ( <NUM_LIT> , endpoint . _creator . port ) <EOL> def test_specifiedPort ( self ) : <EOL> """<STR_LIT>""" <EOL> endpoint = SSHCommandClientEndpoint . newConnection ( <EOL> self . reactor , b"<STR_LIT>" , b"<STR_LIT>" , <EOL> self . hostname , port = <NUM_LIT> ) <EOL> self . assertEqual ( <NUM_LIT> , endpoint . _creator . port ) <EOL> def test_destination ( self ) : <EOL> """<STR_LIT>""" <EOL> endpoint = SSHCommandClientEndpoint . newConnection ( <EOL> self . reactor , b"<STR_LIT>" , self . user , self . hostname , self . port , <EOL> password = self . password , knownHosts = self . knownHosts , <EOL> ui = FixedResponseUI ( False ) ) <EOL> factory = Factory ( ) <EOL> factory . protocol = Protocol <EOL> endpoint . connect ( factory ) <EOL> host , port , factory , timeout , bindAddress = self . reactor . tcpClients [ <NUM_LIT:0> ] <EOL> self . assertEqual ( self . hostname , host ) <EOL> self . assertEqual ( self . port , port ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( self . reactor . tcpClients ) ) <EOL> def test_connectionFailed ( self ) : <EOL> """<STR_LIT>""" <EOL> endpoint = SSHCommandClientEndpoint . newConnection ( <EOL> self . reactor , b"<STR_LIT>" , b"<STR_LIT>" , <EOL> self . hostname , self . port , knownHosts = self . knownHosts , <EOL> ui = FixedResponseUI ( False ) ) <EOL> factory = Factory ( ) <EOL> factory . protocol = Protocol <EOL> d = endpoint . connect ( factory ) <EOL> factory = self . reactor . tcpClients [ <NUM_LIT:0> ] [ <NUM_LIT:2> ] <EOL> factory . clientConnectionFailed ( None , Failure ( ConnectionRefusedError ( ) ) ) <EOL> self . failureResultOf ( d ) . trap ( ConnectionRefusedError ) <EOL> def test_userRejectedHostKey ( self ) : <EOL> """<STR_LIT>""" <EOL> endpoint = SSHCommandClientEndpoint . newConnection ( <EOL> self . reactor , b"<STR_LIT>" , b"<STR_LIT>" , <EOL> self . hostname , self . port , knownHosts = KnownHostsFile ( self . mktemp ( ) ) , <EOL> ui = FixedResponseUI ( False ) ) <EOL> factory = Factory ( ) <EOL> factory . protocol = Protocol <EOL> connected = endpoint . connect ( factory ) <EOL> server , client , pump = self . connectedServerAndClient ( <EOL> self . factory , self . reactor . tcpClients [ <NUM_LIT:0> ] [ <NUM_LIT:2> ] ) <EOL> f = self . failureResultOf ( connected ) <EOL> f . trap ( UserRejectedKey ) <EOL> def test_mismatchedHostKey ( self ) : <EOL> """<STR_LIT>""" <EOL> differentKey = Key . fromString ( privateDSA_openssh ) . public ( ) <EOL> knownHosts = KnownHostsFile ( self . mktemp ( ) ) <EOL> knownHosts . addHostKey ( self . serverAddress . host , differentKey ) <EOL> knownHosts . addHostKey ( self . hostname , differentKey ) <EOL> ui = FixedResponseUI ( True ) <EOL> endpoint = SSHCommandClientEndpoint . newConnection ( <EOL> self . reactor , b"<STR_LIT>" , b"<STR_LIT>" , <EOL> self . hostname , self . port , password = b"<STR_LIT>" , <EOL> knownHosts = knownHosts , ui = ui ) <EOL> factory = Factory ( ) <EOL> factory . protocol = Protocol <EOL> connected = endpoint . connect ( factory ) <EOL> server , client , pump = self . connectedServerAndClient ( <EOL> self . factory , self . reactor . tcpClients [ <NUM_LIT:0> ] [ <NUM_LIT:2> ] ) <EOL> f = self . failureResultOf ( connected ) <EOL> f . trap ( HostKeyChanged ) <EOL> def test_connectionClosedBeforeSecure ( self ) : <EOL> """<STR_LIT>""" <EOL> endpoint = SSHCommandClientEndpoint . newConnection ( <EOL> self . reactor , b"<STR_LIT>" , b"<STR_LIT>" , <EOL> self . hostname , self . port , knownHosts = self . knownHosts , <EOL> ui = FixedResponseUI ( False ) ) <EOL> factory = Factory ( ) <EOL> factory . protocol = Protocol <EOL> d = endpoint . connect ( factory ) <EOL> transport = StringTransport ( ) <EOL> factory = self . reactor . tcpClients [ <NUM_LIT:0> ] [ <NUM_LIT:2> ] <EOL> client = factory . buildProtocol ( None ) <EOL> client . makeConnection ( transport ) <EOL> client . connectionLost ( Failure ( ConnectionDone ( ) ) ) <EOL> self . failureResultOf ( d ) . trap ( ConnectionDone ) <EOL> def test_connectionCancelledBeforeSecure ( self ) : <EOL> """<STR_LIT>""" <EOL> endpoint = SSHCommandClientEndpoint . newConnection ( <EOL> self . reactor , b"<STR_LIT>" , b"<STR_LIT>" , <EOL> self . hostname , self . port , knownHosts = self . knownHosts , <EOL> ui = FixedResponseUI ( False ) ) <EOL> factory = Factory ( ) <EOL> factory . protocol = Protocol <EOL> d = endpoint . connect ( factory ) <EOL> transport = AbortableFakeTransport ( None , isServer = False ) <EOL> factory = self . reactor . tcpClients [ <NUM_LIT:0> ] [ <NUM_LIT:2> ] <EOL> client = factory . buildProtocol ( None ) <EOL> client . makeConnection ( transport ) <EOL> d . cancel ( ) <EOL> self . failureResultOf ( d ) . trap ( CancelledError ) <EOL> self . assertTrue ( transport . aborted ) <EOL> client . connectionLost ( Failure ( ConnectionDone ( ) ) ) <EOL> def test_connectionCancelledBeforeConnected ( self ) : <EOL> """<STR_LIT>""" <EOL> endpoint = SSHCommandClientEndpoint . newConnection ( <EOL> self . reactor , b"<STR_LIT>" , b"<STR_LIT>" , <EOL> self . hostname , self . port , knownHosts = self . knownHosts , <EOL> ui = FixedResponseUI ( False ) ) <EOL> factory = Factory ( ) <EOL> factory . protocol = Protocol <EOL> d = endpoint . connect ( factory ) <EOL> d . cancel ( ) <EOL> self . failureResultOf ( d ) . trap ( ConnectingCancelledError ) <EOL> self . assertTrue ( self . reactor . connectors [ <NUM_LIT:0> ] . stoppedConnecting ) <EOL> def test_passwordAuthenticationFailure ( self ) : <EOL> """<STR_LIT>""" <EOL> endpoint = SSHCommandClientEndpoint . newConnection ( <EOL> self . reactor , b"<STR_LIT>" , b"<STR_LIT>" , <EOL> self . hostname , self . port , password = b"<STR_LIT>" , <EOL> knownHosts = self . knownHosts , ui = FixedResponseUI ( False ) ) <EOL> factory = Factory ( ) <EOL> factory . protocol = Protocol <EOL> connected = endpoint . connect ( factory ) <EOL> server , client , pump = self . connectedServerAndClient ( <EOL> self . factory , self . reactor . tcpClients [ <NUM_LIT:0> ] [ <NUM_LIT:2> ] ) <EOL> self . reactor . advance ( server . service . passwordDelay ) <EOL> pump . flush ( ) <EOL> f = self . failureResultOf ( connected ) <EOL> f . trap ( AuthenticationFailed ) <EOL> self . assertClientTransportState ( client , False ) <EOL> def setupKeyChecker ( self , portal , users ) : <EOL> """<STR_LIT>""" <EOL> mapping = dict ( [ ( k , [ Key . fromString ( v ) . public ( ) ] ) <EOL> for k , v in users . iteritems ( ) ] ) <EOL> checker = SSHPublicKeyChecker ( InMemorySSHKeyDB ( mapping ) ) <EOL> portal . registerChecker ( checker ) <EOL> def test_publicKeyAuthenticationFailure ( self ) : <EOL> """<STR_LIT>""" <EOL> badKey = Key . fromString ( privateRSA_openssh ) <EOL> self . setupKeyChecker ( self . portal , { self . user : privateDSA_openssh } ) <EOL> endpoint = SSHCommandClientEndpoint . newConnection ( <EOL> self . reactor , b"<STR_LIT>" , self . user , <EOL> self . hostname , self . port , keys = [ badKey ] , <EOL> knownHosts = self . knownHosts , ui = FixedResponseUI ( False ) ) <EOL> factory = Factory ( ) <EOL> factory . protocol = Protocol <EOL> connected = endpoint . connect ( factory ) <EOL> server , client , pump = self . connectedServerAndClient ( <EOL> self . factory , self . reactor . tcpClients [ <NUM_LIT:0> ] [ <NUM_LIT:2> ] ) <EOL> f = self . failureResultOf ( connected ) <EOL> f . trap ( AuthenticationFailed ) <EOL> self . assertTrue ( client . transport . disconnecting ) <EOL> def test_authenticationFallback ( self ) : <EOL> """<STR_LIT>""" <EOL> badKey = Key . fromString ( privateRSA_openssh ) <EOL> self . setupKeyChecker ( self . portal , { self . user : privateDSA_openssh } ) <EOL> endpoint = SSHCommandClientEndpoint . newConnection ( <EOL> self . reactor , b"<STR_LIT>" , self . user , self . hostname , self . port , <EOL> keys = [ badKey ] , password = self . password , knownHosts = self . knownHosts , <EOL> ui = FixedResponseUI ( False ) ) <EOL> factory = Factory ( ) <EOL> factory . protocol = Protocol <EOL> connected = endpoint . connect ( factory ) <EOL> self . factory . attemptsBeforeDisconnect += <NUM_LIT:1> <EOL> server , client , pump = self . connectedServerAndClient ( <EOL> self . factory , self . reactor . tcpClients [ <NUM_LIT:0> ] [ <NUM_LIT:2> ] ) <EOL> pump . pump ( ) <EOL> errors = self . flushLoggedErrors ( ConchError ) <EOL> self . assertIn ( <EOL> '<STR_LIT>' , ( errors [ <NUM_LIT:0> ] . value . data , errors [ <NUM_LIT:0> ] . value . value ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( errors ) ) <EOL> f = self . failureResultOf ( connected ) <EOL> f . trap ( ConchError ) <EOL> self . assertEqual ( '<STR_LIT>' , f . value . value ) <EOL> self . assertTrue ( client . transport . disconnecting ) <EOL> def test_publicKeyAuthentication ( self ) : <EOL> """<STR_LIT>""" <EOL> key = Key . fromString ( privateDSA_openssh ) <EOL> self . setupKeyChecker ( self . portal , { self . user : privateDSA_openssh } ) <EOL> self . realm . channelLookup [ b'<STR_LIT>' ] = WorkingExecSession <EOL> endpoint = SSHCommandClientEndpoint . newConnection ( <EOL> self . reactor , b"<STR_LIT>" , self . user , self . hostname , self . port , <EOL> keys = [ key ] , knownHosts = self . knownHosts , ui = FixedResponseUI ( False ) ) <EOL> factory = Factory ( ) <EOL> factory . protocol = Protocol <EOL> connected = endpoint . connect ( factory ) <EOL> server , client , pump = self . connectedServerAndClient ( <EOL> self . factory , self . reactor . tcpClients [ <NUM_LIT:0> ] [ <NUM_LIT:2> ] ) <EOL> protocol = self . successResultOf ( connected ) <EOL> self . assertIsNot ( None , protocol . transport ) <EOL> def test_skipPasswordAuthentication ( self ) : <EOL> """<STR_LIT>""" <EOL> endpoint = SSHCommandClientEndpoint . newConnection ( <EOL> self . reactor , b"<STR_LIT>" , self . user , self . hostname , self . port , <EOL> knownHosts = self . knownHosts , ui = FixedResponseUI ( False ) ) <EOL> factory = Factory ( ) <EOL> factory . protocol = Protocol <EOL> connected = endpoint . connect ( factory ) <EOL> server , client , pump = self . connectedServerAndClient ( <EOL> self . factory , self . reactor . tcpClients [ <NUM_LIT:0> ] [ <NUM_LIT:2> ] ) <EOL> pump . pump ( ) <EOL> f = self . failureResultOf ( connected ) <EOL> f . trap ( AuthenticationFailed ) <EOL> self . assertTrue ( client . transport . disconnecting ) <EOL> def test_agentAuthentication ( self ) : <EOL> """<STR_LIT>""" <EOL> key = Key . fromString ( privateRSA_openssh ) <EOL> agentServer = SSHAgentServer ( ) <EOL> agentServer . factory = Factory ( ) <EOL> agentServer . factory . keys = { key . blob ( ) : ( key , "<STR_LIT>" ) } <EOL> self . setupKeyChecker ( self . portal , { self . user : privateRSA_openssh } ) <EOL> agentEndpoint = SingleUseMemoryEndpoint ( agentServer ) <EOL> endpoint = SSHCommandClientEndpoint . newConnection ( <EOL> self . reactor , b"<STR_LIT>" , self . user , self . hostname , self . port , <EOL> knownHosts = self . knownHosts , ui = FixedResponseUI ( False ) , <EOL> agentEndpoint = agentEndpoint ) <EOL> self . realm . channelLookup [ b'<STR_LIT>' ] = WorkingExecSession <EOL> factory = Factory ( ) <EOL> factory . protocol = Protocol <EOL> connected = endpoint . connect ( factory ) <EOL> server , client , pump = self . connectedServerAndClient ( <EOL> self . factory , self . reactor . tcpClients [ <NUM_LIT:0> ] [ <NUM_LIT:2> ] ) <EOL> for i in range ( <NUM_LIT> ) : <EOL> agentEndpoint . pump . pump ( ) <EOL> pump . pump ( ) <EOL> protocol = self . successResultOf ( connected ) <EOL> self . assertIsNot ( None , protocol . transport ) <EOL> def test_loseConnection ( self ) : <EOL> """<STR_LIT>""" <EOL> self . realm . channelLookup [ b'<STR_LIT>' ] = WorkingExecSession <EOL> endpoint = self . create ( ) <EOL> factory = Factory ( ) <EOL> factory . protocol = Protocol <EOL> connected = endpoint . connect ( factory ) <EOL> server , client , pump = self . finishConnection ( ) <EOL> protocol = self . successResultOf ( connected ) <EOL> closed = self . record ( server , protocol , '<STR_LIT>' , noArgs = True ) <EOL> protocol . transport . loseConnection ( ) <EOL> pump . pump ( ) <EOL> self . assertEqual ( [ None ] , closed ) <EOL> pump . pump ( ) <EOL> self . assertTrue ( client . transport . disconnecting ) <EOL> class ExistingConnectionTests ( TestCase , SSHCommandClientEndpointTestsMixin ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> SSHCommandClientEndpointTestsMixin . setUp ( self ) <EOL> knownHosts = KnownHostsFile ( FilePath ( self . mktemp ( ) ) ) <EOL> knownHosts . addHostKey ( <EOL> self . hostname , self . factory . publicKeys [ '<STR_LIT>' ] ) <EOL> knownHosts . addHostKey ( <EOL> self . serverAddress . host , self . factory . publicKeys [ '<STR_LIT>' ] ) <EOL> self . endpoint = SSHCommandClientEndpoint . newConnection ( <EOL> self . reactor , b"<STR_LIT>" , self . user , self . hostname , self . port , <EOL> password = self . password , knownHosts = knownHosts , <EOL> ui = FixedResponseUI ( False ) ) <EOL> def create ( self ) : <EOL> """<STR_LIT>""" <EOL> factory = Factory ( ) <EOL> factory . protocol = Protocol <EOL> connected = self . endpoint . connect ( factory ) <EOL> channelLookup = self . realm . channelLookup . copy ( ) <EOL> try : <EOL> self . realm . channelLookup [ b'<STR_LIT>' ] = WorkingExecSession <EOL> server , client , pump = self . connectedServerAndClient ( <EOL> self . factory , self . reactor . tcpClients [ <NUM_LIT:0> ] [ <NUM_LIT:2> ] ) <EOL> finally : <EOL> self . realm . channelLookup . clear ( ) <EOL> self . realm . channelLookup . update ( channelLookup ) <EOL> self . _server = server <EOL> self . _client = client <EOL> self . _pump = pump <EOL> protocol = self . successResultOf ( connected ) <EOL> connection = protocol . transport . conn <EOL> return SSHCommandClientEndpoint . existingConnection ( <EOL> connection , b"<STR_LIT>" ) <EOL> def finishConnection ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _pump . pump ( ) <EOL> self . _pump . pump ( ) <EOL> self . _pump . pump ( ) <EOL> self . _pump . pump ( ) <EOL> return self . _server , self . _client , self . _pump <EOL> def assertClientTransportState ( self , client , immediateClose ) : <EOL> """<STR_LIT>""" <EOL> self . assertFalse ( client . transport . disconnecting ) <EOL> self . assertFalse ( client . transport . aborted ) <EOL> class ExistingConnectionHelperTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_interface ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertTrue ( <EOL> verifyClass ( _ISSHConnectionCreator , _ExistingConnectionHelper ) ) <EOL> def test_secureConnection ( self ) : <EOL> """<STR_LIT>""" <EOL> result = object ( ) <EOL> helper = _ExistingConnectionHelper ( result ) <EOL> self . assertIs ( <EOL> result , self . successResultOf ( helper . secureConnection ( ) ) ) <EOL> def test_cleanupConnectionNotImmediately ( self ) : <EOL> """<STR_LIT>""" <EOL> helper = _ExistingConnectionHelper ( object ( ) ) <EOL> helper . cleanupConnection ( object ( ) , False ) <EOL> def test_cleanupConnectionImmediately ( self ) : <EOL> """<STR_LIT>""" <EOL> helper = _ExistingConnectionHelper ( object ( ) ) <EOL> helper . cleanupConnection ( object ( ) , True ) <EOL> class _PTYPath ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , contents ) : <EOL> """<STR_LIT>""" <EOL> self . contents = contents <EOL> def open ( self , mode ) : <EOL> """<STR_LIT>""" <EOL> if mode == "<STR_LIT>" : <EOL> return _ReadFile ( self . contents ) <EOL> raise OSError ( ENOSYS , "<STR_LIT>" ) <EOL> class NewConnectionHelperTests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_interface ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertTrue ( <EOL> verifyClass ( _ISSHConnectionCreator , _NewConnectionHelper ) ) <EOL> def test_defaultPath ( self ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( <EOL> "<STR_LIT>" , _NewConnectionHelper . _KNOWN_HOSTS ) <EOL> def test_defaultKnownHosts ( self ) : <EOL> """<STR_LIT>""" <EOL> result = object ( ) <EOL> self . patch ( _NewConnectionHelper , '<STR_LIT>' , lambda cls : result ) <EOL> helper = _NewConnectionHelper ( <EOL> None , None , None , None , None , None , None , None , None , None ) <EOL> self . assertIs ( result , helper . knownHosts ) <EOL> def test_readExisting ( self ) : <EOL> """<STR_LIT>""" <EOL> key = CommandFactory ( ) . publicKeys [ '<STR_LIT>' ] <EOL> path = FilePath ( self . mktemp ( ) ) <EOL> knownHosts = KnownHostsFile ( path ) <EOL> knownHosts . addHostKey ( "<STR_LIT:127.0.0.1>" , key ) <EOL> knownHosts . save ( ) <EOL> msg ( "<STR_LIT>" % ( path . path , ) ) <EOL> home = os . path . expanduser ( "<STR_LIT>" ) <EOL> default = path . path . replace ( home , "<STR_LIT>" ) <EOL> self . patch ( _NewConnectionHelper , "<STR_LIT>" , default ) <EOL> msg ( "<STR_LIT>" % ( default , ) ) <EOL> loaded = _NewConnectionHelper . _knownHosts ( ) <EOL> self . assertTrue ( loaded . hasHostKey ( "<STR_LIT:127.0.0.1>" , key ) ) <EOL> def test_defaultConsoleUI ( self ) : <EOL> """<STR_LIT>""" <EOL> helper = _NewConnectionHelper ( <EOL> None , None , None , None , None , None , None , None , None , None ) <EOL> self . assertIsInstance ( helper . ui , ConsoleUI ) <EOL> def test_ttyConsoleUI ( self ) : <EOL> """<STR_LIT>""" <EOL> tty = _PTYPath ( b"<STR_LIT:yes>" ) <EOL> helper = _NewConnectionHelper ( <EOL> None , None , None , None , None , None , None , None , None , None , tty ) <EOL> result = self . successResultOf ( helper . ui . prompt ( b"<STR_LIT>" ) ) <EOL> self . assertTrue ( result ) <EOL> def test_nottyUI ( self ) : <EOL> """<STR_LIT>""" <EOL> tty = FilePath ( self . mktemp ( ) ) <EOL> helper = _NewConnectionHelper ( <EOL> None , None , None , None , None , None , None , None , None , None , tty ) <EOL> result = self . successResultOf ( helper . ui . prompt ( b"<STR_LIT>" ) ) <EOL> self . assertFalse ( result ) <EOL> def test_defaultTTYFilename ( self ) : <EOL> """<STR_LIT>""" <EOL> helper = _NewConnectionHelper ( <EOL> None , None , None , None , None , None , None , None , None , None ) <EOL> self . assertEqual ( FilePath ( b"<STR_LIT>" ) , helper . tty ) <EOL> def test_cleanupConnectionNotImmediately ( self ) : <EOL> """<STR_LIT>""" <EOL> helper = _NewConnectionHelper ( <EOL> None , None , None , None , None , None , None , None , None , None ) <EOL> connection = SSHConnection ( ) <EOL> connection . transport = StringTransport ( ) <EOL> helper . cleanupConnection ( connection , False ) <EOL> self . assertTrue ( connection . transport . disconnecting ) <EOL> def test_cleanupConnectionImmediately ( self ) : <EOL> """<STR_LIT>""" <EOL> class Abortable : <EOL> aborted = False <EOL> def abortConnection ( self ) : <EOL> """<STR_LIT>""" <EOL> self . aborted = True <EOL> helper = _NewConnectionHelper ( <EOL> None , None , None , None , None , None , None , None , None , None ) <EOL> connection = SSHConnection ( ) <EOL> connection . transport = SSHClientTransport ( ) <EOL> connection . transport . transport = Abortable ( ) <EOL> helper . cleanupConnection ( connection , True ) <EOL> self . assertTrue ( connection . transport . transport . aborted ) </s>
