in awsiot/mqtt5_client_builder.py [0:0]
def direct_with_custom_authorizer(
auth_username=None,
auth_authorizer_name=None,
auth_authorizer_signature=None,
auth_password=None,
auth_token_key_name=None,
auth_token_value=None,
**kwargs) -> awscrt.mqtt5.Client:
"""
This builder creates an :class:`awscrt.mqtt5.Client`, configured for an MQTT5 Client using a custom
authorizer. This function will set the username, port, and TLS options.
This function takes all :mod:`common arguments<awsiot.mqtt5_client_builder>`
described at the top of this doc, as well as...
Keyword Args:
auth_username (`str`): The username to use with the custom authorizer.
If provided, the username given will be passed when connecting to the custom authorizer.
If not provided, it will check to see if a username has already been set (via username="example")
and will use that instead. Custom authentication parameters will be appended as appropriate
to any supplied username value.
auth_password (`str`): The password to use with the custom authorizer.
If not provided, then no password will be sent in the initial CONNECT packet.
auth_authorizer_name (`str`): Name of the custom authorizer to use.
Required if the endpoint does not have a default custom authorizer associated with it. It is strongly
suggested to URL-encode this value; the SDK will not do so for you.
auth_authorizer_signature (`str`): The digital signature of the token value in the `auth_token_value`
parameter. The signature must be based on the private key associated with the custom authorizer. The
signature must be base64 encoded.
Required if the custom authorizer has signing enabled.
auth_token_key_name (`str`): Key used to extract the custom authorizer token from MQTT username query-string
properties.
Required if the custom authorizer has signing enabled. It is strongly suggested to URL-encode
this value; the SDK will not do so for you.
auth_token_value (`str`): An opaque token value. This value must be signed by the private key associated with
the custom authorizer and the result passed in via the `auth_authorizer_signature` parameter.
Required if the custom authorizer has signing enabled.
"""
_check_required_kwargs(**kwargs)
username_string = ""
if auth_username is None:
if not _get(kwargs, "username") is None:
username_string += _get(kwargs, "username")
else:
username_string += auth_username
if auth_authorizer_name is not None:
username_string = _add_to_username_parameter(
username_string, auth_authorizer_name, "x-amz-customauthorizer-name=")
if auth_authorizer_signature is not None:
encoded_signature = auth_authorizer_signature
if "%" not in encoded_signature:
encoded_signature = urllib.parse.quote(encoded_signature)
username_string = _add_to_username_parameter(
username_string, encoded_signature, "x-amz-customauthorizer-signature=")
if auth_token_key_name is not None and auth_token_value is not None:
username_string = _add_to_username_parameter(username_string, auth_token_value, auth_token_key_name + "=")
kwargs["username"] = username_string
kwargs["password"] = auth_password
tls_ctx_options = awscrt.io.TlsContextOptions()
tls_ctx_options.alpn_list = ["mqtt"]
return _builder(tls_ctx_options=tls_ctx_options,
use_websockets=False,
use_custom_authorizer=True,
**kwargs)