test the support for ssl and sockets the custom key and certificate files used in testssl are generated using libtestcertdatamakesslcerts py other certificates are simply fetched from the internet servers they are meant to authenticate empty crl two keys and certs signed by the same ca for sni tests same certificate as pycacert pem but without extra text in file cert with all kinds of subject alt names not defined in all versions of openssl ubuntu has patched openssl and changed behavior of security level 2 see https bugs python orgissue41561msg389003 assume that any references of ubuntu implies ubuntulike distro the workaround is not required for 18 04 but doesn t hurt either lower security level to 1 and allow all ciphers for tls 1 01 for ctx in ctxs if hasattrctx minimumversion and ctx minimumversion ssl tlsversion tlsv11 ctx setciphersseclevel1 all else def seclevelworkaroundctxs pass def hastlsprotocolprotocol if isinstanceprotocol str assert protocol startswith protocol protocol getattrssl protocol none if protocol is none return false if protocol in ssl protocoltls ssl protocoltlsserver ssl protocoltlsclient autonegotiate protocols are always available return true name protocol name return hastlsversionnamelen protocol functools lrucache def hastlsversionversion if isinstanceversion str version ssl tlsversion membersversion check compile time flags like ssl hastlsv12 if not getattrssl f hasversion name return false if isopenssl300 and version ssl tlsversion tlsv12 bpo43791 3 0 0alpha14 fails with tlsv1alertinternalerror return false check runtime and dynamic crypto policy settings a tls version may be compiled in but disabled by a policy or config option ctx ssl sslcontextssl protocoltlsclient if hasattrctx minimumversion and ctx minimumversion ssl tlsversion minimumsupported and version ctx minimumversion return false if hasattrctx maximumversion and ctx maximumversion ssl tlsversion maximumsupported and version ctx maximumversion return false return true def requirestlsversionversion def decoratorfunc functools wrapsfunc def wrapperargs kw if not hastlsversionversion raise unittest skiptestfversion is not available else return funcargs kw return wrapper return decorator def handleerrorprefix excformat jointraceback formatexceptionsys exception if support verbose sys stdout writeprefix excformat def utcoffset note ignore issues like 1647654 local time utc time utc offset if time daylight and time localtime tmisdst 0 return time altzone seconds return time timezone ignoredeprecation warningshelper ignorewarnings categorydeprecationwarning def testwrapsocketsock certreqsssl certnone cacertsnone ciphersnone certfilenone keyfilenone kwargs if not kwargs getserverside kwargsserverhostname signedcertfilehostname context ssl sslcontextssl protocoltlsclient else context ssl sslcontextssl protocoltlsserver if certreqs is not none if certreqs ssl certnone context checkhostname false context verifymode certreqs if cacerts is not none context loadverifylocationscacerts if certfile is not none or keyfile is not none context loadcertchaincertfile keyfile if ciphers is not none context setciphersciphers return context wrapsocketsock kwargs def testingcontextservercertsignedcertfile serverchaintrue if servercert signedcertfile hostname signedcertfilehostname elif servercert signedcertfile2 hostname signedcertfile2hostname elif servercert nosanfile hostname nosanhostname else raise valueerrorservercert clientcontext ssl sslcontextssl protocoltlsclient clientcontext loadverifylocationssigningca servercontext ssl sslcontextssl protocoltlsserver servercontext loadcertchainservercert if serverchain servercontext loadverifylocationssigningca return clientcontext servercontext hostname class basicsockettestsunittest testcase def testconstantsself ssl certnone ssl certoptional ssl certrequired ssl opcipherserverpreference ssl opsingledhuse ssl opsingleecdhuse ssl opnocompression self assertequalssl hassni true self assertequalssl hasecdh true self assertequalssl hastlsv12 true self assertequalssl hastlsv13 true ssl opnosslv2 ssl opnosslv3 ssl opnotlsv1 ssl opnotlsv13 ssl opnotlsv11 ssl opnotlsv12 self assertequalssl protocoltls ssl protocolsslv23 def testoptionsself gh106687 ssl options values are unsigned integer uint64t for name in dirssl if not name startswith op continue with self subtestoptionname value getattrssl name self assertgreaterequalvalue 0 fssl name def testssltypesself ssltypes ssl sslcontext ssl sslsocket ssl memorybio ssl certificate ssl sslsession ssl sslerror for ssltype in ssltypes with self subtestssltypessltype with self assertraisesregextypeerror immutable type ssltype value none support checkdisallowinstantiationself ssl certificate def testprivateinitself with self assertraisesregextypeerror public constructor with socket socket as s ssl sslsockets def teststrforenumsself make sure that the protocol constants have enumlike string reprs proto ssl protocoltlsclient self assertequalreprproto sslmethod protocoltlsclient r proto value self assertequalstrproto strproto value ctx ssl sslcontextproto self assertisctx protocol proto def testrandomself v ssl randstatus if support verbose sys stdout writen randstatus is d sn v v and sufficient randomness or insufficient randomness if v data ssl randbytes16 self assertequallendata 16 else self assertraisesssl sslerror ssl randbytes 16 negative num is invalid self assertraisesvalueerror ssl randbytes 5 ssl randaddthis is a random string 75 0 ssl randaddbthis is a random bytes object 75 0 ssl randaddbytearraybthis is a random bytearray object 75 0 def testparsecertself note that this uses an unofficial function in ssl c provided solely for this test to exercise the certificate parsing code self assertequal ssl ssl testdecodecertcertfile certfileinfo self assertequal ssl ssl testdecodecertsignedcertfile signedcertfileinfo issue 13034 the subjectaltname in some certificates notably projects developer nokia com 443 wasn t parsed p ssl ssl testdecodecertnokiacert if support verbose sys stdout writen pprint pformatp n self assertequalp subjectaltname dns projects developer nokia com dns projects forum nokia com extra ocsp and aia fields self assertequalp ocsp http ocsp verisign com self assertequalp caissuers http svrintlg3aia verisign comsvrintlg3 cer self assertequalp crldistributionpoints http svrintlg3crl verisign comsvrintlg3 crl def testparsecertcve20195010self p ssl ssl testdecodecerttalosinvalidcrldp if support verbose sys stdout writen pprint pformatp n self assertequal p issuer countryname uk commonname codyca notafter jun 14 18 00 58 2028 gmt notbefore jun 18 18 00 58 2018 gmt serialnumber 02 subject countryname uk commonname codenomiconvm2 test lal cisco com subjectaltname dns codenomiconvm2 test lal cisco com version 3 def testparsecertcve20134238self p ssl ssl testdecodecertnullbytecert if support verbose sys stdout writen pprint pformatp n subject countryname us stateorprovincename oregon localityname beaverton organizationname python software foundation organizationalunitname python core development commonname null python orgx00example org emailaddress pythondevpython org self assertequalp subject subject self assertequalp issuer subject if ssl opensslapiversion 0 9 8 san dns altnull python orgx00example com email nullpython orgx00userexample org uri http null python orgx00http example org ip address 192 0 2 1 ip address 2001 db8 0 0 0 0 0 1 else openssl 0 9 7 doesn t support ipv6 addresses in subjectaltname san dns altnull python orgx00example com email nullpython orgx00userexample org uri http null python orgx00http example org ip address 192 0 2 1 ip address invalid self assertequalp subjectaltname san def testparseallsansself p ssl ssl testdecodecertallsanfile self assertequalp subjectaltname dns allsans othername unsupported othername unsupported email userexample org dns www example org dirname countryname xy localityname castle anthrax organizationname python software foundation commonname dirname example uri https www python org ip address 127 0 0 1 ip address 0 0 0 0 0 0 0 1 registered id 1 2 3 4 5 def testdertopemself with opencafilecacert r as f pem f read d1 ssl pemcerttodercertpem p2 ssl dercerttopemcertd1 d2 ssl pemcerttodercertp2 self assertequald1 d2 if not p2 startswithssl pemheader n self faildertopem didn t include correct header nrn p2 if not p2 endswith n ssl pemfooter n self faildertopem didn t include correct footer nrn p2 def testopensslversionself n ssl opensslversionnumber t ssl opensslversioninfo s ssl opensslversion self assertisinstancen int self assertisinstancet tuple self assertisinstances str some sanity checks follow 1 1 1 self assertgreaterequaln 0x10101000 4 0 self assertlessn 0x40000000 major minor fix patch status t self assertgreaterequalmajor 1 self assertlessmajor 4 self assertgreaterequalminor 0 self assertlessminor 256 self assertgreaterequalfix 0 self assertlessfix 256 self assertgreaterequalpatch 0 self assertlessequalpatch 63 self assertgreaterequalstatus 0 self assertlessequalstatus 15 libresslver flibressl major d if major 3 3 x uses 0xmnn00pp0l opensslver fopenssl major d minor d patch d else opensslver fopenssl major d minor d fix d self asserttrue s startswithopensslver libresslver s t hexn support cpythononly def testrefcycleself issue 7943 an ssl object doesn t create reference cycles with itself s socket socketsocket afinet ss testwrapsockets wr weakref refss with warningshelper checkwarnings resourcewarning del ss self assertequalwr none def testwrappedunconnectedself methods on an unconnected sslsocket propagate the original oserror raise by the underlying socket object s socket socketsocket afinet with testwrapsockets as ss self assertraisesoserror ss recv 1 self assertraisesoserror ss recvinto bytearrayb x self assertraisesoserror ss recvfrom 1 self assertraisesoserror ss recvfrominto bytearrayb x 1 self assertraisesoserror ss send b x self assertraisesoserror ss sendto b x 0 0 0 0 0 self assertraisesnotimplementederror ss dup self assertraisesnotimplementederror ss sendmsg b x 0 0 0 0 0 0 self assertraisesnotimplementederror ss recvmsg 100 self assertraisesnotimplementederror ss recvmsginto bytearray100 def testtimeoutself issue 8524 when creating an ssl socket the timeout of the original socket should be retained for timeout in none 0 0 5 0 s socket socketsocket afinet s settimeouttimeout with testwrapsockets as ss self assertequaltimeout ss gettimeout def testopenssl111deprecationsself options ssl opnotlsv1 ssl opnotlsv11 ssl opnotlsv12 ssl opnotlsv13 protocols ssl protocoltlsv1 ssl protocoltlsv11 ssl protocoltlsv12 ssl protocoltls versions ssl tlsversion sslv3 ssl tlsversion tlsv1 ssl tlsversion tlsv11 for option in options with self subtestoptionoption ctx ssl sslcontextssl protocoltlsclient with self assertwarnsdeprecationwarning as cm ctx options option self assertequal ssl opnosslssl opnotls options are deprecated strcm warning for protocol in protocols if not hastlsprotocolprotocol continue with self subtestprotocolprotocol with self assertwarnsdeprecationwarning as cm ssl sslcontextprotocol self assertequal f ssl protocol name is deprecated strcm warning for version in versions if not hastlsversionversion continue with self subtestversionversion ctx ssl sslcontextssl protocoltlsclient with self assertwarnsdeprecationwarning as cm ctx minimumversion version versiontext s s version class name version name self assertequal f ssl versiontext is deprecated strcm warning def badcerttestself certfile wrapping with an empty cert file self badcerttestnullcert pem def testmalformedcertself wrapping with a badly formatted key syntax error self badcerttestbadkey pem def testserversideself serverhostname doesn t work for server sockets ctx ssl sslcontextssl protocoltlsserver with socket socket as sock self assertraisesvalueerror ctx wrapsocket sock true serverhostnamesome hostname def testunknownchannelbindingself should raise valueerror for unknown type s socket createserver 127 0 0 1 0 c socket socketsocket afinet c connects getsockname with testwrapsocketc dohandshakeonconnectfalse as ss with self assertraisesvalueerror ss getchannelbindingunknowntype s close unittest skipunlesstlsunique in ssl channelbindingtypes tlsunique channel binding not available def testtlsuniquechannelbindingself unconnected should return none for known type s socket socketsocket afinet with testwrapsockets as ss self assertisnoness getchannelbindingtlsunique the same for serverside s socket socketsocket afinet with testwrapsockets serversidetrue certfilecertfile as ss self assertisnoness getchannelbindingtlsunique def testdeallocwarnself ss testwrapsocketsocket socketsocket afinet r reprss with self assertwarnsresourcewarning as cm ss none support gccollect self assertinr strcm warning args0 def testgetdefaultverifypathsself paths ssl getdefaultverifypaths self assertequallenpaths 6 self assertisinstancepaths ssl defaultverifypaths with oshelper environmentvarguard as env envsslcertdir capath envsslcertfile certfile paths ssl getdefaultverifypaths self assertequalpaths cafile certfile self assertequalpaths capath capath unittest skipunlesssys platform win32 windows specific def testenumcertificatesself self asserttruessl enumcertificatesca self asserttruessl enumcertificatesroot self assertraisestypeerror ssl enumcertificates self assertraiseswindowserror ssl enumcertificates trustoids set for storename in ca root store ssl enumcertificatesstorename self assertisinstancestore list for element in store self assertisinstanceelement tuple self assertequallenelement 3 cert enc trust element self assertisinstancecert bytes self assertinenc x509asn pkcs7asn self assertisinstancetrust frozenset set bool if isinstancetrust frozenset set trustoids updatetrust serverauth 1 3 6 1 5 5 7 3 1 self assertinserverauth trustoids unittest skipunlesssys platform win32 windows specific def testenumcrlsself self asserttruessl enumcrlsca self assertraisestypeerror ssl enumcrls self assertraiseswindowserror ssl enumcrls crls ssl enumcrlsca self assertisinstancecrls list for element in crls self assertisinstanceelement tuple self assertequallenelement 2 self assertisinstanceelement0 bytes self assertinelement1 x509asn pkcs7asn def testasn1objectself expected 129 serverauth tls web server authentication 1 3 6 1 5 5 7 3 1 val ssl asn1object 1 3 6 1 5 5 7 3 1 self assertequalval expected self assertequalval nid 129 self assertequalval shortname serverauth self assertequalval longname tls web server authentication self assertequalval oid 1 3 6 1 5 5 7 3 1 self assertisinstanceval ssl asn1object self assertraisesvalueerror ssl asn1object serverauth val ssl asn1object fromnid129 self assertequalval expected self assertisinstanceval ssl asn1object self assertraisesvalueerror ssl asn1object fromnid 1 with self assertraisesregexvalueerror unknown nid 100000 ssl asn1object fromnid100000 for i in range1000 try obj ssl asn1object fromnidi except valueerror pass else self assertisinstanceobj nid int self assertisinstanceobj shortname str self assertisinstanceobj longname str self assertisinstanceobj oid str typenone val ssl asn1object fromname tls web server authentication self assertequalval expected self assertisinstanceval ssl asn1object self assertequalssl asn1object fromname serverauth expected self assertequalssl asn1object fromname 1 3 6 1 5 5 7 3 1 expected with self assertraisesregexvalueerror unknown object serverauth ssl asn1object fromname serverauth def testpurposeenumself val ssl asn1object 1 3 6 1 5 5 7 3 1 self assertisinstancessl purpose serverauth ssl asn1object self assertequalssl purpose serverauth val self assertequalssl purpose serverauth nid 129 self assertequalssl purpose serverauth shortname serverauth self assertequalssl purpose serverauth oid 1 3 6 1 5 5 7 3 1 val ssl asn1object 1 3 6 1 5 5 7 3 2 self assertisinstancessl purpose clientauth ssl asn1object self assertequalssl purpose clientauth val self assertequalssl purpose clientauth nid 130 self assertequalssl purpose clientauth shortname clientauth self assertequalssl purpose clientauth oid 1 3 6 1 5 5 7 3 2 def testunsupporteddtlsself s socket socketsocket afinet socket sockdgram self addcleanups close with self assertraisesnotimplementederror as cx testwrapsockets certreqsssl certnone self assertequalstrcx exception only stream sockets are supported ctx ssl sslcontextssl protocoltlsclient with self assertraisesnotimplementederror as cx ctx wrapsockets self assertequalstrcx exception only stream sockets are supported def certtimeokself timestring timestamp self assertequalssl certtimetosecondstimestring timestamp def certtimefailself timestring with self assertraisesvalueerror ssl certtimetosecondstimestring unittest skipunlessutcoffset local time needs to be different from utc def testcerttimetosecondstimezoneself issue 19940 ssl certtimetoseconds returns wrong results if local timezone is not utc self certtimeokmay 9 00 00 00 2007 gmt 1178668800 0 self certtimeokjan 5 09 34 43 2018 gmt 1515144883 0 def testcerttimetosecondsself timestring jan 5 09 34 43 2018 gmt ts 1515144883 0 self certtimeoktimestring ts accept keyword parameter assert its name self assertequalssl certtimetosecondscerttimetimestring ts accept both e and d space or zero generated by strftime self certtimeokjan 05 09 34 43 2018 gmt ts caseinsensitive self certtimeokjan 5 09 34 43 2018 gmt ts self certtimefailjan 5 09 34 2018 gmt no seconds self certtimefailjan 5 09 34 43 2018 no gmt self certtimefailjan 5 09 34 43 2018 utc not gmt timezone self certtimefailjan 35 09 34 43 2018 gmt invalid day self certtimefailjon 5 09 34 43 2018 gmt invalid month self certtimefailjan 5 24 00 00 2018 gmt invalid hour self certtimefailjan 5 09 60 43 2018 gmt invalid minute newyearts 1230768000 0 leap seconds self certtimeokdec 31 23 59 60 2008 gmt newyearts same timestamp self certtimeokjan 1 00 00 00 2009 gmt newyearts self certtimeokjan 5 09 34 59 2018 gmt 1515144899 allow 60th second even if it is not a leap second self certtimeokjan 5 09 34 60 2018 gmt 1515144900 allow 2nd leap second for compatibility with time strptime self certtimeokjan 5 09 34 61 2018 gmt 1515144901 self certtimefailjan 5 09 34 62 2018 gmt invalid seconds no special treatment for the special value 99991231235959z rfc 5280 self certtimeokdec 31 23 59 59 9999 gmt 253402300799 0 support runwithlocale lcall def testcerttimetosecondslocaleself certtimetoseconds should be locale independent def localfebruaryname return time strftime b 1 2 3 4 5 6 0 0 0 if localfebruaryname lower feb self skiptestlocalespecific month name needs to be different from c locale localeindependent self certtimeokfeb 9 00 00 00 2007 gmt 1170979200 0 self certtimefaillocalfebruaryname 9 00 00 00 2007 gmt def testconnectexerrorself server socket socketsocket afinet self addcleanupserver close port sockethelper bindportserver reserve port but don t listen s testwrapsocketsocket socketsocket afinet certreqsssl certrequired self addcleanups close rc s connectexhost port issue 19919 windows machines or vms hosted on windows machines sometimes return ewouldblock errors errno econnrefused errno ehostunreach errno etimedout errno ewouldblock self assertinrc errors def testreadwritezeroself empty reads and writes now work bpo42854 bpo31711 clientcontext servercontext hostname testingcontext server threadedechoservercontextservercontext with server with clientcontext wrapsocketsocket socket serverhostnamehostname as s s connecthost server port self assertequals recv0 b self assertequals sendb 0 class contexttestsunittest testcase def testconstructorself for protocol in protocols if hastlsprotocolprotocol with warningshelper checkwarnings ctx ssl sslcontextprotocol self assertequalctx protocol protocol with warningshelper checkwarnings ctx ssl sslcontext self assertequalctx protocol ssl protocoltls self assertraisesvalueerror ssl sslcontext 1 self assertraisesvalueerror ssl sslcontext 42 def testciphersself ctx ssl sslcontextssl protocoltlsclient ctx setciphersall ctx setciphersdefault with self assertraisesregexssl sslerror no cipher can be selected ctx setciphers dorothyx unittest skipunlesspyssldefaultciphers 1 test applies only to python default ciphers def testpythonciphersself ctx ssl sslcontextssl protocoltlsclient ciphers ctx getciphers for suite in ciphers name suite name self assertnotinpsk name self assertnotinsrp name self assertnotinmd5 name self assertnotinrc4 name self assertnotin3des name def testgetciphersself ctx ssl sslcontextssl protocoltlsclient ctx setciphers aesgcm names setd name for d in ctx getciphers expected aes128gcmsha256 ecdheecdsaaes128gcmsha256 ecdhersaaes128gcmsha256 dhersaaes128gcmsha256 aes256gcmsha384 ecdheecdsaaes256gcmsha384 ecdhersaaes256gcmsha384 dhersaaes256gcmsha384 intersection names intersectionexpected self assertgreaterequal lenintersection 2 fngot sortednamesnexpected sortedexpected def testoptionsself test default sslcontext options ctx ssl sslcontextssl protocoltlsclient opall opnosslv2 opnosslv3 is the default value default ssl opall ssl opnosslv2 ssl opnosslv3 sslcontext also enables these by default default opnocompression opcipherserverpreference opsingledhuse opsingleecdhuse openablemiddleboxcompat self assertequaldefault ctx options disallow tlsv1 with warningshelper checkwarnings ctx options ssl opnotlsv1 self assertequaldefault ssl opnotlsv1 ctx options allow tlsv1 with warningshelper checkwarnings ctx options ctx options ssl opnotlsv1 self assertequaldefault ctx options clear all options ctx options 0 ubuntu has opnosslv3 forced on by default self assertequal0 ctx options ssl opnosslv3 invalid options with self assertraisesoverflowerror ctx options 1 with self assertraisesoverflowerror ctx options 2 100 with self assertraisestypeerror ctx options abc def testverifymodeprotocolself with warningshelper checkwarnings ctx ssl sslcontextssl protocoltls default value self assertequalctx verifymode ssl certnone ctx verifymode ssl certoptional self assertequalctx verifymode ssl certoptional ctx verifymode ssl certrequired self assertequalctx verifymode ssl certrequired ctx verifymode ssl certnone self assertequalctx verifymode ssl certnone with self assertraisestypeerror ctx verifymode none with self assertraisesvalueerror ctx verifymode 42 ctx ssl sslcontextssl protocoltlsserver self assertequalctx verifymode ssl certnone self assertfalsectx checkhostname ctx ssl sslcontextssl protocoltlsclient self assertequalctx verifymode ssl certrequired self asserttruectx checkhostname def testhostnamecheckscommonnameself ctx ssl sslcontextssl protocoltlsclient self asserttruectx hostnamecheckscommonname if ssl hasnevercheckcommonname ctx hostnamecheckscommonname true self asserttruectx hostnamecheckscommonname ctx hostnamecheckscommonname false self assertfalsectx hostnamecheckscommonname ctx hostnamecheckscommonname true self asserttruectx hostnamecheckscommonname else with self assertraisesattributeerror ctx hostnamecheckscommonname true ignoredeprecation def testminmaxversionself ctx ssl sslcontextssl protocoltlsserver openssl default is minimumsupported however some vendors like fedora override the setting to tls 1 0 minimumrange stock openssl ssl tlsversion minimumsupported fedora 29 uses tls 1 0 by default ssl tlsversion tlsv1 rhel 8 uses tls 1 2 by default ssl tlsversion tlsv12 maximumrange stock openssl ssl tlsversion maximumsupported fedora 32 uses tls 1 3 by default ssl tlsversion tlsv13 self assertin ctx minimumversion minimumrange self assertin ctx maximumversion maximumrange ctx minimumversion ssl tlsversion tlsv11 ctx maximumversion ssl tlsversion tlsv12 self assertequal ctx minimumversion ssl tlsversion tlsv11 self assertequal ctx maximumversion ssl tlsversion tlsv12 ctx minimumversion ssl tlsversion minimumsupported ctx maximumversion ssl tlsversion tlsv1 self assertequal ctx minimumversion ssl tlsversion minimumsupported self assertequal ctx maximumversion ssl tlsversion tlsv1 ctx maximumversion ssl tlsversion maximumsupported self assertequal ctx maximumversion ssl tlsversion maximumsupported ctx maximumversion ssl tlsversion minimumsupported self assertin ctx maximumversion ssl tlsversion tlsv1 ssl tlsversion tlsv11 ssl tlsversion sslv3 ctx minimumversion ssl tlsversion maximumsupported self assertin ctx minimumversion ssl tlsversion tlsv12 ssl tlsversion tlsv13 with self assertraisesvalueerror ctx minimumversion 42 if hastlsprotocolssl protocoltlsv11 ctx ssl sslcontextssl protocoltlsv11 self assertin ctx minimumversion minimumrange self assertequal ctx maximumversion ssl tlsversion maximumsupported with self assertraisesvalueerror ctx minimumversion ssl tlsversion minimumsupported with self assertraisesvalueerror ctx maximumversion ssl tlsversion tlsv1 unittest skipunless hasattrssl sslcontext securitylevel requires openssl 1 1 0 def testsecuritylevelself ctx ssl sslcontextssl protocoltlsclient the default security callback allows for levels between 05 with openssl defaulting to 1 however some vendors override the default value e g debian defaults to 2 securitylevelrange 0 1 openssl default 2 debian 3 4 5 self assertinctx securitylevel securitylevelrange def testverifyflagsself ctx ssl sslcontextssl protocoltlsserver default value tf getattrssl verifyx509trustedfirst 0 self assertequalctx verifyflags ssl verifydefault tf ctx verifyflags ssl verifycrlcheckleaf self assertequalctx verifyflags ssl verifycrlcheckleaf ctx verifyflags ssl verifycrlcheckchain self assertequalctx verifyflags ssl verifycrlcheckchain ctx verifyflags ssl verifydefault self assertequalctx verifyflags ssl verifydefault ctx verifyflags ssl verifyallowproxycerts self assertequalctx verifyflags ssl verifyallowproxycerts supports any value ctx verifyflags ssl verifycrlcheckleaf ssl verifyx509strict self assertequalctx verifyflags ssl verifycrlcheckleaf ssl verifyx509strict with self assertraisestypeerror ctx verifyflags none def testloadcertchainself ctx ssl sslcontextssl protocoltlsserver combined key and cert in a single file ctx loadcertchaincertfile keyfilenone ctx loadcertchaincertfile keyfilecertfile self assertraisestypeerror ctx loadcertchain keyfilecertfile with self assertraisesoserror as cm ctx loadcertchainnonexistingcert self assertequalcm exception errno errno enoent with self assertraisesregexssl sslerror pem lib ctx loadcertchainbadcert with self assertraisesregexssl sslerror pem lib ctx loadcertchainemptycert separate key and cert ctx ssl sslcontextssl protocoltlsserver ctx loadcertchainonlycert onlykey ctx loadcertchaincertfileonlycert keyfileonlykey ctx loadcertchaincertfilebytesonlycert keyfilebytesonlykey with self assertraisesregexssl sslerror pem lib ctx loadcertchainonlycert with self assertraisesregexssl sslerror pem lib ctx loadcertchainonlykey with self assertraisesregexssl sslerror pem lib ctx loadcertchaincertfileonlykey keyfileonlycert mismatching key and cert ctx ssl sslcontextssl protocoltlsserver with self assertraisesregexssl sslerror key values mismatch ctx loadcertchaincafilecacert onlykey password protected key and cert ctx loadcertchaincertfileprotected passwordkeypassword ctx loadcertchaincertfileprotected passwordkeypassword encode ctx loadcertchaincertfileprotected passwordbytearraykeypassword encode ctx loadcertchainonlycert onlykeyprotected keypassword ctx loadcertchainonlycert onlykeyprotected keypassword encode ctx loadcertchainonlycert onlykeyprotected bytearraykeypassword encode with self assertraisesregextypeerror should be a string ctx loadcertchaincertfileprotected passwordtrue with self assertraisesssl sslerror ctx loadcertchaincertfileprotected passwordbadpass with self assertraisesregexvalueerror cannot be longer openssl has a fixed limit on the password buffer pembufsize is generally set to 1kb return a string larger than this ctx loadcertchaincertfileprotected passwordb a 102400 password callback def getpassunicode return keypassword def getpassbytes return keypassword encode def getpassbytearray return bytearraykeypassword encode def getpassbadpass return badpass def getpasshuge return b a 1024 1024 def getpassbadtype return 9 def getpassexception raise exception getpass error class getpasscallable def callself return keypassword def getpassself return keypassword ctx loadcertchaincertfileprotected passwordgetpassunicode ctx loadcertchaincertfileprotected passwordgetpassbytes ctx loadcertchaincertfileprotected passwordgetpassbytearray ctx loadcertchaincertfileprotected passwordgetpasscallable ctx loadcertchaincertfileprotected passwordgetpasscallable getpass with self assertraisesssl sslerror ctx loadcertchaincertfileprotected passwordgetpassbadpass with self assertraisesregexvalueerror cannot be longer ctx loadcertchaincertfileprotected passwordgetpasshuge with self assertraisesregextypeerror must return a string ctx loadcertchaincertfileprotected passwordgetpassbadtype with self assertraisesregexexception getpass error ctx loadcertchaincertfileprotected passwordgetpassexception make sure the password function isn t called if it isn t needed ctx loadcertchaincertfile passwordgetpassexception def testloadverifylocationsself ctx ssl sslcontextssl protocoltlsserver ctx loadverifylocationscertfile ctx loadverifylocationscafilecertfile capathnone ctx loadverifylocationsbytescertfile ctx loadverifylocationscafilebytescertfile capathnone self assertraisestypeerror ctx loadverifylocations self assertraisestypeerror ctx loadverifylocations none none none with self assertraisesoserror as cm ctx loadverifylocationsnonexistingcert self assertequalcm exception errno errno enoent with self assertraisesregexssl sslerror pem lib ctx loadverifylocationsbadcert ctx loadverifylocationscertfile capath ctx loadverifylocationscertfile capathbytescapath issue 10989 crash if the second argument type is invalid self assertraisestypeerror ctx loadverifylocations none true def testloadverifycadataself test cadata with opencafilecacert as f cacertpem f read cacertder ssl pemcerttodercertcacertpem with opencafileneuronio as f neuroniopem f read neuronioder ssl pemcerttodercertneuroniopem test pem ctx ssl sslcontextssl protocoltlsclient self assertequalctx certstorestatsx509ca 0 ctx loadverifylocationscadatacacertpem self assertequalctx certstorestatsx509ca 1 ctx loadverifylocationscadataneuroniopem self assertequalctx certstorestatsx509ca 2 cert already in hash table ctx loadverifylocationscadataneuroniopem self assertequalctx certstorestatsx509ca 2 combined ctx ssl sslcontextssl protocoltlsclient combined n joincacertpem neuroniopem ctx loadverifylocationscadatacombined self assertequalctx certstorestatsx509ca 2 with junk around the certs ctx ssl sslcontextssl protocoltlsclient combined head cacertpem other neuroniopem again neuroniopem tail ctx loadverifylocationscadatan joincombined self assertequalctx certstorestatsx509ca 2 test der ctx ssl sslcontextssl protocoltlsclient ctx loadverifylocationscadatacacertder ctx loadverifylocationscadataneuronioder self assertequalctx certstorestatsx509ca 2 cert already in hash table ctx loadverifylocationscadatacacertder self assertequalctx certstorestatsx509ca 2 combined ctx ssl sslcontextssl protocoltlsclient combined b joincacertder neuronioder ctx loadverifylocationscadatacombined self assertequalctx certstorestatsx509ca 2 error cases ctx ssl sslcontextssl protocoltlsclient self assertraisestypeerror ctx loadverifylocations cadataobject with self assertraisesregex ssl sslerror no start line cadata does not contain a certificate ctx loadverifylocationscadatabroken with self assertraisesregex ssl sslerror not enough data cadata does not contain a certificate ctx loadverifylocationscadatabbroken with self assertraisesssl sslerror ctx loadverifylocationscadatacacertder ba unittest skipifpydebugwin32 avoid mixing debugrelease crt on windows def testloaddhparamsself ctx ssl sslcontextssl protocoltlsserver ctx loaddhparamsdhfile if os name nt ctx loaddhparamsbytesdhfile self assertraisestypeerror ctx loaddhparams self assertraisestypeerror ctx loaddhparams none with self assertraisesfilenotfounderror as cm ctx loaddhparamsnonexistingcert self assertequalcm exception errno errno enoent with self assertraisesssl sslerror as cm ctx loaddhparamscertfile def testsessionstatsself for proto in ssl protocoltlsclient ssl protocoltlsserver ctx ssl sslcontextproto self assertequalctx sessionstats number 0 connect 0 connectgood 0 connectrenegotiate 0 accept 0 acceptgood 0 acceptrenegotiate 0 hits 0 misses 0 timeouts 0 cachefull 0 def testsetdefaultverifypathsself there s not much we can do to test that it acts as expected so just check it doesn t crash or raise an exception ctx ssl sslcontextssl protocoltlsclient ctx setdefaultverifypaths unittest skipunlessssl hasecdh ecdh disabled on this openssl build def testsetecdhcurveself ctx ssl sslcontextssl protocoltlsserver ctx setecdhcurveprime256v1 ctx setecdhcurvebprime256v1 self assertraisestypeerror ctx setecdhcurve self assertraisestypeerror ctx setecdhcurve none self assertraisesvalueerror ctx setecdhcurve foo self assertraisesvalueerror ctx setecdhcurve bfoo def testsnicallbackself ctx ssl sslcontextssl protocoltlsserver setservernamecallback expects a callable or none self assertraisestypeerror ctx setservernamecallback self assertraisestypeerror ctx setservernamecallback 4 self assertraisestypeerror ctx setservernamecallback self assertraisestypeerror ctx setservernamecallback ctx def dummycallbacksock servername ctx pass ctx setservernamecallbacknone ctx setservernamecallbackdummycallback def testsnicallbackrefcycleself reference cycles through the servername callback are detected and cleared ctx ssl sslcontextssl protocoltlsserver def dummycallbacksock servername ctx cyclectx pass ctx setservernamecallbackdummycallback wr weakref refctx del ctx dummycallback gc collect self assertiswr none def testcertstorestatsself ctx ssl sslcontextssl protocoltlsclient self assertequalctx certstorestats x509ca 0 crl 0 x509 0 ctx loadcertchaincertfile self assertequalctx certstorestats x509ca 0 crl 0 x509 0 ctx loadverifylocationscertfile self assertequalctx certstorestats x509ca 0 crl 0 x509 1 ctx loadverifylocationscafilecacert self assertequalctx certstorestats x509ca 1 crl 0 x509 2 def testgetcacertsself ctx ssl sslcontextssl protocoltlsclient self assertequalctx getcacerts certfile is not flagged as x509v3 basic constraints ca true ctx loadverifylocationscertfile self assertequalctx getcacerts but cafilecacert is a ca cert ctx loadverifylocationscafilecacert self assertequalctx getcacerts issuer organizationname root ca organizationalunitname http www cacert org commonname ca cert signing ity emailaddress supportcacert org notafter mar 29 12 29 49 2033 gmt notbefore mar 30 12 29 49 2003 gmt serialnumber 00 crldistributionpoints https www cacert orgrevoke crl subject organizationname root ca organizationalunitname http www cacert org commonname ca cert signing ity emailaddress supportcacert org version 3 with opencafilecacert as f pem f read der ssl pemcerttodercertpem self assertequalctx getcacertstrue der def testloaddefaultcertsself ctx ssl sslcontextssl protocoltlsclient ctx loaddefaultcerts ctx ssl sslcontextssl protocoltlsclient ctx loaddefaultcertsssl purpose serverauth ctx loaddefaultcerts ctx ssl sslcontextssl protocoltlsclient ctx loaddefaultcertsssl purpose clientauth ctx ssl sslcontextssl protocoltlsclient self assertraisestypeerror ctx loaddefaultcerts none self assertraisestypeerror ctx loaddefaultcerts serverauth unittest skipifsys platform win32 notwindows specific def testloaddefaultcertsenvself ctx ssl sslcontextssl protocoltlsclient with oshelper environmentvarguard as env envsslcertdir capath envsslcertfile certfile ctx loaddefaultcerts self assertequalctx certstorestats crl 0 x509 1 x509ca 0 unittest skipunlesssys platform win32 windows specific unittest skipifsupport pydebug debug build does not share environment between crts def testloaddefaultcertsenvwindowsself ctx ssl sslcontextssl protocoltlsclient ctx loaddefaultcerts stats ctx certstorestats ctx ssl sslcontextssl protocoltlsclient with oshelper environmentvarguard as env envsslcertdir capath envsslcertfile certfile ctx loaddefaultcerts statsx509 1 self assertequalctx certstorestats stats def assertcontextoptionsself ctx self assertequalctx options ssl opnosslv2 ssl opnosslv2 if opnocompression 0 self assertequalctx options opnocompression opnocompression if opsingledhuse 0 self assertequalctx options opsingledhuse opsingledhuse if opsingleecdhuse 0 self assertequalctx options opsingleecdhuse opsingleecdhuse if opcipherserverpreference 0 self assertequalctx options opcipherserverpreference opcipherserverpreference self assertequalctx options ssl oplegacyserverconnect 0 if isopenssl300 else ssl oplegacyserverconnect def testcreatedefaultcontextself ctx ssl createdefaultcontext self assertequalctx protocol ssl protocoltlsclient self assertequalctx verifymode ssl certrequired self asserttruectx checkhostname self assertcontextoptionsctx with opensigningca as f cadata f read ctx ssl createdefaultcontextcafilesigningca capathcapath cadatacadata self assertequalctx protocol ssl protocoltlsclient self assertequalctx verifymode ssl certrequired self assertcontextoptionsctx ctx ssl createdefaultcontextssl purpose clientauth self assertequalctx protocol ssl protocoltlsserver self assertequalctx verifymode ssl certnone self assertcontextoptionsctx def testcreatestdlibcontextself ctx ssl createstdlibcontext self assertequalctx protocol ssl protocoltlsclient self assertequalctx verifymode ssl certnone self assertfalsectx checkhostname self assertcontextoptionsctx if hastlsprotocolssl protocoltlsv1 with warningshelper checkwarnings ctx ssl createstdlibcontextssl protocoltlsv1 self assertequalctx protocol ssl protocoltlsv1 self assertequalctx verifymode ssl certnone self assertcontextoptionsctx with warningshelper checkwarnings ctx ssl createstdlibcontext ssl protocoltlsv12 certreqsssl certrequired checkhostnametrue self assertequalctx protocol ssl protocoltlsv12 self assertequalctx verifymode ssl certrequired self asserttruectx checkhostname self assertcontextoptionsctx ctx ssl createstdlibcontextpurposessl purpose clientauth self assertequalctx protocol ssl protocoltlsserver self assertequalctx verifymode ssl certnone self assertcontextoptionsctx def testcheckhostnameself with warningshelper checkwarnings ctx ssl sslcontextssl protocoltls self assertfalsectx checkhostname self assertequalctx verifymode ssl certnone auto set certrequired ctx checkhostname true self asserttruectx checkhostname self assertequalctx verifymode ssl certrequired ctx checkhostname false ctx verifymode ssl certrequired self assertfalsectx checkhostname self assertequalctx verifymode ssl certrequired changing verifymode does not affect checkhostname ctx checkhostname false ctx verifymode ssl certnone ctx checkhostname false self assertfalsectx checkhostname self assertequalctx verifymode ssl certnone auto set ctx checkhostname true self asserttruectx checkhostname self assertequalctx verifymode ssl certrequired ctx checkhostname false ctx verifymode ssl certoptional ctx checkhostname false self assertfalsectx checkhostname self assertequalctx verifymode ssl certoptional keep certoptional ctx checkhostname true self asserttruectx checkhostname self assertequalctx verifymode ssl certoptional cannot set certnone with checkhostname enabled with self assertraisesvalueerror ctx verifymode ssl certnone ctx checkhostname false self assertfalsectx checkhostname ctx verifymode ssl certnone self assertequalctx verifymode ssl certnone def testcontextclientserverself protocoltlsclient has sane defaults ctx ssl sslcontextssl protocoltlsclient self asserttruectx checkhostname self assertequalctx verifymode ssl certrequired protocoltlsserver has different but also sane defaults ctx ssl sslcontextssl protocoltlsserver self assertfalsectx checkhostname self assertequalctx verifymode ssl certnone def testcontextcustomclassself class mysslsocketssl sslsocket pass class mysslobjectssl sslobject pass ctx ssl sslcontextssl protocoltlsserver ctx sslsocketclass mysslsocket ctx sslobjectclass mysslobject with ctx wrapsocketsocket socket serversidetrue as sock self assertisinstancesock mysslsocket obj ctx wrapbiossl memorybio ssl memorybio serversidetrue self assertisinstanceobj mysslobject def testnumtickestself ctx ssl sslcontextssl protocoltlsserver self assertequalctx numtickets 2 ctx numtickets 1 self assertequalctx numtickets 1 ctx numtickets 0 self assertequalctx numtickets 0 with self assertraisesvalueerror ctx numtickets 1 with self assertraisestypeerror ctx numtickets none ctx ssl sslcontextssl protocoltlsclient self assertequalctx numtickets 2 with self assertraisesvalueerror ctx numtickets 1 class sslerrortestsunittest testcase def teststrself the str of a sslerror doesn t include the errno e ssl sslerror1 foo self assertequalstre foo self assertequale errno 1 same for a subclass e ssl sslzeroreturnerror1 foo self assertequalstre foo self assertequale errno 1 unittest skipifpydebugwin32 avoid mixing debugrelease crt on windows def testlibreasonself test the library and reason attributes ctx ssl sslcontextssl protocoltlsclient with self assertraisesssl sslerror as cm ctx loaddhparamscertfile self assertequalcm exception library pem self assertequalcm exception reason nostartline s strcm exception self asserttrues startswithpem nostartline no start line s def testsubclassself check that the appropriate sslerror subclass is raised this only tests one of them ctx ssl sslcontextssl protocoltlsclient ctx checkhostname false ctx verifymode ssl certnone with socket createserver127 0 0 1 0 as s c socket createconnections getsockname c setblockingfalse with ctx wrapsocketc false dohandshakeonconnectfalse as c with self assertraisesssl sslwantreaderror as cm c dohandshake s strcm exception self asserttrues startswiththe operation did not complete read s for compatibility self assertequalcm exception errno ssl sslerrorwantread def testbadserverhostnameself ctx ssl createdefaultcontext with self assertraisesvalueerror ctx wrapbiossl memorybio ssl memorybio serverhostname with self assertraisesvalueerror ctx wrapbiossl memorybio ssl memorybio serverhostname example org with self assertraisestypeerror ctx wrapbiossl memorybio ssl memorybio serverhostnameexample orgx00evil com class memorybiotestsunittest testcase def testreadwriteself bio ssl memorybio bio writeb foo self assertequalbio read b foo self assertequalbio read b bio writeb foo bio writeb bar self assertequalbio read b foobar self assertequalbio read b bio writeb baz self assertequalbio read2 b ba self assertequalbio read1 b z self assertequalbio read1 b def testeofself bio ssl memorybio self assertfalsebio eof self assertequalbio read b self assertfalsebio eof bio writeb foo self assertfalsebio eof bio writeeof self assertfalsebio eof self assertequalbio read2 b fo self assertfalsebio eof self assertequalbio read1 b o self asserttruebio eof self assertequalbio read b self asserttruebio eof def testpendingself bio ssl memorybio self assertequalbio pending 0 bio writeb foo self assertequalbio pending 3 for i in range3 bio read1 self assertequalbio pending 3i1 for i in range3 bio writeb x self assertequalbio pending i1 bio read self assertequalbio pending 0 def testbuffertypesself bio ssl memorybio bio writeb foo self assertequalbio read b foo bio writebytearrayb bar self assertequalbio read b bar bio writememoryviewb baz self assertequalbio read b baz m memoryviewbytearrayb noncontig noncontigwritable m 2 with self assertraisesbuffererror bio writememoryviewnoncontigwritable def testerrortypesself bio ssl memorybio self assertraisestypeerror bio write foo self assertraisestypeerror bio write none self assertraisestypeerror bio write true self assertraisestypeerror bio write 1 class sslobjecttestsunittest testcase def testprivateinitself bio ssl memorybio with self assertraisesregextypeerror public constructor ssl sslobjectbio bio def testunwrapself clientctx serverctx hostname testingcontext cin ssl memorybio cout ssl memorybio sin ssl memorybio sout ssl memorybio client clientctx wrapbiocin cout serverhostnamehostname server serverctx wrapbiosin sout serversidetrue loop on the handshake for a bit to get it settled for in range5 try client dohandshake except ssl sslwantreaderror pass if cout pending sin writecout read try server dohandshake except ssl sslwantreaderror pass if sout pending cin writesout read now the handshakes should be complete don t raise wantreaderror client dohandshake server dohandshake now if we unwrap one side unilaterally it should send closenotify and raise wantreaderror with self assertraisesssl sslwantreaderror client unwrap but server unwrap does not raise because it reads the client s closenotify sin writecout read server unwrap and now that the client gets the server s closenotify it doesn t raise either cin writesout read client unwrap class simplebackgroundtestsunittest testcase this should succeed because we specify the root cert this should fail because we have no verification certs connection failure crashes threadedechoserver so run this in an independent test method issue 11326 check connectex implementation issue 11326 nonblocking connectex should allow handshake to proceed after the socket gets ready ewouldblock under windows einprogress elsewhere wait for connect to finish nonblocking handshake ssl established same as testconnect but with a separately created context same with a server hostname this should succeed because we specify the root cert this should fail because we have no verification certs connection failure crashes threadedechoserver so run this in an independent test method verify server certificates using the capath argument note the subject hashing algorithm has been changed between openssl 0 9 8n and 1 0 0 as a result the capath directory must contain both versions of each certificate same content different filename for this test to be portable across openssl releases same with a bytes capath argument same with der issue 5238 creating a filelike object with makefile shouldn t delay closing the underlying real socket here tested with its file descriptor hence skipping the test under windows the fd is still open closing the ssl socket should close the fd too we store servernamecb arguments to make sure they match the host connection failure crashes threadedechoserver so run this in an independent test method error checking can happen at instantiation or when connecting capath certs are loaded on request check that the context of a connected socket can be replaced a simple io loop call funcargs depending on the error we get wantread or wantwrite move data between the socket and the bios get any data from the outgoing bio irrespective of any error and send it to the socket if there s no error we re done for wantread we need to get data from the socket and put it in the incoming bio if the server shuts down the tcp connection without sending a secure shutdown message this is reported as sslerrorsyscall simulate eof from the transport issue 12065 on a timeout connectex should return the original errno mimicking the behaviour of nonssl sockets should fail a mildly complicated class because we want it to work both with and without the ssl wrapper around the socket connection so that we can test the starttls functionality def initself server connsock addr self server server self running false self sock connsock self addr addr self sock setblockingtrue self sslconn none threading thread initself self daemon true def wrapconnself try self sslconn self server context wrapsocket self sock serversidetrue self server selectedalpnprotocols appendself sslconn selectedalpnprotocol except connectionreseterror brokenpipeerror connectionabortederror as e we treat connectionreseterror as though it were an sslerror openssl on ubuntu abruptly closes the connection when asked to use an unsupported protocol brokenpipeerror is raised in tls 1 3 mode when openssl tries to send session tickets after handshake https github comopensslopensslissues6342 connectionabortederror is raised in tls 1 3 mode when openssl tries to send session tickets after handshake when using winsock self server connerrors appendstre if self server chatty handleerrorn server bad connection attempt from reprself addr n self running false self close return false except ssl sslerror oserror as e oserror may occur with wrong protocols e g both sides use protocoltlsserver xxx various errors can have happened here for example a mismatching protocol version an invalid certificate or a lowlevel bug this should be made more discriminating bpo31323 store the exception as string to prevent a reference leak server connerrors exception traceback self connectionhandler server self server connerrors appendstre if self server chatty handleerrorn server bad connection attempt from reprself addr n bpo44229 bpo43855 bpo44237 and bpo33450 ignore spurious eprototype returned by write on macos see also http erickt github ioblog20141119adventuresindebuggingapotentialosxkernelbug if e errno errno eprototype and sys platform darwin self running false self server stop self close return false else self server sharedciphers appendself sslconn sharedciphers if self server context verifymode ssl certrequired cert self sslconn getpeercert if support verbose and self server chatty sys stdout write client cert is pprint pformatcert n certbinary self sslconn getpeercerttrue if support verbose and self server chatty if certbinary is none sys stdout write client did not provide a certn else sys stdout writef cert binary is lencertbinarybn cipher self sslconn cipher if support verbose and self server chatty sys stdout write server connection cipher is now strcipher n return true def readself if self sslconn return self sslconn read else return self sock recv1024 def writeself bytes if self sslconn return self sslconn writebytes else return self sock sendbytes def closeself if self sslconn self sslconn close else self sock close def runself self running true if not self server starttlsserver if not self wrapconn return while self running try msg self read stripped msg strip if not stripped eof so quit this handler self running false try self sock self sslconn unwrap except oserror many tests shut the tcp connection down without an ssl shutdown this causes unwrap to raise oserror with errno0 pass else self sslconn none self close elif stripped b over if support verbose and self server connectionchatty sys stdout write server client closed connectionn self close return elif self server starttlsserver and stripped b starttls if support verbose and self server connectionchatty sys stdout write server read starttls from client sending ok n self writebokn if not self wrapconn return elif self server starttlsserver and self sslconn and stripped b endtls if support verbose and self server connectionchatty sys stdout write server read endtls from client sending ok n self writebokn self sock self sslconn unwrap self sslconn none if support verbose and self server connectionchatty sys stdout write server connection is now unencrypted n elif stripped b cb tlsunique if support verbose and self server connectionchatty sys stdout write server read cb tlsunique from client sending our cb data n data self sslconn getchannelbindingtlsunique self writereprdata encodeusascii bn elif stripped b pha if support verbose and self server connectionchatty sys stdout write server initiating post handshake authn try self sslconn verifyclientposthandshake except ssl sslerror as e self writerepre encodeusascii bn else self writebokn elif stripped b hascert if self sslconn getpeercert is not none self writeb truen else self writeb falsen elif stripped b getcert cert self sslconn getpeercert self writereprcert encodeusascii bn elif stripped b verifiedchain certs self sslconn sslobj getverifiedchain self writelencerts tobytes1 big bn elif stripped b unverifiedchain certs self sslconn sslobj getunverifiedchain self writelencerts tobytes1 big bn else if support verbose and self server connectionchatty ctype self sslconn and encrypted or unencrypted sys stdout write server read r s sending back r s n msg ctype msg lower ctype self writemsg lower except oserror as e handles sslerror and socket errors if self server chatty and support verbose if isinstancee connectionerror openssl 1 1 1 sometimes raises connectionreseterror when connection is not shut down gracefully print f connection reset by peer self addr else handleerrortest server failure n try self writeberrorn except oserror pass self close self running false normally we d just stop here but for the test harness we want to stop the server self server stop def initself certificatenone sslversionnone certreqsnone cacertsnone chattytrue connectionchattyfalse starttlsserverfalse alpnprotocolsnone ciphersnone contextnone if context self context context else self context ssl sslcontextsslversion if sslversion is not none else ssl protocoltlsserver self context verifymode certreqs if certreqs is not none else ssl certnone if cacerts self context loadverifylocationscacerts if certificate self context loadcertchaincertificate if alpnprotocols self context setalpnprotocolsalpnprotocols if ciphers self context setciphersciphers self chatty chatty self connectionchatty connectionchatty self starttlsserver starttlsserver self sock socket socket self port sockethelper bindportself sock self flag none self active false self selectedalpnprotocols self sharedciphers self connerrors threading thread initself self daemon true def enterself self startthreading event self flag wait return self def exitself args self stop self join def startself flagnone self flag flag threading thread startself def runself self sock settimeout1 0 self sock listen5 self active true if self flag signal an event self flag set while self active try newconn connaddr self sock accept if support verbose and self chatty sys stdout write server new connection from reprconnaddr n handler self connectionhandlerself newconn connaddr handler start handler join except timeouterror as e if support verbose sys stdout writef connection timeout e rn except keyboardinterrupt self stop except baseexception as e if support verbose and self chatty sys stdout write connection handling failed repre n self close def closeself if self sock is not none self sock close self sock none def stopself self active false class asyncoreechoserverthreading thread this one s based on asyncore dispatcher class echoserver asyncore dispatcher class connectionhandlerasyncore dispatcherwithsend def initself conn certfile self socket testwrapsocketconn serversidetrue certfilecertfile dohandshakeonconnectfalse asyncore dispatcherwithsend initself self socket self sslaccepting true self dosslhandshake def readableself if isinstanceself socket ssl sslsocket while self socket pending 0 self handlereadevent return true def dosslhandshakeself try self socket dohandshake except ssl sslwantreaderror ssl sslwantwriteerror return except ssl ssleoferror return self handleclose except ssl sslerror raise except oserror as err if err args0 errno econnaborted return self handleclose else self sslaccepting false def handlereadself if self sslaccepting self dosslhandshake else data self recv1024 if support verbose sys stdout write server read s from clientn reprdata if not data self close else self senddata lower def handlecloseself self close if support verbose sys stdout write server closed connection sn self socket def handleerrorself raise def initself certfile self certfile certfile sock socket socketsocket afinet socket sockstream self port sockethelper bindportsock asyncore dispatcher initself sock self listen5 def handleacceptedself sockobj addr if support verbose sys stdout write server new connection from s sn addr self connectionhandlersockobj self certfile def handleerrorself raise def initself certfile self flag none self active false self server self echoservercertfile self port self server port threading thread initself self daemon true def strself return s s self class name self server def enterself self startthreading event self flag wait return self def exitself args if support verbose sys stdout write cleanup stopping server n self stop if support verbose sys stdout write cleanup joining server thread n self join if support verbose sys stdout write cleanup successfully joined n make sure that connectionhandler is removed from socketmap asyncore closeallignorealltrue def start self flagnone self flag flag threading thread startself def runself self active true if self flag self flag set while self active try asyncore loop1 except pass def stopself self active false self server close def serverparamstestclientcontext servercontext indatabfoon chattytrue connectionchattyfalse sninamenone sessionnone stats server threadedechoservercontextservercontext chattychatty connectionchattyfalse with server with clientcontext wrapsocketsocket socket serverhostnamesniname sessionsession as s s connecthost server port for arg in indata bytearrayindata memoryviewindata if connectionchatty if support verbose sys stdout write client sending r n indata s writearg outdata s read if connectionchatty if support verbose sys stdout write client read rn outdata if outdata indata lower raise assertionerror bad data r d received expected r dn outdata 20 lenoutdata indata 20 lower lenindata s writebovern if connectionchatty if support verbose sys stdout write client closing connection n stats update compression s compression cipher s cipher peercert s getpeercert clientalpnprotocol s selectedalpnprotocol version s version sessionreused s sessionreused session s session s close stats serveralpnprotocols server selectedalpnprotocols stats serversharedciphers server sharedciphers return stats def tryprotocolcomboserverprotocol clientprotocol expectsuccess certsreqsnone serveroptions0 clientoptions0 if certsreqs is none certsreqs ssl certnone certtype ssl certnone certnone ssl certoptional certoptional ssl certrequired certrequired certsreqs if support verbose formatstr expectsuccess and ss sn or ss sn sys stdout writeformatstr ssl getprotocolnameclientprotocol ssl getprotocolnameserverprotocol certtype with warningshelper checkwarnings ignore deprecation warnings clientcontext ssl sslcontextclientprotocol clientcontext options clientoptions servercontext ssl sslcontextserverprotocol servercontext options serveroptions minversion protocoltotlsversion getclientprotocol none if minversion is not none sslcontext minimumversion is only available on recent openssl setter added in openssl 1 1 0 getter added in openssl 1 1 1 and hasattrservercontext minimumversion and serverprotocol ssl protocoltls and servercontext minimumversion minversion if openssl configuration is strict and requires more recent tls version we have to change the minimum to test old tls versions with warningshelper checkwarnings servercontext minimumversion minversion note we must enable all ciphers on the client otherwise an sslv23 client will send an sslv3 hello rather than sslv2 starting from openssl 1 0 0 see issue 8322 if clientcontext protocol ssl protocoltls clientcontext setciphersall seclevelworkaroundservercontext clientcontext for ctx in clientcontext servercontext ctx verifymode certsreqs ctx loadcertchainsignedcertfile ctx loadverifylocationssigningca try stats serverparamstestclientcontext servercontext chattyfalse connectionchattyfalse protocol mismatch can result in either an sslerror or a connection reset by peer error except ssl sslerror if expectsuccess raise except oserror as e if expectsuccess or e errno errno econnreset raise else if not expectsuccess raise assertionerror client protocol s succeeded with server protocol s ssl getprotocolnameclientprotocol ssl getprotocolnameserverprotocol elif expectsuccess is not true and expectsuccess stats version raise assertionerrorversion mismatch expected r got r expectsuccess stats version class threadedtestsunittest testcase support requiresresource walltime def testechoself getpeercert raise valueerror while the handshake isn t done verifydefault should pass verifycrlcheckleaf without a loaded crl file fails now load a crl file the crl file is signed by the ca correct hostname should verify incorrect hostname should raise an exception missing serverhostname arg should cause an exception too default cert has a san load ecc cert correct hostname should verify todo fix tlsv1 3 once sslcontext can restrict signature algorithms only ecdsa certs load ecc and rsa keycert pairs correct hostname should verify correct hostname should verify when specified in several different ways knigsgchen idna2008 pythontest net xnknigsgchenb4a3dun idna2008 pythontest net incorrect hostname should raise an exception connecting when the server rejects the client s certificate launch a server with certrequired and check that trying to connect to it with a wrong client certificate fails load client cert that is not signed by trusted ca require tls client authentication tls 1 3 has different handshake expect either an ssl error about the server rejecting the connection or a lowlevel connection reset which sometimes happens on windows load client cert that is not signed by trusted ca tls 1 3 perform client cert exchange after handshake a brutal shutdown of an ssl server should raise an oserror in the client when attempting handshake listener runs in a thread it sits in an accept until the main thread connects then it rudely closes the socket and sets event listenergone to let the main thread know the socket is gone connecting to an sslv23 server with various client options if support verbose sys stdout writen if hastlsversion sslv3 tryprotocolcombossl protocoltls ssl protocolsslv3 false tryprotocolcombossl protocoltls ssl protocoltls true if hastlsversion tlsv1 tryprotocolcombossl protocoltls ssl protocoltlsv1 tlsv1 if hastlsversion sslv3 tryprotocolcombossl protocoltls ssl protocolsslv3 false ssl certoptional tryprotocolcombossl protocoltls ssl protocoltls true ssl certoptional if hastlsversion tlsv1 tryprotocolcombossl protocoltls ssl protocoltlsv1 tlsv1 ssl certoptional if hastlsversion sslv3 tryprotocolcombossl protocoltls ssl protocolsslv3 false ssl certrequired tryprotocolcombossl protocoltls ssl protocoltls true ssl certrequired if hastlsversion tlsv1 tryprotocolcombossl protocoltls ssl protocoltlsv1 tlsv1 ssl certrequired server with specific ssl options if hastlsversion sslv3 tryprotocolcombossl protocoltls ssl protocolsslv3 false serveroptionsssl opnosslv3 will choose tlsv1 tryprotocolcombossl protocoltls ssl protocoltls true serveroptionsssl opnosslv2 ssl opnosslv3 if hastlsversion tlsv1 tryprotocolcombossl protocoltls ssl protocoltlsv1 false serveroptionsssl opnotlsv1 requirestlsversion sslv3 def testprotocolsslv3self connecting to a tlsv1 server with various client options if support verbose sys stdout writen tryprotocolcombossl protocoltlsv1 ssl protocoltlsv1 tlsv1 tryprotocolcombossl protocoltlsv1 ssl protocoltlsv1 tlsv1 ssl certoptional tryprotocolcombossl protocoltlsv1 ssl protocoltlsv1 tlsv1 ssl certrequired if hastlsversion sslv3 tryprotocolcombossl protocoltlsv1 ssl protocolsslv3 false tryprotocolcombossl protocoltlsv1 ssl protocoltls false clientoptionsssl opnotlsv1 requirestlsversion tlsv11 def testprotocoltlsv11self connecting to a tlsv1 2 server with various client options testing against older tls versions if support verbose sys stdout writen tryprotocolcombossl protocoltlsv12 ssl protocoltlsv12 tlsv1 2 serveroptionsssl opnosslv3ssl opnosslv2 clientoptionsssl opnosslv3ssl opnosslv2 if hastlsversion sslv3 tryprotocolcombossl protocoltlsv12 ssl protocolsslv3 false tryprotocolcombossl protocoltlsv12 ssl protocoltls false clientoptionsssl opnotlsv12 tryprotocolcombossl protocoltls ssl protocoltlsv12 tlsv1 2 if hastlsprotocolssl protocoltlsv1 tryprotocolcombossl protocoltlsv12 ssl protocoltlsv1 false tryprotocolcombossl protocoltlsv1 ssl protocoltlsv12 false if hastlsprotocolssl protocoltlsv11 tryprotocolcombossl protocoltlsv12 ssl protocoltlsv11 false tryprotocolcombossl protocoltlsv11 ssl protocoltlsv12 false def teststarttlsself starttls ok switch to secure mode endtls ok switch back to clear text using socketserver to create and manage ssl connections server makehttpsserverself certfilesignedcertfile try to connect if support verbose sys stdout write n get this test file itself with openfile rb as f d1 f read d2 now fetch the same data from the https server url f https localhost server porttestssl py context ssl createdefaultcontextcafilesigningca f urllib request urlopenurl contextcontext try dlen f info getcontentlength if dlen and intdlen 0 d2 f readintdlen if support verbose sys stdout write client read d bytes from remote server s n lend2 server finally f close self assertequald1 d2 def testasyncoreserverself test recv send and friends if support verbose sys stdout writen server threadedechoservercertfile certreqsssl certnone sslversionssl protocoltlsserver cacertscertfile chattytrue connectionchattyfalse with server s testwrapsocketsocket socket serversidefalse certfilecertfile cacertscertfile certreqsssl certnone s connecthost server port helper methods for standardising recv method signatures def recvinto b bytearrayb0100 count s recvintob return b count def recvfrominto b bytearrayb0100 count addr s recvfromintob return b count name method expect success args return value func sendmethods send s send true len sendto s sendto false some address len sendall s sendall true lambda x none name method whether to expect success args recvmethods recv s recv true recvfrom s recvfrom false some address recvinto recvinto true recvfrominto recvfrominto false dataprefix prefix for methname sendmeth expectsuccess args retvalmeth in sendmethods indata dataprefix methname encode ascii try ret sendmethindata args msg sending with formatmethname self assertequalret retvalmethindata msgmsg outdata s read if outdata indata lower self fail while sending with name s bad data outdata r nout d received expected indata r nin dn format namemethname outdataoutdata 20 noutlenoutdata indataindata 20 ninlenindata except valueerror as e if expectsuccess self fail failed to send with method name s expected to succeed n formatnamemethname if not stre startswithmethname self fail method name s failed with unexpected exception message exp sn format namemethname expe for methname recvmeth expectsuccess args in recvmethods indata dataprefix methname encode ascii try s sendindata outdata recvmethargs if outdata indata lower self fail while receiving with name s bad data outdata r nout d received expected indata r nin dn format namemethname outdataoutdata 20 noutlenoutdata indataindata 20 ninlenindata except valueerror as e if expectsuccess self fail failed to receive with method name s expected to succeed n formatnamemethname if not stre startswithmethname self fail method name s failed with unexpected exception message exp sn format namemethname expe consume data s read read1 buffer is supported even though read1 is not data bdata s senddata buffer bytearraylendata self assertequals read1 buffer lendata self assertequalbuffer data sendall accepts byteslike objects if ctypes is not none ubyte ctypes cubyte lendata byteslike ubyte frombuffercopydata s sendallbyteslike self assertequals read data make sure sendmsg et al are disallowed to avoid inadvertent disclosure of data andor corruption of the encrypted data stream self assertraisesnotimplementederror s dup self assertraisesnotimplementederror s sendmsg bdata self assertraisesnotimplementederror s recvmsg 100 self assertraisesnotimplementederror s recvmsginto bytearray100 s writebovern self assertraisesvalueerror s recv 1 self assertraisesvalueerror s read 1 s close def testrecvzeroself server threadedechoservercertfile self entercontextserver s socket createconnectionhost server port self addcleanups close s testwrapsockets suppressraggedeofsfalse self addcleanups close recvread0 should return no data s sendbdata self assertequals recv0 b self assertequals read0 b self assertequals read bdata should not block if the other end sends no data s setblockingfalse self assertequals recv0 b self assertequals recvintobytearray 0 def testrecvintobufferprotocollenself server threadedechoservercertfile self entercontextserver s socket createconnectionhost server port self addcleanups close s testwrapsockets suppressraggedeofsfalse self addcleanups close s sendbdata buf array array i 0 0 self assertequals recvintobuf 4 self assertequalbytesbuf 4 bdata class bbytearray def lenself 10 s sendbdata buf b6 self assertequals recvintobuf 4 self assertequalbytesbuf bdata00 def testnonblockingsendself server threadedechoservercertfile certreqsssl certnone sslversionssl protocoltlsserver cacertscertfile chattytrue connectionchattyfalse with server s testwrapsocketsocket socket serversidefalse certfilecertfile cacertscertfile certreqsssl certnone s connecthost server port s setblockingfalse if we keep sending data at some point the buffers will be full and the call will block buf bytearray8192 def fillbuffer while true s sendbuf self assertraisesssl sslwantwriteerror ssl sslwantreaderror fillbuffer now read all the output and discard it s setblockingtrue s close def testhandshaketimeoutself issue 5103 ssl handshake must respect the socket timeout server socket socketsocket afinet host 127 0 0 1 port sockethelper bindportserver started threading event finish false def serve server listen started set conns while not finish r w e select selectserver 0 1 if server in r let the socket hang around rather than having it closed by garbage collection conns appendserver accept0 for sock in conns sock close t threading threadtargetserve t start started wait try try c socket socketsocket afinet c settimeout0 2 c connecthost port will attempt handshake and time out self assertraisesregextimeouterror timed out testwrapsocket c finally c close try c socket socketsocket afinet c testwrapsocketc c settimeout0 2 will attempt handshake and time out self assertraisesregextimeouterror timed out c connect host port finally c close finally finish true t join server close def testserveracceptself issue 16357 accept on a sslsocket created through sslcontext wrapsocket clientctx serverctx hostname testingcontext server socket socketsocket afinet host 127 0 0 1 port sockethelper bindportserver server serverctx wrapsocketserver serversidetrue self asserttrueserver serverside evt threading event remote none peer none def serve nonlocal remote peer server listen block on the accept and wait on the connection to close evt set remote peer server accept remote sendremote recv4 t threading threadtargetserve t start client wait until server setup and perform a connect evt wait client clientctx wrapsocket socket socket serverhostnamehostname client connecthostname port client sendb data client recv clientaddr client getsockname client close t join remote close server close sanity checks self assertisinstanceremote ssl sslsocket self assertequalpeer clientaddr def testgetpeercertenotconnself context ssl sslcontextssl protocoltlsclient context checkhostname false with context wrapsocketsocket socket as sock with self assertraisesoserror as cm sock getpeercert self assertequalcm exception errno errno enotconn def testdohandshakeenotconnself context ssl sslcontextssl protocoltlsclient context checkhostname false with context wrapsocketsocket socket as sock with self assertraisesoserror as cm sock dohandshake self assertequalcm exception errno errno enotconn def testnosharedciphersself clientcontext servercontext hostname testingcontext openssl enables all tls 1 3 ciphers enforce tls 1 2 for test clientcontext maximumversion ssl tlsversion tlsv12 force different suites on client and server clientcontext setciphersaes128 servercontext setciphersaes256 with threadedechoservercontextservercontext as server with clientcontext wrapsocketsocket socket serverhostnamehostname as s with self assertraisesoserror s connecthost server port self assertinno shared cipher server connerrors0 def testversionbasicself context ssl sslcontextssl protocoltlsclient context checkhostname false context verifymode ssl certnone with threadedechoservercertfile sslversionssl protocoltlsserver chattyfalse as server with context wrapsocketsocket socket as s self assertiss version none self assertiss sslobj none s connecthost server port self assertequals version tlsv1 3 self assertiss sslobj none self assertiss version none requirestlsversion tlsv13 def testtls13self clientcontext servercontext hostname testingcontext clientcontext minimumversion ssl tlsversion tlsv13 with threadedechoservercontextservercontext as server with clientcontext wrapsocketsocket socket serverhostnamehostname as s s connecthost server port self assertins cipher0 tlsaes256gcmsha384 tlschacha20poly1305sha256 tlsaes128gcmsha256 self assertequals version tlsv1 3 requirestlsversion tlsv12 requirestlsversion tlsv1 ignoredeprecation def testminmaxversiontlsv12self clientcontext servercontext hostname testingcontext client tlsv1 0 to 1 2 clientcontext minimumversion ssl tlsversion tlsv1 clientcontext maximumversion ssl tlsversion tlsv12 server only tlsv1 2 servercontext minimumversion ssl tlsversion tlsv12 servercontext maximumversion ssl tlsversion tlsv12 with threadedechoservercontextservercontext as server with clientcontext wrapsocketsocket socket serverhostnamehostname as s s connecthost server port self assertequals version tlsv1 2 requirestlsversion tlsv11 ignoredeprecation def testminmaxversiontlsv11self clientcontext servercontext hostname testingcontext client 1 0 to 1 2 server 1 0 to 1 1 clientcontext minimumversion ssl tlsversion tlsv1 clientcontext maximumversion ssl tlsversion tlsv12 servercontext minimumversion ssl tlsversion tlsv1 servercontext maximumversion ssl tlsversion tlsv11 seclevelworkaroundclientcontext servercontext with threadedechoservercontextservercontext as server with clientcontext wrapsocketsocket socket serverhostnamehostname as s s connecthost server port self assertequals version tlsv1 1 requirestlsversion tlsv12 requirestlsversion tlsv1 ignoredeprecation def testminmaxversionmismatchself clientcontext servercontext hostname testingcontext client 1 0 server 1 2 mismatch servercontext maximumversion ssl tlsversion tlsv12 servercontext minimumversion ssl tlsversion tlsv12 clientcontext maximumversion ssl tlsversion tlsv1 clientcontext minimumversion ssl tlsversion tlsv1 seclevelworkaroundclientcontext servercontext with threadedechoservercontextservercontext as server with clientcontext wrapsocketsocket socket serverhostnamehostname as s with self assertraisesssl sslerror as e s connecthost server port self assertinalert stre exception requirestlsversion sslv3 def testminmaxversionsslv3self clientcontext servercontext hostname testingcontext servercontext minimumversion ssl tlsversion sslv3 clientcontext minimumversion ssl tlsversion sslv3 clientcontext maximumversion ssl tlsversion sslv3 seclevelworkaroundclientcontext servercontext with threadedechoservercontextservercontext as server with clientcontext wrapsocketsocket socket serverhostnamehostname as s s connecthost server port self assertequals version sslv3 def testdefaultecdhcurveself issue 21015 elliptic curvebased diffie hellman key exchange should be enabled by default on ssl contexts clientcontext servercontext hostname testingcontext tlsv1 3 defaults to pfs key agreement and no longer has kea in cipher name clientcontext maximumversion ssl tlsversion tlsv12 prior to openssl 1 0 0 ecdh ciphers have to be enabled explicitly using the eccdraft cipher alias otherwise our default cipher list should prefer ecdhbased ciphers automatically with threadedechoservercontextservercontext as server with clientcontext wrapsocketsocket socket serverhostnamehostname as s s connecthost server port self assertinecdh s cipher0 unittest skipunlesstlsunique in ssl channelbindingtypes tlsunique channel binding not available def testtlsuniquechannelbindingself get the data check if it is sane and compare with the peers version now again is it really unique check we can get a connection with ephemeral diffiehellman test scenario needs tls 1 2 server secp384r1 client auto server auto client secp384r1 server client curve mismatch selectedalpnprotocol is none unless alpn is used selectedalpnprotocol is none unless alpn is used by the client the hostname was fetched properly and the certificate was changed for the connection certfile4 was selected the callback is called with servernamenone check disabling the callback certificate didn t change returning a tls alert is reflected to the connecting client raising fails the connection with a tls handshake failure alert returning the wrong return type terminates the tls connection with an internal error alert tls 1 3 ciphers are always enabled todo sessions aren t compatible with tlsv1 3 yet first connection without session reuse session another one without session reuse session again todo session reuse does not work with tlsv1 3 session is none before handshake cannot set session after handshake can set session before handshake and before the connection was established cannot reuse session with a different sslcontext correct psk should connect incorrect psk should fail identityhint and clientidentity should be sent to the other side adding client callback to server or vice versa raises an exception test with utf8 identities identityhint is not sent to the client in tls 1 3 pha method just returns true when cert is already available test sometimes fails with eof error test passes as long as server aborts connection with an error receive certificaterequest send empty certificate finish receive alert check certoptional optional doesn t fail when client does not have a cert server doesn t have pha enabled cert is requested in handshake pha doesn t fail if there is already a cert tls 1 2 pha fails for tls 1 3 verify that posthandshakeauth does not implicitly enable cert validation no cert validation and ca on client side server cert has not been validated windows raises permissionerror header 5 lines for tls 1 3 verify behavior of close sockets with received data before to the handshake on timeout just close the listener on windows the tcp rst leads to a connectionreseterror econnreset which linux doesn t appear to surface to userspace if wrapsocket winds up on the if connected path and doing the actual wrapping we get an sslerror from openssl typically wrongversionnumber while appropriate neither is the scenario we re specifically trying to test the way this test is written is known to work on linux we ll skip it anywhere else that it does not present as doing so gh108342 explicitly break the reference cycle if maintaining this conditional winds up being a problem just turn this into an unconditional skip anything but linux the important thing is that our ci has the logic covered this forces an immediate connection close via rst on close gh108342 explicitly break the reference cycle this forces an immediate connection close via rst on close redundant callafteraccept sets solinger on the accepted conn gh108342 explicitly break the reference cycle call clear text http connect not the encrypted https tls connect wrapsocket is called manually below wait for our fault injection server to have done its thing this forces an immediate connection close via rst on close redundant callafteraccept sets solinger on the accepted conn there are lots of reasons this raises as desired long before this test was added sending the request requires a successful tls wrapped socket that fails if the connection is broken it may seem pointless to test this it serves as an illustration of something that we never want to happen properly not happening content types record layer see rfc 8446 section b 1 pseudo content types alert types for tlscontenttype alert messages see rfc 8466 section b 2 message types handshake protocol see rfc 8446 section b 3 this member is assigned dynamically in ssl py test the support for ssl and sockets the custom key and certificate files used in test_ssl are generated using lib test certdata make_ssl_certs py other certificates are simply fetched from the internet servers they are meant to authenticate empty crl two keys and certs signed by the same ca for sni tests same certificate as pycacert pem but without extra text in file cert with all kinds of subject alt names not defined in all versions of openssl ubuntu has patched openssl and changed behavior of security level 2 see https bugs python org issue41561 msg389003 assume that any references of ubuntu implies ubuntu like distro the workaround is not required for 18 04 but doesn t hurt either lower security level to 1 and allow all ciphers for tls 1 0 1 check if a tls protocol is available and enabled param protocol enum ssl _sslmethod member or name return bool auto negotiate protocols are always available check if a tls ssl version is enabled param version tls version name or ssl tlsversion member return bool check compile time flags like ssl has_tlsv1_2 bpo43791 3 0 0 alpha14 fails with tlsv1_alert_internal_error check runtime and dynamic crypto policy settings a tls version may be compiled in but disabled by a policy or config option decorator to skip tests when a required tls version is not available param version tls version name or ssl tlsversion member return note ignore issues like 1647654 local time utc time utc offset seconds create context client_context server_context hostname testing_context gh 106687 ssl options values are unsigned integer uint64_t make sure that the protocol_ constants have enum like string reprs negative num is invalid note that this uses an unofficial function in _ssl c provided solely for this test to exercise the certificate parsing code issue 13034 the subjectaltname in some certificates notably projects developer nokia com 443 wasn t parsed extra ocsp and aia fields openssl 0 9 7 doesn t support ipv6 addresses in subjectaltname some sanity checks follow 1 1 1 4 0 3 x uses 0xmnn00pp0l issue 7943 an ssl object doesn t create reference cycles with itself methods on an unconnected sslsocket propagate the original oserror raise by the underlying socket object issue 8524 when creating an ssl socket the timeout of the original socket should be retained check that trying to use the given client certificate fails wrapping with an empty cert file wrapping with a badly formatted certificate syntax error wrapping with a badly formatted key syntax error server_hostname doesn t work for server sockets should raise valueerror for unknown type unconnected should return none for known type the same for server side issue 19940 ssl cert_time_to_seconds returns wrong results if local timezone is not utc accept keyword parameter assert its name accept both e and d space or zero generated by strftime case insensitive no seconds no gmt not gmt timezone invalid day invalid month invalid hour invalid minute leap seconds same timestamp allow 60th second even if it is not a leap second allow 2nd leap second for compatibility with time strptime invalid seconds no special treatment for the special value 99991231235959z rfc 5280 cert_time_to_seconds should be locale independent locale independent reserve port but don t listen issue 19919 windows machines or vms hosted on windows machines sometimes return ewouldblock empty reads and writes now work bpo 42854 bpo 31711 test default sslcontext options op_all op_no_sslv2 op_no_sslv3 is the default value sslcontext also enables these by default disallow tlsv1 allow tlsv1 clear all options ubuntu has op_no_sslv3 forced on by default invalid options default value openssl default is minimum_supported however some vendors like fedora override the setting to tls 1 0 stock openssl fedora 29 uses tls 1 0 by default rhel 8 uses tls 1 2 by default stock openssl fedora 32 uses tls 1 3 by default the default security callback allows for levels between 0 5 with openssl defaulting to 1 however some vendors override the default value e g debian defaults to 2 openssl default debian default value supports any value combined key and cert in a single file separate key and cert mismatching key and cert password protected key and cert openssl has a fixed limit on the password buffer pem_bufsize is generally set to 1kb return a string larger than this password callback make sure the password function isn t called if it isn t needed issue 10989 crash if the second argument type is invalid test cadata test pem cert already in hash table combined with junk around the certs test der cert already in hash table combined error cases there s not much we can do to test that it acts as expected so just check it doesn t crash or raise an exception set_servername_callback expects a callable or none reference cycles through the servername callback are detected and cleared certfile is not flagged as x509v3 basic constraints ca true but cafile_cacert is a ca cert auto set cert_required changing verify_mode does not affect check_hostname auto set keep cert_optional cannot set cert_none with check_hostname enabled protocol_tls_client has sane defaults protocol_tls_server has different but also sane defaults the str of a sslerror doesn t include the errno same for a subclass test the library and reason attributes check that the appropriate sslerror subclass is raised this only tests one of them for compatibility loop on the handshake for a bit to get it settled now the handshakes should be complete don t raise wantreaderror now if we unwrap one side unilaterally it should send close notify and raise wantreaderror but server unwrap does not raise because it reads the client s close notify and now that the client gets the server s close notify it doesn t raise either tests that connect to a simple server running in the background this should succeed because we specify the root cert this should fail because we have no verification certs connection failure crashes threadedechoserver so run this in an independent test method issue 11326 check connect_ex implementation issue 11326 non blocking connect_ex should allow handshake to proceed after the socket gets ready ewouldblock under windows einprogress elsewhere wait for connect to finish non blocking handshake ssl established same as test_connect but with a separately created context same with a server hostname this should succeed because we specify the root cert this should fail because we have no verification certs connection failure crashes threadedechoserver so run this in an independent test method verify server certificates using the capath argument note the subject hashing algorithm has been changed between openssl 0 9 8n and 1 0 0 as a result the capath directory must contain both versions of each certificate same content different filename for this test to be portable across openssl releases same with a bytes capath argument same with der issue 5238 creating a file like object with makefile shouldn t delay closing the underlying real socket here tested with its file descriptor hence skipping the test under windows the fd is still open closing the ssl socket should close the fd too we store servername_cb arguments to make sure they match the host connection failure crashes threadedechoserver so run this in an independent test method error checking can happen at instantiation or when connecting capath certs are loaded on request check that the context of a connected socket can be replaced a simple io loop call func args depending on the error we get want_read or want_write move data between the socket and the bios get any data from the outgoing bio irrespective of any error and send it to the socket if there s no error we re done for want_read we need to get data from the socket and put it in the incoming bio if the server shuts down the tcp connection without sending a secure shutdown message this is reported as ssl_error_syscall simulate eof from the transport issue 12065 on a timeout connect_ex should return the original errno mimicking the behaviour of non ssl sockets should fail a mildly complicated class because we want it to work both with and without the ssl wrapper around the socket connection so that we can test the starttls functionality we treat connectionreseterror as though it were an sslerror openssl on ubuntu abruptly closes the connection when asked to use an unsupported protocol brokenpipeerror is raised in tls 1 3 mode when openssl tries to send session tickets after handshake https github com openssl openssl issues 6342 connectionabortederror is raised in tls 1 3 mode when openssl tries to send session tickets after handshake when using winsock oserror may occur with wrong protocols e g both sides use protocol_tls_server xxx various errors can have happened here for example a mismatching protocol version an invalid certificate or a low level bug this should be made more discriminating bpo 31323 store the exception as string to prevent a reference leak server conn_errors exception traceback self connectionhandler server bpo 44229 bpo 43855 bpo 44237 and bpo 33450 ignore spurious eprototype returned by write on macos see also http erickt github io blog 2014 11 19 adventures in debugging a potential osx kernel bug eof so quit this handler many tests shut the tcp connection down without an ssl shutdown this causes unwrap to raise oserror with errno 0 handles sslerror and socket errors openssl 1 1 1 sometimes raises connectionreseterror when connection is not shut down gracefully normally we d just stop here but for the test harness we want to stop the server signal an event this one s based on asyncore dispatcher make sure that connectionhandler is removed from socket_map launch a server connect a client to it and try various reads and writes try to ssl connect using client_protocol to server_protocol if expect_success is true assert that the connection succeeds if it s false assert that the connection fails also if expect_success is a string assert that it is the protocol version actually used by the connection ignore deprecation warnings sslcontext minimum_version is only available on recent openssl setter added in openssl 1 1 0 getter added in openssl 1 1 1 if openssl configuration is strict and requires more recent tls version we have to change the minimum to test old tls versions note we must enable all ciphers on the client otherwise an sslv23 client will send an sslv3 hello rather than sslv2 starting from openssl 1 0 0 see issue 8322 protocol mismatch can result in either an sslerror or a connection reset by peer error basic test of an ssl client connecting to a server getpeercert raise valueerror while the handshake isn t done verify_default should pass verify_crl_check_leaf without a loaded crl file fails now load a crl file the crl file is signed by the ca correct hostname should verify incorrect hostname should raise an exception missing server_hostname arg should cause an exception too default cert has a san load ecc cert correct hostname should verify todo fix tlsv1 3 once sslcontext can restrict signature algorithms only ecdsa certs load ecc and rsa key cert pairs correct hostname should verify correct hostname should verify when specified in several different ways königsgäßchen idna2008 pythontest net xn knigsgchen b4a3dun idna2008 pythontest net incorrect hostname should raise an exception connecting when the server rejects the client s certificate launch a server with cert_required and check that trying to connect to it with a wrong client certificate fails load client cert that is not signed by trusted ca require tls client authentication tls 1 3 has different handshake expect either an ssl error about the server rejecting the connection or a low level connection reset which sometimes happens on windows load client cert that is not signed by trusted ca tls 1 3 perform client cert exchange after handshake a brutal shutdown of an ssl server should raise an oserror in the client when attempting handshake listener runs in a thread it sits in an accept until the main thread connects then it rudely closes the socket and sets event listener_gone to let the main thread know the socket is gone connecting to an sslv23 server with various client options server with specific ssl options will choose tlsv1 connecting to an sslv3 server with various client options connecting to a tlsv1 server with various client options connecting to a tlsv1 1 server with various client options testing against older tls versions connecting to a tlsv1 2 server with various client options testing against older tls versions switching from clear text to encrypted and back again starttls ok switch to secure mode endtls ok switch back to clear text using socketserver to create and manage ssl connections try to connect get this test file itself now fetch the same data from the https server check the example asyncore integration test recv send and friends helper methods for standardising recv method signatures name method expect success args return value func name method whether to expect success args consume data read 1 buffer is supported even though read 1 is not sendall accepts bytes like objects make sure sendmsg et al are disallowed to avoid inadvertent disclosure of data and or corruption of the encrypted data stream recv read 0 should return no data should not block if the other end sends no data if we keep sending data at some point the buffers will be full and the call will block now read all the output and discard it issue 5103 ssl handshake must respect the socket timeout let the socket hang around rather than having it closed by garbage collection will attempt handshake and time out will attempt handshake and time out issue 16357 accept on a sslsocket created through sslcontext wrap_socket block on the accept and wait on the connection to close client wait until server setup and perform a connect sanity checks openssl enables all tls 1 3 ciphers enforce tls 1 2 for test force different suites on client and server basic tests for sslsocket version more tests are done in the test_protocol_ methods client tlsv1 0 to 1 2 server only tlsv1 2 client 1 0 to 1 2 server 1 0 to 1 1 client 1 0 server 1 2 mismatch issue 21015 elliptic curve based diffie hellman key exchange should be enabled by default on ssl contexts tlsv1 3 defaults to pfs key agreement and no longer has kea in cipher name prior to openssl 1 0 0 ecdh ciphers have to be enabled explicitly using the eccdraft cipher alias otherwise our default cipher list should prefer ecdh based ciphers automatically test tls unique channel binding get the data check if it is sane true for tlsv1 and compare with the peers version now again is it really unique true for tlsv1 check we can get a connection with ephemeral diffie hellman test scenario needs tls 1 2 server secp384r1 client auto server auto client secp384r1 server client curve mismatch selected_alpn_protocol is none unless alpn is used selected_alpn_protocol is none unless alpn is used by the client the hostname was fetched properly and the certificate was changed for the connection certfile4 was selected the callback is called with server_name none check disabling the callback certificate didn t change returning a tls alert is reflected to the connecting client raising fails the connection with a tls handshake failure alert returning the wrong return type terminates the tls connection with an internal error alert tls 1 3 ciphers are always enabled todo sessions aren t compatible with tlsv1 3 yet first connection without session reuse session another one without session reuse session again todo session reuse does not work with tlsv1 3 session is none before handshake cannot set session after handshake can set session before handshake and before the connection was established cannot re use session with a different sslcontext correct psk should connect incorrect psk should fail identity_hint and client_identity should be sent to the other side adding client callback to server or vice versa raises an exception test with utf 8 identities translation identity hint translation customer identity identity_hint is not sent to the client in tls 1 3 pha method just returns true when cert is already available test sometimes fails with eof error test passes as long as server aborts connection with an error receive certificaterequest send empty certificate finish receive alert check cert_optional optional doesn t fail when client does not have a cert server doesn t have pha enabled cert is requested in handshake pha doesn t fail if there is already a cert tls 1 2 pha fails for tls 1 3 verify that post_handshake_auth does not implicitly enable cert validation no cert validation and ca on client side server cert has not been validated windows raises permissionerror header 5 lines for tls 1 3 verify behavior of close sockets with received data before to the handshake set by run set by run set by start set by start avoid dangling references on timeout just close the listener ssl sslerror inherits from oserror closed protocol error etc expect the full test setup to always work on linux on windows the tcp rst leads to a connectionreseterror econnreset which linux doesn t appear to surface to userspace if wrap_socket winds up on the if connected path and doing the actual wrapping we get an sslerror from openssl typically wrong_version_number while appropriate neither is the scenario we re specifically trying to test the way this test is written is known to work on linux we ll skip it anywhere else that it does not present as doing so gh 108342 explicitly break the reference cycle if maintaining this conditional winds up being a problem just turn this into an unconditional skip anything but linux the important thing is that our ci has the logic covered tell the server thread to continue starts it unittest testcase stops it this forces an immediate connection close via rst on close rst all platforms gh 108342 explicitly break the reference cycle this forces an immediate connection close via rst on close rst tell the server to stop starts it unittest testcase stops it redundant call_after_accept sets so_linger on the accepted conn sslerror inherits from oserror all platforms gh 108342 explicitly break the reference cycle call clear text http connect not the encrypted https tls connect wrap_socket is called manually below wait for our fault injection server to have done its thing this forces an immediate connection close via rst on close rst tell the server to stop starts it unittest testcase stops it redundant call_after_accept sets so_linger on the accepted conn there are lots of reasons this raises as desired long before this test was added sending the request requires a successful tls wrapped socket that fails if the connection is broken it may seem pointless to test this it serves as an illustration of something that we never want to happen properly not happening content types record layer see rfc 8446 section b 1 pseudo content types alert types for tlscontenttype alert messages see rfc 8466 section b 2 message types handshake protocol see rfc 8446 section b 3 this member is assigned dynamically in ssl py
import sys
import unittest
import unittest.mock
from test import support
from test.support import import_helper
from test.support import os_helper
from test.support import socket_helper
from test.support import threading_helper
from test.support import warnings_helper
from test.support import asyncore
import array
import re
import socket
import select
import struct
import time
import enum
import gc
import http.client
import os
import errno
import pprint
import urllib.request
import threading
import traceback
import weakref
import platform
import sysconfig
import functools
try:
    import ctypes
