wrapper module for ssl providing some additional facilities implemented in python written by bill janssen this module provides some more pythonic support for ssl object types sslsocket subtype of socket socket which does ssl over the socket exceptions sslerror exception raised for io errors functions certtimetoseconds convert time string used for certificate notbefore and notafter functions to integer seconds past the epoch the time values returned from time time getservercertificate addr sslversion cacerts timeout retrieve the certificate from the server at the specified address and return it as a pemencoded string integer constants sslerrorzeroreturn sslerrorwantread sslerrorwantwrite sslerrorwantx509lookup sslerrorsyscall sslerrorssl sslerrorwantconnect sslerroreof sslerrorinvaliderrorcode the following group define certificate requirements that one side is allowingrequiring from the other side certnone no certificates from the other side are required or will be looked at if provided certoptional certificates are not required but if provided will be validated and if validation fails the connection will also fail certrequired certificates are required and will be validated and if validation fails the connection will also fail the following constants identify various ssl protocol variants protocolsslv2 protocolsslv3 protocolsslv23 protocoltls protocoltlsclient protocoltlsserver protocoltlsv1 protocoltlsv11 protocoltlsv12 the following constants identify various ssl alert message descriptions as per http www iana orgassignmentstlsparameterstlsparameters xmltlsparameters6 alertdescriptionclosenotify alertdescriptionunexpectedmessage alertdescriptionbadrecordmac alertdescriptionrecordoverflow alertdescriptiondecompressionfailure alertdescriptionhandshakefailure alertdescriptionbadcertificate alertdescriptionunsupportedcertificate alertdescriptioncertificaterevoked alertdescriptioncertificateexpired alertdescriptioncertificateunknown alertdescriptionillegalparameter alertdescriptionunknownca alertdescriptionaccessdenied alertdescriptiondecodeerror alertdescriptiondecrypterror alertdescriptionprotocolversion alertdescriptioninsufficientsecurity alertdescriptioninternalerror alertdescriptionusercancelled alertdescriptionnorenegotiation alertdescriptionunsupportedextension alertdescriptioncertificateunobtainable alertdescriptionunrecognizedname alertdescriptionbadcertificatestatusresponse alertdescriptionbadcertificatehashvalue alertdescriptionunknownpskidentity libressl does not provide randegd 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 matching according to rfc 6125 section 6 4 3 hostnames are compared lowercase for idna both dn and hostname must be encoded as idn alabel ace partial wildcards like www example org multiple wildcards sole wildcard or wildcards in labels other then the leftmost label are not supported and a certificateerror is raised a wildcard must match at least one character speed up common case wo wildcards only match wildcard in leftmost segment no right side no partial wildcard matching wildcard must match at least one char try to convert an ip address to packed binary form supports ipv4 addresses on all platforms and ipv6 on platforms with ipv6 support inetaton also accepts strings like 1 127 1 some also trailing data like 127 0 0 1 whatever not an ipv4 address only accept injective ipnames refuse for short ipv4 notation and additional trailing data afinet6 not available exact matching of ip addresses rfc 6125 explicitly doesn t define an algorithm for this section 1 7 2 out of scope openssl may add a trailing newline to a subjectaltname s ip address commonly with ipv6 addresses strip off trailing n return paths to default cafile and capath environment vars shadow paths asn 1 object identifier lookup create asn1object from openssl numeric id create asn1object from short name long name or oid sslcontext purpose flags with x509v3 extended key usage objects an sslcontext holds various sslrelated configuration options and data such as certificates and possibly a private key windowscertstores ca root sslsocketclass none sslsocket is assigned later sslobjectclass none sslobject is assigned later def newcls protocolnone args kwargs if protocol is none warnings warn ssl sslcontext without protocol argument is deprecated categorydeprecationwarning stacklevel2 protocol protocoltls self sslcontext newcls protocol return self def encodehostnameself hostname if hostname is none return none elif isinstancehostname str return hostname encode idna decode ascii else return hostname decode ascii def wrapsocketself sock serversidefalse dohandshakeonconnecttrue suppressraggedeofstrue serverhostnamenone sessionnone sslsocket class handles serverhostname encoding before it calls ctx wrapsocket return self sslsocketclass create socksock serversideserverside dohandshakeonconnectdohandshakeonconnect suppressraggedeofssuppressraggedeofs serverhostnameserverhostname contextself sessionsession def wrapbioself incoming outgoing serversidefalse serverhostnamenone sessionnone need to encode serverhostname here because wrapbio can only handle ascii str return self sslobjectclass create incoming outgoing serversideserverside serverhostnameself encodehostnameserverhostname sessionsession contextself def setnpnprotocolsself npnprotocols warnings warn ssl npn is deprecated use alpn instead deprecationwarning stacklevel2 protos bytearray for protocol in npnprotocols b bytesprotocol ascii if lenb 0 or lenb 255 raise sslerror npn protocols must be 1 to 255 in length protos appendlenb protos extendb self setnpnprotocolsprotos def setservernamecallbackself servernamecallback if servernamecallback is none self snicallback none else if not callableservernamecallback raise typeerrornot a callable object def shimcbsslobj servername sslctx servername self encodehostnameservername return servernamecallbacksslobj servername sslctx self snicallback shimcb def setalpnprotocolsself alpnprotocols protos bytearray for protocol in alpnprotocols b bytesprotocol ascii if lenb 0 or lenb 255 raise sslerror alpn protocols must be 1 to 255 in length protos appendlenb protos extendb self setalpnprotocolsprotos def loadwindowsstorecertsself storename purpose certs bytearray try for cert encoding trust in enumcertificatesstorename ca certs are never pkcs7 encoded if encoding x509asn if trust is true or purpose oid in trust certs extendcert except permissionerror warnings warnunable to enumerate windows certificate store if certs self loadverifylocationscadatacerts return certs def loaddefaultcertsself purposepurpose serverauth if not isinstancepurpose asn1object raise typeerrorpurpose if sys platform win32 for storename in self windowscertstores self loadwindowsstorecertsstorename purpose self setdefaultverifypaths if hasattrsslcontext minimumversion property def minimumversionself return tlsversionsuper minimumversion minimumversion setter def minimumversionself value if value tlsversion sslv3 self options options opnosslv3 supersslcontext sslcontext minimumversion setself value property def maximumversionself return tlsversionsuper maximumversion maximumversion setter def maximumversionself value supersslcontext sslcontext maximumversion setself value property def optionsself return optionssuper options options setter def optionsself value supersslcontext sslcontext options setself value if hasattrssl hostflagneverchecksubject property def hostnamecheckscommonnameself ncs self hostflags ssl hostflagneverchecksubject return ncs ssl hostflagneverchecksubject hostnamecheckscommonname setter def hostnamecheckscommonnameself value if value self hostflags ssl hostflagneverchecksubject else self hostflags ssl hostflagneverchecksubject else property def hostnamecheckscommonnameself return true property def msgcallbackself inner super msgcallback if inner is not none return inner userfunction else return none msgcallback setter def msgcallbackself callback if callback is none supersslcontext sslcontext msgcallback setself none return if not hasattrcallback call raise typeerrorfcallback is not callable def innerconn direction version contenttype msgtype data try version tlsversionversion except valueerror pass try contenttype tlscontenttypecontenttype except valueerror pass if contenttype tlscontenttype header msgenum tlscontenttype elif contenttype tlscontenttype alert msgenum tlsalerttype else msgenum tlsmessagetype try msgtype msgenummsgtype except valueerror pass return callbackconn direction version contenttype msgtype data inner userfunction callback supersslcontext sslcontext msgcallback setself inner property def protocolself return sslmethodsuper protocol property def verifyflagsself return verifyflagssuper verifyflags verifyflags setter def verifyflagsself value supersslcontext sslcontext verifyflags setself value property def verifymodeself value super verifymode try return verifymodevalue except valueerror return value verifymode setter def verifymodeself value supersslcontext sslcontext verifymode setself value def createdefaultcontextpurposepurpose serverauth cafilenone capathnone cadatanone if not isinstancepurpose asn1object raise typeerrorpurpose sslcontext sets opnosslv2 opnosslv3 opnocompression opcipherserverpreference opsingledhuse and opsingleecdhuse by default if purpose purpose serverauth verify certs and host name in client mode context sslcontextprotocoltlsclient context verifymode certrequired context checkhostname true elif purpose purpose clientauth context sslcontextprotocoltlsserver else raise valueerrorpurpose if cafile or capath or cadata context loadverifylocationscafile capath cadata elif context verifymode certnone no explicit cafile capath or cadata but the verify mode is certoptional or certrequired let s try to load default system root ca certificates for the given purpose this may fail silently context loaddefaultcertspurpose openssl 1 1 1 keylog file if hasattrcontext keylogfilename keylogfile os environ get sslkeylogfile if keylogfile and not sys flags ignoreenvironment context keylogfilename keylogfile return context def createunverifiedcontextprotocolnone certreqscertnone checkhostnamefalse purposepurpose serverauth certfilenone keyfilenone cafilenone capathnone cadatanone if not isinstancepurpose asn1object raise typeerrorpurpose sslcontext sets opnosslv2 opnosslv3 opnocompression opcipherserverpreference opsingledhuse and opsingleecdhuse by default if purpose purpose serverauth verify certs and host name in client mode if protocol is none protocol protocoltlsclient elif purpose purpose clientauth if protocol is none protocol protocoltlsserver else raise valueerrorpurpose context sslcontextprotocol context checkhostname checkhostname if certreqs is not none context verifymode certreqs if checkhostname context checkhostname true if keyfile and not certfile raise valueerrorcertfile must be specified if certfile or keyfile context loadcertchaincertfile keyfile load ca root certs if cafile or capath or cadata context loadverifylocationscafile capath cadata elif context verifymode certnone no explicit cafile capath or cadata but the verify mode is certoptional or certrequired let s try to load default system root ca certificates for the given purpose this may fail silently context loaddefaultcertspurpose openssl 1 1 1 keylog file if hasattrcontext keylogfilename keylogfile os environ get sslkeylogfile if keylogfile and not sys flags ignoreenvironment context keylogfilename keylogfile return context used by http client if no context is explicitly passed createdefaulthttpscontext createdefaultcontext backwards compatibility alias even though it s not a public name createstdlibcontext createunverifiedcontext class sslobject def initself args kwargs raise typeerror fself class name does not have a public fconstructor instances are returned by sslcontext wrapbio classmethod def createcls incoming outgoing serversidefalse serverhostnamenone sessionnone contextnone self cls newcls sslobj context wrapbio incoming outgoing serversideserverside serverhostnameserverhostname ownerself sessionsession self sslobj sslobj return self property def contextself the sslsession for client socket return self sslobj session session setter def sessionself session self sslobj session session property def sessionreusedself whether this is a serverside socket return self sslobj serverside property def serverhostnameself read up to len bytes from the ssl object and return them if buffer is provided read into this buffer and return the number of bytes read write data to the ssl object and return the number of bytes written the data argument must support the buffer interface returns a formatted version of the data in the certificate provided by the other end of the ssl channel return none if no certificate was provided if a certificate was provided but not validated returns verified certificate chain provided by the other end of the ssl channel as a list of derencoded bytes if certificate verification was disabled method acts the same as sslsocket getunverifiedchain returns raw certificate chain provided by the other end of the ssl channel as a list of derencoded bytes return the currently selected npn protocol as a string or none if a next protocol was not negotiated or if npn is not supported by one of the peers warnings warn ssl npn is deprecated use alpn instead deprecationwarning stacklevel2 def selectedalpnprotocolself return the currently selected cipher as a 3tuple name sslversion secretbits return self sslobj cipher def sharedciphersself return self sslobj sharedciphers def compressionself return the number of bytes that can be read immediately return self sslobj pending def dohandshakeself start the ssl shutdown handshake return self sslobj shutdown def getchannelbindingself cbtypetlsunique return a string identifying the protocol version used by the current ssl channel return self sslobj version def verifyclientposthandshakeself return self sslobj verifyclientposthandshake def sslcopydocfunc this class implements a subtype of socket socket that wraps the underlying os socket in an ssl context when necessary and provides read and write methods over that channel def initself args kwargs raise typeerror fself class name does not have a public fconstructor instances are returned by fsslcontext wrapsocket classmethod def createcls sock serversidefalse dohandshakeonconnecttrue suppressraggedeofstrue serverhostnamenone contextnone sessionnone if sock getsockoptsolsocket sotype sockstream raise notimplementederroronly stream sockets are supported if serverside if serverhostname raise valueerrorserverhostname can only be specified in client mode if session is not none raise valueerrorsession can only be specified in client mode if context checkhostname and not serverhostname raise valueerrorcheckhostname requires serverhostname kwargs dict familysock family typesock type protosock proto filenosock fileno self cls newcls kwargs supersslsocket self initkwargs socktimeout sock gettimeout sock detach self context context self session session self closed false self sslobj none self serverside serverside self serverhostname context encodehostnameserverhostname self dohandshakeonconnect dohandshakeonconnect self suppressraggedeofs suppressraggedeofs see if we are connected try self getpeername except oserror as e if e errno errno enotconn raise connected false blocking self getblocking self setblockingfalse try we are not connected so this is not supposed to block but testing revealed otherwise on macos and windows so we do the nonblocking dance regardless our raise when any data is found means consuming the data is harmless notconnprehandshakedata self recv1 except oserror as e einval occurs for recv1 on nonconnected on unix sockets if e errno not in errno enotconn errno einval raise notconnprehandshakedata b self setblockingblocking if notconnprehandshakedata this prevents pending data sent to the socket before it was closed from escaping to the caller who could otherwise presume it came through a successful tls connection reason closed before tls handshake with data in recv buffer notconnprehandshakedataerror sslerrore errno reason add the sslerror attributes that ssl c always adds notconnprehandshakedataerror reason reason notconnprehandshakedataerror library none try self close except oserror pass try raise notconnprehandshakedataerror finally explicitly break the reference cycle notconnprehandshakedataerror none else connected true self settimeoutsocktimeout must come after setblocking calls self connected connected if connected create the ssl object try self sslobj self context wrapsocket self serverside self serverhostname ownerself sessionself session if dohandshakeonconnect timeout self gettimeout if timeout 0 0 nonblocking raise valueerrordohandshakeonconnect should not be specified for nonblocking sockets self dohandshake except oserror valueerror self close raise return self property sslcopydoc def contextself return self context context setter def contextself ctx self context ctx self sslobj context ctx property sslcopydoc def sessionself if self sslobj is not none return self sslobj session session setter def sessionself session self session session if self sslobj is not none self sslobj session session property sslcopydoc def sessionreusedself if self sslobj is not none return self sslobj sessionreused def dupself raise notimplementederrorcan t dup s instances self class name def checkclosedself msgnone raise an exception here if you wish to check for spurious closes pass def checkconnectedself if not self connected getpeername will raise enotconn if the socket is really not connected note that we can be connected even without connected being set e g if connect first returned eagain self getpeername def readself len1024 buffernone write data to the underlying ssl channel returns number of bytes of data actually transmitted self checkclosed if self sslobj is none raise valueerrorwrite on closed or unwrapped ssl socket return self sslobj writedata sslcopydoc def getpeercertself binaryformfalse self checkclosed self checkconnected return self sslobj getpeercertbinaryform sslcopydoc def getverifiedchainself return self sslobj getverifiedchain sslcopydoc def getunverifiedchainself return self sslobj getunverifiedchain sslcopydoc def selectednpnprotocolself self checkclosed warnings warn ssl npn is deprecated use alpn instead deprecationwarning stacklevel2 return none sslcopydoc def selectedalpnprotocolself self checkclosed if self sslobj is none or not ssl hasalpn return none else return self sslobj selectedalpnprotocol sslcopydoc def cipherself self checkclosed if self sslobj is none return none else return self sslobj cipher sslcopydoc def sharedciphersself self checkclosed if self sslobj is none return none else return self sslobj sharedciphers sslcopydoc def compressionself self checkclosed if self sslobj is none return none else return self sslobj compression def sendself data flags0 self checkclosed if self sslobj is not none if flags 0 raise valueerror nonzero flags not allowed in calls to send on s self class return self sslobj writedata else return super senddata flags def sendtoself data flagsoraddr addrnone self checkclosed if self sslobj is not none raise valueerrorsendto not allowed on instances of s self class elif addr is none return super sendtodata flagsoraddr else return super sendtodata flagsoraddr addr def sendmsgself args kwargs ensure programs don t send data unencrypted if they try to use this method raise notimplementederrorsendmsg not allowed on instances of s self class def sendallself data flags0 self checkclosed if self sslobj is not none if flags 0 raise valueerror nonzero flags not allowed in calls to sendall on s self class count 0 with memoryviewdata as view view castb as byteview amount lenbyteview while count amount v self sendbyteviewcount count v else return super sendalldata flags def sendfileself file offset0 countnone if self sslobj is not none return self sendfileusesendfile offset count else os sendfile works with plain sockets only return super sendfilefile offset count def recvself buflen1024 flags0 self checkclosed if self sslobj is not none if flags 0 raise valueerror nonzero flags not allowed in calls to recv on s self class return self readbuflen else return super recvbuflen flags def recvintoself buffer nbytesnone flags0 self checkclosed if nbytes is none if buffer is not none with memoryviewbuffer as view nbytes view nbytes if not nbytes nbytes 1024 else nbytes 1024 if self sslobj is not none if flags 0 raise valueerror nonzero flags not allowed in calls to recvinto on s self class return self readnbytes buffer else return super recvintobuffer nbytes flags def recvfromself buflen1024 flags0 self checkclosed if self sslobj is not none raise valueerrorrecvfrom not allowed on instances of s self class else return super recvfrombuflen flags def recvfromintoself buffer nbytesnone flags0 self checkclosed if self sslobj is not none raise valueerrorrecvfrominto not allowed on instances of s self class else return super recvfromintobuffer nbytes flags def recvmsgself args kwargs raise notimplementederrorrecvmsg not allowed on instances of s self class def recvmsgintoself args kwargs raise notimplementederrorrecvmsginto not allowed on instances of s self class sslcopydoc def pendingself self checkclosed if self sslobj is not none return self sslobj pending else return 0 def shutdownself how self checkclosed self sslobj none super shutdownhow sslcopydoc def unwrapself if self sslobj s self sslobj shutdown self sslobj none return s else raise valueerrorno ssl wrapper around strself sslcopydoc def verifyclientposthandshakeself if self sslobj return self sslobj verifyclientposthandshake else raise valueerrorno ssl wrapper around strself def realcloseself self sslobj none super realclose sslcopydoc def dohandshakeself blockfalse self checkconnected timeout self gettimeout try if timeout 0 0 and block self settimeoutnone self sslobj dohandshake finally self settimeouttimeout def realconnectself addr connectex if self serverside raise valueerrorcan t connect in serverside mode here we assume that the socket is clientside and not connected at the time of the call we connect it then wrap it if self connected or self sslobj is not none raise valueerrorattempt to connect alreadyconnected sslsocket self sslobj self context wrapsocket self false self serverhostname ownerself sessionself session try if connectex rc super connectexaddr else rc none super connectaddr if not rc self connected true if self dohandshakeonconnect self dohandshake return rc except oserror valueerror self sslobj none raise def connectself addr connects to remote addr and then wraps the connection in an ssl channel return self realconnectaddr true def acceptself python does not support forward declaration of types some utility functions return the time in seconds since the epoch given the timestring representing the notbefore or notafter date from a certificate in b d h m s y z strptime format c locale notbefore or notafter dates must use utc rfc 5280 month is one of jan feb mar apr may jun jul aug sep oct nov dec utc should be specified as gmt see asn1timeprint found valid month return an integer the previous mktimebased implementation returned a float fractional seconds are always zero here takes a certificate in binary der format and returns the pem version of it as a string f strbase64 standardb64encodedercertbytes ascii strict ss pemheader ss fi i64 for i in range0 lenf 64 ss appendpemfooter n return n joinss def pemcerttodercertpemcertstring retrieve the certificate from the server at the specified address and return it as a pemencoded string if cacerts is specified validate the server cert against it if sslversion is specified use it in the connection attempt if timeout is specified use it in the connection attempt wrapper module for _ssl providing some additional facilities implemented in python written by bill janssen this module provides some more pythonic support for ssl object types sslsocket subtype of socket socket which does ssl over the socket exceptions sslerror exception raised for i o errors functions cert_time_to_seconds convert time string used for certificate notbefore and notafter functions to integer seconds past the epoch the time values returned from time time get_server_certificate addr ssl_version ca_certs timeout retrieve the certificate from the server at the specified address and return it as a pem encoded string integer constants ssl_error_zero_return ssl_error_want_read ssl_error_want_write ssl_error_want_x509_lookup ssl_error_syscall ssl_error_ssl ssl_error_want_connect ssl_error_eof ssl_error_invalid_error_code the following group define certificate requirements that one side is allowing requiring from the other side cert_none no certificates from the other side are required or will be looked at if provided cert_optional certificates are not required but if provided will be validated and if validation fails the connection will also fail cert_required certificates are required and will be validated and if validation fails the connection will also fail the following constants identify various ssl protocol variants protocol_sslv2 protocol_sslv3 protocol_sslv23 protocol_tls protocol_tls_client protocol_tls_server protocol_tlsv1 protocol_tlsv1_1 protocol_tlsv1_2 the following constants identify various ssl alert message descriptions as per http www iana org assignments tls parameters tls parameters xml tls parameters 6 alert_description_close_notify alert_description_unexpected_message alert_description_bad_record_mac alert_description_record_overflow alert_description_decompression_failure alert_description_handshake_failure alert_description_bad_certificate alert_description_unsupported_certificate alert_description_certificate_revoked alert_description_certificate_expired alert_description_certificate_unknown alert_description_illegal_parameter alert_description_unknown_ca alert_description_access_denied alert_description_decode_error alert_description_decrypt_error alert_description_protocol_version alert_description_insufficient_security alert_description_internal_error alert_description_user_cancelled alert_description_no_renegotiation alert_description_unsupported_extension alert_description_certificate_unobtainable alert_description_unrecognized_name alert_description_bad_certificate_status_response alert_description_bad_certificate_hash_value alert_description_unknown_psk_identity if we can t import it let the error propagate libressl does not provide rand_egd 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 for der to pem translation keep that public name in module namespace matching according to rfc 6125 section 6 4 3 hostnames are compared lower case for idna both dn and hostname must be encoded as idn a label ace partial wildcards like www example org multiple wildcards sole wildcard or wildcards in labels other then the left most label are not supported and a certificateerror is raised a wildcard must match at least one character speed up common case w o wildcards only match wildcard in leftmost segment no right side no partial wildcard matching wildcard must match at least one char try to convert an ip address to packed binary form supports ipv4 addresses on all platforms and ipv6 on platforms with ipv6 support inet_aton also accepts strings like 1 127 1 some also trailing data like 127 0 0 1 whatever not an ipv4 address only accept injective ipnames refuse for short ipv4 notation and additional trailing data af_inet6 not available exact matching of ip addresses rfc 6125 explicitly doesn t define an algorithm for this section 1 7 2 out of scope openssl may add a trailing newline to a subjectaltname s ip address commonly with ipv6 addresses strip off trailing n return paths to default cafile and capath environment vars shadow paths asn 1 object identifier lookup create _asn1object from openssl numeric id create _asn1object from short name long name or oid sslcontext purpose flags with x509v3 extended key usage objects an sslcontext holds various ssl related configuration options and data such as certificates and possibly a private key sslsocket is assigned later sslobject is assigned later sslsocket class handles server_hostname encoding before it calls ctx _wrap_socket need to encode server_hostname here because _wrap_bio can only handle ascii str ca certs are never pkcs 7 encoded tls message callback the message callback provides a debugging hook to analyze tls connections the callback is called for any tls protocol message header handshake alert and more but not for application data due to technical limitations the callback can t be used to filter traffic or to abort a connection any exception raised in the callback is delayed until the handshake read or write operation has been performed def msg_cb conn direction version content_type msg_type data pass conn class sslsocket or class sslobject instance direction read or write version class tlsversion enum member or int for unknown version for a frame header it s the header version content_type class _tlscontenttype enum member or int for unsupported content type msg_type either a class _tlscontenttype enum number for a header message a class _tlsalerttype enum member for an alert message a class _tlsmessagetype enum member for other messages or int for unsupported message types data raw decrypted message content as bytes create a sslcontext object with default settings note the protocol and settings may change anytime without prior deprecation the values represent a fair balance between maximum compatibility and security sslcontext sets op_no_sslv2 op_no_sslv3 op_no_compression op_cipher_server_preference op_single_dh_use and op_single_ecdh_use by default verify certs and host name in client mode no explicit cafile capath or cadata but the verify mode is cert_optional or cert_required let s try to load default system root ca certificates for the given purpose this may fail silently openssl 1 1 1 keylog file create a sslcontext object for python stdlib modules all python stdlib modules shall use this function to create sslcontext objects in order to keep common settings in one place the configuration is less restrict than create_default_context s to increase backward compatibility sslcontext sets op_no_sslv2 op_no_sslv3 op_no_compression op_cipher_server_preference op_single_dh_use and op_single_ecdh_use by default verify certs and host name in client mode load ca root certs no explicit cafile capath or cadata but the verify mode is cert_optional or cert_required let s try to load default system root ca certificates for the given purpose this may fail silently openssl 1 1 1 keylog file used by http client if no context is explicitly passed backwards compatibility alias even though it s not a public name this class implements an interface on top of a low level ssl object as implemented by openssl this object captures the state of an ssl connection but does not provide any network io itself io needs to be performed through separate bio objects which are openssl s io abstraction layer this class does not have a public constructor instances are returned by sslcontext wrap_bio this class is typically used by framework s that want to implement asynchronous io for ssl through memory buffers when compared to sslsocket this object lacks the following features any form of network io including methods such as recv and send the do_handshake_on_connect and suppress_ragged_eofs machinery the sslcontext that is currently in use the sslsession for client socket was the client session reused during handshake whether this is a server side socket the currently set server hostname for sni or none if no server hostname is set read up to len bytes from the ssl object and return them if buffer is provided read into this buffer and return the number of bytes read write data to the ssl object and return the number of bytes written the data argument must support the buffer interface returns a formatted version of the data in the certificate provided by the other end of the ssl channel return none if no certificate was provided if a certificate was provided but not validated returns verified certificate chain provided by the other end of the ssl channel as a list of der encoded bytes if certificate verification was disabled method acts the same as sslsocket get_unverified_chain returns raw certificate chain provided by the other end of the ssl channel as a list of der encoded bytes return the currently selected npn protocol as a string or none if a next protocol was not negotiated or if npn is not supported by one of the peers return the currently selected alpn protocol as a string or none if a next protocol was not negotiated or if alpn is not supported by one of the peers return the currently selected cipher as a 3 tuple name ssl_version secret_bits return a list of ciphers shared by the client during the handshake or none if this is not a valid server connection return the current compression algorithm in use or none if compression was not negotiated or not supported by one of the peers return the number of bytes that can be read immediately start the ssl tls handshake start the ssl shutdown handshake get channel binding data for current connection raise valueerror if the requested cb_type is not supported return bytes of the data or none if the data is not available e g before the handshake return a string identifying the protocol version used by the current ssl channel copy docstring from sslobject to sslsocket this class implements a subtype of socket socket that wraps the underlying os socket in an ssl context when necessary and provides read and write methods over that channel see if we are connected we are not connected so this is not supposed to block but testing revealed otherwise on macos and windows so we do the non blocking dance regardless our raise when any data is found means consuming the data is harmless einval occurs for recv 1 on non connected on unix sockets this prevents pending data sent to the socket before it was closed from escaping to the caller who could otherwise presume it came through a successful tls connection add the sslerror attributes that _ssl c always adds explicitly break the reference cycle must come after setblocking calls create the ssl object non blocking raise an exception here if you wish to check for spurious closes getpeername will raise enotconn if the socket is really not connected note that we can be connected even without _connected being set e g if connect first returned eagain read up to len bytes and return them return zero length string on eof write data to the underlying ssl channel returns number of bytes of data actually transmitted ensure programs don t send data unencrypted if they try to use this method send a file possibly by using os sendfile if this is a clear text socket return the total number of bytes sent os sendfile works with plain sockets only here we assume that the socket is client side and not connected at the time of the call we connect it then wrap it connects to remote addr and then wraps the connection in an ssl channel connects to remote addr and then wraps the connection in an ssl channel accepts a new connection from a remote client and returns a tuple containing that new connection wrapped with a server side ssl channel and the address of the remote client python does not support forward declaration of types some utility functions return the time in seconds since the epoch given the timestring representing the notbefore or notafter date from a certificate in b d h m s y z strptime format c locale notbefore or notafter dates must use utc rfc 5280 month is one of jan feb mar apr may jun jul aug sep oct nov dec utc should be specified as gmt see asn1_time_print note no month fixed gmt found valid month return an integer the previous mktime based implementation returned a float fractional seconds are always zero here takes a certificate in binary der format and returns the pem version of it as a string takes a certificate in ascii pem format and returns the der encoded version of it as a byte sequence retrieve the certificate from the server at the specified address and return it as a pem encoded string if ca_certs is specified validate the server cert against it if ssl_version is specified use it in the connection attempt if timeout is specified use it in the connection attempt
import sys
import os
from collections import namedtuple
from enum import Enum as _Enum, IntEnum as _IntEnum, IntFlag as _IntFlag
from enum import _simple_enum

