AWS iot documentation change
Summary
Added Python code example demonstrating AWS IoT operations including thing management, certificate handling, shadow operations, and rule creation with a complete interactive scenario
Security assessment
The change adds documentation about security features like certificate management (create_keys_and_certificate, attach_thing_principal) and IAM role usage in topic rules, but doesn't address any specific security vulnerability or incident. It demonstrates security best practices for device authentication and authorization.
Diff
diff --git a/iot/latest/developerguide/example_iot_Scenario_section.md b/iot/latest/developerguide/example_iot_Scenario_section.md index ae34c0e3e..049cc0856 100644 --- a//iot/latest/developerguide/example_iot_Scenario_section.md +++ b//iot/latest/developerguide/example_iot_Scenario_section.md @@ -3748,0 +3749,679 @@ There's more on GitHub. Find the complete example and learn how to set up and ru +Python + + +**SDK for Python (Boto3)** + + +###### Note + +There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/iot#code-examples). + +Create an IoT wrapper class to manage operations. + + + class IoTWrapper: + """Encapsulates AWS IoT actions.""" + + def __init__(self, iot_client, iot_data_client=None): + """ + :param iot_client: A Boto3 AWS IoT client. + :param iot_data_client: A Boto3 AWS IoT Data Plane client. + """ + self.iot_client = iot_client + self.iot_data_client = iot_data_client + + @classmethod + def from_client(cls): + iot_client = boto3.client("iot") + iot_data_client = boto3.client("iot-data") + return cls(iot_client, iot_data_client) + + + def create_thing(self, thing_name): + """ + Creates an AWS IoT thing. + + :param thing_name: The name of the thing to create. + :return: The name and ARN of the created thing. + """ + try: + response = self.iot_client.create_thing(thingName=thing_name) + logger.info("Created thing %s.", thing_name) + except ClientError as err: + if err.response["Error"]["Code"] == "ResourceAlreadyExistsException": + logger.info("Thing %s already exists. Skipping creation.", thing_name) + return None + logger.error( + "Couldn't create thing %s. Here's why: %s: %s", + thing_name, + err.response["Error"]["Code"], + err.response["Error"]["Message"], + ) + raise + else: + return response + + + def list_things(self): + """ + Lists AWS IoT things. + + :return: The list of things. + """ + try: + things = [] + paginator = self.iot_client.get_paginator("list_things") + for page in paginator.paginate(): + things.extend(page["things"]) + logger.info("Retrieved %s things.", len(things)) + return things + except ClientError as err: + if err.response["Error"]["Code"] == "ThrottlingException": + logger.error("Request throttled. Please try again later.") + else: + logger.error( + "Couldn't list things. Here's why: %s: %s", + err.response["Error"]["Code"], + err.response["Error"]["Message"], + ) + raise + + + + def create_keys_and_certificate(self): + """ + Creates keys and a certificate for an AWS IoT thing. + + :return: The certificate ID, ARN, and PEM. + """ + try: + response = self.iot_client.create_keys_and_certificate(setAsActive=True) + logger.info("Created certificate %s.", response["certificateId"]) + except ClientError as err: + if err.response["Error"]["Code"] == "ThrottlingException": + logger.error("Request throttled. Please try again later.") + else: + logger.error( + "Couldn't create keys and certificate. Here's why: %s: %s", + err.response["Error"]["Code"], + err.response["Error"]["Message"], + ) + raise + else: + return response + + + def attach_thing_principal(self, thing_name, principal): + """ + Attaches a certificate to an AWS IoT thing. + + :param thing_name: The name of the thing. + :param principal: The ARN of the certificate. + """ + try: + self.iot_client.attach_thing_principal( + thingName=thing_name, principal=principal + ) + logger.info("Attached principal %s to thing %s.", principal, thing_name) + except ClientError as err: + if err.response["Error"]["Code"] == "ResourceNotFoundException": + logger.error("Cannot attach principal. Resource not found.") + return + logger.error( + "Couldn't attach principal to thing. Here's why: %s: %s", + err.response["Error"]["Code"], + err.response["Error"]["Message"], + ) + raise + + + def describe_endpoint(self, endpoint_type="iot:Data-ATS"): + """ + Gets the AWS IoT endpoint. + + :param endpoint_type: The endpoint type. + :return: The endpoint. + """ + try: + response = self.iot_client.describe_endpoint(endpointType=endpoint_type) + logger.info("Retrieved endpoint %s.", response["endpointAddress"]) + except ClientError as err: + if err.response["Error"]["Code"] == "ThrottlingException": + logger.error("Request throttled. Please try again later.") + else: + logger.error( + "Couldn't describe endpoint. Here's why: %s: %s", + err.response["Error"]["Code"], + err.response["Error"]["Message"], + ) + raise + else: + return response["endpointAddress"] + + + def list_certificates(self): + """ + Lists AWS IoT certificates. + + :return: The list of certificates. + """ + try: + certificates = [] + paginator = self.iot_client.get_paginator("list_certificates") + for page in paginator.paginate(): + certificates.extend(page["certificates"]) + logger.info("Retrieved %s certificates.", len(certificates)) + return certificates + except ClientError as err: + if err.response["Error"]["Code"] == "ThrottlingException": + logger.error("Request throttled. Please try again later.") + else: + logger.error( + "Couldn't list certificates. Here's why: %s: %s", + err.response["Error"]["Code"], + err.response["Error"]["Message"], + ) + raise + + + def detach_thing_principal(self, thing_name, principal): + """ + Detaches a certificate from an AWS IoT thing. + + :param thing_name: The name of the thing. + :param principal: The ARN of the certificate. + """ + try: + self.iot_client.detach_thing_principal( + thingName=thing_name, principal=principal + ) + logger.info("Detached principal %s from thing %s.", principal, thing_name) + except ClientError as err: + if err.response["Error"]["Code"] == "ResourceNotFoundException": + logger.error("Cannot detach principal. Resource not found.") + return + logger.error(