except ImportError:
    ctypes = None


ssl = import_helper.import_module("ssl")
import _ssl

from ssl import TLSVersion, _TLSContentType, _TLSMessageType, _TLSAlertType

Py_DEBUG_WIN32 = support.Py_DEBUG and sys.platform == 'win32'

PROTOCOLS = sorted(ssl._PROTOCOL_NAMES)
HOST = socket_helper.HOST
IS_OPENSSL_3_0_0 = ssl.OPENSSL_VERSION_INFO >= (3, 0, 0)
PY_SSL_DEFAULT_CIPHERS = sysconfig.get_config_var('PY_SSL_DEFAULT_CIPHERS')

PROTOCOL_TO_TLS_VERSION = {}
for proto, ver in (
    ("PROTOCOL_SSLv3", "SSLv3"),
    ("PROTOCOL_TLSv1", "TLSv1"),
    ("PROTOCOL_TLSv1_1", "TLSv1_1"),
):
    try:
        proto = getattr(ssl, proto)
        ver = getattr(ssl.TLSVersion, ver)
    except AttributeError:
        continue
    PROTOCOL_TO_TLS_VERSION[proto] = ver

def data_file(*name):
    return os.path.join(os.path.dirname(__file__), "certdata", *name)






CERTFILE = data_file("keycert.pem")
BYTES_CERTFILE = os.fsencode(CERTFILE)
ONLYCERT = data_file("ssl_cert.pem")
ONLYKEY = data_file("ssl_key.pem")
BYTES_ONLYCERT = os.fsencode(ONLYCERT)
BYTES_ONLYKEY = os.fsencode(ONLYKEY)
CERTFILE_PROTECTED = data_file("keycert.passwd.pem")
ONLYKEY_PROTECTED = data_file("ssl_key.passwd.pem")
KEY_PASSWORD = "somepass"
CAPATH = data_file("capath")
BYTES_CAPATH = os.fsencode(CAPATH)
CAFILE_NEURONIO = data_file("capath", "4e1295a3.0")
CAFILE_CACERT = data_file("capath", "5ed36f99.0")