import _ssl             

from _ssl import OPENSSL_VERSION_NUMBER, OPENSSL_VERSION_INFO, OPENSSL_VERSION
from _ssl import _SSLContext, MemoryBIO, SSLSession
from _ssl import (
    SSLError, SSLZeroReturnError, SSLWantReadError, SSLWantWriteError,
    SSLSyscallError, SSLEOFError, SSLCertVerificationError
    )
from _ssl import txt2obj as _txt2obj, nid2obj as _nid2obj
from _ssl import RAND_status, RAND_add, RAND_bytes
try:
    from _ssl import RAND_egd
except ImportError:
    
    pass


from _ssl import (
    HAS_SNI, HAS_ECDH, HAS_NPN, HAS_ALPN, HAS_SSLv2, HAS_SSLv3, HAS_TLSv1,
    HAS_TLSv1_1, HAS_TLSv1_2, HAS_TLSv1_3, HAS_PSK
)
from _ssl import _DEFAULT_CIPHERS, _OPENSSL_API_VERSION

_IntEnum._convert_(
    '_SSLMethod', __name__,
    lambda name: name.startswith('PROTOCOL_') and name != 'PROTOCOL_SSLv23',
    source=_ssl)

_IntFlag._convert_(
    'Options', __name__,
    lambda name: name.startswith('OP_'),
    source=_ssl)

