'TLS13-AES-256-GCM-SHA384:TLS13-CHACHA20-POLY1305-SHA256:'

'TLS13-AES-128-GCM-SHA256:'

'ECDH+AESGCM:ECDH+CHACHA20:DH+AESGCM:DH+CHACHA20:ECDH+AES256:DH+AES256:'

'ECDH+AES128:DH+AES:ECDH+HIGH:DH+HIGH:RSA+AESGCM:RSA+AES:RSA+HIGH:'

'!aNULL:!eNULL:!MD5:!3DES'

'TLS13-AES-256-GCM-SHA384:TLS13-CHACHA20-POLY1305-SHA256:'

'TLS13-AES-128-GCM-SHA256:'

'ECDH+AESGCM:ECDH+CHACHA20:DH+AESGCM:DH+CHACHA20:ECDH+AES256:DH+AES256:'

'ECDH+AES128:DH+AES:ECDH+HIGH:DH+HIGH:RSA+AESGCM:RSA+AES:RSA+HIGH:'

'!aNULL:!eNULL:!MD5:!DSS:!RC4:!3DES'

"""Matching according to RFC 6125, section 6.4.3

pats = []

if not dn:

return False

pieces = dn.split(r'.')

leftmost = pieces[0]

remainder = pieces[1:]

wildcards = leftmost.count('*')

if wildcards > max_wildcards:

# Issue #17980: avoid denials of service by refusing more

# than one wildcard per fragment. A survery of established

# policy among SSL implementations showed it to be a

# reasonable choice.

raise CertificateError(

"too many wildcards in certificate DNS name: " + repr(dn))

# speed up common case w/o wildcards

if not wildcards:

return dn.lower() == hostname.lower()

# RFC 6125, section 6.4.3, subitem 1.

# The client SHOULD NOT attempt to match a presented identifier in which

# the wildcard character comprises a label other than the left-most label.

if leftmost == '*':

# When '*' is a fragment by itself, it matches a non-empty dotless

# fragment.

pats.append('[^.]+')

elif leftmost.startswith('xn--') or hostname.startswith('xn--'):

# RFC 6125, section 6.4.3, subitem 3.

# The client SHOULD NOT attempt to match a presented identifier

# where the wildcard character is embedded within an A-label or

# U-label of an internationalized domain name.

pats.append(re.escape(leftmost))

else:

# Otherwise, '*' matches any dotless string, e.g. www*

pats.append(re.escape(leftmost).replace(r'\*', '[^.]*'))

# add the remaining fragments, ignore any wildcards

for frag in remainder:

pats.append(re.escape(frag))

pat = re.compile(r'\A' + r'\.'.join(pats) + r'\Z', re.IGNORECASE)

"""Verify that *cert* (in decoded format as returned by

if not cert:

raise ValueError("empty or no certificate, match_hostname needs a "

"SSL socket or SSL context with either "

"CERT_OPTIONAL or CERT_REQUIRED")

dnsnames = []

san = cert.get('subjectAltName', ())

for key, value in san:

if key == 'DNS':

if _dnsname_match(value, hostname):

return

dnsnames.append(value)

if not dnsnames:

# The subject is only checked when there is no dNSName entry

# in subjectAltName

for sub in cert.get('subject', ()):

for key, value in sub:

# XXX according to RFC 2818, the most specific Common Name

# must be used.

if key == 'commonName':

if _dnsname_match(value, hostname):

return

dnsnames.append(value)

if len(dnsnames) > 1:

raise CertificateError("hostname %r "

"doesn't match either of %s"

% (hostname, ', '.join(map(repr, dnsnames))))

elif len(dnsnames) == 1:

raise CertificateError("hostname %r "

"doesn't match %r"

% (hostname, dnsnames[0]))

else:

raise CertificateError("no appropriate commonName or "

"""Return paths to default cafile and capath.

parts = _ssl.get_default_verify_paths()

# environment vars shadow 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,

"""ASN.1 object identifier lookup

__slots__ = ()

def __new__(cls, oid):

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

@classmethod

def fromnid(cls, nid):

"""Create _ASN1Object from OpenSSL numeric ID

return super(_ASN1Object, cls).__new__(cls, *_nid2obj(nid))

@classmethod

def fromname(cls, name):

"""Create _ASN1Object from short name, long name or OID

"""An SSLContext holds various SSL-related configuration options and

__slots__ = ('protocol', '__weakref__')

_windows_cert_stores = ("CA", "ROOT")

def __new__(cls, protocol, *args, **kwargs):

self = _SSLContext.__new__(cls, protocol)

if protocol != _SSLv2_IF_EXISTS:

self.set_ciphers(_DEFAULT_CIPHERS)

return self

def __init__(self, protocol):

self.protocol = protocol

def wrap_socket(self, sock, server_side=False,

do_handshake_on_connect=True,

suppress_ragged_eofs=True,

server_hostname=None):

return SSLSocket(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)

def set_npn_protocols(self, npn_protocols):

protos = bytearray()

for protocol in npn_protocols:

b = protocol.encode('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_alpn_protocols(self, alpn_protocols):

protos = bytearray()

for protocol in alpn_protocols:

b = protocol.encode('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):

# CA certs are never PKCS#7 encoded

if encoding == "x509_asn":

if trust is True or purpose.oid in trust:

certs.extend(cert)

except OSError:

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)

capath=None, cadata=None):

"""Create a SSLContext object with default settings.

if not isinstance(purpose, _ASN1Object):

raise TypeError(purpose)

# 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.

context = SSLContext(PROTOCOL_TLS)

if purpose == Purpose.SERVER_AUTH:

# verify certs and host name in client mode

context.verify_mode = CERT_REQUIRED

context.check_hostname = True

elif purpose == Purpose.CLIENT_AUTH:

context.set_ciphers(_RESTRICTED_SERVER_CIPHERS)

if cafile or capath or cadata:

context.load_verify_locations(cafile, capath, cadata)

elif context.verify_mode != CERT_NONE:

# 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.

context.load_default_certs(purpose)

check_hostname=False, purpose=Purpose.SERVER_AUTH,

certfile=None, keyfile=None,

cafile=None, capath=None, cadata=None):

"""Create a SSLContext object for Python stdlib modules

if not isinstance(purpose, _ASN1Object):

raise TypeError(purpose)

# 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.

context = SSLContext(protocol)

if cert_reqs is not None:

context.verify_mode = cert_reqs

context.check_hostname = check_hostname

if keyfile and not certfile:

raise ValueError("certfile must be specified")

if certfile or keyfile:

context.load_cert_chain(certfile, keyfile)

# load CA root certs

if cafile or capath or cadata:

context.load_verify_locations(cafile, capath, cadata)

elif context.verify_mode != CERT_NONE:

# 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.

context.load_default_certs(purpose)

if not sys.flags.ignore_environment:

config_setting = os.environ.get(_https_verify_envvar)

if config_setting == '0':

return _create_unverified_context

"""Verify server HTTPS certificates by default?"""

global _create_default_https_context

if enable:

_create_default_https_context = create_default_context

else:

"""This class implements a subtype of socket.socket that wraps

def __init__(self, sock=None, keyfile=None, certfile=None,

server_side=False, cert_reqs=CERT_NONE,

ssl_version=PROTOCOL_TLS, ca_certs=None,

do_handshake_on_connect=True,

family=AF_INET, type=SOCK_STREAM, proto=0, fileno=None,

suppress_ragged_eofs=True, npn_protocols=None, ciphers=None,

server_hostname=None,

_context=None):

self._makefile_refs = 0

if _context:

self._context = _context

else:

if server_side and not certfile:

raise ValueError("certfile must be specified for server-side "

"operations")

if keyfile and not certfile:

raise ValueError("certfile must be specified")

if certfile and not keyfile:

keyfile = certfile

self._context = SSLContext(ssl_version)

self._context.verify_mode = cert_reqs

if ca_certs:

self._context.load_verify_locations(ca_certs)

if certfile:

self._context.load_cert_chain(certfile, keyfile)

if npn_protocols:

self._context.set_npn_protocols(npn_protocols)

if ciphers:

self._context.set_ciphers(ciphers)

self.keyfile = keyfile

self.certfile = certfile

self.cert_reqs = cert_reqs

self.ssl_version = ssl_version

self.ca_certs = ca_certs

self.ciphers = ciphers

# Can't use sock.type as other flags (such as SOCK_NONBLOCK) get

# mixed in.

if sock.getsockopt(SOL_SOCKET, SO_TYPE) != SOCK_STREAM:

raise NotImplementedError("only stream sockets are supported")

socket.__init__(self, _sock=sock._sock)

# The initializer for socket overrides the methods send(), recv(), etc.

# in the instancce, which we don't need -- but we want to provide the

# methods defined in SSLSocket.

for attr in _delegate_methods:

try:

delattr(self, attr)

except AttributeError:

pass

if server_side and server_hostname:

raise ValueError("server_hostname can only be specified "

"in client mode")

if self._context.check_hostname and not server_hostname:

raise ValueError("check_hostname requires server_hostname")

self.server_side = server_side

self.server_hostname = server_hostname

self.do_handshake_on_connect = do_handshake_on_connect

self.suppress_ragged_eofs = suppress_ragged_eofs

# See if we are connected

try:

self.getpeername()

except socket_error as e:

if e.errno != errno.ENOTCONN:

raise

connected = False

else:

connected = True

self._closed = False

self._sslobj = None

self._connected = connected

if connected:

# create the SSL object

try:

self._sslobj = self._context._wrap_socket(self._sock, server_side,

server_hostname, ssl_sock=self)

if do_handshake_on_connect:

timeout = self.gettimeout()

if timeout == 0.0:

# non-blocking

raise ValueError("do_handshake_on_connect should not be specified for non-blocking sockets")

self.do_handshake()

except (OSError, ValueError):

self.close()

raise

@property

def context(self):

return self._context

@context.setter

def context(self, ctx):

self._context = ctx

self._sslobj.context = ctx

def dup(self):

raise NotImplementedError("Can't dup() %s instances" %

self.__class__.__name__)

def _checkClosed(self, msg=None):

# raise an exception here if you wish to check for spurious closes

pass

def _check_connected(self):

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 read(self, len=1024, buffer=None):

"""Read up to LEN bytes and return them.

self._checkClosed()

if not self._sslobj:

raise ValueError("Read on closed or unwrapped SSL socket.")

try:

if buffer is not None:

v = self._sslobj.read(len, buffer)

else:

v = self._sslobj.read(len)

return v

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

"""Write DATA to the underlying SSL channel. Returns

self._checkClosed()

if not self._sslobj:

raise ValueError("Write on closed or unwrapped SSL socket.")

return self._sslobj.write(data)

def getpeercert(self, binary_form=False):

"""Returns a formatted version of the data in the

self._checkClosed()

self._check_connected()

return self._sslobj.peer_certificate(binary_form)

def selected_npn_protocol(self):

self._checkClosed()

if not self._sslobj or not _ssl.HAS_NPN:

return None

else:

return self._sslobj.selected_npn_protocol()

def selected_alpn_protocol(self):

self._checkClosed()

if not self._sslobj or not _ssl.HAS_ALPN:

return None

else:

return self._sslobj.selected_alpn_protocol()

def cipher(self):

self._checkClosed()

if not self._sslobj:

return None

else:

return self._sslobj.cipher()

def compression(self):

self._checkClosed()

if not self._sslobj:

return None

else:

return self._sslobj.compression()

def send(self, data, flags=0):

self._checkClosed()

if self._sslobj:

if flags != 0:

raise ValueError(

"non-zero flags not allowed in calls to send() on %s" %

self.__class__)

try:

v = self._sslobj.write(data)

except SSLError as x:

if x.args[0] == SSL_ERROR_WANT_READ:

return 0

elif x.args[0] == SSL_ERROR_WANT_WRITE:

return 0

else:

raise

else:

return v

else:

return self._sock.send(data, flags)

def sendto(self, data, flags_or_addr, addr=None):

self._checkClosed()

if self._sslobj:

raise ValueError("sendto not allowed on instances of %s" %

self.__class__)

elif addr is None:

return self._sock.sendto(data, flags_or_addr)

else:

return self._sock.sendto(data, flags_or_addr, addr)

def sendall(self, data, flags=0):

self._checkClosed()

if self._sslobj:

if flags != 0:

raise ValueError(

"non-zero flags not allowed in calls to sendall() on %s" %

self.__class__)

amount = len(data)

count = 0

while (count < amount):

v = self.send(data[count:])

count += v

return amount

else:

return socket.sendall(self, data, flags)

def recv(self, buflen=1024, flags=0):

self._checkClosed()

if self._sslobj:

if flags != 0:

raise ValueError(

"non-zero flags not allowed in calls to recv() on %s" %

self.__class__)

return self.read(buflen)

else:

return self._sock.recv(buflen, flags)

def recv_into(self, buffer, nbytes=None, flags=0):

self._checkClosed()

if buffer and (nbytes is None):

nbytes = len(buffer)

elif nbytes is None:

nbytes = 1024

if self._sslobj:

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 self._sock.recv_into(buffer, nbytes, flags)

def recvfrom(self, buflen=1024, flags=0):

self._checkClosed()

if self._sslobj:

raise ValueError("recvfrom not allowed on instances of %s" %

self.__class__)

else:

return self._sock.recvfrom(buflen, flags)

def recvfrom_into(self, buffer, nbytes=None, flags=0):

self._checkClosed()

if self._sslobj:

raise ValueError("recvfrom_into not allowed on instances of %s" %

self.__class__)

else:

return self._sock.recvfrom_into(buffer, nbytes, flags)

def pending(self):

self._checkClosed()

if self._sslobj:

return self._sslobj.pending()

else:

return 0

def shutdown(self, how):

self._checkClosed()

self._sslobj = None

socket.shutdown(self, how)

def close(self):

if self._makefile_refs < 1:

self._sslobj = None

socket.close(self)

else:

self._makefile_refs -= 1

def unwrap(self):

if self._sslobj:

s = self._sslobj.shutdown()

self._sslobj = None

return s

else:

raise ValueError("No SSL wrapper around " + str(self))

def _real_close(self):

self._sslobj = None

socket._real_close(self)

def do_handshake(self, block=False):

"""Perform a TLS/SSL handshake."""

self._check_connected()

timeout = self.gettimeout()

try:

if timeout == 0.0 and block:

self.settimeout(None)

self._sslobj.do_handshake()

finally:

self.settimeout(timeout)

if self.context.check_hostname:

if not self.server_hostname:

raise ValueError("check_hostname needs server_hostname "

"argument")

match_hostname(self.getpeercert(), self.server_hostname)

def _real_connect(self, addr, connect_ex):

if self.server_side:

raise ValueError("can't connect in server-side mode")

# Here we assume that the socket is client-side, and not

# connected at the time of the call. We connect it, then wrap it.

if self._connected:

raise ValueError("attempt to connect already-connected SSLSocket!")

self._sslobj = self.context._wrap_socket(self._sock, False, self.server_hostname, ssl_sock=self)

try:

if connect_ex:

rc = socket.connect_ex(self, addr)

else:

rc = None

socket.connect(self, 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):

"""Connects to remote ADDR, and then wraps the connection in

self._real_connect(addr, False)

def connect_ex(self, addr):

"""Connects to remote ADDR, and then wraps the connection in

return self._real_connect(addr, True)

def accept(self):

"""Accepts a new connection from a remote client, and returns

newsock, addr = socket.accept(self)

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

def makefile(self, mode='r', bufsize=-1):

"""Make and return a file-like object that

self._makefile_refs += 1

# close=True so as to decrement the reference count when done with

# the file-like object.

return _fileobject(self, mode, bufsize, close=True)

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

"""Get channel binding data for current connection. Raise ValueError

if cb_type not in CHANNEL_BINDING_TYPES:

raise ValueError("Unsupported channel binding type")

if cb_type != "tls-unique":

raise NotImplementedError(

"{0} channel binding type not implemented"

.format(cb_type))

if self._sslobj is None:

return None

return self._sslobj.tls_unique_cb()

def version(self):

"""

if self._sslobj is None:

return None

server_side=False, cert_reqs=CERT_NONE,

ssl_version=PROTOCOL_TLS, ca_certs=None,

do_handshake_on_connect=True,

suppress_ragged_eofs=True,

ciphers=None):

return SSLSocket(sock=sock, keyfile=keyfile, certfile=certfile,

server_side=server_side, cert_reqs=cert_reqs,

ssl_version=ssl_version, ca_certs=ca_certs,

do_handshake_on_connect=do_handshake_on_connect,

suppress_ragged_eofs=suppress_ragged_eofs,

"""Return the time in seconds since the Epoch, given the timestring

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' # NOTE: no month, fixed 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:

# found valid month

tt = strptime(cert_time[3:], time_format)

# 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

f = base64.standard_b64encode(der_cert_bytes).decode('ascii')

return (PEM_HEADER + '\n' +

textwrap.fill(f, 64) + '\n' +

"""Takes a certificate in ASCII PEM format and returns the

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

"""Retrieve the certificate from the server at the specified address,

host, port = addr

if ca_certs is not None:

cert_reqs = CERT_REQUIRED

else:

cert_reqs = CERT_NONE