CERTFILE_INFO = {
    'issuer': ((('countryName', 'XY'),),
               (('localityName', 'Castle Anthrax'),),
               (('organizationName', 'Python Software Foundation'),),
               (('commonName', 'localhost'),)),
    'notAfter': 'Aug 26 14:23:15 2028 GMT',
    'notBefore': 'Aug 29 14:23:15 2018 GMT',
    'serialNumber': '98A7CF88C74A32ED',
    'subject': ((('countryName', 'XY'),),
             (('localityName', 'Castle Anthrax'),),
             (('organizationName', 'Python Software Foundation'),),
             (('commonName', 'localhost'),)),
    'subjectAltName': (('DNS', 'localhost'),),
    'version': 3
}


CRLFILE = data_file("revocation.crl")


SIGNED_CERTFILE = data_file("keycert3.pem")
SIGNED_CERTFILE_HOSTNAME = 'localhost'

SIGNED_CERTFILE_INFO = {
    'OCSP': ('http://testca.pythontest.net/testca/ocsp/',),
    'caIssuers': ('http://testca.pythontest.net/testca/pycacert.cer',),
    'crlDistributionPoints': ('http://testca.pythontest.net/testca/revocation.crl',),
    'issuer': ((('countryName', 'XY'),),
            (('organizationName', 'Python Software Foundation CA'),),
            (('commonName', 'our-ca-server'),)),
    'notAfter': 'Oct 28 14:23:16 2037 GMT',
    'notBefore': 'Aug 29 14:23:16 2018 GMT',
    'serialNumber': 'CB2D80995A69525C',
    'subject': ((('countryName', 'XY'),),
             (('localityName', 'Castle Anthrax'),),
             (('organizationName', 'Python Software Foundation'),),
             (('commonName', 'localhost'),)),
    'subjectAltName': (('DNS', 'localhost'),),
    'version': 3
}

SIGNED_CERTFILE2 = data_file("keycert4.pem")
SIGNED_CERTFILE2_HOSTNAME = 'fakehostname'
SIGNED_CERTFILE_ECC = data_file("keycertecc.pem")
SIGNED_CERTFILE_ECC_HOSTNAME = 'localhost-ecc'


SIGNING_CA = data_file("capath", "ceff1710.0")

ALLSANFILE = data_file("allsans.pem")
IDNSANSFILE = data_file("idnsans.pem")
NOSANFILE = data_file("nosan.pem")
NOSAN_HOSTNAME = 'localhost'

REMOTE_HOST = "self-signed.pythontest.net"

EMPTYCERT = data_file("nullcert.pem")
BADCERT = data_file("badcert.pem")
NONEXISTINGCERT = data_file("XXXnonexisting.pem")
BADKEY = data_file("badkey.pem")
NOKIACERT = data_file("nokia.pem")
NULLBYTECERT = data_file("nullbytecert.pem")
TALOS_INVALID_CRLDP = data_file("talos-2019-0758.pem")

DHFILE = data_file("ffdh3072.pem")
BYTES_DHFILE = os.fsencode(DHFILE)


OP_NO_COMPRESSION = getattr(ssl, "OP_NO_COMPRESSION", 0)
OP_SINGLE_DH_USE = getattr(ssl, "OP_SINGLE_DH_USE", 0)
OP_SINGLE_ECDH_USE = getattr(ssl, "OP_SINGLE_ECDH_USE", 0)
OP_CIPHER_SERVER_PREFERENCE = getattr(ssl, "OP_CIPHER_SERVER_PREFERENCE", 0)
OP_ENABLE_MIDDLEBOX_COMPAT = getattr(ssl, "OP_ENABLE_MIDDLEBOX_COMPAT", 0)



def is_ubuntu():
    try:
        
        
        with open("/etc/os-release", encoding="utf-8") as f:
            return "ubuntu" in f.read()
    except FileNotFoundError:
        return False

if is_ubuntu():
    def seclevel_workaround(*ctxs):
        
        for ctx in ctxs:
            if (
                hasattr(ctx, "minimum_version") and
                ctx.minimum_version <= ssl.TLSVersion.TLSv1_1
            ):
                ctx.set_ciphers("@SECLEVEL=1:ALL")
else:
    def seclevel_workaround(*ctxs):
        pass


def has_tls_protocol(protocol):
    
    if isinstance(protocol, str):
        assert protocol.startswith('PROTOCOL_')
        protocol = getattr(ssl, protocol, None)
        if protocol is None:
            return False
    if protocol in {
        ssl.PROTOCOL_TLS, ssl.PROTOCOL_TLS_SERVER,
        ssl.PROTOCOL_TLS_CLIENT
    }:
        
        return True
    name = protocol.name
    return has_tls_version(name[len('PROTOCOL_'):])


@functools.lru_cache
def has_tls_version(version):
    
    if isinstance(version, str):
        version = ssl.TLSVersion.__members__[version]

    
    if not getattr(ssl, f'HAS_{version.name}'):
        return False

    if IS_OPENSSL_3_0_0 and version < ssl.TLSVersion.TLSv1_2:
        
        return False

    
    
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    if (
            hasattr(ctx, 'minimum_version') and
            ctx.minimum_version != ssl.TLSVersion.MINIMUM_SUPPORTED and
            version < ctx.minimum_version
    ):
        return False
    if (
        hasattr(ctx, 'maximum_version') and
        ctx.maximum_version != ssl.TLSVersion.MAXIMUM_SUPPORTED and
        version > ctx.maximum_version
    ):
        return False

    return True


def requires_tls_version(version):
    
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kw):
            if not has_tls_version(version):
                raise unittest.SkipTest(f"{version} is not available.")
            else:
                return func(*args, **kw)
        return wrapper
    return decorator


def handle_error(prefix):
    exc_format = ' '.join(traceback.format_exception(sys.exception()))
    if support.verbose:
        sys.stdout.write(prefix + exc_format)


def utc_offset(): 
    
    if time.daylight and time.localtime().tm_isdst > 0:
        return -time.altzone  
    return -time.timezone


ignore_deprecation = warnings_helper.ignore_warnings(
    category=DeprecationWarning
)


def test_wrap_socket(sock, *,
                     cert_reqs=ssl.CERT_NONE, ca_certs=None,
                     ciphers=None, certfile=None, keyfile=None,
                     **kwargs):
    if not kwargs.get("server_side"):
        kwargs["server_hostname"] = SIGNED_CERTFILE_HOSTNAME
        context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    else:
        context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    if cert_reqs is not None:
        if cert_reqs == ssl.CERT_NONE:
            context.check_hostname = False
        context.verify_mode = cert_reqs
    if ca_certs is not None:
        context.load_verify_locations(ca_certs)
    if certfile is not None or keyfile is not None:
        context.load_cert_chain(certfile, keyfile)
    if ciphers is not None:
        context.set_ciphers(ciphers)
    return context.wrap_socket(sock, **kwargs)


def testing_context(server_cert=SIGNED_CERTFILE, *, server_chain=True):
    
    if server_cert == SIGNED_CERTFILE:
        hostname = SIGNED_CERTFILE_HOSTNAME
    elif server_cert == SIGNED_CERTFILE2:
        hostname = SIGNED_CERTFILE2_HOSTNAME
    elif server_cert == NOSANFILE:
        hostname = NOSAN_HOSTNAME
    else:
        raise ValueError(server_cert)

    client_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    client_context.load_verify_locations(SIGNING_CA)

    server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    server_context.load_cert_chain(server_cert)
    if server_chain:
        server_context.load_verify_locations(SIGNING_CA)

    return client_context, server_context, hostname


