def check_openssl_supports_tls_version_1_2(**kwargs):
import ssl
try:
openssl_version_tuple = ssl.OPENSSL_VERSION_INFO
if openssl_version_tuple[0] < 1 or openssl_version_tuple[2] < 1:
warnings.warn(
'Currently installed openssl version: %s does not '
'support TLS 1.2, which is required for use of iot-data. '
'Please use python installed with openssl version 1.0.1 or '
'higher.' % (ssl.OPENSSL_VERSION),
UnsupportedTLSVersionWarning
)
# We cannot check the openssl version on python2.6, so we should just
# pass on this conveniency check.
except AttributeError:
pass
python类OPENSSL_VERSION的实例源码
def check_openssl_supports_tls_version_1_2(**kwargs):
import ssl
try:
openssl_version_tuple = ssl.OPENSSL_VERSION_INFO
if openssl_version_tuple < (1, 0, 1):
warnings.warn(
'Currently installed openssl version: %s does not '
'support TLS 1.2, which is required for use of iot-data. '
'Please use python installed with openssl version 1.0.1 or '
'higher.' % (ssl.OPENSSL_VERSION),
UnsupportedTLSVersionWarning
)
# We cannot check the openssl version on python2.6, so we should just
# pass on this conveniency check.
except AttributeError:
pass
def check_openssl_supports_tls_version_1_2(**kwargs):
import ssl
try:
openssl_version_tuple = ssl.OPENSSL_VERSION_INFO
if openssl_version_tuple[0] < 1 or openssl_version_tuple[2] < 1:
warnings.warn(
'Currently installed openssl version: %s does not '
'support TLS 1.2, which is required for use of iot-data. '
'Please use python installed with openssl version 1.0.1 or '
'higher.' % (ssl.OPENSSL_VERSION),
UnsupportedTLSVersionWarning
)
# We cannot check the openssl version on python2.6, so we should just
# pass on this conveniency check.
except AttributeError:
pass
def run_regrtests(*extra_args):
print(ssl.OPENSSL_VERSION)
args = [
sys.executable,
'-Werror', '-bb', # turn warnings into exceptions
'-m', 'test.regrtest',
]
if not extra_args:
args.extend([
'-r', # randomize
'-w', # re-run failed tests with -v
'-u', 'network', # use network
'-u', 'urlfetch', # download test vectors
])
else:
args.extend(extra_args)
args.extend(TESTS)
result = subprocess.call(args)
sys.exit(result)
def run_regrtests(*extra_args):
print(ssl.OPENSSL_VERSION)
args = [
sys.executable,
'-Werror', '-bb', # turn warnings into exceptions
'-m', 'test.regrtest',
]
if not extra_args:
args.extend([
'-r', # randomize
'-w', # re-run failed tests with -v
'-u', 'network', # use network
'-u', 'urlfetch', # download test vectors
])
else:
args.extend(extra_args)
args.extend(TESTS)
result = subprocess.call(args)
sys.exit(result)
def check_openssl_supports_tls_version_1_2(**kwargs):
import ssl
try:
openssl_version_tuple = ssl.OPENSSL_VERSION_INFO
if openssl_version_tuple[0] < 1 or openssl_version_tuple[2] < 1:
warnings.warn(
'Currently installed openssl version: %s does not '
'support TLS 1.2, which is required for use of iot-data. '
'Please use python installed with openssl version 1.0.1 or '
'higher.' % (ssl.OPENSSL_VERSION),
UnsupportedTLSVersionWarning
)
# We cannot check the openssl version on python2.6, so we should just
# pass on this conveniency check.
except AttributeError:
pass
def init_mqtt_client(self):
endpoint = self.iot_client.describe_endpoint()
use_websocket = True if self.credentials else False
endpoint_port = 443 if use_websocket else 8883
self.mqtt_client = AWSIoTMQTTClient("testo", useWebsocket=use_websocket)
self.mqtt_client.configureEndpoint(endpoint['endpointAddress'], endpoint_port)
self.mqtt_client.configureOfflinePublishQueueing(-1)
self.mqtt_client.configureConnectDisconnectTimeout(10)
self.mqtt_client.configureMQTTOperationTimeout(10)
self.configure_credentials()
log.debug("OpenSSL version {}".format(ssl.OPENSSL_VERSION))
log.debug("Connecting MQTT client to {} on port {}...".format(endpoint['endpointAddress'], endpoint_port))
try:
self.mqtt_client.connect()
log.debug("MQTT client connected")
except connectTimeoutException:
log.error("Failed to connect MQTT client - timeout (check policy)")
self.mqtt_client = None
def init_mqtt_client(self):
endpoint = self.iot_client.describe_endpoint()
use_websocket = True if self.credentials else False
endpoint_port = 443 if use_websocket else 8883
self.mqtt_client = AWSIoTMQTTClient("testo", useWebsocket=use_websocket)
self.mqtt_client.configureEndpoint(endpoint['endpointAddress'], endpoint_port)
self.mqtt_client.configureOfflinePublishQueueing(-1)
self.mqtt_client.configureConnectDisconnectTimeout(10)
self.mqtt_client.configureMQTTOperationTimeout(10)
self.configure_credentials()
log.debug("OpenSSL version {}".format(ssl.OPENSSL_VERSION))
log.debug("Connecting MQTT client to {} on port {}...".format(endpoint['endpointAddress'], endpoint_port))
try:
self.mqtt_client.connect()
log.debug("MQTT client connected")
except connectTimeoutException:
log.error("Failed to connect MQTT client - timeout (check policy)")
self.mqtt_client = None
def check_openssl_supports_tls_version_1_2(**kwargs):
import ssl
try:
openssl_version_tuple = ssl.OPENSSL_VERSION_INFO
if openssl_version_tuple[0] < 1 or openssl_version_tuple[2] < 1:
warnings.warn(
'Currently installed openssl version: %s does not '
'support TLS 1.2, which is required for use of iot-data. '
'Please use python installed with openssl version 1.0.1 or '
'higher.' % (ssl.OPENSSL_VERSION),
UnsupportedTLSVersionWarning
)
# We cannot check the openssl version on python2.6, so we should just
# pass on this conveniency check.
except AttributeError:
pass
def check_openssl_supports_tls_version_1_2(**kwargs):
import ssl
try:
openssl_version_tuple = ssl.OPENSSL_VERSION_INFO
if openssl_version_tuple[0] < 1 or openssl_version_tuple[2] < 1:
warnings.warn(
'Currently installed openssl version: %s does not '
'support TLS 1.2, which is required for use of iot-data. '
'Please use python installed with openssl version 1.0.1 or '
'higher.' % (ssl.OPENSSL_VERSION),
UnsupportedTLSVersionWarning
)
# We cannot check the openssl version on python2.6, so we should just
# pass on this conveniency check.
except AttributeError:
pass
handlers.py 文件源码
项目:tf_aws_ecs_instance_draining_on_scale_in
作者: terraform-community-modules
项目源码
文件源码
阅读 22
收藏 0
点赞 0
评论 0
def check_openssl_supports_tls_version_1_2(**kwargs):
import ssl
try:
openssl_version_tuple = ssl.OPENSSL_VERSION_INFO
if openssl_version_tuple[0] < 1 or openssl_version_tuple[2] < 1:
warnings.warn(
'Currently installed openssl version: %s does not '
'support TLS 1.2, which is required for use of iot-data. '
'Please use python installed with openssl version 1.0.1 or '
'higher.' % (ssl.OPENSSL_VERSION),
UnsupportedTLSVersionWarning
)
# We cannot check the openssl version on python2.6, so we should just
# pass on this conveniency check.
except AttributeError:
pass
def test_ssl_ok_using_system_ca(self):
try:
if 'OpenSSL' in ssl.OPENSSL_VERSION:
self._test_ssl(use_system_ca_bundle=True)
else:
raise common.Skipped("OpenSSL not available.")
except SSLUnavailable:
raise common.Skipped("SSL not available.")
def ssl_version_check():
from ssl import OPENSSL_VERSION_INFO as sslv, OPENSSL_VERSION
if not (sslv[0] > 1 or
(sslv[0] == 1 and sslv[1] > 0) or
(sslv[0] == 1 and sslv[1] == 0 and sslv[2] > 0)):
raise Exception('At least SSL v1.0.1 required for TLS v1.2 (found %s)' % OPENSSL_VERSION)
# callback function check helper
def __init__(self):
system = sys.platform.lower()
if system == 'darwin':
self.platform = "mac"
elif system.startswith('linux'):
self.platform = 'linux'
elif system == 'win32':
self.platform = 'windows'
else:
self.platform = None
if self.platform:
self.local_path_field = "local_path_%s" % (self.platform)
else:
self.local_path_field = None
self.py_version = ".".join(str(x) for x in sys.version_info[:2])
# extract the OpenSSL version if we can. The version is only available in Python 2.7 and
# only if we successfully imported ssl
self.ssl_version = "unknown"
try:
self.ssl_version = ssl.OPENSSL_VERSION
except (AttributeError, NameError):
pass