_IntEnum._convert_(
    'AlertDescription', __name__,
    lambda name: name.startswith('ALERT_DESCRIPTION_'),
    source=_ssl)

_IntEnum._convert_(
    'SSLErrorNumber', __name__,
    lambda name: name.startswith('SSL_ERROR_'),
    source=_ssl)

_IntFlag._convert_(
    'VerifyFlags', __name__,
    lambda name: name.startswith('VERIFY_'),
    source=_ssl)

_IntEnum._convert_(
    'VerifyMode', __name__,
    lambda name: name.startswith('CERT_'),
    source=_ssl)

PROTOCOL_SSLv23 = _SSLMethod.PROTOCOL_SSLv23 = _SSLMethod.PROTOCOL_TLS
_PROTOCOL_NAMES = {value: name for name, value in _SSLMethod.__members__.items()}

_SSLv2_IF_EXISTS = getattr(_SSLMethod, 'PROTOCOL_SSLv2', None)


@_simple_enum(_IntEnum)
class TLSVersion:
    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


@_simple_enum(_IntEnum)
class _TLSContentType:
    
    CHANGE_CIPHER_SPEC = 20
    ALERT = 21
    HANDSHAKE = 22
    APPLICATION_DATA = 23
    
    HEADER = 0x100
    INNER_CONTENT_TYPE = 0x101