class BasicSocketTests(unittest.TestCase):

    def test_constants(self):
        ssl.CERT_NONE
        ssl.CERT_OPTIONAL
        ssl.CERT_REQUIRED
        ssl.OP_CIPHER_SERVER_PREFERENCE
        ssl.OP_SINGLE_DH_USE
        ssl.OP_SINGLE_ECDH_USE
        ssl.OP_NO_COMPRESSION
        self.assertEqual(ssl.HAS_SNI, True)
        self.assertEqual(ssl.HAS_ECDH, True)
        self.assertEqual(ssl.HAS_TLSv1_2, True)
        self.assertEqual(ssl.HAS_TLSv1_3, True)
        ssl.OP_NO_SSLv2
        ssl.OP_NO_SSLv3
        ssl.OP_NO_TLSv1
        ssl.OP_NO_TLSv1_3
        ssl.OP_NO_TLSv1_1
        ssl.OP_NO_TLSv1_2
        self.assertEqual(ssl.PROTOCOL_TLS, ssl.PROTOCOL_SSLv23)

    def test_options(self):
        
        for name in dir(ssl):
            if not name.startswith('OP_'):
                continue
            with self.subTest(option=name):
                value = getattr(ssl, name)
                self.assertGreaterEqual(value, 0, f"ssl.{name}")

    def test_ssl_types(self):
        ssl_types = [
            _ssl._SSLContext,
            _ssl._SSLSocket,
            _ssl.MemoryBIO,
            _ssl.Certificate,
            _ssl.SSLSession,
            _ssl.SSLError,
        ]
        for ssl_type in ssl_types:
            with self.subTest(ssl_type=ssl_type):
                with self.assertRaisesRegex(TypeError, "immutable type"):
                    ssl_type.value = None
        support.check_disallow_instantiation(self, _ssl.Certificate)

    def test_private_init(self):
        with self.assertRaisesRegex(TypeError, "public constructor"):
            with socket.socket() as s:
                ssl.SSLSocket(s)

    def test_str_for_enums(self):
        
        
        proto = ssl.PROTOCOL_TLS_CLIENT
        self.assertEqual(repr(proto), '<_SSLMethod.PROTOCOL_TLS_CLIENT: %r>' % proto.value)
        self.assertEqual(str(proto), str(proto.value))
        ctx = ssl.SSLContext(proto)
        self.assertIs(ctx.protocol, proto)

    def test_random(self):
        v = ssl.RAND_status()
        if support.verbose:
            sys.stdout.write("\n RAND_status is %d (%s)\n"
                             % (v, (v and "sufficient randomness") or
                                "insufficient randomness"))

        if v:
            data = ssl.RAND_bytes(16)
            self.assertEqual(len(data), 16)
        else:
            self.assertRaises(ssl.SSLError, ssl.RAND_bytes, 16)

        
        self.assertRaises(ValueError, ssl.RAND_bytes, -5)

        ssl.RAND_add("this is a random string", 75.0)
        ssl.RAND_add(b"this is a random bytes object", 75.0)
        ssl.RAND_add(bytearray(b"this is a random bytearray object"), 75.0)

    def test_parse_cert(self):
        
        
        
        self.assertEqual(
            ssl._ssl._test_decode_cert(CERTFILE),
            CERTFILE_INFO
        )
        self.assertEqual(
            ssl._ssl._test_decode_cert(SIGNED_CERTFILE),
            SIGNED_CERTFILE_INFO
        )

        
        
        p = ssl._ssl._test_decode_cert(NOKIACERT)
        if support.verbose:
            sys.stdout.write("\n" + pprint.pformat(p) + "\n")
        self.assertEqual(p['subjectAltName'],
                         (('DNS', 'projects.developer.nokia.com'),
                          ('DNS', 'projects.forum.nokia.com'))
                        )
        
        self.assertEqual(p['OCSP'], ('http://ocsp.verisign.com',))
        self.assertEqual(p['caIssuers'],
                         ('http://SVRIntl-G3-aia.verisign.com/SVRIntlG3.cer',))
        self.assertEqual(p['crlDistributionPoints'],
                         ('http://SVRIntl-G3-crl.verisign.com/SVRIntlG3.crl',))

    def test_parse_cert_CVE_2019_5010(self):
        p = ssl._ssl._test_decode_cert(TALOS_INVALID_CRLDP)
        if support.verbose:
            sys.stdout.write("\n" + pprint.pformat(p) + "\n")
        self.assertEqual(
            p,
            {
                'issuer': (
                    (('countryName', 'UK'),), (('commonName', 'cody-ca'),)),
                'notAfter': 'Jun 14 18:00:58 2028 GMT',
                'notBefore': 'Jun 18 18:00:58 2018 GMT',
                'serialNumber': '02',
                'subject': ((('countryName', 'UK'),),
                            (('commonName',
                              'codenomicon-vm-2.test.lal.cisco.com'),)),
                'subjectAltName': (
                    ('DNS', 'codenomicon-vm-2.test.lal.cisco.com'),),
                'version': 3
            }
        )

    def test_parse_cert_CVE_2013_4238(self):
        p = ssl._ssl._test_decode_cert(NULLBYTECERT)
        if support.verbose:
            sys.stdout.write("\n" + pprint.pformat(p) + "\n")
        subject = ((('countryName', 'US'),),
                   (('stateOrProvinceName', 'Oregon'),),
                   (('localityName', 'Beaverton'),),
                   (('organizationName', 'Python Software Foundation'),),
                   (('organizationalUnitName', 'Python Core Development'),),
                   (('commonName', 'null.python.org\x00example.org'),),
                   (('emailAddress', 'python-dev@python.org'),))
        self.assertEqual(p['subject'], subject)
        self.assertEqual(p['issuer'], subject)
        if ssl._OPENSSL_API_VERSION >= (0, 9, 8):
            san = (('DNS', 'altnull.python.org\x00example.com'),
                   ('email', 'null@python.org\x00user@example.org'),
                   ('URI', 'http://null.python.org\x00http://example.org'),
                   ('IP Address', '192.0.2.1'),
                   ('IP Address', '2001:DB8:0:0:0:0:0:1'))
        else:
            
            san = (('DNS', 'altnull.python.org\x00example.com'),
                   ('email', 'null@python.org\x00user@example.org'),
                   ('URI', 'http://null.python.org\x00http://example.org'),
                   ('IP Address', '192.0.2.1'),
                   ('IP Address', '<invalid>'))

        self.assertEqual(p['subjectAltName'], san)

    def test_parse_all_sans(self):
        p = ssl._ssl._test_decode_cert(ALLSANFILE)
        self.assertEqual(p['subjectAltName'],
            (
                ('DNS', 'allsans'),
                ('othername', '<unsupported>'),
                ('othername', '<unsupported>'),
                ('email', 'user@example.org'),
                ('DNS', 'www.example.org'),
                ('DirName',
                    ((('countryName', 'XY'),),
                    (('localityName', 'Castle Anthrax'),),
                    (('organizationName', 'Python Software Foundation'),),
                    (('commonName', 'dirname example'),))),
                ('URI', 'https://www.python.org/'),
                ('IP Address', '127.0.0.1'),
                ('IP Address', '0:0:0:0:0:0:0:1'),
                ('Registered ID', '1.2.3.4.5')
            )
        )

    def test_DER_to_PEM(self):
        with open(CAFILE_CACERT, 'r') as f:
            pem = f.read()
        d1 = ssl.PEM_cert_to_DER_cert(pem)
        p2 = ssl.DER_cert_to_PEM_cert(d1)
        d2 = ssl.PEM_cert_to_DER_cert(p2)
        self.assertEqual(d1, d2)
        if not p2.startswith(ssl.PEM_HEADER + '\n'):
            self.fail("DER-to-PEM didn't include correct header:\n%r\n" % p2)
        if not p2.endswith('\n' + ssl.PEM_FOOTER + '\n'):
            self.fail("DER-to-PEM didn't include correct footer:\n%r\n" % p2)

    def test_openssl_version(self):
        n = ssl.OPENSSL_VERSION_NUMBER
        t = ssl.OPENSSL_VERSION_INFO
        s = ssl.OPENSSL_VERSION
        self.assertIsInstance(n, int)
        self.assertIsInstance(t, tuple)
        self.assertIsInstance(s, str)
        
        
        self.assertGreaterEqual(n, 0x10101000)
        
        self.assertLess(n, 0x40000000)
        major, minor, fix, patch, status = t
        self.assertGreaterEqual(major, 1)
        self.assertLess(major, 4)
        self.assertGreaterEqual(minor, 0)
        self.assertLess(minor, 256)
        self.assertGreaterEqual(fix, 0)
        self.assertLess(fix, 256)
        self.assertGreaterEqual(patch, 0)
        self.assertLessEqual(patch, 63)
        self.assertGreaterEqual(status, 0)
        self.assertLessEqual(status, 15)

        libressl_ver = f"LibreSSL {major:d}"
        if major >= 3:
            
            openssl_ver = f"OpenSSL {major:d}.{minor:d}.{patch:d}"
        else:
            openssl_ver = f"OpenSSL {major:d}.{minor:d}.{fix:d}"
        self.assertTrue(
            s.startswith((openssl_ver, libressl_ver)),
            (s, t, hex(n))
        )

    @support.cpython_only
    def test_refcycle(self):
        
        
        s = socket.socket(socket.AF_INET)
        ss = test_wrap_socket(s)
        wr = weakref.ref(ss)
        with warnings_helper.check_warnings(("", ResourceWarning)):
            del ss
        self.assertEqual(wr(), None)

    def test_wrapped_unconnected(self):
        
        
        s = socket.socket(socket.AF_INET)
        with test_wrap_socket(s) as ss:
            self.assertRaises(OSError, ss.recv, 1)
            self.assertRaises(OSError, ss.recv_into, bytearray(b'x'))
            self.assertRaises(OSError, ss.recvfrom, 1)
            self.assertRaises(OSError, ss.recvfrom_into, bytearray(b'x'), 1)
            self.assertRaises(OSError, ss.send, b'x')
            self.assertRaises(OSError, ss.sendto, b'x', ('0.0.0.0', 0))
            self.assertRaises(NotImplementedError, ss.dup)
            self.assertRaises(NotImplementedError, ss.sendmsg,
                              [b'x'], (), 0, ('0.0.0.0', 0))
            self.assertRaises(NotImplementedError, ss.recvmsg, 100)
            self.assertRaises(NotImplementedError, ss.recvmsg_into,
                              [bytearray(100)])

    def test_timeout(self):
        
        
        for timeout in (None, 0.0, 5.0):
            s = socket.socket(socket.AF_INET)
            s.settimeout(timeout)
            with test_wrap_socket(s) as ss:
                self.assertEqual(timeout, ss.gettimeout())

    def test_openssl111_deprecations(self):
        options = [
            ssl.OP_NO_TLSv1,
            ssl.OP_NO_TLSv1_1,
            ssl.OP_NO_TLSv1_2,
            ssl.OP_NO_TLSv1_3
        ]
        protocols = [
            ssl.PROTOCOL_TLSv1,
            ssl.PROTOCOL_TLSv1_1,
            ssl.PROTOCOL_TLSv1_2,
            ssl.PROTOCOL_TLS
        ]
        versions = [
            ssl.TLSVersion.SSLv3,
            ssl.TLSVersion.TLSv1,
            ssl.TLSVersion.TLSv1_1,
        ]

        for option in options:
            with self.subTest(option=option):
                ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
                with self.assertWarns(DeprecationWarning) as cm:
                    ctx.options |= option
                self.assertEqual(
                    'ssl.OP_NO_SSL*/ssl.OP_NO_TLS* options are deprecated',
                    str(cm.warning)
                )

        for protocol in protocols:
            if not has_tls_protocol(protocol):
                continue
            with self.subTest(protocol=protocol):
                with self.assertWarns(DeprecationWarning) as cm:
                    ssl.SSLContext(protocol)
                self.assertEqual(
                    f'ssl.{protocol.name} is deprecated',
                    str(cm.warning)
                )

        for version in versions:
            if not has_tls_version(version):
                continue
            with self.subTest(version=version):
                ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
                with self.assertWarns(DeprecationWarning) as cm:
                    ctx.minimum_version = version
                version_text = '%s.%s' % (version.__class__.__name__, version.name)
                self.assertEqual(
                    f'ssl.{version_text} is deprecated',
                    str(cm.warning)
                )

    def bad_cert_test(self, certfile):
        
        certfile = os.path.join(os.path.dirname(__file__) or os.curdir,
                                "certdata", certfile)
        sock = socket.socket()
        self.addCleanup(sock.close)
        with self.assertRaises(ssl.SSLError):
            test_wrap_socket(sock,
                             certfile=certfile)

    def test_empty_cert(self):
        
        self.bad_cert_test("nullcert.pem")

    def test_malformed_cert(self):
        
        self.bad_cert_test("badcert.pem")

    def test_malformed_key(self):
        
        self.bad_cert_test("badkey.pem")

    def test_server_side(self):
        
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        with socket.socket() as sock:
            self.assertRaises(ValueError, ctx.wrap_socket, sock, True,
                              server_hostname="some.hostname")

    def test_unknown_channel_binding(self):
        
        s = socket.create_server(('127.0.0.1', 0))
        c = socket.socket(socket.AF_INET)
        c.connect(s.getsockname())
        with test_wrap_socket(c, do_handshake_on_connect=False) as ss:
            with self.assertRaises(ValueError):
                ss.get_channel_binding("unknown-type")
        s.close()

    @unittest.skipUnless("tls-unique" in ssl.CHANNEL_BINDING_TYPES,
                         "'tls-unique' channel binding not available")
    def test_tls_unique_channel_binding(self):
        
        s = socket.socket(socket.AF_INET)
        with test_wrap_socket(s) as ss:
            self.assertIsNone(ss.get_channel_binding("tls-unique"))
        
        s = socket.socket(socket.AF_INET)
        with test_wrap_socket(s, server_side=True, certfile=CERTFILE) as ss:
            self.assertIsNone(ss.get_channel_binding("tls-unique"))

    def test_dealloc_warn(self):
        ss = test_wrap_socket(socket.socket(socket.AF_INET))
        r = repr(ss)
        with self.assertWarns(ResourceWarning) as cm:
            ss = None
            support.gc_collect()
        self.assertIn(r, str(cm.warning.args[0]))

    def test_get_default_verify_paths(self):
        paths = ssl.get_default_verify_paths()
        self.assertEqual(len(paths), 6)
        self.assertIsInstance(paths, ssl.DefaultVerifyPaths)

        with os_helper.EnvironmentVarGuard() as env:
            env["SSL_CERT_DIR"] = CAPATH
            env["SSL_CERT_FILE"] = CERTFILE
            paths = ssl.get_default_verify_paths()
            self.assertEqual(paths.cafile, CERTFILE)
            self.assertEqual(paths.capath, CAPATH)

    @unittest.skipUnless(sys.platform == "win32", "Windows specific")
    def test_enum_certificates(self):
        self.assertTrue(ssl.enum_certificates("CA"))
        self.assertTrue(ssl.enum_certificates("ROOT"))

        self.assertRaises(TypeError, ssl.enum_certificates)
        self.assertRaises(WindowsError, ssl.enum_certificates, "")

        trust_oids = set()
        for storename in ("CA", "ROOT"):
            store = ssl.enum_certificates(storename)
            self.assertIsInstance(store, list)
            for element in store:
                self.assertIsInstance(element, tuple)
                self.assertEqual(len(element), 3)
                cert, enc, trust = element
                self.assertIsInstance(cert, bytes)
                self.assertIn(enc, {"x509_asn", "pkcs_7_asn"})
                self.assertIsInstance(trust, (frozenset, set, bool))
                if isinstance(trust, (frozenset, set)):
                    trust_oids.update(trust)

        serverAuth = "1.3.6.1.5.5.7.3.1"
        self.assertIn(serverAuth, trust_oids)

    @unittest.skipUnless(sys.platform == "win32", "Windows specific")
    def test_enum_crls(self):
        self.assertTrue(ssl.enum_crls("CA"))
        self.assertRaises(TypeError, ssl.enum_crls)
        self.assertRaises(WindowsError, ssl.enum_crls, "")

        crls = ssl.enum_crls("CA")
        self.assertIsInstance(crls, list)
        for element in crls:
            self.assertIsInstance(element, tuple)
            self.assertEqual(len(element), 2)
            self.assertIsInstance(element[0], bytes)
            self.assertIn(element[1], {"x509_asn", "pkcs_7_asn"})


    def test_asn1object(self):
        expected = (129, 'serverAuth', 'TLS Web Server Authentication',
                    '1.3.6.1.5.5.7.3.1')

        val = ssl._ASN1Object('1.3.6.1.5.5.7.3.1')
        self.assertEqual(val, expected)
        self.assertEqual(val.nid, 129)
        self.assertEqual(val.shortname, 'serverAuth')
        self.assertEqual(val.longname, 'TLS Web Server Authentication')
        self.assertEqual(val.oid, '1.3.6.1.5.5.7.3.1')
        self.assertIsInstance(val, ssl._ASN1Object)
        self.assertRaises(ValueError, ssl._ASN1Object, 'serverAuth')

        val = ssl._ASN1Object.fromnid(129)
        self.assertEqual(val, expected)
        self.assertIsInstance(val, ssl._ASN1Object)
        self.assertRaises(ValueError, ssl._ASN1Object.fromnid, -1)
        with self.assertRaisesRegex(ValueError, "unknown NID 100000"):
            ssl._ASN1Object.fromnid(100000)
        for i in range(1000):
            try:
                obj = ssl._ASN1Object.fromnid(i)
            except ValueError:
                pass
            else:
                self.assertIsInstance(obj.nid, int)
                self.assertIsInstance(obj.shortname, str)
                self.assertIsInstance(obj.longname, str)
                self.assertIsInstance(obj.oid, (str, type(None)))

        val = ssl._ASN1Object.fromname('TLS Web Server Authentication')
        self.assertEqual(val, expected)
        self.assertIsInstance(val, ssl._ASN1Object)
        self.assertEqual(ssl._ASN1Object.fromname('serverAuth'), expected)
        self.assertEqual(ssl._ASN1Object.fromname('1.3.6.1.5.5.7.3.1'),
                         expected)
        with self.assertRaisesRegex(ValueError, "unknown object 'serverauth'"):
            ssl._ASN1Object.fromname('serverauth')

    def test_purpose_enum(self):
        val = ssl._ASN1Object('1.3.6.1.5.5.7.3.1')
        self.assertIsInstance(ssl.Purpose.SERVER_AUTH, ssl._ASN1Object)
        self.assertEqual(ssl.Purpose.SERVER_AUTH, val)
        self.assertEqual(ssl.Purpose.SERVER_AUTH.nid, 129)
        self.assertEqual(ssl.Purpose.SERVER_AUTH.shortname, 'serverAuth')
        self.assertEqual(ssl.Purpose.SERVER_AUTH.oid,
                              '1.3.6.1.5.5.7.3.1')

        val = ssl._ASN1Object('1.3.6.1.5.5.7.3.2')
        self.assertIsInstance(ssl.Purpose.CLIENT_AUTH, ssl._ASN1Object)
        self.assertEqual(ssl.Purpose.CLIENT_AUTH, val)
        self.assertEqual(ssl.Purpose.CLIENT_AUTH.nid, 130)
        self.assertEqual(ssl.Purpose.CLIENT_AUTH.shortname, 'clientAuth')
        self.assertEqual(ssl.Purpose.CLIENT_AUTH.oid,
                              '1.3.6.1.5.5.7.3.2')

    def test_unsupported_dtls(self):
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.addCleanup(s.close)
        with self.assertRaises(NotImplementedError) as cx:
            test_wrap_socket(s, cert_reqs=ssl.CERT_NONE)
        self.assertEqual(str(cx.exception), "only stream sockets are supported")
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        with self.assertRaises(NotImplementedError) as cx:
            ctx.wrap_socket(s)
        self.assertEqual(str(cx.exception), "only stream sockets are supported")

    def cert_time_ok(self, timestring, timestamp):
        self.assertEqual(ssl.cert_time_to_seconds(timestring), timestamp)

    def cert_time_fail(self, timestring):
        with self.assertRaises(ValueError):
            ssl.cert_time_to_seconds(timestring)

    @unittest.skipUnless(utc_offset(),
                         'local time needs to be different from UTC')
    def test_cert_time_to_seconds_timezone(self):
        
        
        self.cert_time_ok("May  9 00:00:00 2007 GMT", 1178668800.0)
        self.cert_time_ok("Jan  5 09:34:43 2018 GMT", 1515144883.0)

    def test_cert_time_to_seconds(self):
        timestring = "Jan  5 09:34:43 2018 GMT"
        ts = 1515144883.0
        self.cert_time_ok(timestring, ts)
        
        self.assertEqual(ssl.cert_time_to_seconds(cert_time=timestring), ts)
        
        self.cert_time_ok("Jan 05 09:34:43 2018 GMT", ts)
        
        self.cert_time_ok("JaN  5 09:34:43 2018 GmT", ts)
        self.cert_time_fail("Jan  5 09:34 2018 GMT")     
        self.cert_time_fail("Jan  5 09:34:43 2018")      
        self.cert_time_fail("Jan  5 09:34:43 2018 UTC")  
        self.cert_time_fail("Jan 35 09:34:43 2018 GMT")  
        self.cert_time_fail("Jon  5 09:34:43 2018 GMT")  
        self.cert_time_fail("Jan  5 24:00:00 2018 GMT")  
        self.cert_time_fail("Jan  5 09:60:43 2018 GMT")  

        newyear_ts = 1230768000.0
        
        self.cert_time_ok("Dec 31 23:59:60 2008 GMT", newyear_ts)
        
        self.cert_time_ok("Jan  1 00:00:00 2009 GMT", newyear_ts)

        self.cert_time_ok("Jan  5 09:34:59 2018 GMT", 1515144899)
        
        self.cert_time_ok("Jan  5 09:34:60 2018 GMT", 1515144900)
        
        self.cert_time_ok("Jan  5 09:34:61 2018 GMT", 1515144901)
        self.cert_time_fail("Jan  5 09:34:62 2018 GMT")  

        
        
        self.cert_time_ok("Dec 31 23:59:59 9999 GMT", 253402300799.0)

    @support.run_with_locale('LC_ALL', '')
    def test_cert_time_to_seconds_locale(self):
        

        def local_february_name():
            return time.strftime('%b', (1, 2, 3, 4, 5, 6, 0, 0, 0))

        if local_february_name().lower() == 'feb':
            self.skipTest("locale-specific month name needs to be "
                          "different from C locale")

        
        self.cert_time_ok("Feb  9 00:00:00 2007 GMT", 1170979200.0)
        self.cert_time_fail(local_february_name() + "  9 00:00:00 2007 GMT")

    def test_connect_ex_error(self):
        server = socket.socket(socket.AF_INET)
        self.addCleanup(server.close)
        port = socket_helper.bind_port(server)  
        s = test_wrap_socket(socket.socket(socket.AF_INET),
                            cert_reqs=ssl.CERT_REQUIRED)
        self.addCleanup(s.close)
        rc = s.connect_ex((HOST, port))
        
        
        errors = (
            errno.ECONNREFUSED, errno.EHOSTUNREACH, errno.ETIMEDOUT,
            errno.EWOULDBLOCK,
        )
        self.assertIn(rc, errors)

    def test_read_write_zero(self):
        
        client_context, server_context, hostname = testing_context()
        server = ThreadedEchoServer(context=server_context)
        with server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))
                self.assertEqual(s.recv(0), b"")
                self.assertEqual(s.send(b""), 0)


class ContextTests(unittest.TestCase):

    def test_constructor(self):
        for protocol in PROTOCOLS:
            if has_tls_protocol(protocol):
                with warnings_helper.check_warnings():
                    ctx = ssl.SSLContext(protocol)
                self.assertEqual(ctx.protocol, protocol)
        with warnings_helper.check_warnings():
            ctx = ssl.SSLContext()
        self.assertEqual(ctx.protocol, ssl.PROTOCOL_TLS)
        self.assertRaises(ValueError, ssl.SSLContext, -1)
        self.assertRaises(ValueError, ssl.SSLContext, 42)

    def test_ciphers(self):
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ctx.set_ciphers("ALL")
        ctx.set_ciphers("DEFAULT")
        with self.assertRaisesRegex(ssl.SSLError, "No cipher can be selected"):
            ctx.set_ciphers("^$:,;?*'dorothyx")

    @unittest.skipUnless(PY_SSL_DEFAULT_CIPHERS == 1,
                         "Test applies only to Python default ciphers")
    def test_python_ciphers(self):
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ciphers = ctx.get_ciphers()
        for suite in ciphers:
            name = suite['name']
            self.assertNotIn("PSK", name)
            self.assertNotIn("SRP", name)
            self.assertNotIn("MD5", name)
            self.assertNotIn("RC4", name)
            self.assertNotIn("3DES", name)

    def test_get_ciphers(self):
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ctx.set_ciphers('AESGCM')
        names = set(d['name'] for d in ctx.get_ciphers())
        expected = {
            'AES128-GCM-SHA256',
            'ECDHE-ECDSA-AES128-GCM-SHA256',
            'ECDHE-RSA-AES128-GCM-SHA256',
            'DHE-RSA-AES128-GCM-SHA256',
            'AES256-GCM-SHA384',
            'ECDHE-ECDSA-AES256-GCM-SHA384',
            'ECDHE-RSA-AES256-GCM-SHA384',
            'DHE-RSA-AES256-GCM-SHA384',
        }
        intersection = names.intersection(expected)
        self.assertGreaterEqual(
            len(intersection), 2, f"\ngot: {sorted(names)}\nexpected: {sorted(expected)}"
        )

    def test_options(self):
        
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        
        default = (ssl.OP_ALL | ssl.OP_NO_SSLv2 | ssl.OP_NO_SSLv3)
        
        default |= (OP_NO_COMPRESSION | OP_CIPHER_SERVER_PREFERENCE |
                    OP_SINGLE_DH_USE | OP_SINGLE_ECDH_USE |
                    OP_ENABLE_MIDDLEBOX_COMPAT)
        self.assertEqual(default, ctx.options)

        
        with warnings_helper.check_warnings():
            ctx.options |= ssl.OP_NO_TLSv1
        self.assertEqual(default | ssl.OP_NO_TLSv1, ctx.options)

        
        with warnings_helper.check_warnings():
            ctx.options = (ctx.options & ~ssl.OP_NO_TLSv1)
        self.assertEqual(default, ctx.options)

        
        ctx.options = 0
        
        self.assertEqual(0, ctx.options & ~ssl.OP_NO_SSLv3)

        
        with self.assertRaises(OverflowError):
            ctx.options = -1
        with self.assertRaises(OverflowError):
            ctx.options = 2 ** 100
        with self.assertRaises(TypeError):
            ctx.options = "abc"

    def test_verify_mode_protocol(self):
        with warnings_helper.check_warnings():
            ctx = ssl.SSLContext(ssl.PROTOCOL_TLS)
        
        self.assertEqual(ctx.verify_mode, ssl.CERT_NONE)
        ctx.verify_mode = ssl.CERT_OPTIONAL
        self.assertEqual(ctx.verify_mode, ssl.CERT_OPTIONAL)
        ctx.verify_mode = ssl.CERT_REQUIRED
        self.assertEqual(ctx.verify_mode, ssl.CERT_REQUIRED)
        ctx.verify_mode = ssl.CERT_NONE
        self.assertEqual(ctx.verify_mode, ssl.CERT_NONE)
        with self.assertRaises(TypeError):
            ctx.verify_mode = None
        with self.assertRaises(ValueError):
            ctx.verify_mode = 42

        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        self.assertEqual(ctx.verify_mode, ssl.CERT_NONE)
        self.assertFalse(ctx.check_hostname)

        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        self.assertEqual(ctx.verify_mode, ssl.CERT_REQUIRED)
        self.assertTrue(ctx.check_hostname)

    def test_hostname_checks_common_name(self):
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        self.assertTrue(ctx.hostname_checks_common_name)
        if ssl.HAS_NEVER_CHECK_COMMON_NAME:
            ctx.hostname_checks_common_name = True
            self.assertTrue(ctx.hostname_checks_common_name)
            ctx.hostname_checks_common_name = False
            self.assertFalse(ctx.hostname_checks_common_name)
            ctx.hostname_checks_common_name = True
            self.assertTrue(ctx.hostname_checks_common_name)
        else:
            with self.assertRaises(AttributeError):
                ctx.hostname_checks_common_name = True

    @ignore_deprecation
    def test_min_max_version(self):
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        
        
        minimum_range = {
            
            ssl.TLSVersion.MINIMUM_SUPPORTED,
            
            ssl.TLSVersion.TLSv1,
            
            ssl.TLSVersion.TLSv1_2
        }
        maximum_range = {
            
            ssl.TLSVersion.MAXIMUM_SUPPORTED,
            
            ssl.TLSVersion.TLSv1_3
        }

        self.assertIn(
            ctx.minimum_version, minimum_range
        )
        self.assertIn(
            ctx.maximum_version, maximum_range
        )

        ctx.minimum_version = ssl.TLSVersion.TLSv1_1
        ctx.maximum_version = ssl.TLSVersion.TLSv1_2
        self.assertEqual(
            ctx.minimum_version, ssl.TLSVersion.TLSv1_1
        )
        self.assertEqual(
            ctx.maximum_version, ssl.TLSVersion.TLSv1_2
        )

        ctx.minimum_version = ssl.TLSVersion.MINIMUM_SUPPORTED
        ctx.maximum_version = ssl.TLSVersion.TLSv1
        self.assertEqual(
            ctx.minimum_version, ssl.TLSVersion.MINIMUM_SUPPORTED
        )
        self.assertEqual(
            ctx.maximum_version, ssl.TLSVersion.TLSv1
        )

        ctx.maximum_version = ssl.TLSVersion.MAXIMUM_SUPPORTED
        self.assertEqual(
            ctx.maximum_version, ssl.TLSVersion.MAXIMUM_SUPPORTED
        )

        ctx.maximum_version = ssl.TLSVersion.MINIMUM_SUPPORTED
        self.assertIn(
            ctx.maximum_version,
            {ssl.TLSVersion.TLSv1, ssl.TLSVersion.TLSv1_1, ssl.TLSVersion.SSLv3}
        )

        ctx.minimum_version = ssl.TLSVersion.MAXIMUM_SUPPORTED
        self.assertIn(
            ctx.minimum_version,
            {ssl.TLSVersion.TLSv1_2, ssl.TLSVersion.TLSv1_3}
        )

        with self.assertRaises(ValueError):
            ctx.minimum_version = 42

        if has_tls_protocol(ssl.PROTOCOL_TLSv1_1):
            ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1_1)

            self.assertIn(
                ctx.minimum_version, minimum_range
            )
            self.assertEqual(
                ctx.maximum_version, ssl.TLSVersion.MAXIMUM_SUPPORTED
            )
            with self.assertRaises(ValueError):
                ctx.minimum_version = ssl.TLSVersion.MINIMUM_SUPPORTED
            with self.assertRaises(ValueError):
                ctx.maximum_version = ssl.TLSVersion.TLSv1

    @unittest.skipUnless(
        hasattr(ssl.SSLContext, 'security_level'),
        "requires OpenSSL >= 1.1.0"
    )
    def test_security_level(self):
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        
        
        
        security_level_range = {
            0,
            1, 
            2, 
            3,
            4,
            5,
        }
        self.assertIn(ctx.security_level, security_level_range)

    def test_verify_flags(self):
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        
        tf = getattr(ssl, "VERIFY_X509_TRUSTED_FIRST", 0)
        self.assertEqual(ctx.verify_flags, ssl.VERIFY_DEFAULT | tf)
        ctx.verify_flags = ssl.VERIFY_CRL_CHECK_LEAF
        self.assertEqual(ctx.verify_flags, ssl.VERIFY_CRL_CHECK_LEAF)
        ctx.verify_flags = ssl.VERIFY_CRL_CHECK_CHAIN
        self.assertEqual(ctx.verify_flags, ssl.VERIFY_CRL_CHECK_CHAIN)
        ctx.verify_flags = ssl.VERIFY_DEFAULT
        self.assertEqual(ctx.verify_flags, ssl.VERIFY_DEFAULT)
        ctx.verify_flags = ssl.VERIFY_ALLOW_PROXY_CERTS
        self.assertEqual(ctx.verify_flags, ssl.VERIFY_ALLOW_PROXY_CERTS)
        
        ctx.verify_flags = ssl.VERIFY_CRL_CHECK_LEAF | ssl.VERIFY_X509_STRICT
        self.assertEqual(ctx.verify_flags,
                         ssl.VERIFY_CRL_CHECK_LEAF | ssl.VERIFY_X509_STRICT)
        with self.assertRaises(TypeError):
            ctx.verify_flags = None

    def test_load_cert_chain(self):
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        
        ctx.load_cert_chain(CERTFILE, keyfile=None)
        ctx.load_cert_chain(CERTFILE, keyfile=CERTFILE)
        self.assertRaises(TypeError, ctx.load_cert_chain, keyfile=CERTFILE)
        with self.assertRaises(OSError) as cm:
            ctx.load_cert_chain(NONEXISTINGCERT)
        self.assertEqual(cm.exception.errno, errno.ENOENT)
        with self.assertRaisesRegex(ssl.SSLError, "PEM lib"):
            ctx.load_cert_chain(BADCERT)
        with self.assertRaisesRegex(ssl.SSLError, "PEM lib"):
            ctx.load_cert_chain(EMPTYCERT)
        
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        ctx.load_cert_chain(ONLYCERT, ONLYKEY)
        ctx.load_cert_chain(certfile=ONLYCERT, keyfile=ONLYKEY)
        ctx.load_cert_chain(certfile=BYTES_ONLYCERT, keyfile=BYTES_ONLYKEY)
        with self.assertRaisesRegex(ssl.SSLError, "PEM lib"):
            ctx.load_cert_chain(ONLYCERT)
        with self.assertRaisesRegex(ssl.SSLError, "PEM lib"):
            ctx.load_cert_chain(ONLYKEY)
        with self.assertRaisesRegex(ssl.SSLError, "PEM lib"):
            ctx.load_cert_chain(certfile=ONLYKEY, keyfile=ONLYCERT)
        
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        with self.assertRaisesRegex(ssl.SSLError, "key values mismatch"):
            ctx.load_cert_chain(CAFILE_CACERT, ONLYKEY)
        
        ctx.load_cert_chain(CERTFILE_PROTECTED, password=KEY_PASSWORD)
        ctx.load_cert_chain(CERTFILE_PROTECTED, password=KEY_PASSWORD.encode())
        ctx.load_cert_chain(CERTFILE_PROTECTED,
                            password=bytearray(KEY_PASSWORD.encode()))
        ctx.load_cert_chain(ONLYCERT, ONLYKEY_PROTECTED, KEY_PASSWORD)
        ctx.load_cert_chain(ONLYCERT, ONLYKEY_PROTECTED, KEY_PASSWORD.encode())
        ctx.load_cert_chain(ONLYCERT, ONLYKEY_PROTECTED,
                            bytearray(KEY_PASSWORD.encode()))
        with self.assertRaisesRegex(TypeError, "should be a string"):
            ctx.load_cert_chain(CERTFILE_PROTECTED, password=True)
        with self.assertRaises(ssl.SSLError):
            ctx.load_cert_chain(CERTFILE_PROTECTED, password="badpass")
        with self.assertRaisesRegex(ValueError, "cannot be longer"):
            
            
            
            ctx.load_cert_chain(CERTFILE_PROTECTED, password=b'a' * 102400)
        
        def getpass_unicode():
            return KEY_PASSWORD
        def getpass_bytes():
            return KEY_PASSWORD.encode()
        def getpass_bytearray():
            return bytearray(KEY_PASSWORD.encode())
        def getpass_badpass():
            return "badpass"
        def getpass_huge():
            return b'a' * (1024 * 1024)
        def getpass_bad_type():
            return 9
        def getpass_exception():
            raise Exception('getpass error')
        class GetPassCallable:
            def __call__(self):
                return KEY_PASSWORD
            def getpass(self):
                return KEY_PASSWORD
        ctx.load_cert_chain(CERTFILE_PROTECTED, password=getpass_unicode)
        ctx.load_cert_chain(CERTFILE_PROTECTED, password=getpass_bytes)
        ctx.load_cert_chain(CERTFILE_PROTECTED, password=getpass_bytearray)
        ctx.load_cert_chain(CERTFILE_PROTECTED, password=GetPassCallable())
        ctx.load_cert_chain(CERTFILE_PROTECTED,
                            password=GetPassCallable().getpass)
        with self.assertRaises(ssl.SSLError):
            ctx.load_cert_chain(CERTFILE_PROTECTED, password=getpass_badpass)
        with self.assertRaisesRegex(ValueError, "cannot be longer"):
            ctx.load_cert_chain(CERTFILE_PROTECTED, password=getpass_huge)
        with self.assertRaisesRegex(TypeError, "must return a string"):
            ctx.load_cert_chain(CERTFILE_PROTECTED, password=getpass_bad_type)
        with self.assertRaisesRegex(Exception, "getpass error"):
            ctx.load_cert_chain(CERTFILE_PROTECTED, password=getpass_exception)
        
        ctx.load_cert_chain(CERTFILE, password=getpass_exception)

    def test_load_verify_locations(self):
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        ctx.load_verify_locations(CERTFILE)
        ctx.load_verify_locations(cafile=CERTFILE, capath=None)
        ctx.load_verify_locations(BYTES_CERTFILE)
        ctx.load_verify_locations(cafile=BYTES_CERTFILE, capath=None)
        self.assertRaises(TypeError, ctx.load_verify_locations)
        self.assertRaises(TypeError, ctx.load_verify_locations, None, None, None)
        with self.assertRaises(OSError) as cm:
            ctx.load_verify_locations(NONEXISTINGCERT)
        self.assertEqual(cm.exception.errno, errno.ENOENT)
        with self.assertRaisesRegex(ssl.SSLError, "PEM lib"):
            ctx.load_verify_locations(BADCERT)
        ctx.load_verify_locations(CERTFILE, CAPATH)
        ctx.load_verify_locations(CERTFILE, capath=BYTES_CAPATH)

        
        self.assertRaises(TypeError, ctx.load_verify_locations, None, True)

    def test_load_verify_cadata(self):
        
        with open(CAFILE_CACERT) as f:
            cacert_pem = f.read()
        cacert_der = ssl.PEM_cert_to_DER_cert(cacert_pem)
        with open(CAFILE_NEURONIO) as f:
            neuronio_pem = f.read()
        neuronio_der = ssl.PEM_cert_to_DER_cert(neuronio_pem)

        
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        self.assertEqual(ctx.cert_store_stats()["x509_ca"], 0)
        ctx.load_verify_locations(cadata=cacert_pem)
        self.assertEqual(ctx.cert_store_stats()["x509_ca"], 1)
        ctx.load_verify_locations(cadata=neuronio_pem)
        self.assertEqual(ctx.cert_store_stats()["x509_ca"], 2)
        
        ctx.load_verify_locations(cadata=neuronio_pem)
        self.assertEqual(ctx.cert_store_stats()["x509_ca"], 2)

        
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        combined = "\n".join((cacert_pem, neuronio_pem))
        ctx.load_verify_locations(cadata=combined)
        self.assertEqual(ctx.cert_store_stats()["x509_ca"], 2)

        
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        combined = ["head", cacert_pem, "other", neuronio_pem, "again",
                    neuronio_pem, "tail"]
        ctx.load_verify_locations(cadata="\n".join(combined))
        self.assertEqual(ctx.cert_store_stats()["x509_ca"], 2)

        
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ctx.load_verify_locations(cadata=cacert_der)
        ctx.load_verify_locations(cadata=neuronio_der)
        self.assertEqual(ctx.cert_store_stats()["x509_ca"], 2)
        
        ctx.load_verify_locations(cadata=cacert_der)
        self.assertEqual(ctx.cert_store_stats()["x509_ca"], 2)

        
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        combined = b"".join((cacert_der, neuronio_der))
        ctx.load_verify_locations(cadata=combined)
        self.assertEqual(ctx.cert_store_stats()["x509_ca"], 2)

        
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        self.assertRaises(TypeError, ctx.load_verify_locations, cadata=object)

        with self.assertRaisesRegex(
            ssl.SSLError,
            "no start line: cadata does not contain a certificate"
        ):
            ctx.load_verify_locations(cadata="broken")
        with self.assertRaisesRegex(
            ssl.SSLError,
            "not enough data: cadata does not contain a certificate"
        ):
            ctx.load_verify_locations(cadata=b"broken")
        with self.assertRaises(ssl.SSLError):
            ctx.load_verify_locations(cadata=cacert_der + b"A")

    @unittest.skipIf(Py_DEBUG_WIN32, "Avoid mixing debug/release CRT on Windows")
    def test_load_dh_params(self):
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        ctx.load_dh_params(DHFILE)
        if os.name != 'nt':
            ctx.load_dh_params(BYTES_DHFILE)
        self.assertRaises(TypeError, ctx.load_dh_params)
        self.assertRaises(TypeError, ctx.load_dh_params, None)
        with self.assertRaises(FileNotFoundError) as cm:
            ctx.load_dh_params(NONEXISTINGCERT)
        self.assertEqual(cm.exception.errno, errno.ENOENT)
        with self.assertRaises(ssl.SSLError) as cm:
            ctx.load_dh_params(CERTFILE)

    def test_session_stats(self):
        for proto in {ssl.PROTOCOL_TLS_CLIENT, ssl.PROTOCOL_TLS_SERVER}:
            ctx = ssl.SSLContext(proto)
            self.assertEqual(ctx.session_stats(), {
                'number': 0,
                'connect': 0,
                'connect_good': 0,
                'connect_renegotiate': 0,
                'accept': 0,
                'accept_good': 0,
                'accept_renegotiate': 0,
                'hits': 0,
                'misses': 0,
                'timeouts': 0,
                'cache_full': 0,
            })

    def test_set_default_verify_paths(self):
        
        
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ctx.set_default_verify_paths()

    @unittest.skipUnless(ssl.HAS_ECDH, "ECDH disabled on this OpenSSL build")
    def test_set_ecdh_curve(self):
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        ctx.set_ecdh_curve("prime256v1")
        ctx.set_ecdh_curve(b"prime256v1")
        self.assertRaises(TypeError, ctx.set_ecdh_curve)
        self.assertRaises(TypeError, ctx.set_ecdh_curve, None)
        self.assertRaises(ValueError, ctx.set_ecdh_curve, "foo")
        self.assertRaises(ValueError, ctx.set_ecdh_curve, b"foo")

    def test_sni_callback(self):
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)

        
        self.assertRaises(TypeError, ctx.set_servername_callback)
        self.assertRaises(TypeError, ctx.set_servername_callback, 4)
        self.assertRaises(TypeError, ctx.set_servername_callback, "")
        self.assertRaises(TypeError, ctx.set_servername_callback, ctx)

        def dummycallback(sock, servername, ctx):
            pass
        ctx.set_servername_callback(None)
        ctx.set_servername_callback(dummycallback)

    def test_sni_callback_refcycle(self):
        
        
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        def dummycallback(sock, servername, ctx, cycle=ctx):
            pass
        ctx.set_servername_callback(dummycallback)
        wr = weakref.ref(ctx)
        del ctx, dummycallback
        gc.collect()
        self.assertIs(wr(), None)

    def test_cert_store_stats(self):
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        self.assertEqual(ctx.cert_store_stats(),
            {'x509_ca': 0, 'crl': 0, 'x509': 0})
        ctx.load_cert_chain(CERTFILE)
        self.assertEqual(ctx.cert_store_stats(),
            {'x509_ca': 0, 'crl': 0, 'x509': 0})
        ctx.load_verify_locations(CERTFILE)
        self.assertEqual(ctx.cert_store_stats(),
            {'x509_ca': 0, 'crl': 0, 'x509': 1})
        ctx.load_verify_locations(CAFILE_CACERT)
        self.assertEqual(ctx.cert_store_stats(),
            {'x509_ca': 1, 'crl': 0, 'x509': 2})

    def test_get_ca_certs(self):
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        self.assertEqual(ctx.get_ca_certs(), [])
        
        ctx.load_verify_locations(CERTFILE)
        self.assertEqual(ctx.get_ca_certs(), [])
        
        ctx.load_verify_locations(CAFILE_CACERT)
        self.assertEqual(ctx.get_ca_certs(),
            [{'issuer': ((('organizationName', 'Root CA'),),
                         (('organizationalUnitName', 'http://www.cacert.org'),),
                         (('commonName', 'CA Cert Signing Authority'),),
                         (('emailAddress', 'support@cacert.org'),)),
              'notAfter': 'Mar 29 12:29:49 2033 GMT',
              'notBefore': 'Mar 30 12:29:49 2003 GMT',
              'serialNumber': '00',
              'crlDistributionPoints': ('https://www.cacert.org/revoke.crl',),
              'subject': ((('organizationName', 'Root CA'),),
                          (('organizationalUnitName', 'http://www.cacert.org'),),
                          (('commonName', 'CA Cert Signing Authority'),),
                          (('emailAddress', 'support@cacert.org'),)),
              'version': 3}])

        with open(CAFILE_CACERT) as f:
            pem = f.read()
        der = ssl.PEM_cert_to_DER_cert(pem)
        self.assertEqual(ctx.get_ca_certs(True), [der])

    def test_load_default_certs(self):
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ctx.load_default_certs()

        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ctx.load_default_certs(ssl.Purpose.SERVER_AUTH)
        ctx.load_default_certs()

        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ctx.load_default_certs(ssl.Purpose.CLIENT_AUTH)

        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        self.assertRaises(TypeError, ctx.load_default_certs, None)
        self.assertRaises(TypeError, ctx.load_default_certs, 'SERVER_AUTH')

    @unittest.skipIf(sys.platform == "win32", "not-Windows specific")
    def test_load_default_certs_env(self):
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        with os_helper.EnvironmentVarGuard() as env:
            env["SSL_CERT_DIR"] = CAPATH
            env["SSL_CERT_FILE"] = CERTFILE
            ctx.load_default_certs()
            self.assertEqual(ctx.cert_store_stats(), {"crl": 0, "x509": 1, "x509_ca": 0})

    @unittest.skipUnless(sys.platform == "win32", "Windows specific")
    @unittest.skipIf(support.Py_DEBUG,
                     "Debug build does not share environment between CRTs")
    def test_load_default_certs_env_windows(self):
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ctx.load_default_certs()
        stats = ctx.cert_store_stats()

        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        with os_helper.EnvironmentVarGuard() as env:
            env["SSL_CERT_DIR"] = CAPATH
            env["SSL_CERT_FILE"] = CERTFILE
            ctx.load_default_certs()
            stats["x509"] += 1
            self.assertEqual(ctx.cert_store_stats(), stats)

    def _assert_context_options(self, ctx):
        self.assertEqual(ctx.options & ssl.OP_NO_SSLv2, ssl.OP_NO_SSLv2)
        if OP_NO_COMPRESSION != 0:
            self.assertEqual(ctx.options & OP_NO_COMPRESSION,
                             OP_NO_COMPRESSION)
        if OP_SINGLE_DH_USE != 0:
            self.assertEqual(ctx.options & OP_SINGLE_DH_USE,
                             OP_SINGLE_DH_USE)
        if OP_SINGLE_ECDH_USE != 0:
            self.assertEqual(ctx.options & OP_SINGLE_ECDH_USE,
                             OP_SINGLE_ECDH_USE)
        if OP_CIPHER_SERVER_PREFERENCE != 0:
            self.assertEqual(ctx.options & OP_CIPHER_SERVER_PREFERENCE,
                             OP_CIPHER_SERVER_PREFERENCE)
        self.assertEqual(ctx.options & ssl.OP_LEGACY_SERVER_CONNECT,
                         0 if IS_OPENSSL_3_0_0 else ssl.OP_LEGACY_SERVER_CONNECT)

    def test_create_default_context(self):
        ctx = ssl.create_default_context()

        self.assertEqual(ctx.protocol, ssl.PROTOCOL_TLS_CLIENT)
        self.assertEqual(ctx.verify_mode, ssl.CERT_REQUIRED)
        self.assertTrue(ctx.check_hostname)
        self._assert_context_options(ctx)

        with open(SIGNING_CA) as f:
            cadata = f.read()
        ctx = ssl.create_default_context(cafile=SIGNING_CA, capath=CAPATH,
                                         cadata=cadata)
        self.assertEqual(ctx.protocol, ssl.PROTOCOL_TLS_CLIENT)
        self.assertEqual(ctx.verify_mode, ssl.CERT_REQUIRED)
        self._assert_context_options(ctx)

        ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
        self.assertEqual(ctx.protocol, ssl.PROTOCOL_TLS_SERVER)
        self.assertEqual(ctx.verify_mode, ssl.CERT_NONE)
        self._assert_context_options(ctx)

    def test__create_stdlib_context(self):
        ctx = ssl._create_stdlib_context()
        self.assertEqual(ctx.protocol, ssl.PROTOCOL_TLS_CLIENT)
        self.assertEqual(ctx.verify_mode, ssl.CERT_NONE)
        self.assertFalse(ctx.check_hostname)
        self._assert_context_options(ctx)

        if has_tls_protocol(ssl.PROTOCOL_TLSv1):
            with warnings_helper.check_warnings():
                ctx = ssl._create_stdlib_context(ssl.PROTOCOL_TLSv1)
            self.assertEqual(ctx.protocol, ssl.PROTOCOL_TLSv1)
            self.assertEqual(ctx.verify_mode, ssl.CERT_NONE)
            self._assert_context_options(ctx)

        with warnings_helper.check_warnings():
            ctx = ssl._create_stdlib_context(
                ssl.PROTOCOL_TLSv1_2,
                cert_reqs=ssl.CERT_REQUIRED,
                check_hostname=True
            )
        self.assertEqual(ctx.protocol, ssl.PROTOCOL_TLSv1_2)
        self.assertEqual(ctx.verify_mode, ssl.CERT_REQUIRED)
        self.assertTrue(ctx.check_hostname)
        self._assert_context_options(ctx)

        ctx = ssl._create_stdlib_context(purpose=ssl.Purpose.CLIENT_AUTH)
        self.assertEqual(ctx.protocol, ssl.PROTOCOL_TLS_SERVER)
        self.assertEqual(ctx.verify_mode, ssl.CERT_NONE)
        self._assert_context_options(ctx)

    def test_check_hostname(self):
        with warnings_helper.check_warnings():
            ctx = ssl.SSLContext(ssl.PROTOCOL_TLS)
        self.assertFalse(ctx.check_hostname)
        self.assertEqual(ctx.verify_mode, ssl.CERT_NONE)

        
        ctx.check_hostname = True
        self.assertTrue(ctx.check_hostname)
        self.assertEqual(ctx.verify_mode, ssl.CERT_REQUIRED)
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_REQUIRED
        self.assertFalse(ctx.check_hostname)
        self.assertEqual(ctx.verify_mode, ssl.CERT_REQUIRED)

        
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        ctx.check_hostname = False
        self.assertFalse(ctx.check_hostname)
        self.assertEqual(ctx.verify_mode, ssl.CERT_NONE)
        
        ctx.check_hostname = True
        self.assertTrue(ctx.check_hostname)
        self.assertEqual(ctx.verify_mode, ssl.CERT_REQUIRED)

        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_OPTIONAL
        ctx.check_hostname = False
        self.assertFalse(ctx.check_hostname)
        self.assertEqual(ctx.verify_mode, ssl.CERT_OPTIONAL)
        
        ctx.check_hostname = True
        self.assertTrue(ctx.check_hostname)
        self.assertEqual(ctx.verify_mode, ssl.CERT_OPTIONAL)

        
        with self.assertRaises(ValueError):
            ctx.verify_mode = ssl.CERT_NONE
        ctx.check_hostname = False
        self.assertFalse(ctx.check_hostname)
        ctx.verify_mode = ssl.CERT_NONE
        self.assertEqual(ctx.verify_mode, ssl.CERT_NONE)

    def test_context_client_server(self):
        
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        self.assertTrue(ctx.check_hostname)
        self.assertEqual(ctx.verify_mode, ssl.CERT_REQUIRED)

        
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        self.assertFalse(ctx.check_hostname)
        self.assertEqual(ctx.verify_mode, ssl.CERT_NONE)

    def test_context_custom_class(self):
        class MySSLSocket(ssl.SSLSocket):
            pass

        class MySSLObject(ssl.SSLObject):
            pass

        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        ctx.sslsocket_class = MySSLSocket
        ctx.sslobject_class = MySSLObject

        with ctx.wrap_socket(socket.socket(), server_side=True) as sock:
            self.assertIsInstance(sock, MySSLSocket)
        obj = ctx.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(), server_side=True)
        self.assertIsInstance(obj, MySSLObject)

    def test_num_tickest(self):
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        self.assertEqual(ctx.num_tickets, 2)
        ctx.num_tickets = 1
        self.assertEqual(ctx.num_tickets, 1)
        ctx.num_tickets = 0
        self.assertEqual(ctx.num_tickets, 0)
        with self.assertRaises(ValueError):
            ctx.num_tickets = -1
        with self.assertRaises(TypeError):
            ctx.num_tickets = None

        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        self.assertEqual(ctx.num_tickets, 2)
        with self.assertRaises(ValueError):
            ctx.num_tickets = 1


class SSLErrorTests(unittest.TestCase):

    def test_str(self):
        
        e = ssl.SSLError(1, "foo")
        self.assertEqual(str(e), "foo")
        self.assertEqual(e.errno, 1)
        
        e = ssl.SSLZeroReturnError(1, "foo")
        self.assertEqual(str(e), "foo")
        self.assertEqual(e.errno, 1)

    @unittest.skipIf(Py_DEBUG_WIN32, "Avoid mixing debug/release CRT on Windows")
    def test_lib_reason(self):
        
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        with self.assertRaises(ssl.SSLError) as cm:
            ctx.load_dh_params(CERTFILE)
        self.assertEqual(cm.exception.library, 'PEM')
        self.assertEqual(cm.exception.reason, 'NO_START_LINE')
        s = str(cm.exception)
        self.assertTrue(s.startswith("[PEM: NO_START_LINE] no start line"), s)

    def test_subclass(self):
        
        
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        with socket.create_server(("127.0.0.1", 0)) as s:
            c = socket.create_connection(s.getsockname())
            c.setblocking(False)
            with ctx.wrap_socket(c, False, do_handshake_on_connect=False) as c:
                with self.assertRaises(ssl.SSLWantReadError) as cm:
                    c.do_handshake()
                s = str(cm.exception)
                self.assertTrue(s.startswith("The operation did not complete (read)"), s)
                
                self.assertEqual(cm.exception.errno, ssl.SSL_ERROR_WANT_READ)


    def test_bad_server_hostname(self):
        ctx = ssl.create_default_context()
        with self.assertRaises(ValueError):
            ctx.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(),
                         server_hostname="")
        with self.assertRaises(ValueError):
            ctx.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(),
                         server_hostname=".example.org")
        with self.assertRaises(TypeError):
            ctx.wrap_bio(ssl.MemoryBIO(), ssl.MemoryBIO(),
                         server_hostname="example.org\x00evil.com")


class MemoryBIOTests(unittest.TestCase):

    def test_read_write(self):
        bio = ssl.MemoryBIO()
        bio.write(b'foo')
        self.assertEqual(bio.read(), b'foo')
        self.assertEqual(bio.read(), b'')
        bio.write(b'foo')
        bio.write(b'bar')
        self.assertEqual(bio.read(), b'foobar')
        self.assertEqual(bio.read(), b'')
        bio.write(b'baz')
        self.assertEqual(bio.read(2), b'ba')
        self.assertEqual(bio.read(1), b'z')
        self.assertEqual(bio.read(1), b'')

    def test_eof(self):
        bio = ssl.MemoryBIO()
        self.assertFalse(bio.eof)
        self.assertEqual(bio.read(), b'')
        self.assertFalse(bio.eof)
        bio.write(b'foo')
        self.assertFalse(bio.eof)
        bio.write_eof()
        self.assertFalse(bio.eof)
        self.assertEqual(bio.read(2), b'fo')
        self.assertFalse(bio.eof)
        self.assertEqual(bio.read(1), b'o')
        self.assertTrue(bio.eof)
        self.assertEqual(bio.read(), b'')
        self.assertTrue(bio.eof)

    def test_pending(self):
        bio = ssl.MemoryBIO()
        self.assertEqual(bio.pending, 0)
        bio.write(b'foo')
        self.assertEqual(bio.pending, 3)
        for i in range(3):
            bio.read(1)
            self.assertEqual(bio.pending, 3-i-1)
        for i in range(3):
            bio.write(b'x')
            self.assertEqual(bio.pending, i+1)
        bio.read()
        self.assertEqual(bio.pending, 0)

    def test_buffer_types(self):
        bio = ssl.MemoryBIO()
        bio.write(b'foo')
        self.assertEqual(bio.read(), b'foo')
        bio.write(bytearray(b'bar'))
        self.assertEqual(bio.read(), b'bar')
        bio.write(memoryview(b'baz'))
        self.assertEqual(bio.read(), b'baz')
        m = memoryview(bytearray(b'noncontig'))
        noncontig_writable = m[::-2]
        with self.assertRaises(BufferError):
            bio.write(memoryview(noncontig_writable))

    def test_error_types(self):
        bio = ssl.MemoryBIO()
        self.assertRaises(TypeError, bio.write, 'foo')
        self.assertRaises(TypeError, bio.write, None)
        self.assertRaises(TypeError, bio.write, True)
        self.assertRaises(TypeError, bio.write, 1)


class SSLObjectTests(unittest.TestCase):
    def test_private_init(self):
        bio = ssl.MemoryBIO()
        with self.assertRaisesRegex(TypeError, "public constructor"):
            ssl.SSLObject(bio, bio)

    def test_unwrap(self):
        client_ctx, server_ctx, hostname = testing_context()
        c_in = ssl.MemoryBIO()
        c_out = ssl.MemoryBIO()
        s_in = ssl.MemoryBIO()
        s_out = ssl.MemoryBIO()
        client = client_ctx.wrap_bio(c_in, c_out, server_hostname=hostname)
        server = server_ctx.wrap_bio(s_in, s_out, server_side=True)

        
        for _ in range(5):
            try:
                client.do_handshake()
            except ssl.SSLWantReadError:
                pass
            if c_out.pending:
                s_in.write(c_out.read())
            try:
                server.do_handshake()
            except ssl.SSLWantReadError:
                pass
            if s_out.pending:
                c_in.write(s_out.read())
        
        client.do_handshake()
        server.do_handshake()

        
        
        with self.assertRaises(ssl.SSLWantReadError):
            client.unwrap()

        
        
        s_in.write(c_out.read())
        server.unwrap()

        
        
        c_in.write(s_out.read())
        client.unwrap()

class SimpleBackgroundTests(unittest.TestCase):
    

    def setUp(self):
        self.server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        self.server_context.load_cert_chain(SIGNED_CERTFILE)
        server = ThreadedEchoServer(context=self.server_context)
        self.enterContext(server)
        self.server_addr = (HOST, server.port)

    def test_connect(self):
        with test_wrap_socket(socket.socket(socket.AF_INET),
                            cert_reqs=ssl.CERT_NONE) as s:
            s.connect(self.server_addr)
            self.assertEqual({}, s.getpeercert())
            self.assertFalse(s.server_side)

        
        with test_wrap_socket(socket.socket(socket.AF_INET),
                            cert_reqs=ssl.CERT_REQUIRED,
                            ca_certs=SIGNING_CA) as s:
            s.connect(self.server_addr)
            self.assertTrue(s.getpeercert())
            self.assertFalse(s.server_side)

    def test_connect_fail(self):
        
        
        
        s = test_wrap_socket(socket.socket(socket.AF_INET),
                            cert_reqs=ssl.CERT_REQUIRED)
        self.addCleanup(s.close)
        self.assertRaisesRegex(ssl.SSLError, "certificate verify failed",
                               s.connect, self.server_addr)

    def test_connect_ex(self):
        
        s = test_wrap_socket(socket.socket(socket.AF_INET),
                            cert_reqs=ssl.CERT_REQUIRED,
                            ca_certs=SIGNING_CA)
        self.addCleanup(s.close)
        self.assertEqual(0, s.connect_ex(self.server_addr))
        self.assertTrue(s.getpeercert())

    def test_non_blocking_connect_ex(self):
        
        
        s = test_wrap_socket(socket.socket(socket.AF_INET),
                            cert_reqs=ssl.CERT_REQUIRED,
                            ca_certs=SIGNING_CA,
                            do_handshake_on_connect=False)
        self.addCleanup(s.close)
        s.setblocking(False)
        rc = s.connect_ex(self.server_addr)
        
        self.assertIn(rc, (0, errno.EINPROGRESS, errno.EWOULDBLOCK))
        
        select.select([], [s], [], 5.0)
        
        while True:
            try:
                s.do_handshake()
                break
            except ssl.SSLWantReadError:
                select.select([s], [], [], 5.0)
            except ssl.SSLWantWriteError:
                select.select([], [s], [], 5.0)
        
        self.assertTrue(s.getpeercert())

    def test_connect_with_context(self):
        
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        with ctx.wrap_socket(socket.socket(socket.AF_INET)) as s:
            s.connect(self.server_addr)
            self.assertEqual({}, s.getpeercert())
        
        with ctx.wrap_socket(socket.socket(socket.AF_INET),
                            server_hostname="dummy") as s:
            s.connect(self.server_addr)
        ctx.verify_mode = ssl.CERT_REQUIRED
        
        ctx.load_verify_locations(SIGNING_CA)
        with ctx.wrap_socket(socket.socket(socket.AF_INET)) as s:
            s.connect(self.server_addr)
            cert = s.getpeercert()
            self.assertTrue(cert)

    def test_connect_with_context_fail(self):
        
        
        
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        s = ctx.wrap_socket(
            socket.socket(socket.AF_INET),
            server_hostname=SIGNED_CERTFILE_HOSTNAME
        )
        self.addCleanup(s.close)
        self.assertRaisesRegex(ssl.SSLError, "certificate verify failed",
                                s.connect, self.server_addr)

    def test_connect_capath(self):
        
        
        
        
        
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ctx.load_verify_locations(capath=CAPATH)
        with ctx.wrap_socket(socket.socket(socket.AF_INET),
                             server_hostname=SIGNED_CERTFILE_HOSTNAME) as s:
            s.connect(self.server_addr)
            cert = s.getpeercert()
            self.assertTrue(cert)

        
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ctx.load_verify_locations(capath=BYTES_CAPATH)
        with ctx.wrap_socket(socket.socket(socket.AF_INET),
                             server_hostname=SIGNED_CERTFILE_HOSTNAME) as s:
            s.connect(self.server_addr)
            cert = s.getpeercert()
            self.assertTrue(cert)

    def test_connect_cadata(self):
        with open(SIGNING_CA) as f:
            pem = f.read()
        der = ssl.PEM_cert_to_DER_cert(pem)
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ctx.load_verify_locations(cadata=pem)
        with ctx.wrap_socket(socket.socket(socket.AF_INET),
                             server_hostname=SIGNED_CERTFILE_HOSTNAME) as s:
            s.connect(self.server_addr)
            cert = s.getpeercert()
            self.assertTrue(cert)

        
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ctx.load_verify_locations(cadata=der)
        with ctx.wrap_socket(socket.socket(socket.AF_INET),
                             server_hostname=SIGNED_CERTFILE_HOSTNAME) as s:
            s.connect(self.server_addr)
            cert = s.getpeercert()
            self.assertTrue(cert)

    @unittest.skipIf(os.name == "nt", "Can't use a socket as a file under Windows")
    def test_makefile_close(self):
        
        
        
        ss = test_wrap_socket(socket.socket(socket.AF_INET))
        ss.connect(self.server_addr)
        fd = ss.fileno()
        f = ss.makefile()
        f.close()
        
        os.read(fd, 0)
        
        ss.close()
        gc.collect()
        with self.assertRaises(OSError) as e:
            os.read(fd, 0)
        self.assertEqual(e.exception.errno, errno.EBADF)

    def test_non_blocking_handshake(self):
        s = socket.socket(socket.AF_INET)
        s.connect(self.server_addr)
        s.setblocking(False)
        s = test_wrap_socket(s,
                            cert_reqs=ssl.CERT_NONE,
                            do_handshake_on_connect=False)
        self.addCleanup(s.close)
        count = 0
        while True:
            try:
                count += 1
                s.do_handshake()
                break
            except ssl.SSLWantReadError:
                select.select([s], [], [])
            except ssl.SSLWantWriteError:
                select.select([], [s], [])
        if support.verbose:
            sys.stdout.write("\nNeeded %d calls to do_handshake() to establish session.\n" % count)

    def test_get_server_certificate(self):
        _test_get_server_certificate(self, *self.server_addr, cert=SIGNING_CA)

    def test_get_server_certificate_sni(self):
        host, port = self.server_addr
        server_names = []

        
        def servername_cb(ssl_sock, server_name, initial_context):
            server_names.append(server_name)
        self.server_context.set_servername_callback(servername_cb)

        pem = ssl.get_server_certificate((host, port))
        if not pem:
            self.fail("No server certificate on %s:%s!" % (host, port))

        pem = ssl.get_server_certificate((host, port), ca_certs=SIGNING_CA)
        if not pem:
            self.fail("No server certificate on %s:%s!" % (host, port))
        if support.verbose:
            sys.stdout.write("\nVerified certificate for %s:%s is\n%s\n" % (host, port, pem))

        self.assertEqual(server_names, [host, host])

    def test_get_server_certificate_fail(self):
        
        
        _test_get_server_certificate_fail(self, *self.server_addr)

    def test_get_server_certificate_timeout(self):
        def servername_cb(ssl_sock, server_name, initial_context):
            time.sleep(0.2)
        self.server_context.set_servername_callback(servername_cb)

        with self.assertRaises(socket.timeout):
            ssl.get_server_certificate(self.server_addr, ca_certs=SIGNING_CA,
                                       timeout=0.1)

    def test_ciphers(self):
        with test_wrap_socket(socket.socket(socket.AF_INET),
                             cert_reqs=ssl.CERT_NONE, ciphers="ALL") as s:
            s.connect(self.server_addr)
        with test_wrap_socket(socket.socket(socket.AF_INET),
                             cert_reqs=ssl.CERT_NONE, ciphers="DEFAULT") as s:
            s.connect(self.server_addr)
        
        with self.assertRaisesRegex(ssl.SSLError, "No cipher can be selected"):
            with socket.socket(socket.AF_INET) as sock:
                s = test_wrap_socket(sock,
                                    cert_reqs=ssl.CERT_NONE, ciphers="^$:,;?*'dorothyx")
                s.connect(self.server_addr)

    def test_get_ca_certs_capath(self):
        
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ctx.load_verify_locations(capath=CAPATH)
        self.assertEqual(ctx.get_ca_certs(), [])
        with ctx.wrap_socket(socket.socket(socket.AF_INET),
                             server_hostname='localhost') as s:
            s.connect(self.server_addr)
            cert = s.getpeercert()
            self.assertTrue(cert)
        self.assertEqual(len(ctx.get_ca_certs()), 1)

    def test_context_setget(self):
        
        ctx1 = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ctx1.load_verify_locations(capath=CAPATH)
        ctx2 = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ctx2.load_verify_locations(capath=CAPATH)
        s = socket.socket(socket.AF_INET)
        with ctx1.wrap_socket(s, server_hostname='localhost') as ss:
            ss.connect(self.server_addr)
            self.assertIs(ss.context, ctx1)
            self.assertIs(ss._sslobj.context, ctx1)
            ss.context = ctx2
            self.assertIs(ss.context, ctx2)
            self.assertIs(ss._sslobj.context, ctx2)

    def ssl_io_loop(self, sock, incoming, outgoing, func, *args, **kwargs):
        
        
        timeout = kwargs.get('timeout', support.SHORT_TIMEOUT)
        count = 0
        for _ in support.busy_retry(timeout):
            errno = None
            count += 1
            try:
                ret = func(*args)
            except ssl.SSLError as e:
                if e.errno not in (ssl.SSL_ERROR_WANT_READ,
                                   ssl.SSL_ERROR_WANT_WRITE):
                    raise
                errno = e.errno
            
            
            buf = outgoing.read()
            sock.sendall(buf)
            
            
            if errno is None:
                break
            elif errno == ssl.SSL_ERROR_WANT_READ:
                buf = sock.recv(32768)
                if buf:
                    incoming.write(buf)
                else:
                    incoming.write_eof()
        if support.verbose:
            sys.stdout.write("Needed %d calls to complete %s().\n"
                             % (count, func.__name__))
        return ret

    def test_bio_handshake(self):
        sock = socket.socket(socket.AF_INET)
        self.addCleanup(sock.close)
        sock.connect(self.server_addr)
        incoming = ssl.MemoryBIO()
        outgoing = ssl.MemoryBIO()
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        self.assertTrue(ctx.check_hostname)
        self.assertEqual(ctx.verify_mode, ssl.CERT_REQUIRED)
        ctx.load_verify_locations(SIGNING_CA)
        sslobj = ctx.wrap_bio(incoming, outgoing, False,
                              SIGNED_CERTFILE_HOSTNAME)
        self.assertIs(sslobj._sslobj.owner, sslobj)
        self.assertIsNone(sslobj.cipher())
        self.assertIsNone(sslobj.version())
        self.assertIsNone(sslobj.shared_ciphers())
        self.assertRaises(ValueError, sslobj.getpeercert)
        if 'tls-unique' in ssl.CHANNEL_BINDING_TYPES:
            self.assertIsNone(sslobj.get_channel_binding('tls-unique'))
        self.ssl_io_loop(sock, incoming, outgoing, sslobj.do_handshake)
        self.assertTrue(sslobj.cipher())
        self.assertIsNone(sslobj.shared_ciphers())
        self.assertIsNotNone(sslobj.version())
        self.assertTrue(sslobj.getpeercert())
        if 'tls-unique' in ssl.CHANNEL_BINDING_TYPES:
            self.assertTrue(sslobj.get_channel_binding('tls-unique'))
        try:
            self.ssl_io_loop(sock, incoming, outgoing, sslobj.unwrap)
        except ssl.SSLSyscallError:
            
            
            pass
        self.assertRaises(ssl.SSLError, sslobj.write, b'foo')

    def test_bio_read_write_data(self):
        sock = socket.socket(socket.AF_INET)
        self.addCleanup(sock.close)
        sock.connect(self.server_addr)
        incoming = ssl.MemoryBIO()
        outgoing = ssl.MemoryBIO()
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        sslobj = ctx.wrap_bio(incoming, outgoing, False)
        self.ssl_io_loop(sock, incoming, outgoing, sslobj.do_handshake)
        req = b'FOO\n'
        self.ssl_io_loop(sock, incoming, outgoing, sslobj.write, req)
        buf = self.ssl_io_loop(sock, incoming, outgoing, sslobj.read, 1024)
        self.assertEqual(buf, b'foo\n')
        self.ssl_io_loop(sock, incoming, outgoing, sslobj.unwrap)

    def test_transport_eof(self):
        client_context, server_context, hostname = testing_context()
        with socket.socket(socket.AF_INET) as sock:
            sock.connect(self.server_addr)
            incoming = ssl.MemoryBIO()
            outgoing = ssl.MemoryBIO()
            sslobj = client_context.wrap_bio(incoming, outgoing,
                                             server_hostname=hostname)
            self.ssl_io_loop(sock, incoming, outgoing, sslobj.do_handshake)

            
            incoming.write_eof()
            self.assertRaises(ssl.SSLEOFError, sslobj.read)


@support.requires_resource('network')
class NetworkedTests(unittest.TestCase):

    def test_timeout_connect_ex(self):
        
        
        with socket_helper.transient_internet(REMOTE_HOST):
            s = test_wrap_socket(socket.socket(socket.AF_INET),
                                cert_reqs=ssl.CERT_REQUIRED,
                                do_handshake_on_connect=False)
            self.addCleanup(s.close)
            s.settimeout(0.0000001)
            rc = s.connect_ex((REMOTE_HOST, 443))
            if rc == 0:
                self.skipTest("REMOTE_HOST responded too quickly")
            elif rc == errno.ENETUNREACH:
                self.skipTest("Network unreachable.")
            self.assertIn(rc, (errno.EAGAIN, errno.EWOULDBLOCK))

    @unittest.skipUnless(socket_helper.IPV6_ENABLED, 'Needs IPv6')
    @support.requires_resource('walltime')
    def test_get_server_certificate_ipv6(self):
        with socket_helper.transient_internet('ipv6.google.com'):
            _test_get_server_certificate(self, 'ipv6.google.com', 443)
            _test_get_server_certificate_fail(self, 'ipv6.google.com', 443)


def _test_get_server_certificate(test, host, port, cert=None):
    pem = ssl.get_server_certificate((host, port))
    if not pem:
        test.fail("No server certificate on %s:%s!" % (host, port))

    pem = ssl.get_server_certificate((host, port), ca_certs=cert)
    if not pem:
        test.fail("No server certificate on %s:%s!" % (host, port))
    if support.verbose:
        sys.stdout.write("\nVerified certificate for %s:%s is\n%s\n" % (host, port ,pem))

def _test_get_server_certificate_fail(test, host, port):
    try:
        pem = ssl.get_server_certificate((host, port), ca_certs=CERTFILE)
    except ssl.SSLError as x:
        
        if support.verbose:
            sys.stdout.write("%s\n" % x)
    else:
        test.fail("Got server certificate %s for %s:%s!" % (pem, host, port))


from test.ssl_servers import make_https_server

class ThreadedEchoServer(threading.Thread):

    class ConnectionHandler(threading.Thread):

        

        def __init__(self, server, connsock, addr):
            self.server = server
            self.running = False
            self.sock = connsock
            self.addr = addr
            self.sock.setblocking(True)
            self.sslconn = None
            threading.Thread.__init__(self)
            self.daemon = True

        def wrap_conn(self):
            try:
                self.sslconn = self.server.context.wrap_socket(
                    self.sock, server_side=True)
                self.server.selected_alpn_protocols.append(self.sslconn.selected_alpn_protocol())
            except (ConnectionResetError, BrokenPipeError, ConnectionAbortedError) as e:
                
                
                
                
                
                
                
                
                
                
                self.server.conn_errors.append(str(e))
                if self.server.chatty:
                    handle_error("\n server:  bad connection attempt from " + repr(self.addr) + ":\n")
                self.running = False
                self.close()
                return False
            except (ssl.SSLError, OSError) as e:
                
                
                
                
                
                
                
                
                
                
                self.server.conn_errors.append(str(e))
                if self.server.chatty:
                    handle_error("\n server:  bad connection attempt from " + repr(self.addr) + ":\n")

                
                
                
                if e.errno != errno.EPROTOTYPE and sys.platform != "darwin":
                    self.running = False
                    self.server.stop()
                    self.close()
                return False
            else:
                self.server.shared_ciphers.append(self.sslconn.shared_ciphers())
                if self.server.context.verify_mode == ssl.CERT_REQUIRED:
                    cert = self.sslconn.getpeercert()
                    if support.verbose and self.server.chatty:
                        sys.stdout.write(" client cert is " + pprint.pformat(cert) + "\n")
                    cert_binary = self.sslconn.getpeercert(True)
                    if support.verbose and self.server.chatty:
                        if cert_binary is None:
                            sys.stdout.write(" client did not provide a cert\n")
                        else:
                            sys.stdout.write(f" cert binary is {len(cert_binary)}b\n")
                cipher = self.sslconn.cipher()
                if support.verbose and self.server.chatty:
                    sys.stdout.write(" server: connection cipher is now " + str(cipher) + "\n")
                return True

        def read(self):
            if self.sslconn:
                return self.sslconn.read()
            else:
                return self.sock.recv(1024)

        def write(self, bytes):
            if self.sslconn:
                return self.sslconn.write(bytes)
            else:
                return self.sock.send(bytes)

        def close(self):
            if self.sslconn:
                self.sslconn.close()
            else:
                self.sock.close()

        def run(self):
            self.running = True
            if not self.server.starttls_server:
                if not self.wrap_conn():
                    return
            while self.running:
                try:
                    msg = self.read()
                    stripped = msg.strip()
                    if not stripped:
                        
                        self.running = False
                        try:
                            self.sock = self.sslconn.unwrap()
                        except OSError:
                            
                            
                            
                            pass
                        else:
                            self.sslconn = None
                        self.close()
                    elif stripped == b'over':
                        if support.verbose and self.server.connectionchatty:
                            sys.stdout.write(" server: client closed connection\n")
                        self.close()
                        return
                    elif (self.server.starttls_server and
                          stripped == b'STARTTLS'):
                        if support.verbose and self.server.connectionchatty:
                            sys.stdout.write(" server: read STARTTLS from client, sending OK...\n")
                        self.write(b"OK\n")
                        if not self.wrap_conn():
                            return
                    elif (self.server.starttls_server and self.sslconn
                          and stripped == b'ENDTLS'):
                        if support.verbose and self.server.connectionchatty:
                            sys.stdout.write(" server: read ENDTLS from client, sending OK...\n")
                        self.write(b"OK\n")
                        self.sock = self.sslconn.unwrap()
                        self.sslconn = None
                        if support.verbose and self.server.connectionchatty:
                            sys.stdout.write(" server: connection is now unencrypted...\n")
                    elif stripped == b'CB tls-unique':
                        if support.verbose and self.server.connectionchatty:
                            sys.stdout.write(" server: read CB tls-unique from client, sending our CB data...\n")
                        data = self.sslconn.get_channel_binding("tls-unique")
                        self.write(repr(data).encode("us-ascii") + b"\n")
                    elif stripped == b'PHA':
                        if support.verbose and self.server.connectionchatty:
                            sys.stdout.write(" server: initiating post handshake auth\n")
                        try:
                            self.sslconn.verify_client_post_handshake()
                        except ssl.SSLError as e:
                            self.write(repr(e).encode("us-ascii") + b"\n")
                        else:
                            self.write(b"OK\n")
                    elif stripped == b'HASCERT':
                        if self.sslconn.getpeercert() is not None:
                            self.write(b'TRUE\n')
                        else:
                            self.write(b'FALSE\n')
                    elif stripped == b'GETCERT':
                        cert = self.sslconn.getpeercert()
                        self.write(repr(cert).encode("us-ascii") + b"\n")
                    elif stripped == b'VERIFIEDCHAIN':
                        certs = self.sslconn._sslobj.get_verified_chain()
                        self.write(len(certs).to_bytes(1, "big") + b"\n")
                    elif stripped == b'UNVERIFIEDCHAIN':
                        certs = self.sslconn._sslobj.get_unverified_chain()
                        self.write(len(certs).to_bytes(1, "big") + b"\n")
                    else:
                        if (support.verbose and
                            self.server.connectionchatty):
                            ctype = (self.sslconn and "encrypted") or "unencrypted"
                            sys.stdout.write(" server: read %r (%s), sending back %r (%s)...\n"
                                             % (msg, ctype, msg.lower(), ctype))
                        self.write(msg.lower())
                except OSError as e:
                    
                    if self.server.chatty and support.verbose:
                        if isinstance(e, ConnectionError):
                            
                            
                            
                            print(
                                f" Connection reset by peer: {self.addr}"
                            )
                        else:
                            handle_error("Test server failure:\n")
                    try:
                        self.write(b"ERROR\n")
                    except OSError:
                        pass
                    self.close()
                    self.running = False

                    
                    
                    self.server.stop()

    def __init__(self, certificate=None, ssl_version=None,
                 certreqs=None, cacerts=None,
                 chatty=True, connectionchatty=False, starttls_server=False,
                 alpn_protocols=None,
                 ciphers=None, context=None):
        if context:
            self.context = context
        else:
            self.context = ssl.SSLContext(ssl_version
                                          if ssl_version is not None
                                          else ssl.PROTOCOL_TLS_SERVER)
            self.context.verify_mode = (certreqs if certreqs is not None
                                        else ssl.CERT_NONE)
            if cacerts:
                self.context.load_verify_locations(cacerts)
            if certificate:
                self.context.load_cert_chain(certificate)
            if alpn_protocols:
                self.context.set_alpn_protocols(alpn_protocols)
            if ciphers:
                self.context.set_ciphers(ciphers)
        self.chatty = chatty
        self.connectionchatty = connectionchatty
        self.starttls_server = starttls_server
        self.sock = socket.socket()
        self.port = socket_helper.bind_port(self.sock)
        self.flag = None
        self.active = False
        self.selected_alpn_protocols = []
        self.shared_ciphers = []
        self.conn_errors = []
        threading.Thread.__init__(self)
        self.daemon = True

    def __enter__(self):
        self.start(threading.Event())
        self.flag.wait()
        return self

    def __exit__(self, *args):
        self.stop()
        self.join()

    def start(self, flag=None):
        self.flag = flag
        threading.Thread.start(self)

    def run(self):
        self.sock.settimeout(1.0)
        self.sock.listen(5)
        self.active = True
        if self.flag:
            
            self.flag.set()
        while self.active:
            try:
                newconn, connaddr = self.sock.accept()
                if support.verbose and self.chatty:
                    sys.stdout.write(' server:  new connection from '
                                     + repr(connaddr) + '\n')
                handler = self.ConnectionHandler(self, newconn, connaddr)
                handler.start()
                handler.join()
            except TimeoutError as e:
                if support.verbose:
                    sys.stdout.write(f' connection timeout {e!r}\n')
            except KeyboardInterrupt:
                self.stop()
            except BaseException as e:
                if support.verbose and self.chatty:
                    sys.stdout.write(
                        ' connection handling failed: ' + repr(e) + '\n')

        self.close()

    def close(self):
        if self.sock is not None:
            self.sock.close()
            self.sock = None

    def stop(self):
        self.active = False

class AsyncoreEchoServer(threading.Thread):

    

    class EchoServer (asyncore.dispatcher):

        class ConnectionHandler(asyncore.dispatcher_with_send):

            def __init__(self, conn, certfile):
                self.socket = test_wrap_socket(conn, server_side=True,
                                              certfile=certfile,
                                              do_handshake_on_connect=False)
                asyncore.dispatcher_with_send.__init__(self, self.socket)
                self._ssl_accepting = True
                self._do_ssl_handshake()

            def readable(self):
                if isinstance(self.socket, ssl.SSLSocket):
                    while self.socket.pending() > 0:
                        self.handle_read_event()
                return True

            def _do_ssl_handshake(self):
                try:
                    self.socket.do_handshake()
                except (ssl.SSLWantReadError, ssl.SSLWantWriteError):
                    return
                except ssl.SSLEOFError:
                    return self.handle_close()
                except ssl.SSLError:
                    raise
                except OSError as err:
                    if err.args[0] == errno.ECONNABORTED:
                        return self.handle_close()
                else:
                    self._ssl_accepting = False

            def handle_read(self):
                if self._ssl_accepting:
                    self._do_ssl_handshake()
                else:
                    data = self.recv(1024)
                    if support.verbose:
                        sys.stdout.write(" server:  read %s from client\n" % repr(data))
                    if not data:
                        self.close()
                    else:
                        self.send(data.lower())

            def handle_close(self):
                self.close()
                if support.verbose:
                    sys.stdout.write(" server:  closed connection %s\n" % self.socket)

            def handle_error(self):
                raise

        def __init__(self, certfile):
            self.certfile = certfile
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.port = socket_helper.bind_port(sock, '')
            asyncore.dispatcher.__init__(self, sock)
            self.listen(5)

        def handle_accepted(self, sock_obj, addr):
            if support.verbose:
                sys.stdout.write(" server:  new connection from %s:%s\n" %addr)
            self.ConnectionHandler(sock_obj, self.certfile)

        def handle_error(self):
            raise

    def __init__(self, certfile):
        self.flag = None
        self.active = False
        self.server = self.EchoServer(certfile)
        self.port = self.server.port
        threading.Thread.__init__(self)
        self.daemon = True

    def __str__(self):
        return "<%s %s>" % (self.__class__.__name__, self.server)

    def __enter__(self):
        self.start(threading.Event())
        self.flag.wait()
        return self

    def __exit__(self, *args):
        if support.verbose:
            sys.stdout.write(" cleanup: stopping server.\n")
        self.stop()
        if support.verbose:
            sys.stdout.write(" cleanup: joining server thread.\n")
        self.join()
        if support.verbose:
            sys.stdout.write(" cleanup: successfully joined.\n")
        
        asyncore.close_all(ignore_all=True)

    def start (self, flag=None):
        self.flag = flag
        threading.Thread.start(self)

    def run(self):
        self.active = True
        if self.flag:
            self.flag.set()
        while self.active:
            try:
                asyncore.loop(1)
            except:
                pass

    def stop(self):
        self.active = False
        self.server.close()

def server_params_test(client_context, server_context, indata=b"FOO\n",
                       chatty=True, connectionchatty=False, sni_name=None,
                       session=None):
    
    stats = {}
    server = ThreadedEchoServer(context=server_context,
                                chatty=chatty,
                                connectionchatty=False)
    with server:
        with client_context.wrap_socket(socket.socket(),
                server_hostname=sni_name, session=session) as s:
            s.connect((HOST, server.port))
            for arg in [indata, bytearray(indata), memoryview(indata)]:
                if connectionchatty:
                    if support.verbose:
                        sys.stdout.write(
                            " client:  sending %r...\n" % indata)
                s.write(arg)
                outdata = s.read()
                if connectionchatty:
                    if support.verbose:
                        sys.stdout.write(" client:  read %r\n" % outdata)
                if outdata != indata.lower():
                    raise AssertionError(
                        "bad data <<%r>> (%d) received; expected <<%r>> (%d)\n"
                        % (outdata[:20], len(outdata),
                           indata[:20].lower(), len(indata)))
            s.write(b"over\n")
            if connectionchatty:
                if support.verbose:
                    sys.stdout.write(" client:  closing connection.\n")
            stats.update({
                'compression': s.compression(),
                'cipher': s.cipher(),
                'peercert': s.getpeercert(),
                'client_alpn_protocol': s.selected_alpn_protocol(),
                'version': s.version(),
                'session_reused': s.session_reused,
                'session': s.session,
            })
            s.close()
        stats['server_alpn_protocols'] = server.selected_alpn_protocols
        stats['server_shared_ciphers'] = server.shared_ciphers
    return stats

def try_protocol_combo(server_protocol, client_protocol, expect_success,
                       certsreqs=None, server_options=0, client_options=0):
    
    if certsreqs is None:
        certsreqs = ssl.CERT_NONE
    certtype = {
        ssl.CERT_NONE: "CERT_NONE",
        ssl.CERT_OPTIONAL: "CERT_OPTIONAL",
        ssl.CERT_REQUIRED: "CERT_REQUIRED",
    }[certsreqs]
    if support.verbose:
        formatstr = (expect_success and " %s->%s %s\n") or " {%s->%s} %s\n"
        sys.stdout.write(formatstr %
                         (ssl.get_protocol_name(client_protocol),
                          ssl.get_protocol_name(server_protocol),
                          certtype))

    with warnings_helper.check_warnings():
        
        client_context = ssl.SSLContext(client_protocol)
        client_context.options |= client_options
        server_context = ssl.SSLContext(server_protocol)
        server_context.options |= server_options

    min_version = PROTOCOL_TO_TLS_VERSION.get(client_protocol, None)
    if (min_version is not None
        
        
        and hasattr(server_context, 'minimum_version')
        and server_protocol == ssl.PROTOCOL_TLS
        and server_context.minimum_version > min_version
    ):
        
        
        with warnings_helper.check_warnings():
            server_context.minimum_version = min_version

    
    
    
    if client_context.protocol == ssl.PROTOCOL_TLS:
        client_context.set_ciphers("ALL")

    seclevel_workaround(server_context, client_context)

    for ctx in (client_context, server_context):
        ctx.verify_mode = certsreqs
        ctx.load_cert_chain(SIGNED_CERTFILE)
        ctx.load_verify_locations(SIGNING_CA)
    try:
        stats = server_params_test(client_context, server_context,
                                   chatty=False, connectionchatty=False)
    
    
    except ssl.SSLError:
        if expect_success:
            raise
    except OSError as e:
        if expect_success or e.errno != errno.ECONNRESET:
            raise
    else:
        if not expect_success:
            raise AssertionError(
                "Client protocol %s succeeded with server protocol %s!"
                % (ssl.get_protocol_name(client_protocol),
                   ssl.get_protocol_name(server_protocol)))
        elif (expect_success is not True
              and expect_success != stats['version']):
            raise AssertionError("version mismatch: expected %r, got %r"
                                 % (expect_success, stats['version']))


