try:

# Assume that any references of "ubuntu" implies Ubuntu-like distro

# The workaround is not required for 18.04, but doesn't hurt either.

with open("/etc/os-release", encoding="utf-8") as f:

return "ubuntu" in f.read()

except FileNotFoundError:

"""Lower security level to '1' and allow all ciphers for TLS 1.0/1"""

for ctx in ctxs:

if (

hasattr(ctx, "minimum_version") and

ctx.minimum_version <= ssl.TLSVersion.TLSv1_1 and

ctx.security_level > 1

):

"""Check if a TLS protocol is available and enabled

if isinstance(protocol, str):

assert protocol.startswith('PROTOCOL_')

protocol = getattr(ssl, protocol, None)

if protocol is None:

return False

if protocol in {

ssl.PROTOCOL_TLS, ssl.PROTOCOL_TLS_SERVER,

ssl.PROTOCOL_TLS_CLIENT

}:

# auto-negotiate protocols are always available

return True

name = protocol.name

"""Check if a TLS/SSL version is enabled

if isinstance(version, str):

version = ssl.TLSVersion.__members__[version]

# check compile time flags like ssl.HAS_TLSv1_2

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

return False

if IS_OPENSSL_3_0_0 and version < ssl.TLSVersion.TLSv1_2:

# bpo43791: 3.0.0-alpha14 fails with TLSV1_ALERT_INTERNAL_ERROR

return False

# check runtime and dynamic crypto policy settings. A TLS version may

# be compiled in but disabled by a policy or config option.

ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)

if (

hasattr(ctx, 'minimum_version') and

ctx.minimum_version != ssl.TLSVersion.MINIMUM_SUPPORTED and

version < ctx.minimum_version

):

return False

if (

hasattr(ctx, 'maximum_version') and

ctx.maximum_version != ssl.TLSVersion.MAXIMUM_SUPPORTED and

version > ctx.maximum_version

):

return False

"""Decorator to skip tests when a required TLS version is not available

def decorator(func):

@functools.wraps(func)

def wrapper(*args, **kw):

if not has_tls_version(version):

raise unittest.SkipTest(f"{version} is not available.")

else:

return func(*args, **kw)

return wrapper

# local time = utc time + utc offset

if time.daylight and time.localtime().tm_isdst > 0:

return -time.altzone # seconds

*,

server_side=False,

check_hostname=None,

cert_reqs=ssl.CERT_NONE,

ca_certs=None, certfile=None, keyfile=None,

ciphers=None, ciphersuites=None,

min_version=None, max_version=None,

):

if server_side:

context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)

else:

context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)

if check_hostname is None:

if cert_reqs == ssl.CERT_NONE:

context.check_hostname = False

else:

context.check_hostname = check_hostname

if cert_reqs is not None:

context.verify_mode = cert_reqs

if ca_certs is not None:

context.load_verify_locations(ca_certs)

if certfile is not None or keyfile is not None:

context.load_cert_chain(certfile, keyfile)

if ciphers is not None:

context.set_ciphers(ciphers)

if ciphersuites is not None:

context.set_ciphersuites(ciphersuites)

if min_version is not None:

context.minimum_version = min_version

if max_version is not None:

context.maximum_version = max_version

sock,

*,

server_side=False,

check_hostname=None,

cert_reqs=ssl.CERT_NONE,

ca_certs=None, certfile=None, keyfile=None,

ciphers=None, ciphersuites=None,

min_version=None, max_version=None,

**kwargs,

):

context = make_test_context(

server_side=server_side,

check_hostname=check_hostname,

cert_reqs=cert_reqs,

ca_certs=ca_certs, certfile=certfile, keyfile=keyfile,

ciphers=ciphers, ciphersuites=ciphersuites,

min_version=min_version, max_version=max_version,

)

if not server_side:

kwargs.setdefault("server_hostname", SIGNED_CERTFILE_HOSTNAME)

client_cert=None):

"""Create context

global _TEST_CONTEXT

if USE_SAME_TEST_CONTEXT:

if _TEST_CONTEXT is not None:

return _TEST_CONTEXT

if server_cert == SIGNED_CERTFILE:

hostname = SIGNED_CERTFILE_HOSTNAME

elif server_cert == SIGNED_CERTFILE2:

hostname = SIGNED_CERTFILE2_HOSTNAME

elif server_cert == NOSANFILE:

hostname = NOSAN_HOSTNAME

else:

raise ValueError(server_cert)

client_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)

client_context.load_verify_locations(SIGNING_CA)

server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)

server_context.load_cert_chain(server_cert)

if server_chain:

server_context.load_verify_locations(SIGNING_CA)

if client_cert:

client_context.load_cert_chain(client_cert)

server_context.verify_mode = ssl.CERT_REQUIRED

if USE_SAME_TEST_CONTEXT:

if _TEST_CONTEXT is not None:

_TEST_CONTEXT = client_context, server_context, hostname

"""Call do_handshake() on the sslobject and return the sent data.

data, attempt = None, 0

while not data and attempt < max_retry:

with contextlib.suppress(ssl.SSLWantReadError):

sslobject.do_handshake()

data = outgoing.read()

attempt += 1

def test_constants(self):

ssl.CERT_NONE

ssl.CERT_OPTIONAL

ssl.CERT_REQUIRED

ssl.OP_CIPHER_SERVER_PREFERENCE

ssl.OP_SINGLE_DH_USE

ssl.OP_SINGLE_ECDH_USE

ssl.OP_NO_COMPRESSION

self.assertEqual(ssl.HAS_SNI, True)

self.assertEqual(ssl.HAS_ECDH, True)

self.assertIsInstance(ssl.HAS_TLSv1_2, bool)

self.assertEqual(ssl.HAS_TLSv1_3, True)

ssl.OP_NO_SSLv2

ssl.OP_NO_SSLv3

ssl.OP_NO_TLSv1

ssl.OP_NO_TLSv1_3

ssl.OP_NO_TLSv1_1

ssl.OP_NO_TLSv1_2

self.assertEqual(ssl.PROTOCOL_TLS, ssl.PROTOCOL_SSLv23)

def test_options(self):

# gh-106687: SSL options values are unsigned integer (uint64_t)

for name in dir(ssl):

if not name.startswith('OP_'):

continue

with self.subTest(option=name):

value = getattr(ssl, name)

self.assertGreaterEqual(value, 0, f"ssl.{name}")

def test_ssl_types(self):

ssl_types = [

_ssl._SSLContext,

_ssl._SSLSocket,

_ssl.MemoryBIO,

_ssl.Certificate,

_ssl.SSLSession,

_ssl.SSLError,

]

for ssl_type in ssl_types:

with self.subTest(ssl_type=ssl_type):

with self.assertRaisesRegex(TypeError, "immutable type"):

ssl_type.value = None

support.check_disallow_instantiation(self, _ssl.Certificate)

def test_private_init(self):

with self.assertRaisesRegex(TypeError, "public constructor"):

with socket.socket() as s:

ssl.SSLSocket(s)

def test_str_for_enums(self):

# Make sure that the PROTOCOL_* constants have enum-like string

# reprs.

proto = ssl.PROTOCOL_TLS_CLIENT

self.assertEqual(repr(proto), '<_SSLMethod.PROTOCOL_TLS_CLIENT: %r>' % proto.value)

self.assertEqual(str(proto), str(proto.value))

ctx = ssl.SSLContext(proto)

self.assertIs(ctx.protocol, proto)

def test_random(self):

v = ssl.RAND_status()

if support.verbose:

sys.stdout.write("\n RAND_status is %d (%s)\n"

% (v, (v and "sufficient randomness") or

"insufficient randomness"))

if v:

data = ssl.RAND_bytes(16)

self.assertEqual(len(data), 16)

else:

self.assertRaises(ssl.SSLError, ssl.RAND_bytes, 16)

# negative num is invalid

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

ssl.RAND_add("this is a random string", 75.0)

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

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

def test_parse_cert(self):

self.maxDiff = None

# note that this uses an 'unofficial' function in _ssl.c,

# provided solely for this test, to exercise the certificate

# parsing code

self.assertEqual(

ssl._ssl._test_decode_cert(CERTFILE),

CERTFILE_INFO

)

self.assertEqual(

ssl._ssl._test_decode_cert(SIGNED_CERTFILE),

SIGNED_CERTFILE_INFO

)

# Issue #13034: the subjectAltName in some certificates

# (notably projects.developer.nokia.com:443) wasn't parsed

p = ssl._ssl._test_decode_cert(NOKIACERT)

if support.verbose:

sys.stdout.write("\n" + pprint.pformat(p) + "\n")

self.assertEqual(p['subjectAltName'],

(('DNS', 'projects.developer.nokia.com'),

('DNS', 'projects.forum.nokia.com'))

)

# extra OCSP and AIA fields

self.assertEqual(p['OCSP'], ('http://ocsp.verisign.com',))

self.assertEqual(p['caIssuers'],

('http://SVRIntl-G3-aia.verisign.com/SVRIntlG3.cer',))

self.assertEqual(p['crlDistributionPoints'],

('http://SVRIntl-G3-crl.verisign.com/SVRIntlG3.crl',))

def test_parse_cert_CVE_2019_5010(self):

p = ssl._ssl._test_decode_cert(TALOS_INVALID_CRLDP)

if support.verbose:

sys.stdout.write("\n" + pprint.pformat(p) + "\n")

self.assertEqual(

p,

{

'issuer': (

(('countryName', 'UK'),), (('commonName', 'cody-ca'),)),

'notAfter': 'Jun 14 18:00:58 2028 GMT',

'notBefore': 'Jun 18 18:00:58 2018 GMT',

'serialNumber': '02',

'subject': ((('countryName', 'UK'),),

(('commonName',

'codenomicon-vm-2.test.lal.cisco.com'),)),

'subjectAltName': (

('DNS', 'codenomicon-vm-2.test.lal.cisco.com'),),

'version': 3

}

)

def test_parse_cert_CVE_2013_4238(self):

p = ssl._ssl._test_decode_cert(NULLBYTECERT)

if support.verbose:

sys.stdout.write("\n" + pprint.pformat(p) + "\n")

subject = ((('countryName', 'US'),),

(('stateOrProvinceName', 'Oregon'),),

(('localityName', 'Beaverton'),),

(('organizationName', 'Python Software Foundation'),),

(('organizationalUnitName', 'Python Core Development'),),

(('commonName', 'null.python.org\x00example.org'),),

(('emailAddress', 'python-dev@python.org'),))

self.assertEqual(p['subject'], subject)

self.assertEqual(p['issuer'], subject)

if ssl._OPENSSL_API_VERSION >= (0, 9, 8):

san = (('DNS', 'altnull.python.org\x00example.com'),

('email', 'null@python.org\x00user@example.org'),

('URI', 'http://null.python.org\x00http://example.org'),

('IP Address', '192.0.2.1'),

('IP Address', '2001:DB8:0:0:0:0:0:1'))

else:

# OpenSSL 0.9.7 doesn't support IPv6 addresses in subjectAltName

san = (('DNS', 'altnull.python.org\x00example.com'),

('email', 'null@python.org\x00user@example.org'),

('URI', 'http://null.python.org\x00http://example.org'),

('IP Address', '192.0.2.1'),

('IP Address', '<invalid>'))

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

def test_parse_all_sans(self):

p = ssl._ssl._test_decode_cert(ALLSANFILE)

self.assertEqual(p['subjectAltName'],

(

('DNS', 'allsans'),

('othername', '<unsupported>'),

('othername', '<unsupported>'),

('email', 'user@example.org'),

('DNS', 'www.example.org'),

('DirName',

((('countryName', 'XY'),),

(('localityName', 'Castle Anthrax'),),

(('organizationName', 'Python Software Foundation'),),

(('commonName', 'dirname example'),))),

('URI', 'https://www.python.org/'),

('IP Address', '127.0.0.1'),

('IP Address', '0:0:0:0:0:0:0:1'),

('Registered ID', '1.2.3.4.5')

)

)

def test_DER_to_PEM(self):

with open(CAFILE_CACERT, 'r') as f:

pem = f.read()

d1 = ssl.PEM_cert_to_DER_cert(pem)

p2 = ssl.DER_cert_to_PEM_cert(d1)

d2 = ssl.PEM_cert_to_DER_cert(p2)

self.assertEqual(d1, d2)

if not p2.startswith(ssl.PEM_HEADER + '\n'):

self.fail("DER-to-PEM didn't include correct header:\n%r\n" % p2)

if not p2.endswith('\n' + ssl.PEM_FOOTER + '\n'):

self.fail("DER-to-PEM didn't include correct footer:\n%r\n" % p2)

def test_openssl_version(self):

n = ssl.OPENSSL_VERSION_NUMBER

t = ssl.OPENSSL_VERSION_INFO

s = ssl.OPENSSL_VERSION

self.assertIsInstance(n, int)

self.assertIsInstance(t, tuple)

self.assertIsInstance(s, str)

# Some sanity checks follow

# >= 1.1.1

self.assertGreaterEqual(n, 0x10101000)

# < 5.0

self.assertLess(n, 0x50000000)

major, minor, fix, patch, status = t

self.assertGreaterEqual(major, 1)

self.assertLess(major, 5)

self.assertGreaterEqual(minor, 0)

self.assertLess(minor, 256)

self.assertGreaterEqual(fix, 0)

self.assertLess(fix, 256)

self.assertGreaterEqual(patch, 0)

self.assertLessEqual(patch, 63)

self.assertGreaterEqual(status, 0)

self.assertLessEqual(status, 15)

libressl_ver = f"LibreSSL {major:d}"

if major >= 3:

# 3.x uses 0xMNN00PP0L

openssl_ver = f"OpenSSL {major:d}.{minor:d}.{patch:d}"

else:

openssl_ver = f"OpenSSL {major:d}.{minor:d}.{fix:d}"

self.assertStartsWith(

s, (openssl_ver, libressl_ver, "AWS-LC"),

(t, hex(n))

)

@support.cpython_only

def test_refcycle(self):

# Issue #7943: an SSL object doesn't create reference cycles with

# itself.

s = socket.socket(socket.AF_INET)

ss = test_wrap_socket(s)

wr = weakref.ref(ss)

with warnings_helper.check_warnings(("", ResourceWarning)):

del ss

self.assertEqual(wr(), None)

def test_wrapped_unconnected(self):

# Methods on an unconnected SSLSocket propagate the original

# OSError raise by the underlying socket object.

s = socket.socket(socket.AF_INET)

with test_wrap_socket(s) as ss:

self.assertRaises(OSError, ss.recv, 1)

self.assertRaises(OSError, ss.recv_into, bytearray(b'x'))

self.assertRaises(OSError, ss.recvfrom, 1)

self.assertRaises(OSError, ss.recvfrom_into, bytearray(b'x'), 1)

self.assertRaises(OSError, ss.send, b'x')

self.assertRaises(OSError, ss.sendto, b'x', ('0.0.0.0', 0))

self.assertRaises(NotImplementedError, ss.dup)

self.assertRaises(NotImplementedError, ss.sendmsg,

[b'x'], (), 0, ('0.0.0.0', 0))

self.assertRaises(NotImplementedError, ss.recvmsg, 100)

self.assertRaises(NotImplementedError, ss.recvmsg_into,

[bytearray(100)])

def test_timeout(self):

# Issue #8524: when creating an SSL socket, the timeout of the

# original socket should be retained.

for timeout in (None, 0.0, 5.0):

s = socket.socket(socket.AF_INET)

s.settimeout(timeout)

with test_wrap_socket(s) as ss:

self.assertEqual(timeout, ss.gettimeout())

def test_openssl111_deprecations(self):

options = [

ssl.OP_NO_TLSv1,

ssl.OP_NO_TLSv1_1,

ssl.OP_NO_TLSv1_2,

ssl.OP_NO_TLSv1_3

]

protocols = []

if hasattr(ssl, 'PROTOCOL_TLSv1'):

protocols.append(ssl.PROTOCOL_TLSv1)

if hasattr(ssl, 'PROTOCOL_TLSv1_1'):

protocols.append(ssl.PROTOCOL_TLSv1_1)

if hasattr(ssl, 'PROTOCOL_TLSv1_2'):

protocols.append(ssl.PROTOCOL_TLSv1_2)

protocols.append(ssl.PROTOCOL_TLS)

versions = [

ssl.TLSVersion.SSLv3,

ssl.TLSVersion.TLSv1,

ssl.TLSVersion.TLSv1_1,

]

for option in options:

with self.subTest(option=option):

ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)

with self.assertWarns(DeprecationWarning) as cm:

ctx.options |= option

self.assertEqual(

'ssl.OP_NO_SSL*/ssl.OP_NO_TLS* options are deprecated',

str(cm.warning)

)

for protocol in protocols:

if not has_tls_protocol(protocol):

continue

with self.subTest(protocol=protocol):

with self.assertWarns(DeprecationWarning) as cm:

ssl.SSLContext(protocol)

self.assertEqual(

f'ssl.{protocol.name} is deprecated',

str(cm.warning)

)

for version in versions:

if not has_tls_version(version):

continue

with self.subTest(version=version):

ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)

with self.assertWarns(DeprecationWarning) as cm:

ctx.minimum_version = version

version_text = '%s.%s' % (version.__class__.__name__, version.name)

self.assertEqual(

f'ssl.{version_text} is deprecated',

str(cm.warning)

)

def bad_cert_test(self, certfile):

"""Check that trying to use the given client certificate fails"""

certfile = os.path.join(os.path.dirname(__file__) or os.curdir,

"certdata", certfile)

sock = socket.socket()

self.addCleanup(sock.close)

with self.assertRaises(ssl.SSLError):

test_wrap_socket(sock,

certfile=certfile)

def test_empty_cert(self):

"""Wrapping with an empty cert file"""

self.bad_cert_test("nullcert.pem")

def test_malformed_cert(self):

"""Wrapping with a badly formatted certificate (syntax error)"""

self.bad_cert_test("badcert.pem")

def test_malformed_key(self):

"""Wrapping with a badly formatted key (syntax error)"""

self.bad_cert_test("badkey.pem")

def test_server_side(self):

# server_hostname doesn't work for server sockets

ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)

with socket.socket() as sock:

self.assertRaises(ValueError, ctx.wrap_socket, sock, True,

server_hostname="some.hostname")

def test_unknown_channel_binding(self):

# should raise ValueError for unknown type

s = socket.create_server(('127.0.0.1', 0))

c = socket.socket(socket.AF_INET)

c.connect(s.getsockname())

with test_wrap_socket(c, do_handshake_on_connect=False) as ss:

with self.assertRaises(ValueError):

ss.get_channel_binding("unknown-type")

s.close()

@unittest.skipUnless("tls-unique" in ssl.CHANNEL_BINDING_TYPES,

def test_tls_unique_channel_binding(self):

# unconnected should return None for known type

s = socket.socket(socket.AF_INET)

with test_wrap_socket(s) as ss:

self.assertIsNone(ss.get_channel_binding("tls-unique"))

# the same for server-side

s = socket.socket(socket.AF_INET)

with test_wrap_socket(s, server_side=True, certfile=CERTFILE) as ss:

self.assertIsNone(ss.get_channel_binding("tls-unique"))

def test_dealloc_warn(self):

ss = test_wrap_socket(socket.socket(socket.AF_INET))

r = repr(ss)

with self.assertWarns(ResourceWarning) as cm:

ss = None

support.gc_collect()

self.assertIn(r, str(cm.warning.args[0]))

def test_get_default_verify_paths(self):

paths = ssl.get_default_verify_paths()

self.assertEqual(len(paths), 6)

self.assertIsInstance(paths, ssl.DefaultVerifyPaths)

with os_helper.EnvironmentVarGuard() as env:

env["SSL_CERT_DIR"] = CAPATH

env["SSL_CERT_FILE"] = CERTFILE

paths = ssl.get_default_verify_paths()

self.assertEqual(paths.cafile, CERTFILE)

self.assertEqual(paths.capath, CAPATH)

@unittest.skipUnless(sys.platform == "win32", "Windows specific")

def test_enum_certificates(self):

self.assertTrue(ssl.enum_certificates("CA"))

self.assertTrue(ssl.enum_certificates("ROOT"))

self.assertRaises(TypeError, ssl.enum_certificates)

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

trust_oids = set()

for storename in ("CA", "ROOT"):

store = ssl.enum_certificates(storename)

self.assertIsInstance(store, list)

for element in store:

self.assertIsInstance(element, tuple)

self.assertEqual(len(element), 3)

cert, enc, trust = element

self.assertIsInstance(cert, bytes)

self.assertIn(enc, {"x509_asn", "pkcs_7_asn"})

self.assertIsInstance(trust, (frozenset, set, bool))

if isinstance(trust, (frozenset, set)):

trust_oids.update(trust)

serverAuth = "1.3.6.1.5.5.7.3.1"

self.assertIn(serverAuth, trust_oids)

@unittest.skipUnless(sys.platform == "win32", "Windows specific")

def test_enum_crls(self):

self.assertTrue(ssl.enum_crls("CA"))

self.assertRaises(TypeError, ssl.enum_crls)

self.assertRaises(WindowsError, ssl.enum_crls, "")

crls = ssl.enum_crls("CA")

self.assertIsInstance(crls, list)

for element in crls:

self.assertIsInstance(element, tuple)

self.assertEqual(len(element), 2)

self.assertIsInstance(element[0], bytes)

self.assertIn(element[1], {"x509_asn", "pkcs_7_asn"})

def test_asn1object(self):

expected = (129, 'serverAuth', 'TLS Web Server Authentication',

'1.3.6.1.5.5.7.3.1')

val = ssl._ASN1Object('1.3.6.1.5.5.7.3.1')

self.assertEqual(val, expected)

self.assertEqual(val.nid, 129)

self.assertEqual(val.shortname, 'serverAuth')

self.assertEqual(val.longname, 'TLS Web Server Authentication')

self.assertEqual(val.oid, '1.3.6.1.5.5.7.3.1')

self.assertIsInstance(val, ssl._ASN1Object)

self.assertRaises(ValueError, ssl._ASN1Object, 'serverAuth')

val = ssl._ASN1Object.fromnid(129)

self.assertEqual(val, expected)

self.assertIsInstance(val, ssl._ASN1Object)

self.assertRaises(ValueError, ssl._ASN1Object.fromnid, -1)

with self.assertRaisesRegex(ValueError, "unknown NID 100000"):

ssl._ASN1Object.fromnid(100000)

for i in range(1000):

try:

obj = ssl._ASN1Object.fromnid(i)

except ValueError:

pass

else:

self.assertIsInstance(obj.nid, int)

self.assertIsInstance(obj.shortname, str)

self.assertIsInstance(obj.longname, str)

self.assertIsInstance(obj.oid, (str, type(None)))

val = ssl._ASN1Object.fromname('TLS Web Server Authentication')

self.assertEqual(val, expected)

self.assertIsInstance(val, ssl._ASN1Object)

self.assertEqual(ssl._ASN1Object.fromname('serverAuth'), expected)

self.assertEqual(ssl._ASN1Object.fromname('1.3.6.1.5.5.7.3.1'),

expected)

with self.assertRaisesRegex(ValueError, "unknown object 'serverauth'"):

ssl._ASN1Object.fromname('serverauth')

def test_purpose_enum(self):

val = ssl._ASN1Object('1.3.6.1.5.5.7.3.1')

self.assertIsInstance(ssl.Purpose.SERVER_AUTH, ssl._ASN1Object)

self.assertEqual(ssl.Purpose.SERVER_AUTH, val)

self.assertEqual(ssl.Purpose.SERVER_AUTH.nid, 129)

self.assertEqual(ssl.Purpose.SERVER_AUTH.shortname, 'serverAuth')

self.assertEqual(ssl.Purpose.SERVER_AUTH.oid,

'1.3.6.1.5.5.7.3.1')

val = ssl._ASN1Object('1.3.6.1.5.5.7.3.2')

self.assertIsInstance(ssl.Purpose.CLIENT_AUTH, ssl._ASN1Object)

self.assertEqual(ssl.Purpose.CLIENT_AUTH, val)

self.assertEqual(ssl.Purpose.CLIENT_AUTH.nid, 130)

self.assertEqual(ssl.Purpose.CLIENT_AUTH.shortname, 'clientAuth')

self.assertEqual(ssl.Purpose.CLIENT_AUTH.oid,

'1.3.6.1.5.5.7.3.2')

def test_unsupported_dtls(self):

s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

self.addCleanup(s.close)

with self.assertRaises(NotImplementedError) as cx:

test_wrap_socket(s, cert_reqs=ssl.CERT_NONE)

self.assertEqual(str(cx.exception), "only stream sockets are supported")

ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)

with self.assertRaises(NotImplementedError) as cx:

ctx.wrap_socket(s)

self.assertEqual(str(cx.exception), "only stream sockets are supported")

def cert_time_ok(self, timestring, timestamp):

self.assertEqual(ssl.cert_time_to_seconds(timestring), timestamp)

def cert_time_fail(self, timestring):

with self.assertRaises(ValueError):

ssl.cert_time_to_seconds(timestring)

@unittest.skipUnless(utc_offset(),

def test_cert_time_to_seconds_timezone(self):

# Issue #19940: ssl.cert_time_to_seconds() returns wrong

# results if local timezone is not UTC

self.cert_time_ok("May 9 00:00:00 2007 GMT", 1178668800.0)

self.cert_time_ok("Jan 5 09:34:43 2018 GMT", 1515144883.0)

def test_cert_time_to_seconds(self):

timestring = "Jan 5 09:34:43 2018 GMT"

ts = 1515144883.0

self.cert_time_ok(timestring, ts)

# accept keyword parameter, assert its name

self.assertEqual(ssl.cert_time_to_seconds(cert_time=timestring), ts)

# accept both %e and %d (space or zero generated by strftime)

self.cert_time_ok("Jan 05 09:34:43 2018 GMT", ts)

# case-insensitive

self.cert_time_ok("JaN 5 09:34:43 2018 GmT", ts)

self.cert_time_fail("Jan 5 09:34 2018 GMT") # no seconds

self.cert_time_fail("Jan 5 09:34:43 2018") # no GMT

self.cert_time_fail("Jan 5 09:34:43 2018 UTC") # not GMT timezone

self.cert_time_fail("Jan 35 09:34:43 2018 GMT") # invalid day

self.cert_time_fail("Jon 5 09:34:43 2018 GMT") # invalid month

self.cert_time_fail("Jan 5 24:00:00 2018 GMT") # invalid hour

self.cert_time_fail("Jan 5 09:60:43 2018 GMT") # invalid minute

newyear_ts = 1230768000.0

# leap seconds

self.cert_time_ok("Dec 31 23:59:60 2008 GMT", newyear_ts)

# same timestamp

self.cert_time_ok("Jan 1 00:00:00 2009 GMT", newyear_ts)

self.cert_time_ok("Jan 5 09:34:59 2018 GMT", 1515144899)

# allow 60th second (even if it is not a leap second)

self.cert_time_ok("Jan 5 09:34:60 2018 GMT", 1515144900)

# allow 2nd leap second for compatibility with time.strptime()

self.cert_time_ok("Jan 5 09:34:61 2018 GMT", 1515144901)

self.cert_time_fail("Jan 5 09:34:62 2018 GMT") # invalid seconds

# no special treatment for the special value:

# 99991231235959Z (rfc 5280)

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

@support.run_with_locale('LC_ALL', '')

def test_cert_time_to_seconds_locale(self):

# `cert_time_to_seconds()` should be locale independent

def local_february_name():

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

if local_february_name().lower() == 'feb':

self.skipTest("locale-specific month name needs to be "

"different from C locale")

# locale-independent

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

self.cert_time_fail(local_february_name() + " 9 00:00:00 2007 GMT")

def test_connect_ex_error(self):

server = socket.socket(socket.AF_INET)

self.addCleanup(server.close)

port = socket_helper.bind_port(server) # Reserve port but don't listen

s = test_wrap_socket(socket.socket(socket.AF_INET),

cert_reqs=ssl.CERT_REQUIRED)

self.addCleanup(s.close)

rc = s.connect_ex((HOST, port))

# Issue #19919: Windows machines or VMs hosted on Windows

# machines sometimes return EWOULDBLOCK.

errors = (

errno.ECONNREFUSED, errno.EHOSTUNREACH, errno.ETIMEDOUT,

errno.EWOULDBLOCK,

)

self.assertIn(rc, errors)

def test_read_write_zero(self):

# empty reads and writes now work, bpo-42854, bpo-31711

client_context, server_context, hostname = testing_context()

server = ThreadedEchoServer(context=server_context)

with server:

with client_context.wrap_socket(socket.socket(),

server_hostname=hostname) as s:

s.connect((HOST, server.port))

self.assertEqual(s.recv(0), b"")

def test_constructor(self):

for protocol in PROTOCOLS: