# local time = utc time + utc offset

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

return -time.altzone # seconds

# Some versions of OpenSSL ignore seconds, see #18207

# 0.9.8.i

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

fmt = "%b %d %H:%M:%S %Y GMT"

dt = datetime.datetime.strptime(cert_time, fmt)

dt = dt.replace(second=0)

cert_time = dt.strftime(fmt)

# %d adds leading zero but ASN1_TIME_print() uses leading space

if cert_time[4] == "0":

cert_time = cert_time[:4] + " " + cert_time[5:]

if hasattr(ssl, 'PROTOCOL_SSLv2'):

@functools.wraps(func)

def f(*args, **kwargs):

try:

ssl.SSLContext(ssl.PROTOCOL_SSLv2)

except ssl.SSLError:

if (ssl.OPENSSL_VERSION_INFO == (0, 9, 8, 15, 15) and

platform.linux_distribution() == ('debian', 'squeeze/sid', '')):

raise unittest.SkipTest("Patched Ubuntu OpenSSL breaks behaviour")

return func(*args, **kwargs)

return f

else:

cert_reqs=ssl.CERT_NONE, ca_certs=None,

ciphers=None, certfile=None, keyfile=None,

**kwargs):

context = ssl.SSLContext(ssl_version)

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)

def test_constants(self):

ssl.CERT_NONE

ssl.CERT_OPTIONAL

ssl.CERT_REQUIRED

ssl.OP_CIPHER_SERVER_PREFERENCE

ssl.OP_SINGLE_DH_USE

if ssl.HAS_ECDH:

ssl.OP_SINGLE_ECDH_USE

if ssl.OPENSSL_VERSION_INFO >= (1, 0):

ssl.OP_NO_COMPRESSION

self.assertIn(ssl.HAS_SNI, {True, False})

self.assertIn(ssl.HAS_ECDH, {True, False})

def test_str_for_enums(self):

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

# reprs.

proto = ssl.PROTOCOL_TLS

self.assertEqual(str(proto), '_SSLMethod.PROTOCOL_TLS')

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

data, is_cryptographic = ssl.RAND_pseudo_bytes(16)

self.assertEqual(len(data), 16)

self.assertEqual(is_cryptographic, v == 1)

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)

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

if hasattr(ssl, 'RAND_egd'):

self.assertRaises(TypeError, ssl.RAND_egd, 1)

self.assertRaises(TypeError, ssl.RAND_egd, 'foo', 1)

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)

@unittest.skipUnless(os.name == 'posix', 'requires posix')

def test_random_fork(self):

status = ssl.RAND_status()

if not status:

self.fail("OpenSSL's PRNG has insufficient randomness")

rfd, wfd = os.pipe()

pid = os.fork()

if pid == 0:

try:

os.close(rfd)

child_random = ssl.RAND_pseudo_bytes(16)[0]

self.assertEqual(len(child_random), 16)

os.write(wfd, child_random)

os.close(wfd)

except BaseException:

os._exit(1)

else:

os._exit(0)

else:

os.close(wfd)

self.addCleanup(os.close, rfd)

_, status = os.waitpid(pid, 0)

self.assertEqual(status, 0)

child_random = os.read(rfd, 16)

self.assertEqual(len(child_random), 16)

parent_random = ssl.RAND_pseudo_bytes(16)[0]

self.assertEqual(len(parent_random), 16)

self.assertNotEqual(child_random, parent_random)

def test_parse_cert(self):

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

# provided solely for this test, to exercise the certificate

# parsing code

p = ssl._ssl._test_decode_cert(CERTFILE)

if support.verbose:

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

self.assertEqual(p['issuer'],

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

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

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

(('commonName', 'localhost'),))

)

# Note the next three asserts will fail if the keys are regenerated

self.assertEqual(p['notAfter'], asn1time('Oct 5 23:01:56 2020 GMT'))

self.assertEqual(p['notBefore'], asn1time('Oct 8 23:01:56 2010 GMT'))

self.assertEqual(p['serialNumber'], 'D7C7381919AFC24E')

self.assertEqual(p['subject'],

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

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

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

(('commonName', 'localhost'),))

)

self.assertEqual(p['subjectAltName'], (('DNS', 'localhost'),))