class ThreadedTests(unittest.TestCase):

    @support.requires_resource('walltime')
    def test_echo(self):
        
        if support.verbose:
            sys.stdout.write("\n")

        client_context, server_context, hostname = testing_context()

        with self.subTest(client=ssl.PROTOCOL_TLS_CLIENT, server=ssl.PROTOCOL_TLS_SERVER):
            server_params_test(client_context=client_context,
                               server_context=server_context,
                               chatty=True, connectionchatty=True,
                               sni_name=hostname)

        client_context.check_hostname = False
        with self.subTest(client=ssl.PROTOCOL_TLS_SERVER, server=ssl.PROTOCOL_TLS_CLIENT):
            with self.assertRaises(ssl.SSLError) as e:
                server_params_test(client_context=server_context,
                                   server_context=client_context,
                                   chatty=True, connectionchatty=True,
                                   sni_name=hostname)
            self.assertIn(
                'Cannot create a client socket with a PROTOCOL_TLS_SERVER context',
                str(e.exception)
            )

        with self.subTest(client=ssl.PROTOCOL_TLS_SERVER, server=ssl.PROTOCOL_TLS_SERVER):
            with self.assertRaises(ssl.SSLError) as e:
                server_params_test(client_context=server_context,
                                   server_context=server_context,
                                   chatty=True, connectionchatty=True)
            self.assertIn(
                'Cannot create a client socket with a PROTOCOL_TLS_SERVER context',
                str(e.exception)
            )

        with self.subTest(client=ssl.PROTOCOL_TLS_CLIENT, server=ssl.PROTOCOL_TLS_CLIENT):
            with self.assertRaises(ssl.SSLError) as e:
                server_params_test(client_context=server_context,
                                   server_context=client_context,
                                   chatty=True, connectionchatty=True)
            self.assertIn(
                'Cannot create a client socket with a PROTOCOL_TLS_SERVER context',
                str(e.exception))

    def test_getpeercert(self):
        if support.verbose:
            sys.stdout.write("\n")

        client_context, server_context, hostname = testing_context()
        server = ThreadedEchoServer(context=server_context, chatty=False)
        with server:
            with client_context.wrap_socket(socket.socket(),
                                            do_handshake_on_connect=False,
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))
                
                
                with self.assertRaises(ValueError):
                    s.getpeercert()
                s.do_handshake()
                cert = s.getpeercert()
                self.assertTrue(cert, "Can't get peer certificate.")
                cipher = s.cipher()
                if support.verbose:
                    sys.stdout.write(pprint.pformat(cert) + '\n')
                    sys.stdout.write("Connection cipher is " + str(cipher) + '.\n')
                if 'subject' not in cert:
                    self.fail("No subject field in certificate: %s." %
                              pprint.pformat(cert))
                if ((('organizationName', 'Python Software Foundation'),)
                    not in cert['subject']):
                    self.fail(
                        "Missing or invalid 'organizationName' field in certificate subject; "
                        "should be 'Python Software Foundation'.")
                self.assertIn('notBefore', cert)
                self.assertIn('notAfter', cert)
                before = ssl.cert_time_to_seconds(cert['notBefore'])
                after = ssl.cert_time_to_seconds(cert['notAfter'])
                self.assertLess(before, after)

    def test_crl_check(self):
        if support.verbose:
            sys.stdout.write("\n")

        client_context, server_context, hostname = testing_context()

        tf = getattr(ssl, "VERIFY_X509_TRUSTED_FIRST", 0)
        self.assertEqual(client_context.verify_flags, ssl.VERIFY_DEFAULT | tf)

        
        server = ThreadedEchoServer(context=server_context, chatty=True)
        with server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))
                cert = s.getpeercert()
                self.assertTrue(cert, "Can't get peer certificate.")

        
        client_context.verify_flags |= ssl.VERIFY_CRL_CHECK_LEAF

        server = ThreadedEchoServer(context=server_context, chatty=True)
        with server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                with self.assertRaisesRegex(ssl.SSLError,
                                            "certificate verify failed"):
                    s.connect((HOST, server.port))

        
        client_context.load_verify_locations(CRLFILE)

        server = ThreadedEchoServer(context=server_context, chatty=True)
        with server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))
                cert = s.getpeercert()
                self.assertTrue(cert, "Can't get peer certificate.")

    def test_check_hostname(self):
        if support.verbose:
            sys.stdout.write("\n")

        client_context, server_context, hostname = testing_context()

        
        server = ThreadedEchoServer(context=server_context, chatty=True)
        with server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))
                cert = s.getpeercert()
                self.assertTrue(cert, "Can't get peer certificate.")

        
        server = ThreadedEchoServer(context=server_context, chatty=True)
        with server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname="invalid") as s:
                with self.assertRaisesRegex(
                        ssl.CertificateError,
                        "Hostname mismatch, certificate is not valid for 'invalid'."):
                    s.connect((HOST, server.port))

        
        server = ThreadedEchoServer(context=server_context, chatty=True)
        with server:
            with socket.socket() as s:
                with self.assertRaisesRegex(ValueError,
                                            "check_hostname requires server_hostname"):
                    client_context.wrap_socket(s)

    @unittest.skipUnless(
        ssl.HAS_NEVER_CHECK_COMMON_NAME, "test requires hostname_checks_common_name"
    )
    def test_hostname_checks_common_name(self):
        client_context, server_context, hostname = testing_context()
        assert client_context.hostname_checks_common_name
        client_context.hostname_checks_common_name = False

        
        server = ThreadedEchoServer(context=server_context, chatty=True)
        with server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))

        client_context, server_context, hostname = testing_context(NOSANFILE)
        client_context.hostname_checks_common_name = False
        server = ThreadedEchoServer(context=server_context, chatty=True)
        with server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                with self.assertRaises(ssl.SSLCertVerificationError):
                    s.connect((HOST, server.port))

    def test_ecc_cert(self):
        client_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        client_context.load_verify_locations(SIGNING_CA)
        client_context.set_ciphers('ECDHE:ECDSA:!NULL:!aRSA')
        hostname = SIGNED_CERTFILE_ECC_HOSTNAME

        server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        
        server_context.load_cert_chain(SIGNED_CERTFILE_ECC)

        
        server = ThreadedEchoServer(context=server_context, chatty=True)
        with server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))
                cert = s.getpeercert()
                self.assertTrue(cert, "Can't get peer certificate.")
                cipher = s.cipher()[0].split('-')
                self.assertTrue(cipher[:2], ('ECDHE', 'ECDSA'))

    def test_dual_rsa_ecc(self):
        client_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        client_context.load_verify_locations(SIGNING_CA)
        
        
        client_context.maximum_version = ssl.TLSVersion.TLSv1_2
        
        client_context.set_ciphers('ECDHE:ECDSA:!NULL:!aRSA')
        hostname = SIGNED_CERTFILE_ECC_HOSTNAME

        server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        
        server_context.load_cert_chain(SIGNED_CERTFILE_ECC)
        server_context.load_cert_chain(SIGNED_CERTFILE)

        
        server = ThreadedEchoServer(context=server_context, chatty=True)
        with server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))
                cert = s.getpeercert()
                self.assertTrue(cert, "Can't get peer certificate.")
                cipher = s.cipher()[0].split('-')
                self.assertTrue(cipher[:2], ('ECDHE', 'ECDSA'))

    def test_check_hostname_idn(self):
        if support.verbose:
            sys.stdout.write("\n")

        server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        server_context.load_cert_chain(IDNSANSFILE)

        context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        context.verify_mode = ssl.CERT_REQUIRED
        context.check_hostname = True
        context.load_verify_locations(SIGNING_CA)

        
        
        idn_hostnames = [
            ('könig.idn.pythontest.net',
             'xn--knig-5qa.idn.pythontest.net'),
            ('xn--knig-5qa.idn.pythontest.net',
             'xn--knig-5qa.idn.pythontest.net'),
            (b'xn--knig-5qa.idn.pythontest.net',
             'xn--knig-5qa.idn.pythontest.net'),

            ('königsgäßchen.idna2003.pythontest.net',
             'xn--knigsgsschen-lcb0w.idna2003.pythontest.net'),
            ('xn--knigsgsschen-lcb0w.idna2003.pythontest.net',
             'xn--knigsgsschen-lcb0w.idna2003.pythontest.net'),
            (b'xn--knigsgsschen-lcb0w.idna2003.pythontest.net',
             'xn--knigsgsschen-lcb0w.idna2003.pythontest.net'),

            
            
            ('xn--knigsgchen-b4a3dun.idna2008.pythontest.net',
             'xn--knigsgchen-b4a3dun.idna2008.pythontest.net'),
            (b'xn--knigsgchen-b4a3dun.idna2008.pythontest.net',
             'xn--knigsgchen-b4a3dun.idna2008.pythontest.net'),

        ]
        for server_hostname, expected_hostname in idn_hostnames:
            server = ThreadedEchoServer(context=server_context, chatty=True)
            with server:
                with context.wrap_socket(socket.socket(),
                                         server_hostname=server_hostname) as s:
                    self.assertEqual(s.server_hostname, expected_hostname)
                    s.connect((HOST, server.port))
                    cert = s.getpeercert()
                    self.assertEqual(s.server_hostname, expected_hostname)
                    self.assertTrue(cert, "Can't get peer certificate.")

        
        server = ThreadedEchoServer(context=server_context, chatty=True)
        with server:
            with context.wrap_socket(socket.socket(),
                                     server_hostname="python.example.org") as s:
                with self.assertRaises(ssl.CertificateError):
                    s.connect((HOST, server.port))

    def test_wrong_cert_tls12(self):
        
        client_context, server_context, hostname = testing_context()
        
        client_context.load_cert_chain(CERTFILE)
        
        server_context.verify_mode = ssl.CERT_REQUIRED
        
        client_context.maximum_version = ssl.TLSVersion.TLSv1_2

        server = ThreadedEchoServer(
            context=server_context, chatty=True, connectionchatty=True,
        )

        with server, \
                client_context.wrap_socket(socket.socket(),
                                           server_hostname=hostname) as s:
            try:
                
                
                
                s.connect((HOST, server.port))
            except ssl.SSLError as e:
                if support.verbose:
                    sys.stdout.write("\nSSLError is %r\n" % e)
            except OSError as e:
                if e.errno != errno.ECONNRESET:
                    raise
                if support.verbose:
                    sys.stdout.write("\nsocket.error is %r\n" % e)
            else:
                self.fail("Use of invalid cert should have failed!")

    @requires_tls_version('TLSv1_3')
    def test_wrong_cert_tls13(self):
        client_context, server_context, hostname = testing_context()
        
        client_context.load_cert_chain(CERTFILE)
        server_context.verify_mode = ssl.CERT_REQUIRED
        server_context.minimum_version = ssl.TLSVersion.TLSv1_3
        client_context.minimum_version = ssl.TLSVersion.TLSv1_3

        server = ThreadedEchoServer(
            context=server_context, chatty=True, connectionchatty=True,
        )
        with server, \
             client_context.wrap_socket(socket.socket(),
                                        server_hostname=hostname,
                                        suppress_ragged_eofs=False) as s:
            s.connect((HOST, server.port))
            with self.assertRaisesRegex(
                ssl.SSLError,
                'alert unknown ca|EOF occurred'
            ):
                
                s.write(b'data')
                s.read(1000)
                s.write(b'should have failed already')
                s.read(1000)

    def test_rude_shutdown(self):
        
        listener_ready = threading.Event()
        listener_gone = threading.Event()

        s = socket.socket()
        port = socket_helper.bind_port(s, HOST)

        
        
        
        
        def listener():
            s.listen()
            listener_ready.set()
            newsock, addr = s.accept()
            newsock.close()
            s.close()
            listener_gone.set()

        def connector():
            listener_ready.wait()
            with socket.socket() as c:
                c.connect((HOST, port))
                listener_gone.wait()
                try:
                    ssl_sock = test_wrap_socket(c)
                except OSError:
                    pass
                else:
                    self.fail('connecting to closed SSL socket should have failed')

        t = threading.Thread(target=listener)
        t.start()
        try:
            connector()
        finally:
            t.join()

    def test_ssl_cert_verify_error(self):
        if support.verbose:
            sys.stdout.write("\n")

        server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        server_context.load_cert_chain(SIGNED_CERTFILE)

        context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)

        server = ThreadedEchoServer(context=server_context, chatty=True)
        with server:
            with context.wrap_socket(socket.socket(),
                                     server_hostname=SIGNED_CERTFILE_HOSTNAME) as s:
                try:
                    s.connect((HOST, server.port))
                except ssl.SSLError as e:
                    msg = 'unable to get local issuer certificate'
                    self.assertIsInstance(e, ssl.SSLCertVerificationError)
                    self.assertEqual(e.verify_code, 20)
                    self.assertEqual(e.verify_message, msg)
                    self.assertIn(msg, repr(e))
                    self.assertIn('certificate verify failed', repr(e))

    def test_PROTOCOL_TLS(self):
        
        if support.verbose:
            sys.stdout.write("\n")
        if has_tls_version('SSLv3'):
            try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_SSLv3, False)
        try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_TLS, True)
        if has_tls_version('TLSv1'):
            try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_TLSv1, 'TLSv1')

        if has_tls_version('SSLv3'):
            try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_SSLv3, False, ssl.CERT_OPTIONAL)
        try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_TLS, True, ssl.CERT_OPTIONAL)
        if has_tls_version('TLSv1'):
            try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_TLSv1, 'TLSv1', ssl.CERT_OPTIONAL)

        if has_tls_version('SSLv3'):
            try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_SSLv3, False, ssl.CERT_REQUIRED)
        try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_TLS, True, ssl.CERT_REQUIRED)
        if has_tls_version('TLSv1'):
            try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_TLSv1, 'TLSv1', ssl.CERT_REQUIRED)

        
        if has_tls_version('SSLv3'):
            try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_SSLv3, False,
                           server_options=ssl.OP_NO_SSLv3)
        
        try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_TLS, True,
                           server_options=ssl.OP_NO_SSLv2 | ssl.OP_NO_SSLv3)
        if has_tls_version('TLSv1'):
            try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_TLSv1, False,
                               server_options=ssl.OP_NO_TLSv1)

    @requires_tls_version('SSLv3')
    def test_protocol_sslv3(self):
        
        if support.verbose:
            sys.stdout.write("\n")
        try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv3, 'SSLv3')
        try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv3, 'SSLv3', ssl.CERT_OPTIONAL)
        try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_SSLv3, 'SSLv3', ssl.CERT_REQUIRED)
        try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_TLS, False,
                           client_options=ssl.OP_NO_SSLv3)
        try_protocol_combo(ssl.PROTOCOL_SSLv3, ssl.PROTOCOL_TLSv1, False)

    @requires_tls_version('TLSv1')
    def test_protocol_tlsv1(self):
        
        if support.verbose:
            sys.stdout.write("\n")
        try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_TLSv1, 'TLSv1')
        try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_TLSv1, 'TLSv1', ssl.CERT_OPTIONAL)
        try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_TLSv1, 'TLSv1', ssl.CERT_REQUIRED)
        if has_tls_version('SSLv3'):
            try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_SSLv3, False)
        try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_TLS, False,
                           client_options=ssl.OP_NO_TLSv1)

    @requires_tls_version('TLSv1_1')
    def test_protocol_tlsv1_1(self):
        
        if support.verbose:
            sys.stdout.write("\n")
        try_protocol_combo(ssl.PROTOCOL_TLSv1_1, ssl.PROTOCOL_TLSv1_1, 'TLSv1.1')
        if has_tls_version('SSLv3'):
            try_protocol_combo(ssl.PROTOCOL_TLSv1_1, ssl.PROTOCOL_SSLv3, False)
        try_protocol_combo(ssl.PROTOCOL_TLSv1_1, ssl.PROTOCOL_TLS, False,
                           client_options=ssl.OP_NO_TLSv1_1)

        try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_TLSv1_1, 'TLSv1.1')
        try_protocol_combo(ssl.PROTOCOL_TLSv1_1, ssl.PROTOCOL_TLSv1_2, False)
        try_protocol_combo(ssl.PROTOCOL_TLSv1_2, ssl.PROTOCOL_TLSv1_1, False)

    @requires_tls_version('TLSv1_2')
    def test_protocol_tlsv1_2(self):
        
        if support.verbose:
            sys.stdout.write("\n")
        try_protocol_combo(ssl.PROTOCOL_TLSv1_2, ssl.PROTOCOL_TLSv1_2, 'TLSv1.2',
                           server_options=ssl.OP_NO_SSLv3|ssl.OP_NO_SSLv2,
                           client_options=ssl.OP_NO_SSLv3|ssl.OP_NO_SSLv2,)
        if has_tls_version('SSLv3'):
            try_protocol_combo(ssl.PROTOCOL_TLSv1_2, ssl.PROTOCOL_SSLv3, False)
        try_protocol_combo(ssl.PROTOCOL_TLSv1_2, ssl.PROTOCOL_TLS, False,
                           client_options=ssl.OP_NO_TLSv1_2)

        try_protocol_combo(ssl.PROTOCOL_TLS, ssl.PROTOCOL_TLSv1_2, 'TLSv1.2')
        if has_tls_protocol(ssl.PROTOCOL_TLSv1):
            try_protocol_combo(ssl.PROTOCOL_TLSv1_2, ssl.PROTOCOL_TLSv1, False)
            try_protocol_combo(ssl.PROTOCOL_TLSv1, ssl.PROTOCOL_TLSv1_2, False)
        if has_tls_protocol(ssl.PROTOCOL_TLSv1_1):
            try_protocol_combo(ssl.PROTOCOL_TLSv1_2, ssl.PROTOCOL_TLSv1_1, False)
            try_protocol_combo(ssl.PROTOCOL_TLSv1_1, ssl.PROTOCOL_TLSv1_2, False)

    def test_starttls(self):
        
        msgs = (b"msg 1", b"MSG 2", b"STARTTLS", b"MSG 3", b"msg 4", b"ENDTLS", b"msg 5", b"msg 6")

        server = ThreadedEchoServer(CERTFILE,
                                    starttls_server=True,
                                    chatty=True,
                                    connectionchatty=True)
        wrapped = False
        with server:
            s = socket.socket()
            s.setblocking(True)
            s.connect((HOST, server.port))
            if support.verbose:
                sys.stdout.write("\n")
            for indata in msgs:
                if support.verbose:
                    sys.stdout.write(
                        " client:  sending %r...\n" % indata)
                if wrapped:
                    conn.write(indata)
                    outdata = conn.read()
                else:
                    s.send(indata)
                    outdata = s.recv(1024)
                msg = outdata.strip().lower()
                if indata == b"STARTTLS" and msg.startswith(b"ok"):
                    
                    if support.verbose:
                        sys.stdout.write(
                            " client:  read %r from server, starting TLS...\n"
                            % msg)
                    conn = test_wrap_socket(s)
                    wrapped = True
                elif indata == b"ENDTLS" and msg.startswith(b"ok"):
                    
                    if support.verbose:
                        sys.stdout.write(
                            " client:  read %r from server, ending TLS...\n"
                            % msg)
                    s = conn.unwrap()
                    wrapped = False
                else:
                    if support.verbose:
                        sys.stdout.write(
                            " client:  read %r from server\n" % msg)
            if support.verbose:
                sys.stdout.write(" client:  closing connection.\n")
            if wrapped:
                conn.write(b"over\n")
            else:
                s.send(b"over\n")
            if wrapped:
                conn.close()
            else:
                s.close()

    def test_socketserver(self):
        
        server = make_https_server(self, certfile=SIGNED_CERTFILE)
        
        if support.verbose:
            sys.stdout.write('\n')
        
        with open(__file__, 'rb') as f:
            d1 = f.read()
        d2 = ''
        
        url = f'https://localhost:{server.port}/test_ssl.py'
        context = ssl.create_default_context(cafile=SIGNING_CA)
        f = urllib.request.urlopen(url, context=context)
        try:
            dlen = f.info().get("content-length")
            if dlen and (int(dlen) > 0):
                d2 = f.read(int(dlen))
                if support.verbose:
                    sys.stdout.write(
                        " client: read %d bytes from remote server '%s'\n"
                        % (len(d2), server))
        finally:
            f.close()
        self.assertEqual(d1, d2)

    def test_asyncore_server(self):
        
        if support.verbose:
            sys.stdout.write("\n")

        indata = b"FOO\n"
        server = AsyncoreEchoServer(CERTFILE)
        with server:
            s = test_wrap_socket(socket.socket())
            s.connect(('127.0.0.1', server.port))
            if support.verbose:
                sys.stdout.write(
                    " client:  sending %r...\n" % indata)
            s.write(indata)
            outdata = s.read()
            if support.verbose:
                sys.stdout.write(" client:  read %r\n" % outdata)
            if outdata != indata.lower():
                self.fail(
                    "bad data <<%r>> (%d) received; expected <<%r>> (%d)\n"
                    % (outdata[:20], len(outdata),
                       indata[:20].lower(), len(indata)))
            s.write(b"over\n")
            if support.verbose:
                sys.stdout.write(" client:  closing connection.\n")
            s.close()
            if support.verbose:
                sys.stdout.write(" client:  connection closed.\n")

    def test_recv_send(self):
        
        if support.verbose:
            sys.stdout.write("\n")

        server = ThreadedEchoServer(CERTFILE,
                                    certreqs=ssl.CERT_NONE,
                                    ssl_version=ssl.PROTOCOL_TLS_SERVER,
                                    cacerts=CERTFILE,
                                    chatty=True,
                                    connectionchatty=False)
        with server:
            s = test_wrap_socket(socket.socket(),
                                server_side=False,
                                certfile=CERTFILE,
                                ca_certs=CERTFILE,
                                cert_reqs=ssl.CERT_NONE)
            s.connect((HOST, server.port))
            
            def _recv_into():
                b = bytearray(b"\0"*100)
                count = s.recv_into(b)
                return b[:count]

            def _recvfrom_into():
                b = bytearray(b"\0"*100)
                count, addr = s.recvfrom_into(b)
                return b[:count]

            
            send_methods = [
                ('send', s.send, True, [], len),
                ('sendto', s.sendto, False, ["some.address"], len),
                ('sendall', s.sendall, True, [], lambda x: None),
            ]
            
            recv_methods = [
                ('recv', s.recv, True, []),
                ('recvfrom', s.recvfrom, False, ["some.address"]),
                ('recv_into', _recv_into, True, []),
                ('recvfrom_into', _recvfrom_into, False, []),
            ]
            data_prefix = "PREFIX_"

            for (meth_name, send_meth, expect_success, args,
                    ret_val_meth) in send_methods:
                indata = (data_prefix + meth_name).encode('ascii')
                try:
                    ret = send_meth(indata, *args)
                    msg = "sending with {}".format(meth_name)
                    self.assertEqual(ret, ret_val_meth(indata), msg=msg)
                    outdata = s.read()
                    if outdata != indata.lower():
                        self.fail(
                            "While sending with <<{name:s}>> bad data "
                            "<<{outdata:r}>> ({nout:d}) received; "
                            "expected <<{indata:r}>> ({nin:d})\n".format(
                                name=meth_name, outdata=outdata[:20],
                                nout=len(outdata),
                                indata=indata[:20], nin=len(indata)
                            )
                        )
                except ValueError as e:
                    if expect_success:
                        self.fail(
                            "Failed to send with method <<{name:s}>>; "
                            "expected to succeed.\n".format(name=meth_name)
                        )
                    if not str(e).startswith(meth_name):
                        self.fail(
                            "Method <<{name:s}>> failed with unexpected "
                            "exception message: {exp:s}\n".format(
                                name=meth_name, exp=e
                            )
                        )

            for meth_name, recv_meth, expect_success, args in recv_methods:
                indata = (data_prefix + meth_name).encode('ascii')
                try:
                    s.send(indata)
                    outdata = recv_meth(*args)
                    if outdata != indata.lower():
                        self.fail(
                            "While receiving with <<{name:s}>> bad data "
                            "<<{outdata:r}>> ({nout:d}) received; "
                            "expected <<{indata:r}>> ({nin:d})\n".format(
                                name=meth_name, outdata=outdata[:20],
                                nout=len(outdata),
                                indata=indata[:20], nin=len(indata)
                            )
                        )
                except ValueError as e:
                    if expect_success:
                        self.fail(
                            "Failed to receive with method <<{name:s}>>; "
                            "expected to succeed.\n".format(name=meth_name)
                        )
                    if not str(e).startswith(meth_name):
                        self.fail(
                            "Method <<{name:s}>> failed with unexpected "
                            "exception message: {exp:s}\n".format(
                                name=meth_name, exp=e
                            )
                        )
                    
                    s.read()

            
            data = b"data"
            s.send(data)
            buffer = bytearray(len(data))
            self.assertEqual(s.read(-1, buffer), len(data))
            self.assertEqual(buffer, data)

            
            if ctypes is not None:
                ubyte = ctypes.c_ubyte * len(data)
                byteslike = ubyte.from_buffer_copy(data)
                s.sendall(byteslike)
                self.assertEqual(s.read(), data)

            
            
            
            self.assertRaises(NotImplementedError, s.dup)
            self.assertRaises(NotImplementedError, s.sendmsg, [b"data"])
            self.assertRaises(NotImplementedError, s.recvmsg, 100)
            self.assertRaises(NotImplementedError,
                              s.recvmsg_into, [bytearray(100)])
            s.write(b"over\n")

            self.assertRaises(ValueError, s.recv, -1)
            self.assertRaises(ValueError, s.read, -1)

            s.close()

    def test_recv_zero(self):
        server = ThreadedEchoServer(CERTFILE)
        self.enterContext(server)
        s = socket.create_connection((HOST, server.port))
        self.addCleanup(s.close)
        s = test_wrap_socket(s, suppress_ragged_eofs=False)
        self.addCleanup(s.close)

        
        s.send(b"data")
        self.assertEqual(s.recv(0), b"")
        self.assertEqual(s.read(0), b"")
        self.assertEqual(s.read(), b"data")

        
        s.setblocking(False)
        self.assertEqual(s.recv(0), b"")
        self.assertEqual(s.recv_into(bytearray()), 0)

    def test_recv_into_buffer_protocol_len(self):
        server = ThreadedEchoServer(CERTFILE)
        self.enterContext(server)
        s = socket.create_connection((HOST, server.port))
        self.addCleanup(s.close)
        s = test_wrap_socket(s, suppress_ragged_eofs=False)
        self.addCleanup(s.close)

        s.send(b"data")
        buf = array.array('I', [0, 0])
        self.assertEqual(s.recv_into(buf), 4)
        self.assertEqual(bytes(buf)[:4], b"data")

        class B(bytearray):
            def __len__(self):
                1/0
        s.send(b"data")
        buf = B(6)
        self.assertEqual(s.recv_into(buf), 4)
        self.assertEqual(bytes(buf), b"data\0\0")

    def test_nonblocking_send(self):
        server = ThreadedEchoServer(CERTFILE,
                                    certreqs=ssl.CERT_NONE,
                                    ssl_version=ssl.PROTOCOL_TLS_SERVER,
                                    cacerts=CERTFILE,
                                    chatty=True,
                                    connectionchatty=False)
        with server:
            s = test_wrap_socket(socket.socket(),
                                server_side=False,
                                certfile=CERTFILE,
                                ca_certs=CERTFILE,
                                cert_reqs=ssl.CERT_NONE)
            s.connect((HOST, server.port))
            s.setblocking(False)

            
            
            buf = bytearray(8192)
            def fill_buffer():
                while True:
                    s.send(buf)
            self.assertRaises((ssl.SSLWantWriteError,
                               ssl.SSLWantReadError), fill_buffer)

            
            s.setblocking(True)
            s.close()

    def test_handshake_timeout(self):
        
        server = socket.socket(socket.AF_INET)
        host = "127.0.0.1"
        port = socket_helper.bind_port(server)
        started = threading.Event()
        finish = False

        def serve():
            server.listen()
            started.set()
            conns = []
            while not finish:
                r, w, e = select.select([server], [], [], 0.1)
                if server in r:
                    
                    
                    conns.append(server.accept()[0])
            for sock in conns:
                sock.close()

        t = threading.Thread(target=serve)
        t.start()
        started.wait()

        try:
            try:
                c = socket.socket(socket.AF_INET)
                c.settimeout(0.2)
                c.connect((host, port))
                
                self.assertRaisesRegex(TimeoutError, "timed out",
                                       test_wrap_socket, c)
            finally:
                c.close()
            try:
                c = socket.socket(socket.AF_INET)
                c = test_wrap_socket(c)
                c.settimeout(0.2)
                
                self.assertRaisesRegex(TimeoutError, "timed out",
                                       c.connect, (host, port))
            finally:
                c.close()
        finally:
            finish = True
            t.join()
            server.close()

    def test_server_accept(self):
        
        
        client_ctx, server_ctx, hostname = testing_context()
        server = socket.socket(socket.AF_INET)
        host = "127.0.0.1"
        port = socket_helper.bind_port(server)
        server = server_ctx.wrap_socket(server, server_side=True)
        self.assertTrue(server.server_side)

        evt = threading.Event()
        remote = None
        peer = None
        def serve():
            nonlocal remote, peer
            server.listen()
            
            evt.set()
            remote, peer = server.accept()
            remote.send(remote.recv(4))

        t = threading.Thread(target=serve)
        t.start()
        
        evt.wait()
        client = client_ctx.wrap_socket(
            socket.socket(), server_hostname=hostname
        )
        client.connect((hostname, port))
        client.send(b'data')
        client.recv()
        client_addr = client.getsockname()
        client.close()
        t.join()
        remote.close()
        server.close()
        
        self.assertIsInstance(remote, ssl.SSLSocket)
        self.assertEqual(peer, client_addr)

    def test_getpeercert_enotconn(self):
        context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        context.check_hostname = False
        with context.wrap_socket(socket.socket()) as sock:
            with self.assertRaises(OSError) as cm:
                sock.getpeercert()
            self.assertEqual(cm.exception.errno, errno.ENOTCONN)

    def test_do_handshake_enotconn(self):
        context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        context.check_hostname = False
        with context.wrap_socket(socket.socket()) as sock:
            with self.assertRaises(OSError) as cm:
                sock.do_handshake()
            self.assertEqual(cm.exception.errno, errno.ENOTCONN)

    def test_no_shared_ciphers(self):
        client_context, server_context, hostname = testing_context()
        
        client_context.maximum_version = ssl.TLSVersion.TLSv1_2
        
        client_context.set_ciphers("AES128")
        server_context.set_ciphers("AES256")
        with ThreadedEchoServer(context=server_context) as server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                with self.assertRaises(OSError):
                    s.connect((HOST, server.port))
        self.assertIn("no shared cipher", server.conn_errors[0])

    def test_version_basic(self):
        
        context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        context.check_hostname = False
        context.verify_mode = ssl.CERT_NONE
        with ThreadedEchoServer(CERTFILE,
                                ssl_version=ssl.PROTOCOL_TLS_SERVER,
                                chatty=False) as server:
            with context.wrap_socket(socket.socket()) as s:
                self.assertIs(s.version(), None)
                self.assertIs(s._sslobj, None)
                s.connect((HOST, server.port))
                self.assertEqual(s.version(), 'TLSv1.3')
            self.assertIs(s._sslobj, None)
            self.assertIs(s.version(), None)

    @requires_tls_version('TLSv1_3')
    def test_tls1_3(self):
        client_context, server_context, hostname = testing_context()
        client_context.minimum_version = ssl.TLSVersion.TLSv1_3
        with ThreadedEchoServer(context=server_context) as server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))
                self.assertIn(s.cipher()[0], {
                    'TLS_AES_256_GCM_SHA384',
                    'TLS_CHACHA20_POLY1305_SHA256',
                    'TLS_AES_128_GCM_SHA256',
                })
                self.assertEqual(s.version(), 'TLSv1.3')

    @requires_tls_version('TLSv1_2')
    @requires_tls_version('TLSv1')
    @ignore_deprecation
    def test_min_max_version_tlsv1_2(self):
        client_context, server_context, hostname = testing_context()
        
        client_context.minimum_version = ssl.TLSVersion.TLSv1
        client_context.maximum_version = ssl.TLSVersion.TLSv1_2
        
        server_context.minimum_version = ssl.TLSVersion.TLSv1_2
        server_context.maximum_version = ssl.TLSVersion.TLSv1_2

        with ThreadedEchoServer(context=server_context) as server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))
                self.assertEqual(s.version(), 'TLSv1.2')

    @requires_tls_version('TLSv1_1')
    @ignore_deprecation
    def test_min_max_version_tlsv1_1(self):
        client_context, server_context, hostname = testing_context()
        
        client_context.minimum_version = ssl.TLSVersion.TLSv1
        client_context.maximum_version = ssl.TLSVersion.TLSv1_2
        server_context.minimum_version = ssl.TLSVersion.TLSv1
        server_context.maximum_version = ssl.TLSVersion.TLSv1_1
        seclevel_workaround(client_context, server_context)

        with ThreadedEchoServer(context=server_context) as server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))
                self.assertEqual(s.version(), 'TLSv1.1')

    @requires_tls_version('TLSv1_2')
    @requires_tls_version('TLSv1')
    @ignore_deprecation
    def test_min_max_version_mismatch(self):
        client_context, server_context, hostname = testing_context()
        
        server_context.maximum_version = ssl.TLSVersion.TLSv1_2
        server_context.minimum_version = ssl.TLSVersion.TLSv1_2
        client_context.maximum_version = ssl.TLSVersion.TLSv1
        client_context.minimum_version = ssl.TLSVersion.TLSv1
        seclevel_workaround(client_context, server_context)

        with ThreadedEchoServer(context=server_context) as server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                with self.assertRaises(ssl.SSLError) as e:
                    s.connect((HOST, server.port))
                self.assertIn("alert", str(e.exception))

    @requires_tls_version('SSLv3')
    def test_min_max_version_sslv3(self):
        client_context, server_context, hostname = testing_context()
        server_context.minimum_version = ssl.TLSVersion.SSLv3
        client_context.minimum_version = ssl.TLSVersion.SSLv3
        client_context.maximum_version = ssl.TLSVersion.SSLv3
        seclevel_workaround(client_context, server_context)

        with ThreadedEchoServer(context=server_context) as server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))
                self.assertEqual(s.version(), 'SSLv3')

    def test_default_ecdh_curve(self):
        
        
        client_context, server_context, hostname = testing_context()
        
        
        client_context.maximum_version = ssl.TLSVersion.TLSv1_2
        
        
        
        
        with ThreadedEchoServer(context=server_context) as server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))
                self.assertIn("ECDH", s.cipher()[0])

    @unittest.skipUnless("tls-unique" in ssl.CHANNEL_BINDING_TYPES,
                         "'tls-unique' channel binding not available")
    def test_tls_unique_channel_binding(self):
        
        if support.verbose:
            sys.stdout.write("\n")

        client_context, server_context, hostname = testing_context()

        server = ThreadedEchoServer(context=server_context,
                                    chatty=True,
                                    connectionchatty=False)

        with server:
            with client_context.wrap_socket(
                    socket.socket(),
                    server_hostname=hostname) as s:
                s.connect((HOST, server.port))
                
                cb_data = s.get_channel_binding("tls-unique")
                if support.verbose:
                    sys.stdout.write(
                        " got channel binding data: {0!r}\n".format(cb_data))

                
                self.assertIsNotNone(cb_data)
                if s.version() == 'TLSv1.3':
                    self.assertEqual(len(cb_data), 48)
                else:
                    self.assertEqual(len(cb_data), 12)  

                
                s.write(b"CB tls-unique\n")
                peer_data_repr = s.read().strip()
                self.assertEqual(peer_data_repr,
                                 repr(cb_data).encode("us-ascii"))

            
            with client_context.wrap_socket(
                    socket.socket(),
                    server_hostname=hostname) as s:
                s.connect((HOST, server.port))
                new_cb_data = s.get_channel_binding("tls-unique")
                if support.verbose:
                    sys.stdout.write(
                        "got another channel binding data: {0!r}\n".format(
                            new_cb_data)
                    )
                
                self.assertNotEqual(cb_data, new_cb_data)
                self.assertIsNotNone(cb_data)
                if s.version() == 'TLSv1.3':
                    self.assertEqual(len(cb_data), 48)
                else:
                    self.assertEqual(len(cb_data), 12)  
                s.write(b"CB tls-unique\n")
                peer_data_repr = s.read().strip()
                self.assertEqual(peer_data_repr,
                                 repr(new_cb_data).encode("us-ascii"))

    def test_compression(self):
        client_context, server_context, hostname = testing_context()
        stats = server_params_test(client_context, server_context,
                                   chatty=True, connectionchatty=True,
                                   sni_name=hostname)
        if support.verbose:
            sys.stdout.write(" got compression: {!r}\n".format(stats['compression']))
        self.assertIn(stats['compression'], { None, 'ZLIB', 'RLE' })

    @unittest.skipUnless(hasattr(ssl, 'OP_NO_COMPRESSION'),
                         "ssl.OP_NO_COMPRESSION needed for this test")
    def test_compression_disabled(self):
        client_context, server_context, hostname = testing_context()
        client_context.options |= ssl.OP_NO_COMPRESSION
        server_context.options |= ssl.OP_NO_COMPRESSION
        stats = server_params_test(client_context, server_context,
                                   chatty=True, connectionchatty=True,
                                   sni_name=hostname)
        self.assertIs(stats['compression'], None)

    def test_legacy_server_connect(self):
        client_context, server_context, hostname = testing_context()
        client_context.options |= ssl.OP_LEGACY_SERVER_CONNECT
        server_params_test(client_context, server_context,
                                   chatty=True, connectionchatty=True,
                                   sni_name=hostname)

    def test_no_legacy_server_connect(self):
        client_context, server_context, hostname = testing_context()
        client_context.options &= ~ssl.OP_LEGACY_SERVER_CONNECT
        server_params_test(client_context, server_context,
                                   chatty=True, connectionchatty=True,
                                   sni_name=hostname)

    @unittest.skipIf(Py_DEBUG_WIN32, "Avoid mixing debug/release CRT on Windows")
    def test_dh_params(self):
        
        client_context, server_context, hostname = testing_context()
        
        client_context.maximum_version = ssl.TLSVersion.TLSv1_2
        server_context.load_dh_params(DHFILE)
        server_context.set_ciphers("kEDH")
        server_context.maximum_version = ssl.TLSVersion.TLSv1_2
        stats = server_params_test(client_context, server_context,
                                   chatty=True, connectionchatty=True,
                                   sni_name=hostname)
        cipher = stats["cipher"][0]
        parts = cipher.split("-")
        if "ADH" not in parts and "EDH" not in parts and "DHE" not in parts:
            self.fail("Non-DH cipher: " + cipher[0])

    def test_ecdh_curve(self):
        
        client_context, server_context, hostname = testing_context()

        server_context.set_ecdh_curve("secp384r1")
        server_context.set_ciphers("ECDHE:!eNULL:!aNULL")
        server_context.minimum_version = ssl.TLSVersion.TLSv1_2
        stats = server_params_test(client_context, server_context,
                                   chatty=True, connectionchatty=True,
                                   sni_name=hostname)

        
        client_context, server_context, hostname = testing_context()
        client_context.set_ecdh_curve("secp384r1")
        server_context.set_ciphers("ECDHE:!eNULL:!aNULL")
        server_context.minimum_version = ssl.TLSVersion.TLSv1_2
        stats = server_params_test(client_context, server_context,
                                   chatty=True, connectionchatty=True,
                                   sni_name=hostname)

        
        client_context, server_context, hostname = testing_context()
        client_context.set_ecdh_curve("prime256v1")
        server_context.set_ecdh_curve("secp384r1")
        server_context.set_ciphers("ECDHE:!eNULL:!aNULL")
        server_context.minimum_version = ssl.TLSVersion.TLSv1_2
        with self.assertRaises(ssl.SSLError):
            server_params_test(client_context, server_context,
                               chatty=True, connectionchatty=True,
                               sni_name=hostname)

    def test_selected_alpn_protocol(self):
        
        client_context, server_context, hostname = testing_context()
        stats = server_params_test(client_context, server_context,
                                   chatty=True, connectionchatty=True,
                                   sni_name=hostname)
        self.assertIs(stats['client_alpn_protocol'], None)

    def test_selected_alpn_protocol_if_server_uses_alpn(self):
        
        client_context, server_context, hostname = testing_context()
        server_context.set_alpn_protocols(['foo', 'bar'])
        stats = server_params_test(client_context, server_context,
                                   chatty=True, connectionchatty=True,
                                   sni_name=hostname)
        self.assertIs(stats['client_alpn_protocol'], None)

    def test_alpn_protocols(self):
        server_protocols = ['foo', 'bar', 'milkshake']
        protocol_tests = [
            (['foo', 'bar'], 'foo'),
            (['bar', 'foo'], 'foo'),
            (['milkshake'], 'milkshake'),
            (['http/3.0', 'http/4.0'], None)
        ]
        for client_protocols, expected in protocol_tests:
            client_context, server_context, hostname = testing_context()
            server_context.set_alpn_protocols(server_protocols)
            client_context.set_alpn_protocols(client_protocols)

            try:
                stats = server_params_test(client_context,
                                           server_context,
                                           chatty=True,
                                           connectionchatty=True,
                                           sni_name=hostname)
            except ssl.SSLError as e:
                stats = e

            msg = "failed trying %s (s) and %s (c).\n" \
                "was expecting %s, but got %%s from the %%s" \
                    % (str(server_protocols), str(client_protocols),
                        str(expected))
            client_result = stats['client_alpn_protocol']
            self.assertEqual(client_result, expected,
                             msg % (client_result, "client"))
            server_result = stats['server_alpn_protocols'][-1] \
                if len(stats['server_alpn_protocols']) else 'nothing'
            self.assertEqual(server_result, expected,
                             msg % (server_result, "server"))

    def test_npn_protocols(self):
        assert not ssl.HAS_NPN

    def sni_contexts(self):
        server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        server_context.load_cert_chain(SIGNED_CERTFILE)
        other_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        other_context.load_cert_chain(SIGNED_CERTFILE2)
        client_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        client_context.load_verify_locations(SIGNING_CA)
        return server_context, other_context, client_context

    def check_common_name(self, stats, name):
        cert = stats['peercert']
        self.assertIn((('commonName', name),), cert['subject'])

    def test_sni_callback(self):
        calls = []
        server_context, other_context, client_context = self.sni_contexts()

        client_context.check_hostname = False

        def servername_cb(ssl_sock, server_name, initial_context):
            calls.append((server_name, initial_context))
            if server_name is not None:
                ssl_sock.context = other_context
        server_context.set_servername_callback(servername_cb)

        stats = server_params_test(client_context, server_context,
                                   chatty=True,
                                   sni_name='supermessage')
        
        
        self.assertEqual(calls, [("supermessage", server_context)])
        
        self.check_common_name(stats, 'fakehostname')

        calls = []
        
        stats = server_params_test(client_context, server_context,
                                   chatty=True,
                                   sni_name=None)
        self.assertEqual(calls, [(None, server_context)])
        self.check_common_name(stats, SIGNED_CERTFILE_HOSTNAME)

        
        calls = []
        server_context.set_servername_callback(None)

        stats = server_params_test(client_context, server_context,
                                   chatty=True,
                                   sni_name='notfunny')
        
        self.check_common_name(stats, SIGNED_CERTFILE_HOSTNAME)
        self.assertEqual(calls, [])

    def test_sni_callback_alert(self):
        
        server_context, other_context, client_context = self.sni_contexts()

        def cb_returning_alert(ssl_sock, server_name, initial_context):
            return ssl.ALERT_DESCRIPTION_ACCESS_DENIED
        server_context.set_servername_callback(cb_returning_alert)
        with self.assertRaises(ssl.SSLError) as cm:
            stats = server_params_test(client_context, server_context,
                                       chatty=False,
                                       sni_name='supermessage')
        self.assertEqual(cm.exception.reason, 'TLSV1_ALERT_ACCESS_DENIED')

    def test_sni_callback_raising(self):
        
        server_context, other_context, client_context = self.sni_contexts()

        def cb_raising(ssl_sock, server_name, initial_context):
            1/0
        server_context.set_servername_callback(cb_raising)

        with support.catch_unraisable_exception() as catch:
            with self.assertRaises(ssl.SSLError) as cm:
                stats = server_params_test(client_context, server_context,
                                           chatty=False,
                                           sni_name='supermessage')

            self.assertEqual(cm.exception.reason,
                             'SSLV3_ALERT_HANDSHAKE_FAILURE')
            self.assertEqual(catch.unraisable.exc_type, ZeroDivisionError)

    def test_sni_callback_wrong_return_type(self):
        
        
        server_context, other_context, client_context = self.sni_contexts()

        def cb_wrong_return_type(ssl_sock, server_name, initial_context):
            return "foo"
        server_context.set_servername_callback(cb_wrong_return_type)

        with support.catch_unraisable_exception() as catch:
            with self.assertRaises(ssl.SSLError) as cm:
                stats = server_params_test(client_context, server_context,
                                           chatty=False,
                                           sni_name='supermessage')


            self.assertEqual(cm.exception.reason, 'TLSV1_ALERT_INTERNAL_ERROR')
            self.assertEqual(catch.unraisable.exc_type, TypeError)

    def test_shared_ciphers(self):
        client_context, server_context, hostname = testing_context()
        client_context.set_ciphers("AES128:AES256")
        server_context.set_ciphers("AES256:eNULL")
        expected_algs = [
            "AES256", "AES-256",
            
            "TLS_CHACHA20", "TLS_AES",
        ]

        stats = server_params_test(client_context, server_context,
                                   sni_name=hostname)
        ciphers = stats['server_shared_ciphers'][0]
        self.assertGreater(len(ciphers), 0)
        for name, tls_version, bits in ciphers:
            if not any(alg in name for alg in expected_algs):
                self.fail(name)

    def test_read_write_after_close_raises_valuerror(self):
        client_context, server_context, hostname = testing_context()
        server = ThreadedEchoServer(context=server_context, chatty=False)

        with server:
            s = client_context.wrap_socket(socket.socket(),
                                           server_hostname=hostname)
            s.connect((HOST, server.port))
            s.close()

            self.assertRaises(ValueError, s.read, 1024)
            self.assertRaises(ValueError, s.write, b'hello')

    def test_sendfile(self):
        TEST_DATA = b"x" * 512
        with open(os_helper.TESTFN, 'wb') as f:
            f.write(TEST_DATA)
        self.addCleanup(os_helper.unlink, os_helper.TESTFN)
        client_context, server_context, hostname = testing_context()
        server = ThreadedEchoServer(context=server_context, chatty=False)
        with server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))
                with open(os_helper.TESTFN, 'rb') as file:
                    s.sendfile(file)
                    self.assertEqual(s.recv(1024), TEST_DATA)

    def test_session(self):
        client_context, server_context, hostname = testing_context()
        
        client_context.maximum_version = ssl.TLSVersion.TLSv1_2

        
        stats = server_params_test(client_context, server_context,
                                   sni_name=hostname)
        session = stats['session']
        self.assertTrue(session.id)
        self.assertGreater(session.time, 0)
        self.assertGreater(session.timeout, 0)
        self.assertTrue(session.has_ticket)
        self.assertGreater(session.ticket_lifetime_hint, 0)
        self.assertFalse(stats['session_reused'])
        sess_stat = server_context.session_stats()
        self.assertEqual(sess_stat['accept'], 1)
        self.assertEqual(sess_stat['hits'], 0)

        
        stats = server_params_test(client_context, server_context,
                                   session=session, sni_name=hostname)
        sess_stat = server_context.session_stats()
        self.assertEqual(sess_stat['accept'], 2)
        self.assertEqual(sess_stat['hits'], 1)
        self.assertTrue(stats['session_reused'])
        session2 = stats['session']
        self.assertEqual(session2.id, session.id)
        self.assertEqual(session2, session)
        self.assertIsNot(session2, session)
        self.assertGreaterEqual(session2.time, session.time)
        self.assertGreaterEqual(session2.timeout, session.timeout)

        
        stats = server_params_test(client_context, server_context,
                                   sni_name=hostname)
        self.assertFalse(stats['session_reused'])
        session3 = stats['session']
        self.assertNotEqual(session3.id, session.id)
        self.assertNotEqual(session3, session)
        sess_stat = server_context.session_stats()
        self.assertEqual(sess_stat['accept'], 3)
        self.assertEqual(sess_stat['hits'], 1)

        
        stats = server_params_test(client_context, server_context,
                                   session=session, sni_name=hostname)
        self.assertTrue(stats['session_reused'])
        session4 = stats['session']
        self.assertEqual(session4.id, session.id)
        self.assertEqual(session4, session)
        self.assertGreaterEqual(session4.time, session.time)
        self.assertGreaterEqual(session4.timeout, session.timeout)
        sess_stat = server_context.session_stats()
        self.assertEqual(sess_stat['accept'], 4)
        self.assertEqual(sess_stat['hits'], 2)

    def test_session_handling(self):
        client_context, server_context, hostname = testing_context()
        client_context2, _, _ = testing_context()

        
        client_context.maximum_version = ssl.TLSVersion.TLSv1_2
        client_context2.maximum_version = ssl.TLSVersion.TLSv1_2

        server = ThreadedEchoServer(context=server_context, chatty=False)
        with server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                
                self.assertEqual(s.session, None)
                self.assertEqual(s.session_reused, None)
                s.connect((HOST, server.port))
                session = s.session
                self.assertTrue(session)
                with self.assertRaises(TypeError) as e:
                    s.session = object
                self.assertEqual(str(e.exception), 'Value is not a SSLSession.')

            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))
                
                with self.assertRaises(ValueError) as e:
                    s.session = session
                self.assertEqual(str(e.exception),
                                 'Cannot set session after handshake.')

            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                
                
                s.session = session
                s.connect((HOST, server.port))
                self.assertEqual(s.session.id, session.id)
                self.assertEqual(s.session, session)
                self.assertEqual(s.session_reused, True)

            with client_context2.wrap_socket(socket.socket(),
                                             server_hostname=hostname) as s:
                
                with self.assertRaises(ValueError) as e:
                    s.session = session
                    s.connect((HOST, server.port))
                self.assertEqual(str(e.exception),
                                 'Session refers to a different SSLContext.')

    @requires_tls_version('TLSv1_2')
    @unittest.skipUnless(ssl.HAS_PSK, 'TLS-PSK disabled on this OpenSSL build')
    def test_psk(self):
        psk = bytes.fromhex('deadbeef')

        client_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        client_context.check_hostname = False
        client_context.verify_mode = ssl.CERT_NONE
        client_context.maximum_version = ssl.TLSVersion.TLSv1_2
        client_context.set_ciphers('PSK')
        client_context.set_psk_client_callback(lambda hint: (None, psk))

        server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        server_context.maximum_version = ssl.TLSVersion.TLSv1_2
        server_context.set_ciphers('PSK')
        server_context.set_psk_server_callback(lambda identity: psk)

        
        server = ThreadedEchoServer(context=server_context)
        with server:
            with client_context.wrap_socket(socket.socket()) as s:
                s.connect((HOST, server.port))

        
        incorrect_psk = bytes.fromhex('cafebabe')
        client_context.set_psk_client_callback(lambda hint: (None, incorrect_psk))
        server = ThreadedEchoServer(context=server_context)
        with server:
            with client_context.wrap_socket(socket.socket()) as s:
                with self.assertRaises(ssl.SSLError):
                    s.connect((HOST, server.port))

        
        identity_hint = 'identity-hint'
        client_identity = 'client-identity'

        def client_callback(hint):
            self.assertEqual(hint, identity_hint)
            return client_identity, psk

        def server_callback(identity):
            self.assertEqual(identity, client_identity)
            return psk

        client_context.set_psk_client_callback(client_callback)
        server_context.set_psk_server_callback(server_callback, identity_hint)
        server = ThreadedEchoServer(context=server_context)
        with server:
            with client_context.wrap_socket(socket.socket()) as s:
                s.connect((HOST, server.port))

        
        with self.assertRaisesRegex(ssl.SSLError, 'Cannot add PSK server callback'):
            client_context.set_psk_server_callback(server_callback, identity_hint)
        with self.assertRaisesRegex(ssl.SSLError, 'Cannot add PSK client callback'):
            server_context.set_psk_client_callback(client_callback)

        
        identity_hint = '身份暗示'  
        client_identity = '客户身份'  

        client_context.set_psk_client_callback(client_callback)
        server_context.set_psk_server_callback(server_callback, identity_hint)
        server = ThreadedEchoServer(context=server_context)
        with server:
            with client_context.wrap_socket(socket.socket()) as s:
                s.connect((HOST, server.port))

    @requires_tls_version('TLSv1_3')
    @unittest.skipUnless(ssl.HAS_PSK, 'TLS-PSK disabled on this OpenSSL build')
    def test_psk_tls1_3(self):
        psk = bytes.fromhex('deadbeef')
        identity_hint = 'identity-hint'
        client_identity = 'client-identity'

        def client_callback(hint):
            
            self.assertIsNone(hint)
            return client_identity, psk

        def server_callback(identity):
            self.assertEqual(identity, client_identity)
            return psk

        client_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        client_context.check_hostname = False
        client_context.verify_mode = ssl.CERT_NONE
        client_context.minimum_version = ssl.TLSVersion.TLSv1_3
        client_context.set_ciphers('PSK')
        client_context.set_psk_client_callback(client_callback)

        server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        server_context.minimum_version = ssl.TLSVersion.TLSv1_3
        server_context.set_ciphers('PSK')
        server_context.set_psk_server_callback(server_callback, identity_hint)

        server = ThreadedEchoServer(context=server_context)
        with server:
            with client_context.wrap_socket(socket.socket()) as s:
                s.connect((HOST, server.port))


