Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_with_mtls_pkcs12(self):
opt = TlsContextOptions.create_client_with_mtls_pkcs12(
'test/resources/unittests.p12', '1234')
ctx = ClientTlsContext(opt)
def test_init_defaults(self):
opt = TlsContextOptions()
ctx = ClientTlsContext(opt)
def test_override_default_trust_store_dir(self):
opt = TlsContextOptions()
opt.override_default_trust_store_from_path('test/resources', None)
ctx = ClientTlsContext(opt)
def _new_client_connection(self, secure, proxy_options=None):
if secure:
tls_ctx_opt = TlsContextOptions()
tls_ctx_opt.override_default_trust_store_from_path(None, 'test/resources/unittests.crt')
tls_ctx = ClientTlsContext(tls_ctx_opt)
tls_conn_opt = tls_ctx.new_connection_options()
tls_conn_opt.set_server_name(self.hostname)
else:
tls_conn_opt = None
event_loop_group = EventLoopGroup()
host_resolver = DefaultHostResolver(event_loop_group)
bootstrap = ClientBootstrap(event_loop_group, host_resolver)
connection_future = HttpClientConnection.new(host_name=self.hostname,
port=self.port,
bootstrap=bootstrap,
tls_connection_options=tls_conn_opt,
proxy_options=proxy_options)
return connection_future.result(self.timeout)
def test_server_name(self):
opt = TlsContextOptions()
ctx = ClientTlsContext(opt)
conn_opt = TlsConnectionOptions(ctx)
conn_opt.set_server_name('localhost')
This builder creates an awscrt.mqtt.Connection, configured for an MQTT connection over websockets,
with a custom function to transform the websocket handshake request before it is sent to the server.
Arguments:
websocket_handshake_transform: Function with signature:
(awscrt.mqtt.WebsocketHandshakeTransformArgs) -> None
Function is called each time a websocket connection is attempted.
The function may modify the websocket handshake request, and MUST call set_done() when complete.
See awscrt.mqtt.WebsocketHandshakeTransformArgs for more info.
websocket_proxy_options (awscrt.http.HttpProxyOptions): If specified, a proxy is used when connecting.
All other required and optional arguments are explained in this module's docs.
"""
_check_required_kwargs(**kwargs)
tls_ctx_options = awscrt.io.TlsContextOptions()
return _builder(tls_ctx_options=tls_ctx_options,
use_websockets=True,
websocket_handshake_transform=websocket_handshake_transform,
websocket_proxy_options=websocket_proxy_options,
**kwargs)
def mtls_from_bytes(cert_bytes, pri_key_bytes, **kwargs):
"""
This builder creates an awscrt.mqtt.Connection, configured for an mTLS MQTT connection to AWS IoT.
TLS arguments are passed as in-memory bytes.
Arguments:
cert_bytes (bytes): Certificate file.
pri_key_bytes (bytes): Private key.
All other required and optional arguments are explained in this module's docs.
"""
_check_required_kwargs(**kwargs)
tls_ctx_options = awscrt.io.TlsContextOptions.create_client_with_mtls(cert_bytes, pri_key_bytes)
return _builder(tls_ctx_options, **kwargs)
def create_server_pkcs12(pkcs12_filepath, pkcs12_password):
assert isinstance_str(pkcs12_filepath)
assert isinstance_str(pkcs12_password)
opt = TlsContextOptions()
opt.pkcs12_filepath = pkcs12_filepath
opt.pkcs12_password = pkcs12_password
opt.verify_peer = False
return opt