# 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_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\n'))

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\n'),

('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

# >= 0.9

self.assertGreaterEqual(n, 0x900000)

# < 3.0

self.assertLess(n, 0x30000000)

major, minor, fix, patch, status = t

self.assertGreaterEqual(major, 0)

self.assertLess(major, 3)

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)

# Version string as returned by {Open,Libre}SSL, the format might change

if IS_LIBRESSL:

self.assertTrue(s.startswith("LibreSSL {:d}".format(major)),

(s, t, hex(n)))

else:

self.assertTrue(s.startswith("OpenSSL {:d}.{:d}.{:d}".format(major, minor, fix)),

(s, 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 support.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))

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_errors_sslwrap(self):

sock = socket.socket()

self.assertRaisesRegex(ValueError,

"certfile must be specified",

ssl.wrap_socket, sock, keyfile=CERTFILE)

self.assertRaisesRegex(ValueError,

"certfile must be specified for server-side operations",

ssl.wrap_socket, sock, server_side=True)

self.assertRaisesRegex(ValueError,

"certfile must be specified for server-side operations",

ssl.wrap_socket, sock, server_side=True, certfile="")

with ssl.wrap_socket(sock, server_side=True, certfile=CERTFILE) as s:

self.assertRaisesRegex(ValueError, "can't connect in server-side mode",

s.connect, (HOST, 8080))

with self.assertRaises(OSError) as cm:

with socket.socket() as sock:

ssl.wrap_socket(sock, certfile=NONEXISTINGCERT)

self.assertEqual(cm.exception.errno, errno.ENOENT)

with self.assertRaises(OSError) as cm:

with socket.socket() as sock:

ssl.wrap_socket(sock,

certfile=CERTFILE, keyfile=NONEXISTINGCERT)

self.assertEqual(cm.exception.errno, errno.ENOENT)

with self.assertRaises(OSError) as cm:

with socket.socket() as sock:

ssl.wrap_socket(sock,

certfile=NONEXISTINGCERT, keyfile=NONEXISTINGCERT)

self.assertEqual(cm.exception.errno, errno.ENOENT)

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,

certfile)

sock = socket.socket()

self.addCleanup(sock.close)

with self.assertRaises(ssl.SSLError):

test_wrap_socket(sock,

certfile=certfile,

ssl_version=ssl.PROTOCOL_TLSv1)

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_match_hostname(self):

def ok(cert, hostname):

ssl.match_hostname(cert, hostname)

def fail(cert, hostname):

self.assertRaises(ssl.CertificateError,

ssl.match_hostname, cert, hostname)

# -- Hostname matching --

cert = {'subject': ((('commonName', 'example.com'),),)}

ok(cert, 'example.com')

ok(cert, 'ExAmple.cOm')

fail(cert, 'www.example.com')

fail(cert, '.example.com')

fail(cert, 'example.org')

fail(cert, 'exampleXcom')

cert = {'subject': ((('commonName', '*.a.com'),),)}

ok(cert, 'foo.a.com')

fail(cert, 'bar.foo.a.com')

fail(cert, 'a.com')

fail(cert, 'Xa.com')

fail(cert, '.a.com')

# only match one left-most wildcard

cert = {'subject': ((('commonName', 'f*.com'),),)}

ok(cert, 'foo.com')

ok(cert, 'f.com')

fail(cert, 'bar.com')

fail(cert, 'foo.a.com')

fail(cert, 'bar.foo.com')

# NULL bytes are bad, CVE-2013-4073

cert = {'subject': ((('commonName',

'null.python.org\x00example.org'),),)}

ok(cert, 'null.python.org\x00example.org') # or raise an error?

fail(cert, 'example.org')

fail(cert, 'null.python.org')

# error cases with wildcards

cert = {'subject': ((('commonName', '*.*.a.com'),),)}

fail(cert, 'bar.foo.a.com')

fail(cert, 'a.com')

fail(cert, 'Xa.com')

fail(cert, '.a.com')

cert = {'subject': ((('commonName', 'a.*.com'),),)}

fail(cert, 'a.foo.com')

fail(cert, 'a..com')

fail(cert, 'a.com')

# wildcard doesn't match IDNA prefix 'xn--'

idna = 'püthon.python.org'.encode("idna").decode("ascii")

cert = {'subject': ((('commonName', idna),),)}

ok(cert, idna)

cert = {'subject': ((('commonName', 'x*.python.org'),),)}

fail(cert, idna)

cert = {'subject': ((('commonName', 'xn--p*.python.org'),),)}

fail(cert, idna)

# wildcard in first fragment and IDNA A-labels in sequent fragments

# are supported.

idna = 'www*.pythön.org'.encode("idna").decode("ascii")

cert = {'subject': ((('commonName', idna),),)}

ok(cert, 'www.pythön.org'.encode("idna").decode("ascii"))

ok(cert, 'www1.pythön.org'.encode("idna").decode("ascii"))

fail(cert, 'ftp.pythön.org'.encode("idna").decode("ascii"))

fail(cert, 'pythön.org'.encode("idna").decode("ascii"))

# Slightly fake real-world example

cert = {'notAfter': 'Jun 26 21:41:46 2011 GMT',

'subject': ((('commonName', 'linuxfrz.org'),),),

'subjectAltName': (('DNS', 'linuxfr.org'),

('DNS', 'linuxfr.com'),

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

ok(cert, 'linuxfr.org')

ok(cert, 'linuxfr.com')

# Not a "DNS" entry

fail(cert, '<unsupported>')

# When there is a subjectAltName, commonName isn't used

fail(cert, 'linuxfrz.org')

# A pristine real-world example

cert = {'notAfter': 'Dec 18 23:59:59 2011 GMT',

'subject': ((('countryName', 'US'),),

(('stateOrProvinceName', 'California'),),

(('localityName', 'Mountain View'),),

(('organizationName', 'Google Inc'),),

(('commonName', 'mail.google.com'),))}

ok(cert, 'mail.google.com')

fail(cert, 'gmail.com')

# Only commonName is considered

fail(cert, 'California')

# -- IPv4 matching --

cert = {'subject': ((('commonName', 'example.com'),),),

'subjectAltName': (('DNS', 'example.com'),

('IP Address', '10.11.12.13'),

('IP Address', '14.15.16.17'))}

ok(cert, '10.11.12.13')

ok(cert, '14.15.16.17')

fail(cert, '14.15.16.18')

fail(cert, 'example.net')

# -- IPv6 matching --

cert = {'subject': ((('commonName', 'example.com'),),),

'subjectAltName': (('DNS', 'example.com'),

('IP Address', '2001:0:0:0:0:0:0:CAFE\n'),

('IP Address', '2003:0:0:0:0:0:0:BABA\n'))}

ok(cert, '2001::cafe')

ok(cert, '2003::baba')

fail(cert, '2003::bebe')

fail(cert, 'example.net')

# -- Miscellaneous --

# Neither commonName nor subjectAltName

cert = {'notAfter': 'Dec 18 23:59:59 2011 GMT',

'subject': ((('countryName', 'US'),),

(('stateOrProvinceName', 'California'),),

(('localityName', 'Mountain View'),),

(('organizationName', 'Google Inc'),))}

fail(cert, 'mail.google.com')

# No DNS entry in subjectAltName but a commonName

cert = {'notAfter': 'Dec 18 23:59:59 2099 GMT',

'subject': ((('countryName', 'US'),),

(('stateOrProvinceName', 'California'),),

(('localityName', 'Mountain View'),),

(('commonName', 'mail.google.com'),)),

'subjectAltName': (('othername', 'blabla'), )}

ok(cert, 'mail.google.com')

# No DNS entry subjectAltName and no commonName

cert = {'notAfter': 'Dec 18 23:59:59 2099 GMT',

'subject': ((('countryName', 'US'),),

(('stateOrProvinceName', 'California'),),

(('localityName', 'Mountain View'),),

(('organizationName', 'Google Inc'),)),

'subjectAltName': (('othername', 'blabla'),)}

fail(cert, 'google.com')

# Empty cert / no cert

self.assertRaises(ValueError, ssl.match_hostname, None, 'example.com')

self.assertRaises(ValueError, ssl.match_hostname, {}, 'example.com')

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

# wildcard per fragment.

cert = {'subject': ((('commonName', 'a*b.com'),),)}

ok(cert, 'axxb.com')

cert = {'subject': ((('commonName', 'a*b.co*'),),)}

fail(cert, 'axxb.com')

cert = {'subject': ((('commonName', 'a*b*.com'),),)}

with self.assertRaises(ssl.CertificateError) as cm:

ssl.match_hostname(cert, 'axxbxxc.com')

self.assertIn("too many wildcards", str(cm.exception))

def test_server_side(self):

# server_hostname doesn't work for server sockets

ctx = ssl.SSLContext(ssl.PROTOCOL_SSLv23)

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.socket(socket.AF_INET)

s.bind(('127.0.0.1', 0))

s.listen()

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 support.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, (set, bool))

if isinstance(trust, 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_SSLv23)

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 treatement 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 = support.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,

)

@skip_if_broken_ubuntu_ssl

def test_constructor(self):

for protocol in PROTOCOLS:

ssl.SSLContext(protocol)

ctx = ssl.SSLContext()

self.assertEqual(ctx.protocol, ssl.PROTOCOL_TLS)

self.assertRaises(ValueError, ssl.SSLContext, -1)

self.assertRaises(ValueError, ssl.SSLContext, 42)

@skip_if_broken_ubuntu_ssl

def test_protocol(self):

for proto in PROTOCOLS:

ctx = ssl.SSLContext(proto)

self.assertEqual(ctx.protocol, proto)

def test_ciphers(self):

ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1)

ctx.set_ciphers("ALL")

ctx.set_ciphers("DEFAULT")

with self.assertRaisesRegex(ssl.SSLError, "No cipher can be selected"):

ctx.set_ciphers("^$:,;?*'dorothyx")

@unittest.skipIf(ssl.OPENSSL_VERSION_INFO < (1, 0, 2, 0, 0), 'OpenSSL too old')

def test_get_ciphers(self):

ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1)

ctx.set_ciphers('AESGCM')

names = set(d['name'] for d in ctx.get_ciphers())

self.assertIn('AES256-GCM-SHA384', names)

self.assertIn('AES128-GCM-SHA256', names)

@skip_if_broken_ubuntu_ssl

def test_options(self):

ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1)

# OP_ALL | OP_NO_SSLv2 | OP_NO_SSLv3 is the default value

default = (ssl.OP_ALL | ssl.OP_NO_SSLv2 | ssl.OP_NO_SSLv3)

# SSLContext also enables these by default

default |= (OP_NO_COMPRESSION | OP_CIPHER_SERVER_PREFERENCE |

OP_SINGLE_DH_USE | OP_SINGLE_ECDH_USE)

self.assertEqual(default, ctx.options)

ctx.options |= ssl.OP_NO_TLSv1

self.assertEqual(default | ssl.OP_NO_TLSv1, ctx.options)

if can_clear_options():

ctx.options = (ctx.options & ~ssl.OP_NO_TLSv1)

self.assertEqual(default, ctx.options)

ctx.options = 0

# Ubuntu has OP_NO_SSLv3 forced on by default

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

else:

with self.assertRaises(ValueError):

ctx.options = 0

def test_verify_mode(self):

ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1)

# Default value

self.assertEqual(ctx.verify_mode, ssl.CERT_NONE)

ctx.verify_mode = ssl.CERT_OPTIONAL

self.assertEqual(ctx.verify_mode, ssl.CERT_OPTIONAL)

ctx.verify_mode = ssl.CERT_REQUIRED

self.assertEqual(ctx.verify_mode, ssl.CERT_REQUIRED)

ctx.verify_mode = ssl.CERT_NONE

self.assertEqual(ctx.verify_mode, ssl.CERT_NONE)

with self.assertRaises(TypeError):

ctx.verify_mode = None

with self.assertRaises(ValueError):

ctx.verify_mode = 42

@unittest.skipUnless(have_verify_flags(),

def test_verify_flags(self):

ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1)

# default value

tf = getattr(ssl, "VERIFY_X509_TRUSTED_FIRST", 0)

self.assertEqual(ctx.verify_flags, ssl.VERIFY_DEFAULT | tf)

ctx.verify_flags = ssl.VERIFY_CRL_CHECK_LEAF

self.assertEqual(ctx.verify_flags, ssl.VERIFY_CRL_CHECK_LEAF)

ctx.verify_flags = ssl.VERIFY_CRL_CHECK_CHAIN

self.assertEqual(ctx.verify_flags, ssl.VERIFY_CRL_CHECK_CHAIN)

ctx.verify_flags = ssl.VERIFY_DEFAULT

self.assertEqual(ctx.verify_flags, ssl.VERIFY_DEFAULT)

# supports any value

ctx.verify_flags = ssl.VERIFY_CRL_CHECK_LEAF | ssl.VERIFY_X509_STRICT

self.assertEqual(ctx.verify_flags,

ssl.VERIFY_CRL_CHECK_LEAF | ssl.VERIFY_X509_STRICT)

with self.assertRaises(TypeError):

ctx.verify_flags = None

def test_load_cert_chain(self):

ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1)

# Combined key and cert in a single file

ctx.load_cert_chain(CERTFILE, keyfile=None)

ctx.load_cert_chain(CERTFILE, keyfile=CERTFILE)

self.assertRaises(TypeError, ctx.load_cert_chain, keyfile=CERTFILE)

with self.assertRaises(OSError) as cm:

ctx.load_cert_chain(NONEXISTINGCERT)

self.assertEqual(cm.exception.errno, errno.ENOENT)

with self.assertRaisesRegex(ssl.SSLError, "PEM lib"):

ctx.load_cert_chain(BADCERT)

with self.assertRaisesRegex(ssl.SSLError, "PEM lib"):

ctx.load_cert_chain(EMPTYCERT)

# Separate key and cert

ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1)

ctx.load_cert_chain(ONLYCERT, ONLYKEY)

ctx.load_cert_chain(certfile=ONLYCERT, keyfile=ONLYKEY)

ctx.load_cert_chain(certfile=BYTES_ONLYCERT, keyfile=BYTES_ONLYKEY)

with self.assertRaisesRegex(ssl.SSLError, "PEM lib"):

ctx.load_cert_chain(ONLYCERT)

with self.assertRaisesRegex(ssl.SSLError, "PEM lib"):

ctx.load_cert_chain(ONLYKEY)

with self.assertRaisesRegex(ssl.SSLError, "PEM lib"):

ctx.load_cert_chain(certfile=ONLYKEY, keyfile=ONLYCERT)

# Mismatching key and cert

ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1)

with self.assertRaisesRegex(ssl.SSLError, "key values mismatch"):

ctx.load_cert_chain(CAFILE_CACERT, ONLYKEY)

# Password protected key and cert

ctx.load_cert_chain(CERTFILE_PROTECTED, password=KEY_PASSWORD)

ctx.load_cert_chain(CERTFILE_PROTECTED, password=KEY_PASSWORD.encode())

ctx.load_cert_chain(CERTFILE_PROTECTED,

password=bytearray(KEY_PASSWORD.encode()))

ctx.load_cert_chain(ONLYCERT, ONLYKEY_PROTECTED, KEY_PASSWORD)

ctx.load_cert_chain(ONLYCERT, ONLYKEY_PROTECTED, KEY_PASSWORD.encode())

ctx.load_cert_chain(ONLYCERT, ONLYKEY_PROTECTED,

bytearray(KEY_PASSWORD.encode()))

with self.assertRaisesRegex(TypeError, "should be a string"):

ctx.load_cert_chain(CERTFILE_PROTECTED, password=True)

with self.assertRaises(ssl.SSLError):

ctx.load_cert_chain(CERTFILE_PROTECTED, password="badpass")

with self.assertRaisesRegex(ValueError, "cannot be longer"):

# openssl has a fixed limit on the password buffer.

# PEM_BUFSIZE is generally set to 1kb.

# Return a string larger than this.

ctx.load_cert_chain(CERTFILE_PROTECTED, password=b'a' * 102400)

# Password callback

def getpass_unicode():

return KEY_PASSWORD

def getpass_bytes():

return KEY_PASSWORD.encode()

def getpass_bytearray():

return bytearray(KEY_PASSWORD.encode())

def getpass_badpass():

return "badpass"

def getpass_huge():

return b'a' * (1024 * 1024)

def getpass_bad_type():