[XMc@sdZddlmZddlmZmZmZmZddlm Z ddl m Z ddl m Z mZddlmZddlmZmZdd lZdd lZy:dd lmamZdd lmadd lmZWn!ek r d ZenXe jedZ de j!fdYZ"de j!fdYZ#dej$fdYZ%dej$fdYZ&dej$fdYZ'dZ(dZ)dd0dYZ*td k rdtj,fdYZ-ndeej.fd YZ/d!ej.fd"YZ0d#e0fd$YZ1d%ej.fd&YZ2d'ej.e*fd(YZ3d)d1d*YZ4d+ej.fd,YZ5d-ej.fd.YZ6ej7ed d krx/e/e0e1e2e3e5e6gD]Z8d/e8_9qWn[d S(2s Tests for twisted SSL support. i(tunittest(tprotocoltreactort interfacestdefer(tbasic(tutil(tgetClasstfullyQualifiedName(tplatform(tWriteDataTestCasetProperlyCloseFilesMixinN(tSSLtcrypto(tssl(tClientTLSContextcCsdaadS(N(tNoneR R(((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyt_noSSLss server.pemtUnintelligentProtocolcBsMeZdZdddgZddgZdZdZdZd ZRS( s  @ivar deferred: a deferred that will fire at connection lost. @type deferred: L{defer.Deferred} @cvar pretext: text sent before TLS is set up. @type pretext: C{str} @cvar posttext: text sent after TLS is set up. @type posttext: C{str} s first lineslast thing before tls startstSTARTTLSsfirst thing after tls startedslast thing evercCstj|_dS(N(RtDeferredtdeferred(tself((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyt__init__6scCs%x|jD]}|j|q WdS(N(tpretexttsendLine(Rtl((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pytconnectionMade:scCs]|dkrY|jjt|jjx|jD]}|j|q2W|jjndS(NtREADY(t transporttstartTLSRtfactorytclienttposttextRtloseConnection(RtlineR((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyt lineReceived?s  cCs|jjddS(N(RtcallbackR(Rtreason((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pytconnectionLostGs( t__name__t __module__t__doc__RR!RRR$R'(((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyR"s      t LineCollectorcBs>eZdZedZdZdZdZdZRS(sJ @ivar deferred: a deferred that will fire at connection lost. @type deferred: L{defer.Deferred} @ivar doTLS: whether the protocol is initiate TLS or not. @type doTLS: C{bool} @ivar fillBuffer: if set to True, it will send lots of data once C{STARTTLS} is received. @type fillBuffer: C{bool} cCs%||_||_tj|_dS(N(tdoTLSt fillBufferRRR(RR,R-((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyRYs  cCsd|j_g|j_dS(Nt(Rtrawdatatlines(R((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyR_s cCs|jjj||dkr|jrSx(tdD]}|jddq5Wn|jd|jrtdtdt}|j j ||jj q|j ndS(NRitXiRtprivateKeyFileNametcertificateFileName( RR0tappendR-trangeRR,tServerTLSContexttcertPathRRtservert setRawMode(RR#txtctx((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyR$ds     cCs#|jj|7_|jjdS(N(RR/RR"(Rtdata((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pytrawDataReceiveduscCs|jjddS(N(RR%R(RR&((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyR'zs( R(R)R*tFalseRRR$R=R'(((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyR+Ls      tSingleLineServerProtocolcBseZdZdZRS(sK A protocol that sends a single line of data at C{connectionMade}. cCs!|jjd|jjdS(Ns+OK (RtwritetgetPeerCertificate(R((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyRs(R(R)R*R(((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyR?stRecordingClientProtocolcBs)eZdZdZdZdZRS(sv @ivar deferred: a deferred that will fire with first received content. @type deferred: L{defer.Deferred} cCstj|_dS(N(RRR(R((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyRscCs|jjdS(N(RRA(R((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyRscCs|jj|dS(N(RR%(RR<((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyt dataReceiveds(R(R)R*RRRC(((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyRBs  t ImmediatelyDisconnectingProtocolcBs eZdZdZdZRS(s A protocol that disconnect immediately on connection. It fires the C{connectionDisconnected} deferred of its factory on connetion lost. cCs|jjdS(N(RR"(R((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyRscCs|jjjddS(N(RtconnectionDisconnectedR%R(RR&((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyR's(R(R)R*RR'(((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyRDs cCstj}|jtjdtj}|j}||_||_|j||j |dtj }|j d|j d|j d|j|j|j|j|j|j|j |d|||fS(s Create a certificate for given C{organization} and C{organizationalUnit}. @return: a tuple of (key, request, certificate) objects. itmd5iii<(R tPKeyt generate_keytTYPE_RSAtX509Reqt get_subjecttOtOUt set_pubkeytsigntX509tset_serial_numbertgmtime_adj_notBeforetgmtime_adj_notAftert set_issuert set_subjectt get_pubkey(t organizationtorganizationalUnittpkeytreqtsubjecttcert((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pytgenerateCertificateObjectss"          c Cst||\}}}xd|tjfd|tjfd|tjfgD]Y\}}}tjj||f} t| d} | j |tj || j qLWdS(s Create certificate files key, req and cert prefixed by C{basename} for given C{organization} and C{organizationalUnit}. tkeyRZR\twN( R]R tdump_privatekeytdump_certificate_requesttdump_certificatetostextseptjointfileR@t FILETYPE_PEMtclose( tbasenameRWRXRYRZR\texttobjtdumpFunctfNametfObj((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pytgenerateCertificateFiless"tContextGeneratingMixincBs eZdZdZdZRS(s` Offer methods to create L{ssl.DefaultOpenSSLContextFactory} for both client and server. @ivar clientBase: prefix of client certificate files. @type clientBase: C{str} @ivar serverBase: prefix of server certificate files. @type serverBase: C{str} @ivar clientCtxFactory: a generated context factory to be used in C{reactor.connectSSL}. @type clientCtxFactory: L{ssl.DefaultOpenSSLContextFactory} @ivar serverCtxFactory: a generated context factory to be used in C{reactor.listenSSL}. @type serverCtxFactory: L{ssl.DefaultOpenSSLContextFactory} cOsb|j}t|||tjtjj|dftjj|df||}||fS(NR^R\(tmktempRoRtDefaultOpenSSLContextFactoryRcRdRe(RtorgtorgUnittargstkwArgstbasetserverCtxFactory((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pytmakeContextFactorys  cCs@|j||\|_|_|j||\|_|_dS(N(Ryt clientBasetclientCtxFactoryt serverBaseRx(Rt clientArgst clientKwArgst serverArgst serverKwArgs((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pytsetupServerAndClients(R(R)R*RyR(((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyRps R6cBseZdZeZdZRS(sU A context factory with a default method set to L{SSL.TLSv1_METHOD}. cOs'tj|dtisClientR(((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyR6stStolenTCPTestCasecBs8eZdZdZdZdZdZdZRS(sc For SSL transports, test many of the same things which are tested for TCP transports. cCsCtjjttj}|j}tj|||d|S(sY Create an SSL server with a certificate using L{IReactorSSL.listenSSL}. t interface( RtPrivateCertificatetloadPEMRfR7treadtoptionsRt listenSSL(Rtaddresst portNumberRR\tcontextFactory((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyt createServers cCstj}|j|||S(sG Create an SSL client using L{IReactorSSL.connectSSL}. (RtCertificateOptionst connectSSL(RRRt clientCreatorR((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyt connectClient s cCstjS(s Return L{SSL.Error} as the expected error type which will be raised by a write to the L{OpenSSL.SSL.Connection} object after it has been closed. (R tError(R((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pytgetHandleExceptionType(ss0twisted.internet.iocpreactor.reactor.IOCPReactorcCsAttt}tjdkr:||jkr:tjSdgS(s Return the argument L{SSL.Error} will be constructed with for this case. This is basically just a random OpenSSL implementation detail. It would be better if this test worked in a way which did not require this. twin32s SSL routinest SSL_writesprotocol is shutdown(s SSL routinesRsprotocol is shutdown(RRRR tgetTypet_iocpterrnot WSAENOTSOCK(Rtname((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pytgetHandleErrorCode3s!(R(R)R*RRRRR(((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyRs   t TLSTestCasecBsPeZdZeZdZdZdZedZ dZ dZ dZ RS(s Tests for startTLS support. @ivar fillBuffer: forwarded to L{LineCollector.fillBuffer} @type fillBuffer: C{bool} cCsN|jjdk r%|jjjn|jjdk rJ|jjjndS(N(t clientProtoRRR"t serverProto(R((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyttearDownXscs|_tj}|_fd|_|r@t|_n t|_|_tj }|_ fd|_|rt|_n t|_t j d|dd}|j |jt jd|jj|tjjjgS(s Helper method to run TLS tests. @param clientProto: protocol instance attached to the client connection. @param serverProto: protocol instance attached to the server connection. @param clientIsServer: flag indicated if client should initiate startTLS instead of server. @return: a L{defer.Deferred} that will fire when both connections are lost. csS(N(((R(s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pytoscsS(N(((R(s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyRwsiRs 127.0.0.1(RRt ClientFactoryt clientFactoryR>R8tTrueR Rt ServerFactoryt serverFactoryRt listenTCPt addCleanupt stopListeningt connectTCPtgetHosttportRt gatherResultsR(RRRtclientIsServertcftsfR((RRs9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyt_runTest_s       cs=fd}jtttj}|j|S(s~ Test for server and client startTLS: client should received data both before and after the startTLS. cs$jjjtjtjdS(N(t assertEqualsRR0RRR!(tignore(R(s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pytchecks (RRR+RR-t addCallback(RRtd((Rs9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyttest_TLSs cs=fd}jtttj}|j|S(s Test for server startTLS not followed by a startTLS in client: the data received after server startTLS should be received as raw. cs3jjjtjjjjddS(NsNo encrypted bytes received(RRR0RRt failUnlessR/(tignored(R(s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyRs   (RRR+R>R-R(RRR((Rs9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyt test_unTLSs cs@fd}jttjtt}|j|S(s: Test startTLS first initiated by client. cs$jjjtjtjdS(N(RRR0RRR!(R(R(s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyRs (RR+RR-RR(RRR((Rs9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyttest_backwardsTLSsN( R(R)R*R>R-RRRRRRRR(((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyRKs  &  tSpammyTLSTestCasecBseZdZeZRS(sA Test TLS features with bytes sitting in the out buffer. (R(R)R*RR-(((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyRstBufferingTestCasecBs&eZdZdZdZdZRS(cCsN|jjdk r%|jjjn|jjdk rJ|jjjndS(N(RRRR"R(R((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyRscst|_t|_tj}tj}|_fd|_fd|_tj t t }tj }t j d||dd}|j|jt jd|jj||jj|jdS(NcsS(N(((R(s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyRscsS(N(((R(s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyRsiRs 127.0.0.1s+OK (R?RRBRRRRR RRrR7tClientContextFactoryRRRRRRRRRR(RR8R tsCTXtcCTXR((RRs9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyttest_openSSLBufferings   N(R(R)RRRRR(((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyRs tConnectionLostTestCasecBs#eZdZdZdZRS(csd}j||dfi||dfitj}tj|_tjd|j_}tj}t |_t j |_ tj d|jj|j}|j jfdS(Nstwisted.test.test_ssls, clients, serveris 127.0.0.1cs jjS(N(t serverPortR(t ignoredResult(R(s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyRs(RRRtProtocolRRRxRRRDRRRERRRR{R(RRstserverProtocolFactoryRtclientProtocolFactorytclientConnector((Rs9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyttestImmediateDisconnects       c sGd}|j||dfi||dfid}|jjjtj|tj}tj |j _ tj }fd|_t jd||j|_}tj}tj |j _ tj}fd|_t jd|jj||j}tj||gd t} | j|jS( Nstwisted.test.test_ssls, clients, servercWstS(N(R>(ta((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pytverifyscsS(N(((tserverProtocol(s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyRsicsS(N(((tclientProtocol(s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyRss 127.0.0.1t consumeErrors(RR{t getContextt set_verifyR t VERIFY_PEERRRRRR%R'RRRRxRRRRRt DeferredListRRt _cbLostConns( RRsRtserverConnLostRRtclientConnLostRRtdl((RRs9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyttestFailedVerifys,           cCs|\\}}\}}|j||j|tjg}tjrjddlm}|j|n|j||j||j j S(Ni(tConnectionLost( tfailIfR RR t isWindowsttwisted.internet.errorRR4ttrapRR(RtresultstsSuccesstsResulttcSuccesstcResulttacceptableErrorsR((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyRs     (R(R)RRR(((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyRs  t FakeContextcBs2eZdZdZdZdZdZRS(sK L{OpenSSL.SSL.Context} double which can more easily be inspected. cCs||_d|_dS(Ni(t_methodt_options(Rtmethod((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyR5s cCs|j|O_dS(N(R(RR((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyt set_options:scCsdS(N((RtfileName((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pytuse_certificate_file>scCsdS(N((RR((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pytuse_privatekey_fileBs(R(R)R*RRRR(((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyR1s    t!DefaultOpenSSLContextFactoryTestscBs2eZdZdZdZdZdZRS(s8 Tests for L{ssl.DefaultOpenSSLContextFactory}. cCs1tjttdt|_|jj|_dS(Nt_contextFactory(RRrR7RRRtcontext(R((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pytsetUpKscCsk|j|jjtj|j|jjtj@|j|jjtj @|j|jjtj @dS(s L{ssl.DefaultOpenSSLContextFactory.getContext} returns an SSL context which can use SSLv3 or TLSv1 but not SSLv2. N( t assertEqualRRR t SSLv23_METHODt assertTrueRt OP_NO_SSLv2t assertFalset OP_NO_SSLv3t OP_NO_TLSv1(R((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyt test_methodSscCs&|jtjtjt|jdS(s Instantiating L{ssl.DefaultOpenSSLContextFactory} with a certificate filename which does not identify an existing file results in the initializer raising L{OpenSSL.SSL.Error}. N(t assertRaisesR RRRrR7Rq(R((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyttest_missingCertificateFilecscCs&|jtjtj|jtdS(s Instantiating L{ssl.DefaultOpenSSLContextFactory} with a private key filename which does not identify an existing file results in the initializer raising L{OpenSSL.SSL.Error}. N(RR RRRrRqR7(R((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyttest_missingPrivateKeyFilens(R(R)R*RRRR(((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyRGs    tClientContextFactoryTestscBs eZdZdZdZRS(s0 Tests for L{ssl.ClientContextFactory}. cCs1tj|_t|j_|jj|_dS(N(RRRRRRR(R((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyR~s cCsk|j|jjtj|j|jjtj@|j|jjtj @|j|jjtj @dS(s~ L{ssl.ClientContextFactory.getContext} returns a context which can use SSLv3 or TLSv1 but not SSLv2. N( RRRR RRRRRRR(R((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyRs(R(R)R*RR(((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pyRzs s2Reactor does not support SSL, cannot run SSL tests(((:R*t twisted.trialRttwisted.internetRRRRttwisted.protocolsRttwisted.pythonRttwisted.python.reflectRRttwisted.python.runtimeR ttwisted.test.test_tcpR R RcRtOpenSSLR R Rttwisted.test.ssl_helpersRt ImportErrorRtsibpatht__file__R7t LineReceiverRR+RR?RBRDR]RoRpRRrR6tTestCaseRRRRRRRRt IReactorSSLttCasetskip(((s9/usr/lib/python2.7/dist-packages/twisted/test/test_ssl.pytsR"   *3   (  ;i#R3