@_simple_enum(_IntEnum)
class _TLSAlertType:
    
    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


@_simple_enum(_IntEnum)
class _TLSMessageType:
    
    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


if sys.platform == "win32":
    from _ssl import enum_certificates, enum_crls

from socket import socket, SOCK_STREAM, create_connection
from socket import SOL_SOCKET, SO_TYPE, _GLOBAL_DEFAULT_TIMEOUT
import socket as _socket
import base64        
import errno
import warnings


socket_error = OSError  

CHANNEL_BINDING_TYPES = ['tls-unique']

HAS_NEVER_CHECK_COMMON_NAME = hasattr(_ssl, 'HOSTFLAG_NEVER_CHECK_SUBJECT')


_RESTRICTED_SERVER_CIPHERS = _DEFAULT_CIPHERS

CertificateError = SSLCertVerificationError


def _dnsname_match(dn, hostname):
    
    if not dn:
        return False

    wildcards = dn.count('*')
    
    if not wildcards:
        return dn.lower() == hostname.lower()

    if wildcards > 1:
        raise CertificateError(
            "too many wildcards in certificate DNS name: {!r}.".format(dn))

    dn_leftmost, sep, dn_remainder = dn.partition('.')

    if '*' in dn_remainder:
        
        raise CertificateError(
            "wildcard can only be present in the leftmost label: "
            "{!r}.".format(dn))

    if not sep:
        
        raise CertificateError(
            "sole wildcard without additional labels are not support: "
            "{!r}.".format(dn))

    if dn_leftmost != '*':
        
        raise CertificateError(
            "partial wildcards in leftmost label are not supported: "
            "{!r}.".format(dn))

    hostname_leftmost, sep, hostname_remainder = hostname.partition('.')
    if not hostname_leftmost or not sep:
        
        return False
    return dn_remainder.lower() == hostname_remainder.lower()


