-
Notifications
You must be signed in to change notification settings - Fork 100
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
4 changed files
with
145 additions
and
95 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,82 @@ | ||
| # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. | ||
| # SPDX-License-Identifier: Apache-2.0. | ||
|
|
||
| import sys | ||
|
|
||
| import boto3 | ||
|
|
||
|
|
||
| def create_iot_thing(thing_name, region, policy_name, certificate_path, key_path, thing_group=None): | ||
| """ Create IoT thing along with policy and credentials. """ | ||
|
|
||
| iot_client = boto3.client('iot', region_name=region) | ||
|
|
||
| print(f"Creating thing '{thing_name}'", file=sys.stderr) | ||
|
|
||
| iot_client.create_thing(thingName=thing_name) | ||
| if thing_group: | ||
| iot_client.add_thing_to_thing_group(thingGroupName=thing_group, thingName=thing_name) | ||
|
|
||
| try: | ||
| print("Creating certificate", file=sys.stderr) | ||
| create_cert_response = iot_client.create_keys_and_certificate( | ||
| setAsActive=True | ||
| ) | ||
|
|
||
| f = open(certificate_path, "w") | ||
| f.write(create_cert_response['certificatePem']) | ||
| f.close() | ||
|
|
||
| f = open(key_path, "w") | ||
| f.write(create_cert_response['keyPair']['PrivateKey']) | ||
| f.close() | ||
|
|
||
| certificate_arn = create_cert_response['certificateArn'] | ||
|
|
||
| print("Attaching policy to certificate", file=sys.stderr) | ||
| iot_client.attach_policy(policyName=policy_name, target=certificate_arn) | ||
|
|
||
| print("Attaching certificate to thing", file=sys.stderr) | ||
| iot_client.attach_thing_principal(thingName=thing_name, principal=certificate_arn) | ||
| except Exception: | ||
| try: | ||
| iot_client.delete_thing(thingName=thing_name) | ||
| except Exception: | ||
| print("ERROR: Could not delete thing", file=sys.stderr) | ||
| raise | ||
|
|
||
| print("IoT thing created successfully", file=sys.stderr) | ||
|
|
||
|
|
||
| def delete_iot_thing(thing_name, region): | ||
| """ Delete IoT thing and all its principals. """ | ||
|
|
||
| try: | ||
| iot_client = boto3.client('iot', region_name=region) | ||
| except Exception as e: | ||
| print(f"ERROR: Could not make Boto3 client. Credentials likely could not be sourced", file=sys.stderr) | ||
| raise | ||
|
|
||
| # Detach and delete thing's principals. | ||
| try: | ||
| thing_principals = iot_client.list_thing_principals(thingName=thing_name) | ||
| print(f"Detaching and deleting principals: {thing_principals}", file=sys.stderr) | ||
| for principal in thing_principals["principals"]: | ||
| certificate_id = principal.split("/")[1] | ||
| iot_client.detach_thing_principal(thingName=thing_name, principal=principal) | ||
| iot_client.update_certificate(certificateId=certificate_id, newStatus='INACTIVE') | ||
| iot_client.delete_certificate(certificateId=certificate_id, forceDelete=True) | ||
| except Exception: | ||
| print("ERROR: Could not delete certificate for IoT thing {thing_name}, probably thing does not exist", | ||
| file=sys.stderr) | ||
| raise | ||
|
|
||
| # Delete thing. | ||
| try: | ||
| iot_client.delete_thing(thingName=thing_name) | ||
| except Exception: | ||
| raise | ||
|
|
||
| print("IoT thing deleted successfully", file=sys.stderr) | ||
|
|
||
| return 0 |
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,41 @@ | ||
| # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. | ||
| # SPDX-License-Identifier: Apache-2.0. | ||
|
|
||
| import argparse | ||
| import uuid | ||
| import os | ||
| import sys | ||
| import run_in_ci | ||
| import ci_iot_thing | ||
|
|
||
|
|
||
| def main(): | ||
| argument_parser = argparse.ArgumentParser( | ||
| description="Run Fleet Provisioning sample in CI") | ||
| argument_parser.add_argument("--file", required=True, help="Configuration file to pull CI data from") | ||
| argument_parser.add_argument( | ||
| "--input-uuid", required=False, | ||
| help="Use specified UUID for thing name instead of a random one") | ||
| argument_parser.add_argument( | ||
| "--thing-name-prefix", required=False, default="", | ||
| help="Prefix for a thing name, should be the same that Fleet Provisioning template uses") | ||
| argument_parser.add_argument( | ||
| "--region", required=False, default="us-east-1", help="The name of the region to use") | ||
| parsed_commands = argument_parser.parse_args() | ||
|
|
||
| cfg_file = parsed_commands.file | ||
| thing_uuid = parsed_commands.input_uuid if parsed_commands.input_uuid else str(uuid.uuid4()) | ||
|
|
||
| # Perform fleet provisioning. If it's successful, a newly created thing should appear. | ||
| test_result = run_in_ci.setup_and_launch(cfg_file, thing_uuid) | ||
|
|
||
| # Delete a thing created by fleet provisioning. | ||
| # NOTE We want to try to delete thing even if test was unsuccessful. | ||
| thing_name = parsed_commands.thing_name_prefix + thing_uuid | ||
| delete_result = ci_iot_thing.delete_iot_thing(thing_name, parsed_commands.region) | ||
|
|
||
| if test_result != 0 or delete_result != 0: | ||
| sys.exit(-1) | ||
|
|
||
| if __name__ == "__main__": | ||
| main() |