@unittest.skipUnless(has_tls_version('TLSv1_3'), "Test needs TLS 1.3")
class TestPostHandshakeAuth(unittest.TestCase):
    def test_pha_setter(self):
        protocols = [
            ssl.PROTOCOL_TLS_SERVER, ssl.PROTOCOL_TLS_CLIENT
        ]
        for protocol in protocols:
            ctx = ssl.SSLContext(protocol)
            self.assertEqual(ctx.post_handshake_auth, False)

            ctx.post_handshake_auth = True
            self.assertEqual(ctx.post_handshake_auth, True)

            ctx.verify_mode = ssl.CERT_REQUIRED
            self.assertEqual(ctx.verify_mode, ssl.CERT_REQUIRED)
            self.assertEqual(ctx.post_handshake_auth, True)

            ctx.post_handshake_auth = False
            self.assertEqual(ctx.verify_mode, ssl.CERT_REQUIRED)
            self.assertEqual(ctx.post_handshake_auth, False)

            ctx.verify_mode = ssl.CERT_OPTIONAL
            ctx.post_handshake_auth = True
            self.assertEqual(ctx.verify_mode, ssl.CERT_OPTIONAL)
            self.assertEqual(ctx.post_handshake_auth, True)

    def test_pha_required(self):
        client_context, server_context, hostname = testing_context()
        server_context.post_handshake_auth = True
        server_context.verify_mode = ssl.CERT_REQUIRED
        client_context.post_handshake_auth = True
        client_context.load_cert_chain(SIGNED_CERTFILE)

        server = ThreadedEchoServer(context=server_context, chatty=False)
        with server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))
                s.write(b'HASCERT')
                self.assertEqual(s.recv(1024), b'FALSE\n')
                s.write(b'PHA')
                self.assertEqual(s.recv(1024), b'OK\n')
                s.write(b'HASCERT')
                self.assertEqual(s.recv(1024), b'TRUE\n')
                
                s.write(b'PHA')
                self.assertEqual(s.recv(1024), b'OK\n')
                s.write(b'GETCERT')
                cert_text = s.recv(4096).decode('us-ascii')
                self.assertIn('Python Software Foundation CA', cert_text)

    def test_pha_required_nocert(self):
        client_context, server_context, hostname = testing_context()
        server_context.post_handshake_auth = True
        server_context.verify_mode = ssl.CERT_REQUIRED
        client_context.post_handshake_auth = True

        def msg_cb(conn, direction, version, content_type, msg_type, data):
            if support.verbose and content_type == _TLSContentType.ALERT:
                info = (conn, direction, version, content_type, msg_type, data)
                sys.stdout.write(f"TLS: {info!r}\n")

        server_context._msg_callback = msg_cb
        client_context._msg_callback = msg_cb

        server = ThreadedEchoServer(context=server_context, chatty=True)
        with server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname,
                                            suppress_ragged_eofs=False) as s:
                s.connect((HOST, server.port))
                s.write(b'PHA')
                
                
                with self.assertRaisesRegex(
                    ssl.SSLError,
                    '(certificate required|EOF occurred)'
                ):
                    
                    data = s.recv(1024)
                    self.assertEqual(data, b'OK\n')

                    
                    s.write(b'HASCERT')

                    
                    s.recv(1024)

    def test_pha_optional(self):
        if support.verbose:
            sys.stdout.write("\n")

        client_context, server_context, hostname = testing_context()
        server_context.post_handshake_auth = True
        server_context.verify_mode = ssl.CERT_REQUIRED
        client_context.post_handshake_auth = True
        client_context.load_cert_chain(SIGNED_CERTFILE)

        
        server_context.verify_mode = ssl.CERT_OPTIONAL
        server = ThreadedEchoServer(context=server_context, chatty=False)
        with server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))
                s.write(b'HASCERT')
                self.assertEqual(s.recv(1024), b'FALSE\n')
                s.write(b'PHA')
                self.assertEqual(s.recv(1024), b'OK\n')
                s.write(b'HASCERT')
                self.assertEqual(s.recv(1024), b'TRUE\n')

    def test_pha_optional_nocert(self):
        if support.verbose:
            sys.stdout.write("\n")

        client_context, server_context, hostname = testing_context()
        server_context.post_handshake_auth = True
        server_context.verify_mode = ssl.CERT_OPTIONAL
        client_context.post_handshake_auth = True

        server = ThreadedEchoServer(context=server_context, chatty=False)
        with server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))
                s.write(b'HASCERT')
                self.assertEqual(s.recv(1024), b'FALSE\n')
                s.write(b'PHA')
                self.assertEqual(s.recv(1024), b'OK\n')
                
                s.write(b'HASCERT')
                self.assertEqual(s.recv(1024), b'FALSE\n')

    def test_pha_no_pha_client(self):
        client_context, server_context, hostname = testing_context()
        server_context.post_handshake_auth = True
        server_context.verify_mode = ssl.CERT_REQUIRED
        client_context.load_cert_chain(SIGNED_CERTFILE)

        server = ThreadedEchoServer(context=server_context, chatty=False)
        with server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))
                with self.assertRaisesRegex(ssl.SSLError, 'not server'):
                    s.verify_client_post_handshake()
                s.write(b'PHA')
                self.assertIn(b'extension not received', s.recv(1024))

    def test_pha_no_pha_server(self):
        
        client_context, server_context, hostname = testing_context()
        server_context.verify_mode = ssl.CERT_REQUIRED
        client_context.post_handshake_auth = True
        client_context.load_cert_chain(SIGNED_CERTFILE)

        server = ThreadedEchoServer(context=server_context, chatty=False)
        with server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))
                s.write(b'HASCERT')
                self.assertEqual(s.recv(1024), b'TRUE\n')
                
                s.write(b'PHA')
                self.assertEqual(s.recv(1024), b'OK\n')
                s.write(b'HASCERT')
                self.assertEqual(s.recv(1024), b'TRUE\n')

    def test_pha_not_tls13(self):
        
        client_context, server_context, hostname = testing_context()
        server_context.verify_mode = ssl.CERT_REQUIRED
        client_context.maximum_version = ssl.TLSVersion.TLSv1_2
        client_context.post_handshake_auth = True
        client_context.load_cert_chain(SIGNED_CERTFILE)

        server = ThreadedEchoServer(context=server_context, chatty=False)
        with server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))
                
                s.write(b'PHA')
                self.assertIn(b'WRONG_SSL_VERSION', s.recv(1024))

    def test_bpo37428_pha_cert_none(self):
        
        
        hostname = SIGNED_CERTFILE_HOSTNAME
        client_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        client_context.post_handshake_auth = True
        client_context.load_cert_chain(SIGNED_CERTFILE)
        
        client_context.check_hostname = False
        client_context.verify_mode = ssl.CERT_NONE

        server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        server_context.load_cert_chain(SIGNED_CERTFILE)
        server_context.load_verify_locations(SIGNING_CA)
        server_context.post_handshake_auth = True
        server_context.verify_mode = ssl.CERT_REQUIRED

        server = ThreadedEchoServer(context=server_context, chatty=False)
        with server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))
                s.write(b'HASCERT')
                self.assertEqual(s.recv(1024), b'FALSE\n')
                s.write(b'PHA')
                self.assertEqual(s.recv(1024), b'OK\n')
                s.write(b'HASCERT')
                self.assertEqual(s.recv(1024), b'TRUE\n')
                
                self.assertEqual(s.getpeercert(), {})

    def test_internal_chain_client(self):
        client_context, server_context, hostname = testing_context(
            server_chain=False
        )
        server = ThreadedEchoServer(context=server_context, chatty=False)
        with server:
            with client_context.wrap_socket(
                socket.socket(),
                server_hostname=hostname
            ) as s:
                s.connect((HOST, server.port))
                vc = s._sslobj.get_verified_chain()
                self.assertEqual(len(vc), 2)
                ee, ca = vc
                uvc = s._sslobj.get_unverified_chain()
                self.assertEqual(len(uvc), 1)

                self.assertEqual(ee, uvc[0])
                self.assertEqual(hash(ee), hash(uvc[0]))
                self.assertEqual(repr(ee), repr(uvc[0]))

                self.assertNotEqual(ee, ca)
                self.assertNotEqual(hash(ee), hash(ca))
                self.assertNotEqual(repr(ee), repr(ca))
                self.assertNotEqual(ee.get_info(), ca.get_info())
                self.assertIn("CN=localhost", repr(ee))
                self.assertIn("CN=our-ca-server", repr(ca))

                pem = ee.public_bytes(_ssl.ENCODING_PEM)
                der = ee.public_bytes(_ssl.ENCODING_DER)
                self.assertIsInstance(pem, str)
                self.assertIn("-----BEGIN CERTIFICATE-----", pem)
                self.assertIsInstance(der, bytes)
                self.assertEqual(
                    ssl.PEM_cert_to_DER_cert(pem), der
                )

    def test_internal_chain_server(self):
        client_context, server_context, hostname = testing_context()
        client_context.load_cert_chain(SIGNED_CERTFILE)
        server_context.verify_mode = ssl.CERT_REQUIRED
        server_context.maximum_version = ssl.TLSVersion.TLSv1_2

        server = ThreadedEchoServer(context=server_context, chatty=False)
        with server:
            with client_context.wrap_socket(
                socket.socket(),
                server_hostname=hostname
            ) as s:
                s.connect((HOST, server.port))
                s.write(b'VERIFIEDCHAIN\n')
                res = s.recv(1024)
                self.assertEqual(res, b'\x02\n')
                s.write(b'UNVERIFIEDCHAIN\n')
                res = s.recv(1024)
                self.assertEqual(res, b'\x02\n')


HAS_KEYLOG = hasattr(ssl.SSLContext, 'keylog_filename')
requires_keylog = unittest.skipUnless(
    HAS_KEYLOG, 'test requires OpenSSL 1.1.1 with keylog callback')

class TestSSLDebug(unittest.TestCase):

    def keylog_lines(self, fname=os_helper.TESTFN):
        with open(fname) as f:
            return len(list(f))

    @requires_keylog
    @unittest.skipIf(Py_DEBUG_WIN32, "Avoid mixing debug/release CRT on Windows")
    def test_keylog_defaults(self):
        self.addCleanup(os_helper.unlink, os_helper.TESTFN)
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
        self.assertEqual(ctx.keylog_filename, None)

        self.assertFalse(os.path.isfile(os_helper.TESTFN))
        ctx.keylog_filename = os_helper.TESTFN
        self.assertEqual(ctx.keylog_filename, os_helper.TESTFN)
        self.assertTrue(os.path.isfile(os_helper.TESTFN))
        self.assertEqual(self.keylog_lines(), 1)

        ctx.keylog_filename = None
        self.assertEqual(ctx.keylog_filename, None)

        with self.assertRaises((IsADirectoryError, PermissionError)):
            
            ctx.keylog_filename = os.path.dirname(
                os.path.abspath(os_helper.TESTFN))

        with self.assertRaises(TypeError):
            ctx.keylog_filename = 1

    @requires_keylog
    @unittest.skipIf(Py_DEBUG_WIN32, "Avoid mixing debug/release CRT on Windows")
    def test_keylog_filename(self):
        self.addCleanup(os_helper.unlink, os_helper.TESTFN)
        client_context, server_context, hostname = testing_context()

        client_context.keylog_filename = os_helper.TESTFN
        server = ThreadedEchoServer(context=server_context, chatty=False)
        with server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))
        
        self.assertEqual(self.keylog_lines(), 6)

        client_context.keylog_filename = None
        server_context.keylog_filename = os_helper.TESTFN
        server = ThreadedEchoServer(context=server_context, chatty=False)
        with server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))
        self.assertGreaterEqual(self.keylog_lines(), 11)

        client_context.keylog_filename = os_helper.TESTFN
        server_context.keylog_filename = os_helper.TESTFN
        server = ThreadedEchoServer(context=server_context, chatty=False)
        with server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))
        self.assertGreaterEqual(self.keylog_lines(), 21)

        client_context.keylog_filename = None
        server_context.keylog_filename = None

    @requires_keylog
    @unittest.skipIf(sys.flags.ignore_environment,
                     "test is not compatible with ignore_environment")
    @unittest.skipIf(Py_DEBUG_WIN32, "Avoid mixing debug/release CRT on Windows")
    def test_keylog_env(self):
        self.addCleanup(os_helper.unlink, os_helper.TESTFN)
        with unittest.mock.patch.dict(os.environ):
            os.environ['SSLKEYLOGFILE'] = os_helper.TESTFN
            self.assertEqual(os.environ['SSLKEYLOGFILE'], os_helper.TESTFN)

            ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
            self.assertEqual(ctx.keylog_filename, None)

            ctx = ssl.create_default_context()
            self.assertEqual(ctx.keylog_filename, os_helper.TESTFN)

            ctx = ssl._create_stdlib_context()
            self.assertEqual(ctx.keylog_filename, os_helper.TESTFN)

    def test_msg_callback(self):
        client_context, server_context, hostname = testing_context()

        def msg_cb(conn, direction, version, content_type, msg_type, data):
            pass

        self.assertIs(client_context._msg_callback, None)
        client_context._msg_callback = msg_cb
        self.assertIs(client_context._msg_callback, msg_cb)
        with self.assertRaises(TypeError):
            client_context._msg_callback = object()

    def test_msg_callback_tls12(self):
        client_context, server_context, hostname = testing_context()
        client_context.maximum_version = ssl.TLSVersion.TLSv1_2

        msg = []

        def msg_cb(conn, direction, version, content_type, msg_type, data):
            self.assertIsInstance(conn, ssl.SSLSocket)
            self.assertIsInstance(data, bytes)
            self.assertIn(direction, {'read', 'write'})
            msg.append((direction, version, content_type, msg_type))

        client_context._msg_callback = msg_cb

        server = ThreadedEchoServer(context=server_context, chatty=False)
        with server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))

        self.assertIn(
            ("read", TLSVersion.TLSv1_2, _TLSContentType.HANDSHAKE,
             _TLSMessageType.SERVER_KEY_EXCHANGE),
            msg
        )
        self.assertIn(
            ("write", TLSVersion.TLSv1_2, _TLSContentType.CHANGE_CIPHER_SPEC,
             _TLSMessageType.CHANGE_CIPHER_SPEC),
            msg
        )

    def test_msg_callback_deadlock_bpo43577(self):
        client_context, server_context, hostname = testing_context()
        server_context2 = testing_context()[1]

        def msg_cb(conn, direction, version, content_type, msg_type, data):
            pass

        def sni_cb(sock, servername, ctx):
            sock.context = server_context2

        server_context._msg_callback = msg_cb
        server_context.sni_callback = sni_cb

        server = ThreadedEchoServer(context=server_context, chatty=False)
        with server:
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))
            with client_context.wrap_socket(socket.socket(),
                                            server_hostname=hostname) as s:
                s.connect((HOST, server.port))


def set_socket_so_linger_on_with_zero_timeout(sock):
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack('ii', 1, 0))


class TestPreHandshakeClose(unittest.TestCase):
    

    class SingleConnectionTestServerThread(threading.Thread):

        def __init__(self, *, name, call_after_accept, timeout=None):
            self.call_after_accept = call_after_accept
            self.received_data = b''  
            self.wrap_error = None  
            self.listener = None  
            self.port = None  
            if timeout is None:
                self.timeout = support.SHORT_TIMEOUT
            else:
                self.timeout = timeout
            super().__init__(name=name)

        def __enter__(self):
            self.start()
            return self

        def __exit__(self, *args):
            try:
                if self.listener:
                    self.listener.close()
            except OSError:
                pass
            self.join()
            self.wrap_error = None  

        def start(self):
            self.ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
            self.ssl_ctx.verify_mode = ssl.CERT_REQUIRED
            self.ssl_ctx.load_verify_locations(cafile=ONLYCERT)
            self.ssl_ctx.load_cert_chain(certfile=ONLYCERT, keyfile=ONLYKEY)
            self.listener = socket.socket()
            self.port = socket_helper.bind_port(self.listener)
            self.listener.settimeout(self.timeout)
            self.listener.listen(1)
            super().start()

        def run(self):
            try:
                conn, address = self.listener.accept()
            except TimeoutError:
                
                return
            finally:
                self.listener.close()

            with conn:
                if self.call_after_accept(conn):
                    return
                try:
                    tls_socket = self.ssl_ctx.wrap_socket(conn, server_side=True)
                except OSError as err:  
                    self.wrap_error = err
                else:
                    try:
                        self.received_data = tls_socket.recv(400)
                    except OSError:
                        pass  

    def non_linux_skip_if_other_okay_error(self, err):
        if sys.platform == "linux":
            return  
        if (isinstance(err, ConnectionResetError) or
            (isinstance(err, OSError) and err.errno == errno.EINVAL) or
            re.search('wrong.version.number', getattr(err, "reason", ""), re.I)):
            
            
            
            
            
            
            
            
            try:
                self.skipTest(f"Could not recreate conditions on {sys.platform}:"
                              f" {err=}")
            finally:
                
                err = None

        
        
        

    def test_preauth_data_to_tls_server(self):
        server_accept_called = threading.Event()
        ready_for_server_wrap_socket = threading.Event()

        def call_after_accept(unused):
            server_accept_called.set()
            if not ready_for_server_wrap_socket.wait(support.SHORT_TIMEOUT):
                raise RuntimeError("wrap_socket event never set, test may fail.")
            return False  

        server = self.SingleConnectionTestServerThread(
                call_after_accept=call_after_accept,
                name="preauth_data_to_tls_server")
        self.enterContext(server)  

        with socket.socket() as client:
            client.connect(server.listener.getsockname())
            
            set_socket_so_linger_on_with_zero_timeout(client)
            client.setblocking(False)

            server_accept_called.wait()
            client.send(b"DELETE /data HTTP/1.0\r\n\r\n")
            client.close()  

        ready_for_server_wrap_socket.set()
        server.join()

        wrap_error = server.wrap_error
        server.wrap_error = None
        try:
            self.assertEqual(b"", server.received_data)
            self.assertIsInstance(wrap_error, OSError)  
            self.non_linux_skip_if_other_okay_error(wrap_error)
            self.assertIsInstance(wrap_error, ssl.SSLError)
            self.assertIn("before TLS handshake with data", wrap_error.args[1])
            self.assertIn("before TLS handshake with data", wrap_error.reason)
            self.assertNotEqual(0, wrap_error.args[0])
            self.assertIsNone(wrap_error.library, msg="attr must exist")
        finally:
            
            wrap_error = None
            server = None

    def test_preauth_data_to_tls_client(self):
        server_can_continue_with_wrap_socket = threading.Event()
        client_can_continue_with_wrap_socket = threading.Event()

        def call_after_accept(conn_to_client):
            if not server_can_continue_with_wrap_socket.wait(support.SHORT_TIMEOUT):
                print("ERROR: test client took too long")

            
            set_socket_so_linger_on_with_zero_timeout(conn_to_client)
            conn_to_client.send(
                    b"HTTP/1.0 307 Temporary Redirect\r\n"
                    b"Location: https://example.com/someone-elses-server\r\n"
                    b"\r\n")
            conn_to_client.close()  
            client_can_continue_with_wrap_socket.set()
            return True  

        server = self.SingleConnectionTestServerThread(
                call_after_accept=call_after_accept,
                name="preauth_data_to_tls_client")
        self.enterContext(server)  
        
        set_socket_so_linger_on_with_zero_timeout(server.listener)

        with socket.socket() as client:
            client.connect(server.listener.getsockname())
            server_can_continue_with_wrap_socket.set()

            if not client_can_continue_with_wrap_socket.wait(support.SHORT_TIMEOUT):
                self.fail("test server took too long")
            ssl_ctx = ssl.create_default_context()
            try:
                tls_client = ssl_ctx.wrap_socket(
                        client, server_hostname="localhost")
            except OSError as err:  
                wrap_error = err
                received_data = b""
            else:
                wrap_error = None
                received_data = tls_client.recv(400)
                tls_client.close()

        server.join()
        try:
            self.assertEqual(b"", received_data)
            self.assertIsInstance(wrap_error, OSError)  
            self.non_linux_skip_if_other_okay_error(wrap_error)
            self.assertIsInstance(wrap_error, ssl.SSLError)
            self.assertIn("before TLS handshake with data", wrap_error.args[1])
            self.assertIn("before TLS handshake with data", wrap_error.reason)
            self.assertNotEqual(0, wrap_error.args[0])
            self.assertIsNone(wrap_error.library, msg="attr must exist")
        finally:
            
            wrap_error = None
            server = None

    def test_https_client_non_tls_response_ignored(self):
        server_responding = threading.Event()

        class SynchronizedHTTPSConnection(http.client.HTTPSConnection):
            def connect(self):
                
                
                http.client.HTTPConnection.connect(self)

                
                if not server_responding.wait(support.SHORT_TIMEOUT) and support.verbose:
                    sys.stdout.write("server_responding event never set.")
                self.sock = self._context.wrap_socket(
                        self.sock, server_hostname=self.host)

        def call_after_accept(conn_to_client):
            
            set_socket_so_linger_on_with_zero_timeout(conn_to_client)
            conn_to_client.send(
                    b"HTTP/1.0 402 Payment Required\r\n"
                    b"\r\n")
            conn_to_client.close()  
            server_responding.set()
            return True  

        timeout = 2.0
        server = self.SingleConnectionTestServerThread(
                call_after_accept=call_after_accept,
                name="non_tls_http_RST_responder",
                timeout=timeout)
        self.enterContext(server)  
        
        set_socket_so_linger_on_with_zero_timeout(server.listener)

        connection = SynchronizedHTTPSConnection(
                server.listener.getsockname()[0],
                port=server.port,
                context=ssl.create_default_context(),
                timeout=timeout,
        )

        
        
        
        
        
        with self.assertRaises(OSError):
            connection.request("HEAD", "/test", headers={"Host": "localhost"})
            response = connection.getresponse()

        server.join()


class TestEnumerations(unittest.TestCase):

    def test_tlsversion(self):
        class CheckedTLSVersion(enum.IntEnum):
            MINIMUM_SUPPORTED = _ssl.PROTO_MINIMUM_SUPPORTED
            SSLv3 = _ssl.PROTO_SSLv3
            TLSv1 = _ssl.PROTO_TLSv1
            TLSv1_1 = _ssl.PROTO_TLSv1_1
            TLSv1_2 = _ssl.PROTO_TLSv1_2
            TLSv1_3 = _ssl.PROTO_TLSv1_3
            MAXIMUM_SUPPORTED = _ssl.PROTO_MAXIMUM_SUPPORTED
        enum._test_simple_enum(CheckedTLSVersion, TLSVersion)

    def test_tlscontenttype(self):
        class Checked_TLSContentType(enum.IntEnum):
            
            CHANGE_CIPHER_SPEC = 20
            ALERT = 21
            HANDSHAKE = 22
            APPLICATION_DATA = 23
            
            HEADER = 0x100
            INNER_CONTENT_TYPE = 0x101
        enum._test_simple_enum(Checked_TLSContentType, _TLSContentType)

    def test_tlsalerttype(self):
        class Checked_TLSAlertType(enum.IntEnum):
            
            CLOSE_NOTIFY = 0
            UNEXPECTED_MESSAGE = 10
            BAD_RECORD_MAC = 20
            DECRYPTION_FAILED = 21
            RECORD_OVERFLOW = 22
            DECOMPRESSION_FAILURE = 30
            HANDSHAKE_FAILURE = 40
            NO_CERTIFICATE = 41
            BAD_CERTIFICATE = 42
            UNSUPPORTED_CERTIFICATE = 43
            CERTIFICATE_REVOKED = 44
            CERTIFICATE_EXPIRED = 45
            CERTIFICATE_UNKNOWN = 46
            ILLEGAL_PARAMETER = 47
            UNKNOWN_CA = 48
            ACCESS_DENIED = 49
            DECODE_ERROR = 50
            DECRYPT_ERROR = 51
            EXPORT_RESTRICTION = 60
            PROTOCOL_VERSION = 70
            INSUFFICIENT_SECURITY = 71
            INTERNAL_ERROR = 80
            INAPPROPRIATE_FALLBACK = 86
            USER_CANCELED = 90
            NO_RENEGOTIATION = 100
            MISSING_EXTENSION = 109
            UNSUPPORTED_EXTENSION = 110
            CERTIFICATE_UNOBTAINABLE = 111
            UNRECOGNIZED_NAME = 112
            BAD_CERTIFICATE_STATUS_RESPONSE = 113
            BAD_CERTIFICATE_HASH_VALUE = 114
            UNKNOWN_PSK_IDENTITY = 115
            CERTIFICATE_REQUIRED = 116
            NO_APPLICATION_PROTOCOL = 120
        enum._test_simple_enum(Checked_TLSAlertType, _TLSAlertType)

    def test_tlsmessagetype(self):
        class Checked_TLSMessageType(enum.IntEnum):
            
            HELLO_REQUEST = 0
            CLIENT_HELLO = 1
            SERVER_HELLO = 2
            HELLO_VERIFY_REQUEST = 3
            NEWSESSION_TICKET = 4
            END_OF_EARLY_DATA = 5
            HELLO_RETRY_REQUEST = 6
            ENCRYPTED_EXTENSIONS = 8
            CERTIFICATE = 11
            SERVER_KEY_EXCHANGE = 12
            CERTIFICATE_REQUEST = 13
            SERVER_DONE = 14
            CERTIFICATE_VERIFY = 15
            CLIENT_KEY_EXCHANGE = 16
            FINISHED = 20
            CERTIFICATE_URL = 21
            CERTIFICATE_STATUS = 22
            SUPPLEMENTAL_DATA = 23
            KEY_UPDATE = 24
            NEXT_PROTO = 67
            MESSAGE_HASH = 254
            CHANGE_CIPHER_SPEC = 0x0101
        enum._test_simple_enum(Checked_TLSMessageType, _TLSMessageType)

    def test_sslmethod(self):
        Checked_SSLMethod = enum._old_convert_(
                enum.IntEnum, '_SSLMethod', 'ssl',
                lambda name: name.startswith('PROTOCOL_') and name != 'PROTOCOL_SSLv23',
                source=ssl._ssl,
                )
        
        Checked_SSLMethod.PROTOCOL_SSLv23 = Checked_SSLMethod.PROTOCOL_TLS
        enum._test_simple_enum(Checked_SSLMethod, ssl._SSLMethod)

    def test_options(self):
        CheckedOptions = enum._old_convert_(
                enum.IntFlag, 'Options', 'ssl',
                lambda name: name.startswith('OP_'),
                source=ssl._ssl,
                )
        enum._test_simple_enum(CheckedOptions, ssl.Options)

    def test_alertdescription(self):
        CheckedAlertDescription = enum._old_convert_(
                enum.IntEnum, 'AlertDescription', 'ssl',
                lambda name: name.startswith('ALERT_DESCRIPTION_'),
                source=ssl._ssl,
                )
        enum._test_simple_enum(CheckedAlertDescription, ssl.AlertDescription)

    def test_sslerrornumber(self):
        Checked_SSLErrorNumber = enum._old_convert_(
                enum.IntEnum, 'SSLErrorNumber', 'ssl',
                lambda name: name.startswith('SSL_ERROR_'),
                source=ssl._ssl,
                )
        enum._test_simple_enum(Checked_SSLErrorNumber, ssl.SSLErrorNumber)

    def test_verifyflags(self):
        CheckedVerifyFlags = enum._old_convert_(
                enum.IntFlag, 'VerifyFlags', 'ssl',
                lambda name: name.startswith('VERIFY_'),
                source=ssl._ssl,
                )
        enum._test_simple_enum(CheckedVerifyFlags, ssl.VerifyFlags)

    def test_verifymode(self):
        CheckedVerifyMode = enum._old_convert_(
                enum.IntEnum, 'VerifyMode', 'ssl',
                lambda name: name.startswith('CERT_'),
                source=ssl._ssl,
                )
        enum._test_simple_enum(CheckedVerifyMode, ssl.VerifyMode)


def setUpModule():
    if support.verbose:
        plats = {
            'Mac': platform.mac_ver,
            'Windows': platform.win32_ver,
        }
        for name, func in plats.items():
            plat = func()
            if plat and plat[0]:
                plat = '%s %r' % (name, plat)
                break
        else:
            plat = repr(platform.platform())
        print("test_ssl: testing with %r %r" %
            (ssl.OPENSSL_VERSION, ssl.OPENSSL_VERSION_INFO))
        print("          under %s" % plat)
        print("          HAS_SNI = %r" % ssl.HAS_SNI)
        print("          OP_ALL = 0x%8x" % ssl.OP_ALL)
        try:
            print("          OP_NO_TLSv1_1 = 0x%8x" % ssl.OP_NO_TLSv1_1)
        except AttributeError:
            pass

    for filename in [
        CERTFILE, BYTES_CERTFILE,
        ONLYCERT, ONLYKEY, BYTES_ONLYCERT, BYTES_ONLYKEY,
        SIGNED_CERTFILE, SIGNED_CERTFILE2, SIGNING_CA,
        BADCERT, BADKEY, EMPTYCERT]:
        if not os.path.exists(filename):
            raise support.TestFailed("Can't read certificate file %r" % filename)

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


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