def _inet_paton(ipname):
    
    
    
    try:
        addr = _socket.inet_aton(ipname)
    except OSError:
        
        pass
    else:
        if _socket.inet_ntoa(addr) == ipname:
            
            return addr
        else:
            
            raise ValueError(
                "{!r} is not a quad-dotted IPv4 address.".format(ipname)
            )

    try:
        return _socket.inet_pton(_socket.AF_INET6, ipname)
    except OSError:
        raise ValueError("{!r} is neither an IPv4 nor an IP6 "
                         "address.".format(ipname))
    except AttributeError:
        
        pass

    raise ValueError("{!r} is not an IPv4 address.".format(ipname))


def _ipaddress_match(cert_ipaddress, host_ip):
    
    
    
    ip = _inet_paton(cert_ipaddress.rstrip())
    return ip == host_ip


DefaultVerifyPaths = namedtuple("DefaultVerifyPaths",
    "cafile capath openssl_cafile_env openssl_cafile openssl_capath_env "
    "openssl_capath")

def get_default_verify_paths():
    
    parts = _ssl.get_default_verify_paths()

    
    cafile = os.environ.get(parts[0], parts[1])
    capath = os.environ.get(parts[2], parts[3])

    return DefaultVerifyPaths(cafile if os.path.isfile(cafile) else None,
                              capath if os.path.isdir(capath) else None,
                              *parts)


class _ASN1Object(namedtuple("_ASN1Object", "nid shortname longname oid")):
    
    __slots__ = ()

    def __new__(cls, oid):
        return super().__new__(cls, *_txt2obj(oid, name=False))

    @classmethod
    def fromnid(cls, nid):
        
        return super().__new__(cls, *_nid2obj(nid))

    @classmethod
    def fromname(cls, name):
        
        return super().__new__(cls, *_txt2obj(name, name=True))


class Purpose(_ASN1Object, _Enum):
    
    SERVER_AUTH = '1.3.6.1.5.5.7.3.1'
    CLIENT_AUTH = '1.3.6.1.5.5.7.3.2'


