# Test the support for SSL and sockets

import contextlib

import sys

import unittest

import unittest.mock

from ast import literal_eval

from threading import Thread

from test import support

from test.support import import_helper

from test.support import os_helper

from test.support import socket_helper

from test.support import threading_helper

from test.support import warnings_helper

from test.support import asyncore

import array

import re

import socket

import select

import struct

import time

import enum

import gc

import http.client

import os

import errno

import pprint

import urllib.request

import threading

import traceback

import weakref

import platform

import sysconfig

import functools

from contextlib import nullcontext

try:

import ctypes

except ImportError:

ctypes = None

ssl = import_helper.import_module("ssl")

import _ssl

from ssl import Purpose, TLSVersion, _TLSContentType, _TLSMessageType, _TLSAlertType

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

PROTOCOLS = sorted(ssl._PROTOCOL_NAMES)

HOST = socket_helper.HOST

IS_AWS_LC = "AWS-LC" in ssl.OPENSSL_VERSION

IS_OPENSSL_3_0_0 = ssl.OPENSSL_VERSION_INFO >= (3, 0, 0)

CAN_GET_SELECTED_OPENSSL_GROUP = ssl.OPENSSL_VERSION_INFO >= (3, 2)

CAN_IGNORE_UNKNOWN_OPENSSL_GROUPS = ssl.OPENSSL_VERSION_INFO >= (3, 3)

CAN_GET_AVAILABLE_OPENSSL_GROUPS = ssl.OPENSSL_VERSION_INFO >= (3, 5)

CAN_GET_AVAILABLE_OPENSSL_SIGALGS = ssl.OPENSSL_VERSION_INFO >= (3, 4)

CAN_SET_CLIENT_SIGALGS = not IS_AWS_LC

CAN_IGNORE_UNKNOWN_OPENSSL_SIGALGS = ssl.OPENSSL_VERSION_INFO >= (3, 3)

CAN_GET_SELECTED_OPENSSL_SIGALG = ssl.OPENSSL_VERSION_INFO >= (3, 5)

PY_SSL_DEFAULT_CIPHERS = sysconfig.get_config_var('PY_SSL_DEFAULT_CIPHERS')

HAS_KEYLOG = hasattr(ssl.SSLContext, 'keylog_filename')

requires_keylog = unittest.skipUnless(

HAS_KEYLOG, 'test requires OpenSSL 1.1.1 with keylog callback')

CAN_SET_KEYLOG = HAS_KEYLOG and os.name != "nt"

requires_keylog_setter = unittest.skipUnless(

CAN_SET_KEYLOG,

"cannot set 'keylog_filename' on Windows"

)

PROTOCOL_TO_TLS_VERSION = {}

for proto, ver in (

("PROTOCOL_SSLv3", "SSLv3"),

("PROTOCOL_TLSv1", "TLSv1"),

("PROTOCOL_TLSv1_1", "TLSv1_1"),

):

try:

proto = getattr(ssl, proto)

ver = getattr(ssl.TLSVersion, ver)

except AttributeError:

continue

PROTOCOL_TO_TLS_VERSION[proto] = ver

def data_file(*name):

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

# The custom key and certificate files used in test_ssl are generated

# using Lib/test/certdata/make_ssl_certs.py.

# Other certificates are simply fetched from the internet servers they

# are meant to authenticate.

CERTFILE = data_file("keycert.pem")

BYTES_CERTFILE = os.fsencode(CERTFILE)

ONLYCERT = data_file("ssl_cert.pem")

ONLYKEY = data_file("ssl_key.pem")

BYTES_ONLYCERT = os.fsencode(ONLYCERT)

BYTES_ONLYKEY = os.fsencode(ONLYKEY)

CERTFILE_PROTECTED = data_file("keycert.passwd.pem")

ONLYKEY_PROTECTED = data_file("ssl_key.passwd.pem")

KEY_PASSWORD = "somepass"

CAPATH = data_file("capath")

BYTES_CAPATH = os.fsencode(CAPATH)

CAFILE_NEURONIO = data_file("capath", "4e1295a3.0")

CAFILE_CACERT = data_file("capath", "5ed36f99.0")

with open(data_file('keycert.pem.reference')) as file:

CERTFILE_INFO = literal_eval(file.read())

# empty CRL

CRLFILE = data_file("revocation.crl")

# Two keys and certs signed by the same CA (for SNI tests)

SIGNED_CERTFILE = data_file("keycert3.pem")

SINGED_CERTFILE_ONLY = data_file("cert3.pem")

SIGNED_CERTFILE_HOSTNAME = 'localhost'

with open(data_file('keycert3.pem.reference')) as file:

SIGNED_CERTFILE_INFO = literal_eval(file.read())

SIGNED_CERTFILE2 = data_file("keycert4.pem")

SIGNED_CERTFILE2_HOSTNAME = 'fakehostname'

SIGNED_CERTFILE_ECC = data_file("keycertecc.pem")

SIGNED_CERTFILE_ECC_HOSTNAME = 'localhost-ecc'

# A custom testcase, extracted from `rfc5280::aki::leaf-missing-aki` in x509-limbo:

# The leaf (server) certificate has no AKI, which is forbidden under RFC 5280.