class SSLContext(_SSLContext):
    
    _windows_cert_stores = ("CA", "ROOT")

    sslsocket_class = None  
    sslobject_class = None  

    def __new__(cls, protocol=None, *args, **kwargs):
        if protocol is None:
            warnings.warn(
                "ssl.SSLContext() without protocol argument is deprecated.",
                category=DeprecationWarning,
                stacklevel=2
            )
            protocol = PROTOCOL_TLS
        self = _SSLContext.__new__(cls, protocol)
        return self

    def _encode_hostname(self, hostname):
        if hostname is None:
            return None
        elif isinstance(hostname, str):
            return hostname.encode('idna').decode('ascii')
        else:
            return hostname.decode('ascii')

    def wrap_socket(self, sock, server_side=False,
                    do_handshake_on_connect=True,
                    suppress_ragged_eofs=True,
                    server_hostname=None, session=None):
        
        
        return self.sslsocket_class._create(
            sock=sock,
            server_side=server_side,
            do_handshake_on_connect=do_handshake_on_connect,
            suppress_ragged_eofs=suppress_ragged_eofs,
            server_hostname=server_hostname,
            context=self,
            session=session
        )

    def wrap_bio(self, incoming, outgoing, server_side=False,
                 server_hostname=None, session=None):
        
        
        return self.sslobject_class._create(
            incoming, outgoing, server_side=server_side,
            server_hostname=self._encode_hostname(server_hostname),
            session=session, context=self,
        )

    def set_npn_protocols(self, npn_protocols):
        warnings.warn(
            "ssl NPN is deprecated, use ALPN instead",
            DeprecationWarning,
            stacklevel=2
        )
        protos = bytearray()
        for protocol in npn_protocols:
            b = bytes(protocol, 'ascii')
            if len(b) == 0 or len(b) > 255:
                raise SSLError('NPN protocols must be 1 to 255 in length')
            protos.append(len(b))
            protos.extend(b)

        self._set_npn_protocols(protos)

    def set_servername_callback(self, server_name_callback):
        if server_name_callback is None:
            self.sni_callback = None
        else:
            if not callable(server_name_callback):
                raise TypeError("not a callable object")

            def shim_cb(sslobj, servername, sslctx):
                servername = self._encode_hostname(servername)
                return server_name_callback(sslobj, servername, sslctx)

            self.sni_callback = shim_cb

    def set_alpn_protocols(self, alpn_protocols):
        protos = bytearray()
        for protocol in alpn_protocols:
            b = bytes(protocol, 'ascii')
            if len(b) == 0 or len(b) > 255:
                raise SSLError('ALPN protocols must be 1 to 255 in length')
            protos.append(len(b))
            protos.extend(b)

        self._set_alpn_protocols(protos)

    def _load_windows_store_certs(self, storename, purpose):
        certs = bytearray()
        try:
            for cert, encoding, trust in enum_certificates(storename):
                
                if encoding == "x509_asn":
                    if trust is True or purpose.oid in trust:
                        certs.extend(cert)
        except PermissionError:
            warnings.warn("unable to enumerate Windows certificate store")
        if certs:
            self.load_verify_locations(cadata=certs)
        return certs

    def load_default_certs(self, purpose=Purpose.SERVER_AUTH):
        if not isinstance(purpose, _ASN1Object):
            raise TypeError(purpose)
        if sys.platform == "win32":
            for storename in self._windows_cert_stores:
                self._load_windows_store_certs(storename, purpose)
        self.set_default_verify_paths()

    if hasattr(_SSLContext, 'minimum_version'):
        @property
        def minimum_version(self):
            return TLSVersion(super().minimum_version)

        @minimum_version.setter
        def minimum_version(self, value):
            if value == TLSVersion.SSLv3:
                self.options &= ~Options.OP_NO_SSLv3
            super(SSLContext, SSLContext).minimum_version.__set__(self, value)

        @property
        def maximum_version(self):
            return TLSVersion(super().maximum_version)

        @maximum_version.setter
        def maximum_version(self, value):
            super(SSLContext, SSLContext).maximum_version.__set__(self, value)

    @property
    def options(self):
        return Options(super().options)

    @options.setter
    def options(self, value):
        super(SSLContext, SSLContext).options.__set__(self, value)

    if hasattr(_ssl, 'HOSTFLAG_NEVER_CHECK_SUBJECT'):
        @property
        def hostname_checks_common_name(self):
            ncs = self._host_flags & _ssl.HOSTFLAG_NEVER_CHECK_SUBJECT
            return ncs != _ssl.HOSTFLAG_NEVER_CHECK_SUBJECT

        @hostname_checks_common_name.setter
        def hostname_checks_common_name(self, value):
            if value:
                self._host_flags &= ~_ssl.HOSTFLAG_NEVER_CHECK_SUBJECT
            else:
                self._host_flags |= _ssl.HOSTFLAG_NEVER_CHECK_SUBJECT
    else:
        @property
        def hostname_checks_common_name(self):
            return True

    @property
    def _msg_callback(self):
        
        inner = super()._msg_callback
        if inner is not None:
            return inner.user_function
        else:
            return None

    @_msg_callback.setter
    def _msg_callback(self, callback):
        if callback is None:
            super(SSLContext, SSLContext)._msg_callback.__set__(self, None)
            return

        if not hasattr(callback, '__call__'):
            raise TypeError(f"{callback} is not callable.")

        def inner(conn, direction, version, content_type, msg_type, data):
            try:
                version = TLSVersion(version)
            except ValueError:
                pass

            try:
                content_type = _TLSContentType(content_type)
            except ValueError:
                pass

            if content_type == _TLSContentType.HEADER:
                msg_enum = _TLSContentType
            elif content_type == _TLSContentType.ALERT:
                msg_enum = _TLSAlertType
            else:
                msg_enum = _TLSMessageType
            try:
                msg_type = msg_enum(msg_type)
            except ValueError:
                pass

            return callback(conn, direction, version,
                            content_type, msg_type, data)

        inner.user_function = callback

        super(SSLContext, SSLContext)._msg_callback.__set__(self, inner)

    @property
    def protocol(self):
        return _SSLMethod(super().protocol)

    @property
    def verify_flags(self):
        return VerifyFlags(super().verify_flags)

    @verify_flags.setter
    def verify_flags(self, value):
        super(SSLContext, SSLContext).verify_flags.__set__(self, value)

    @property
    def verify_mode(self):
        value = super().verify_mode
        try:
            return VerifyMode(value)
        except ValueError:
            return value

    @verify_mode.setter
    def verify_mode(self, value):
        super(SSLContext, SSLContext).verify_mode.__set__(self, value)


def create_default_context(purpose=Purpose.SERVER_AUTH, *, cafile=None,
                           capath=None, cadata=None):
    
    if not isinstance(purpose, _ASN1Object):
        raise TypeError(purpose)

    
    
    
    if purpose == Purpose.SERVER_AUTH:
        
        context = SSLContext(PROTOCOL_TLS_CLIENT)
        context.verify_mode = CERT_REQUIRED
        context.check_hostname = True
    elif purpose == Purpose.CLIENT_AUTH:
        context = SSLContext(PROTOCOL_TLS_SERVER)
    else:
        raise ValueError(purpose)

    if cafile or capath or cadata:
        context.load_verify_locations(cafile, capath, cadata)
    elif context.verify_mode != CERT_NONE:
        
        
        
        context.load_default_certs(purpose)
    
    if hasattr(context, 'keylog_filename'):
        keylogfile = os.environ.get('SSLKEYLOGFILE')
        if keylogfile and not sys.flags.ignore_environment:
            context.keylog_filename = keylogfile
    return context

def _create_unverified_context(protocol=None, *, cert_reqs=CERT_NONE,
                           check_hostname=False, purpose=Purpose.SERVER_AUTH,
                           certfile=None, keyfile=None,
                           cafile=None, capath=None, cadata=None):
    
    if not isinstance(purpose, _ASN1Object):
        raise TypeError(purpose)

    
    
    
    if purpose == Purpose.SERVER_AUTH:
        
        if protocol is None:
            protocol = PROTOCOL_TLS_CLIENT
    elif purpose == Purpose.CLIENT_AUTH:
        if protocol is None:
            protocol = PROTOCOL_TLS_SERVER
    else:
        raise ValueError(purpose)

    context = SSLContext(protocol)
    context.check_hostname = check_hostname
    if cert_reqs is not None:
        context.verify_mode = cert_reqs
    if check_hostname:
        context.check_hostname = True

    if keyfile and not certfile:
        raise ValueError("certfile must be specified")
    if certfile or keyfile:
        context.load_cert_chain(certfile, keyfile)

    
    if cafile or capath or cadata:
        context.load_verify_locations(cafile, capath, cadata)
    elif context.verify_mode != CERT_NONE:
        
        
        
        context.load_default_certs(purpose)
    
    if hasattr(context, 'keylog_filename'):
        keylogfile = os.environ.get('SSLKEYLOGFILE')
        if keylogfile and not sys.flags.ignore_environment:
            context.keylog_filename = keylogfile
    return context


_create_default_https_context = create_default_context



_create_stdlib_context = _create_unverified_context


class SSLObject:
    
    def __init__(self, *args, **kwargs):
        raise TypeError(
            f"{self.__class__.__name__} does not have a public "
            f"constructor. Instances are returned by SSLContext.wrap_bio()."
        )

    @classmethod
    def _create(cls, incoming, outgoing, server_side=False,
                 server_hostname=None, session=None, context=None):
        self = cls.__new__(cls)
        sslobj = context._wrap_bio(
            incoming, outgoing, server_side=server_side,
            server_hostname=server_hostname,
            owner=self, session=session
        )
        self._sslobj = sslobj
        return self

    @property
    def context(self):
        
        return self._sslobj.context

    @context.setter
    def context(self, ctx):
        self._sslobj.context = ctx

    @property
    def session(self):
        
        return self._sslobj.session

    @session.setter
    def session(self, session):
        self._sslobj.session = session

    @property
    def session_reused(self):
        
        return self._sslobj.session_reused

    @property
    def server_side(self):
        
        return self._sslobj.server_side

    @property
    def server_hostname(self):
        
        return self._sslobj.server_hostname

    def read(self, len=1024, buffer=None):
        
        if buffer is not None:
            v = self._sslobj.read(len, buffer)
        else:
            v = self._sslobj.read(len)
        return v

    def write(self, data):
        
        return self._sslobj.write(data)

    def getpeercert(self, binary_form=False):
        
        return self._sslobj.getpeercert(binary_form)

    def get_verified_chain(self):
        
        chain = self._sslobj.get_verified_chain()

        if chain is None:
            return []

        return [cert.public_bytes(_ssl.ENCODING_DER) for cert in chain]

    def get_unverified_chain(self):
        
        chain = self._sslobj.get_unverified_chain()

        if chain is None:
            return []

        return [cert.public_bytes(_ssl.ENCODING_DER) for cert in chain]

    def selected_npn_protocol(self):
        
        warnings.warn(
            "ssl NPN is deprecated, use ALPN instead",
            DeprecationWarning,
            stacklevel=2
        )

    def selected_alpn_protocol(self):
        
        return self._sslobj.selected_alpn_protocol()

    def cipher(self):
        
        return self._sslobj.cipher()

    def shared_ciphers(self):
        
        return self._sslobj.shared_ciphers()

    def compression(self):
        
        return self._sslobj.compression()

    def pending(self):
        
        return self._sslobj.pending()

    def do_handshake(self):
        
        self._sslobj.do_handshake()

    def unwrap(self):
        
        return self._sslobj.shutdown()

    def get_channel_binding(self, cb_type="tls-unique"):
        
        return self._sslobj.get_channel_binding(cb_type)

    def version(self):
        
        return self._sslobj.version()

    def verify_client_post_handshake(self):
        return self._sslobj.verify_client_post_handshake()


def _sslcopydoc(func):
    
    func.__doc__ = getattr(SSLObject, func.__name__).__doc__
    return func


class SSLSocket(socket):
    

    def __init__(self, *args, **kwargs):
        raise TypeError(
            f"{self.__class__.__name__} does not have a public "
            f"constructor. Instances are returned by "
            f"SSLContext.wrap_socket()."
        )

    @classmethod
    def _create(cls, sock, server_side=False, do_handshake_on_connect=True,
                suppress_ragged_eofs=True, server_hostname=None,
                context=None, session=None):
        if sock.getsockopt(SOL_SOCKET, SO_TYPE) != SOCK_STREAM:
            raise NotImplementedError("only stream sockets are supported")
        if server_side:
            if server_hostname:
                raise ValueError("server_hostname can only be specified "
                                 "in client mode")
            if session is not None:
                raise ValueError("session can only be specified in "
                                 "client mode")
        if context.check_hostname and not server_hostname:
            raise ValueError("check_hostname requires server_hostname")

        kwargs = dict(
            family=sock.family, type=sock.type, proto=sock.proto,
            fileno=sock.fileno()
        )
        self = cls.__new__(cls, **kwargs)
        super(SSLSocket, self).__init__(**kwargs)
        sock_timeout = sock.gettimeout()
        sock.detach()

        self._context = context
        self._session = session
        self._closed = False
        self._sslobj = None
        self.server_side = server_side
        self.server_hostname = context._encode_hostname(server_hostname)
        self.do_handshake_on_connect = do_handshake_on_connect
        self.suppress_ragged_eofs = suppress_ragged_eofs

        
        try:
            self.getpeername()
        except OSError as e:
            if e.errno != errno.ENOTCONN:
                raise
            connected = False
            blocking = self.getblocking()
            self.setblocking(False)
            try:
                
                
                
                
                notconn_pre_handshake_data = self.recv(1)
            except OSError as e:
                
                if e.errno not in (errno.ENOTCONN, errno.EINVAL):
                    raise
                notconn_pre_handshake_data = b''
            self.setblocking(blocking)
            if notconn_pre_handshake_data:
                
                
                
                reason = "Closed before TLS handshake with data in recv buffer."
                notconn_pre_handshake_data_error = SSLError(e.errno, reason)
                
                notconn_pre_handshake_data_error.reason = reason
                notconn_pre_handshake_data_error.library = None
                try:
                    self.close()
                except OSError:
                    pass
                try:
                    raise notconn_pre_handshake_data_error
                finally:
                    
                    notconn_pre_handshake_data_error = None
        else:
            connected = True

        self.settimeout(sock_timeout)  
        self._connected = connected
        if connected:
            
            try:
                self._sslobj = self._context._wrap_socket(
                    self, server_side, self.server_hostname,
                    owner=self, session=self._session,
                )
                if do_handshake_on_connect:
                    timeout = self.gettimeout()
                    if timeout == 0.0:
                        
                        raise ValueError("do_handshake_on_connect should not be specified for non-blocking sockets")
                    self.do_handshake()
            except (OSError, ValueError):
                self.close()
                raise
        return self

    @property
    @_sslcopydoc
    def context(self):
        return self._context

    @context.setter
    def context(self, ctx):
        self._context = ctx
        self._sslobj.context = ctx

    @property
    @_sslcopydoc
    def session(self):
        if self._sslobj is not None:
            return self._sslobj.session

    @session.setter
    def session(self, session):
        self._session = session
        if self._sslobj is not None:
            self._sslobj.session = session

    @property
    @_sslcopydoc
    def session_reused(self):
        if self._sslobj is not None:
            return self._sslobj.session_reused

    def dup(self):
        raise NotImplementedError("Can't dup() %s instances" %
                                  self.__class__.__name__)

    def _checkClosed(self, msg=None):
        
        pass

    def _check_connected(self):
        if not self._connected:
            
            
            
            
            self.getpeername()

    def read(self, len=1024, buffer=None):
        

        self._checkClosed()
        if self._sslobj is None:
            raise ValueError("Read on closed or unwrapped SSL socket.")
        try:
            if buffer is not None:
                return self._sslobj.read(len, buffer)
            else:
                return self._sslobj.read(len)
        except SSLError as x:
            if x.args[0] == SSL_ERROR_EOF and self.suppress_ragged_eofs:
                if buffer is not None:
                    return 0
                else:
                    return b''
            else:
                raise

    def write(self, data):
        

        self._checkClosed()
        if self._sslobj is None:
            raise ValueError("Write on closed or unwrapped SSL socket.")
        return self._sslobj.write(data)

    @_sslcopydoc
    def getpeercert(self, binary_form=False):
        self._checkClosed()
        self._check_connected()
        return self._sslobj.getpeercert(binary_form)

    @_sslcopydoc
    def get_verified_chain(self):
        return self._sslobj.get_verified_chain()

    @_sslcopydoc
    def get_unverified_chain(self):
        return self._sslobj.get_unverified_chain()

    @_sslcopydoc
    def selected_npn_protocol(self):
        self._checkClosed()
        warnings.warn(
            "ssl NPN is deprecated, use ALPN instead",
            DeprecationWarning,
            stacklevel=2
        )
        return None

    @_sslcopydoc
    def selected_alpn_protocol(self):
        self._checkClosed()
        if self._sslobj is None or not _ssl.HAS_ALPN:
            return None
        else:
            return self._sslobj.selected_alpn_protocol()

    @_sslcopydoc
    def cipher(self):
        self._checkClosed()
        if self._sslobj is None:
            return None
        else:
            return self._sslobj.cipher()

    @_sslcopydoc
    def shared_ciphers(self):
        self._checkClosed()
        if self._sslobj is None:
            return None
        else:
            return self._sslobj.shared_ciphers()

    @_sslcopydoc
    def compression(self):
        self._checkClosed()
        if self._sslobj is None:
            return None
        else:
            return self._sslobj.compression()

    def send(self, data, flags=0):
        self._checkClosed()
        if self._sslobj is not None:
            if flags != 0:
                raise ValueError(
                    "non-zero flags not allowed in calls to send() on %s" %
                    self.__class__)
            return self._sslobj.write(data)
        else:
            return super().send(data, flags)

    def sendto(self, data, flags_or_addr, addr=None):
        self._checkClosed()
        if self._sslobj is not None:
            raise ValueError("sendto not allowed on instances of %s" %
                             self.__class__)
        elif addr is None:
            return super().sendto(data, flags_or_addr)
        else:
            return super().sendto(data, flags_or_addr, addr)

    def sendmsg(self, *args, **kwargs):
        
        
        raise NotImplementedError("sendmsg not allowed on instances of %s" %
                                  self.__class__)

    def sendall(self, data, flags=0):
        self._checkClosed()
        if self._sslobj is not None:
            if flags != 0:
                raise ValueError(
                    "non-zero flags not allowed in calls to sendall() on %s" %
                    self.__class__)
            count = 0
            with memoryview(data) as view, view.cast("B") as byte_view:
                amount = len(byte_view)
                while count < amount:
                    v = self.send(byte_view[count:])
                    count += v
        else:
            return super().sendall(data, flags)

    def sendfile(self, file, offset=0, count=None):
        
        if self._sslobj is not None:
            return self._sendfile_use_send(file, offset, count)
        else:
            
            return super().sendfile(file, offset, count)

    def recv(self, buflen=1024, flags=0):
        self._checkClosed()
        if self._sslobj is not None:
            if flags != 0:
                raise ValueError(
                    "non-zero flags not allowed in calls to recv() on %s" %
                    self.__class__)
            return self.read(buflen)
        else:
            return super().recv(buflen, flags)

    def recv_into(self, buffer, nbytes=None, flags=0):
        self._checkClosed()
        if nbytes is None:
            if buffer is not None:
                with memoryview(buffer) as view:
                    nbytes = view.nbytes
                if not nbytes:
                    nbytes = 1024
            else:
                nbytes = 1024
        if self._sslobj is not None:
            if flags != 0:
                raise ValueError(
                  "non-zero flags not allowed in calls to recv_into() on %s" %
                  self.__class__)
            return self.read(nbytes, buffer)
        else:
            return super().recv_into(buffer, nbytes, flags)

    def recvfrom(self, buflen=1024, flags=0):
        self._checkClosed()
        if self._sslobj is not None:
            raise ValueError("recvfrom not allowed on instances of %s" %
                             self.__class__)
        else:
            return super().recvfrom(buflen, flags)

    def recvfrom_into(self, buffer, nbytes=None, flags=0):
        self._checkClosed()
        if self._sslobj is not None:
            raise ValueError("recvfrom_into not allowed on instances of %s" %
                             self.__class__)
        else:
            return super().recvfrom_into(buffer, nbytes, flags)

    def recvmsg(self, *args, **kwargs):
        raise NotImplementedError("recvmsg not allowed on instances of %s" %
                                  self.__class__)

    def recvmsg_into(self, *args, **kwargs):
        raise NotImplementedError("recvmsg_into not allowed on instances of "
                                  "%s" % self.__class__)

    @_sslcopydoc
    def pending(self):
        self._checkClosed()
        if self._sslobj is not None:
            return self._sslobj.pending()
        else:
            return 0

    def shutdown(self, how):
        self._checkClosed()
        self._sslobj = None
        super().shutdown(how)

    @_sslcopydoc
    def unwrap(self):
        if self._sslobj:
            s = self._sslobj.shutdown()
            self._sslobj = None
            return s
        else:
            raise ValueError("No SSL wrapper around " + str(self))

    @_sslcopydoc
    def verify_client_post_handshake(self):
        if self._sslobj:
            return self._sslobj.verify_client_post_handshake()
        else:
            raise ValueError("No SSL wrapper around " + str(self))

    def _real_close(self):
        self._sslobj = None
        super()._real_close()

    @_sslcopydoc
    def do_handshake(self, block=False):
        self._check_connected()
        timeout = self.gettimeout()
        try:
            if timeout == 0.0 and block:
                self.settimeout(None)
            self._sslobj.do_handshake()
        finally:
            self.settimeout(timeout)

    def _real_connect(self, addr, connect_ex):
        if self.server_side:
            raise ValueError("can't connect in server-side mode")
        
        
        if self._connected or self._sslobj is not None:
            raise ValueError("attempt to connect already-connected SSLSocket!")
        self._sslobj = self.context._wrap_socket(
            self, False, self.server_hostname,
            owner=self, session=self._session
        )
        try:
            if connect_ex:
                rc = super().connect_ex(addr)
            else:
                rc = None
                super().connect(addr)
            if not rc:
                self._connected = True
                if self.do_handshake_on_connect:
                    self.do_handshake()
            return rc
        except (OSError, ValueError):
            self._sslobj = None
            raise

    def connect(self, addr):
        
        self._real_connect(addr, False)

    def connect_ex(self, addr):
        
        return self._real_connect(addr, True)

    def accept(self):
        

        newsock, addr = super().accept()
        newsock = self.context.wrap_socket(newsock,
                    do_handshake_on_connect=self.do_handshake_on_connect,
                    suppress_ragged_eofs=self.suppress_ragged_eofs,
                    server_side=True)
        return newsock, addr

    @_sslcopydoc
    def get_channel_binding(self, cb_type="tls-unique"):
        if self._sslobj is not None:
            return self._sslobj.get_channel_binding(cb_type)
        else:
            if cb_type not in CHANNEL_BINDING_TYPES:
                raise ValueError(
                    "{0} channel binding type not implemented".format(cb_type)
                )
            return None

    @_sslcopydoc
    def version(self):
        if self._sslobj is not None:
            return self._sslobj.version()
        else:
            return None



SSLContext.sslsocket_class = SSLSocket
SSLContext.sslobject_class = SSLObject




def cert_time_to_seconds(cert_time):
    
    from time import strptime
    from calendar import timegm

    months = (
        "Jan","Feb","Mar","Apr","May","Jun",
        "Jul","Aug","Sep","Oct","Nov","Dec"
    )
    time_format = ' %d %H:%M:%S %Y GMT' 
    try:
        month_number = months.index(cert_time[:3].title()) + 1
    except ValueError:
        raise ValueError('time data %r does not match '
                         'format "%%b%s"' % (cert_time, time_format))
    else:
        
        tt = strptime(cert_time[3:], time_format)
        
        
        return timegm((tt[0], month_number) + tt[2:6])

PEM_HEADER = "-----BEGIN CERTIFICATE-----"
PEM_FOOTER = "-----END CERTIFICATE-----"

def DER_cert_to_PEM_cert(der_cert_bytes):
    

    f = str(base64.standard_b64encode(der_cert_bytes), 'ASCII', 'strict')
    ss = [PEM_HEADER]
    ss += [f[i:i+64] for i in range(0, len(f), 64)]
    ss.append(PEM_FOOTER + '\n')
    return '\n'.join(ss)

def PEM_cert_to_DER_cert(pem_cert_string):
    

    if not pem_cert_string.startswith(PEM_HEADER):
        raise ValueError("Invalid PEM encoding; must start with %s"
                         % PEM_HEADER)
    if not pem_cert_string.strip().endswith(PEM_FOOTER):
        raise ValueError("Invalid PEM encoding; must end with %s"
                         % PEM_FOOTER)
    d = pem_cert_string.strip()[len(PEM_HEADER):-len(PEM_FOOTER)]
    return base64.decodebytes(d.encode('ASCII', 'strict'))

def get_server_certificate(addr, ssl_version=PROTOCOL_TLS_CLIENT,
                           ca_certs=None, timeout=_GLOBAL_DEFAULT_TIMEOUT):
    

    host, port = addr
    if ca_certs is not None:
        cert_reqs = CERT_REQUIRED
    else:
        cert_reqs = CERT_NONE
    context = _create_stdlib_context(ssl_version,
                                     cert_reqs=cert_reqs,
                                     cafile=ca_certs)
    with create_connection(addr, timeout=timeout) as sock:
        with context.wrap_socket(sock, server_hostname=host) as sslsock:
            dercert = sslsock.getpeercert(True)
    return DER_cert_to_PEM_cert(dercert)

def get_protocol_name(protocol_code):
    return _PROTOCOL_NAMES.get(protocol_code, '<unknown>')