# See: https://x509-limbo.com/testcases/rfc5280/#rfc5280akileaf-missing-aki

LEAF_MISSING_AKI_CERTFILE = data_file("leaf-missing-aki.keycert.pem")

LEAF_MISSING_AKI_CERTFILE_HOSTNAME = "example.com"

LEAF_MISSING_AKI_CA = data_file("leaf-missing-aki.ca.pem")

# Same certificate as pycacert.pem, but without extra text in file

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

# cert with all kinds of subject alt names

ALLSANFILE = data_file("allsans.pem")

IDNSANSFILE = data_file("idnsans.pem")

NOSANFILE = data_file("nosan.pem")

NOSAN_HOSTNAME = 'localhost'

REMOTE_HOST = "self-signed.pythontest.net"

EMPTYCERT = data_file("nullcert.pem")

BADCERT = data_file("badcert.pem")

NONEXISTINGCERT = data_file("XXXnonexisting.pem")

NONEXISTINGKEY = data_file("XXXnonexistingkey.pem")

BADKEY = data_file("badkey.pem")

NOKIACERT = data_file("nokia.pem")

NULLBYTECERT = data_file("nullbytecert.pem")

TALOS_INVALID_CRLDP = data_file("talos-2019-0758.pem")

DHFILE = data_file("ffdh3072.pem")

BYTES_DHFILE = os.fsencode(DHFILE)

# Not defined in all versions of OpenSSL

OP_NO_COMPRESSION = getattr(ssl, "OP_NO_COMPRESSION", 0)

OP_SINGLE_DH_USE = getattr(ssl, "OP_SINGLE_DH_USE", 0)

OP_SINGLE_ECDH_USE = getattr(ssl, "OP_SINGLE_ECDH_USE", 0)

OP_CIPHER_SERVER_PREFERENCE = getattr(ssl, "OP_CIPHER_SERVER_PREFERENCE", 0)

OP_ENABLE_MIDDLEBOX_COMPAT = getattr(ssl, "OP_ENABLE_MIDDLEBOX_COMPAT", 0)

# Ubuntu has patched OpenSSL and changed behavior of security level 2

# see https://bugs.python.org/issue41561#msg389003

def is_ubuntu():

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:

return False

if is_ubuntu():

def seclevel_workaround(*ctxs):

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

):

ctx.set_ciphers("@SECLEVEL=1:ALL")

else:

def seclevel_workaround(*ctxs):

pass

def has_tls_protocol(protocol):

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

:param protocol: enum ssl._SSLMethod member or name

:return: bool

"""

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

return has_tls_version(name[len('PROTOCOL_'):])

@functools.lru_cache

def has_tls_version(version):

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

:param version: TLS version name or ssl.TLSVersion member

:return: bool

"""

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

return True

def requires_tls_version(version):

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

:param version: TLS version name or ssl.TLSVersion member

:return:

"""

def decorator(func):

@functools.wraps(func)

def wrapper(*args, **kw):

if not has_tls_version(version):

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

else:

return func(*args, **kw)

return wrapper

return decorator

def handle_error(prefix):

exc_format = ' '.join(traceback.format_exception(sys.exception()))

if support.verbose:

sys.stdout.write(prefix + exc_format)

def utc_offset(): #NOTE: ignore issues like #1647654

# local time = utc time + utc offset

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

return -time.altzone # seconds

return -time.timezone

ignore_deprecation = warnings_helper.ignore_warnings(

category=DeprecationWarning

)

def make_test_context(

*,

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

return context

def test_wrap_socket(

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)

return context.wrap_socket(sock, server_side=server_side, **kwargs)

USE_SAME_TEST_CONTEXT = False

_TEST_CONTEXT = None

def testing_context(server_cert=SIGNED_CERTFILE, *, server_chain=True,

client_cert=None):

"""Create context

client_context, server_context, hostname = testing_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

return client_context, server_context, hostname

def do_ssl_object_handshake(sslobject, outgoing, max_retry=25):

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

If do_handshake() fails more than *max_retry* times, return None.

"""

data, attempt = None, 0

while not data and attempt < max_retry:

with contextlib.suppress(ssl.SSLWantReadError):

sslobject.do_handshake()

data = outgoing.read()

attempt += 1

return data

class BasicSocketTests(unittest.TestCase):

def test_constants(self):

ssl.CERT_NONE

ssl.CERT_OPTIONAL

ssl.CERT_REQUIRED

ssl.OP_CIPHER_SERVER_PREFERENCE

ssl.OP_SINGLE_DH_USE

ssl.OP_SINGLE_ECDH_USE

ssl.OP_NO_COMPRESSION

self.assertEqual(ssl.HAS_SNI, True)

self.assertEqual(ssl.HAS_ECDH, True)

self.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', '[email protected]'),))

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', '[email protected]\[email protected]'),

('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', '[email protected]\[email protected]'),

('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', '[email protected]'),

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

"'tls-unique' channel binding not available")

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(),

'local time needs to be different from UTC')

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

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

class ContextTests(unittest.TestCase):

def test_constructor(self):

for protocol in PROTOCOLS:

if has_tls_protocol(protocol):

with warnings_helper.check_warnings():

ctx = ssl.SSLContext(protocol)