Simple Python 3 script to create an AWS IoT OTA job

create_aws_iot_ota.py

import sys, argparse
import boto3
import pathlib
from pathlib import Path
import random

parser = argparse.ArgumentParser(description='Script to start OTA update')
parser.add_argument("--aws_profile", help="Profile name created using aws configure", default="default", required=False)
parser.add_argument("--s3bucket", help="S3 bucket to store firmware updates", required=True)
parser.add_argument("--s3prefix", help="S3 bucket prefix to store firmware updates", required=False)
parser.add_argument("--ota_role", help="Role for OTA updates", required=True)
parser.add_argument("--codelocation", help="base folder location (can be relative)", required=True)
parser.add_argument("--appversion", help="version of the image being uploade. The appversion value should follow the format APP_VERSION_MAJOR-APP_VERSION_MINOR-APP_VERSION_BUILD that is appended to the filename of the file being uploaded",default="0-0-0",required=True)
parser.add_argument("--otasigningprofile", help="Signing profile to be used", required=True)
parser.add_argument("--thingname", help="thing name", required=True)
args=parser.parse_args()

class AWS_IoT_OTA:

    def __init__(self):
        boto3.setup_default_session(profile_name=args.aws_profile)
        self.Session = boto3.session.Session()
        self.aws_region = self.Session.region_name

        self.aws_account=boto3.client('sts').get_caller_identity().get('Account')
        print("region: {}, account: {}".format(self.aws_region, self.aws_account))

        self.s3 = boto3.resource('s3')

        iam = boto3.resource('iam')
        ota_role = iam.Role(args.ota_role)
        self.ota_role_arn = ota_role.arn

        iot = boto3.client('iot')
        iot_thing = iot.describe_thing(thingName=args.thingname)        
        self.iot_thing_arn = iot_thing["thingArn"]        

    def BuildFirmwareFileNames(self):
        self.DEMOS_PATH=Path(args.codelocation)
        self.BUILD_PATH=self.DEMOS_PATH / Path("build")
        # We Should have the versions stored at this point. Build the App name
        self.OTA_APP_NAME="ota_" + args.appversion + ".bin"
        self.OTA_APP_FULL_NAME=self.BUILD_PATH / Path(self.OTA_APP_NAME)

        self.BUILD_FILE_FULL_NAME=self.BUILD_PATH/Path("aws_demos.bin")

        if(args.s3prefix):
            self.OTA_APP_NAME = ("{}/{}".format(args.s3prefix,self.OTA_APP_NAME))

        print ("OTA_APP_NAME: {}".format(self.OTA_APP_NAME))
        print ("Local Bin full path: {}".format(self.BUILD_FILE_FULL_NAME))

    # Copy the file to the s3 bucket
    def CopyFirmwareFileToS3(self):

        try:
            s3_client = boto3.client('s3')
            # check bucket versioning
            response = s3_client.get_bucket_versioning( Bucket=args.s3bucket )

            versioning_enabled = response["Status"]

            if(not versioning_enabled):
                raise Exception("S3 bucket versioning not enabled!")

            self.s3.meta.client.upload_file( str(self.BUILD_FILE_FULL_NAME), args.s3bucket, str(self.OTA_APP_NAME))

        except self.s3_client.exceptions.NoSuchBucket as e:
            print("Bucket {} not found".format(args.s3bucket))
            sys.exit        
        except Exception as e:
            print("Error uploading file to s3: %s", e)
            sys.exit        

    # Get the latest version
    def GetLatestS3FileVersion(self):
        try: 
            versions=self.s3.meta.client.list_object_versions(Bucket=args.s3bucket, Prefix=self.OTA_APP_NAME)['Versions']
            latestversion = [x for x in versions if x['IsLatest']==True]
            self.latestVersionId=latestversion[0]['VersionId']
            print("Using version %s" % self.latestVersionId)
        except Exception as e:
            print("Error getting versions: %s" % e)
            sys.exit

    # Verify signing profile
    def VerifySigningProfile(self):
        try:
            signer = boto3.client('signer')
            profiles = signer.list_signing_profiles()['profiles']

            foundProfile=False
            afrProfile=None
            print("Searching for profile %s" % args.otasigningprofile)

            if len(profiles) > 0:
              for profile in profiles:
                if profile['profileName'] == args.otasigningprofile:
                    foundProfile = True
                    afrProfile = profile
            
            if (afrProfile != None):
                foundProfile=True
                print("Found Profile %s in account" % args.otasigningprofile)

            if(not foundProfile):
                raise Exception("Error getting signing profile: {}".format(args.otasigningprofile))

        except Exception as e:
            print("Error getting signing profiles: {}".format(e))
            sys.exit


    def CreateOTAJob(self):
        
        # Create OTA job
        try:
            iot = boto3.client('iot')
            randomSeed=random.randint(1, 65535)
            #Initialize the template to use
            files=[{
                'fileName': self.OTA_APP_NAME,
                    'fileVersion': '1',
                    'fileLocation': {
                        's3Location': {
                            'bucket': args.s3bucket,
                            'key': self.OTA_APP_NAME,
                            'version': self.latestVersionId
                        }
                    },
                    'codeSigning':{
                        'startSigningJobParameter':{
                            'signingProfileName': args.otasigningprofile,
                            'destination': {
                                's3Destination': {
                                    'bucket': args.s3bucket
                                }
                            }
                        }
                    }    
                }] 

            target = self.iot_thing_arn            
            updateId="update-"+str(randomSeed)+"-"+args.appversion.replace(".","_")

            print ("Files for update: %s" % files)
            
            ota_update=iot.create_ota_update(
                otaUpdateId=updateId,
                targetSelection='SNAPSHOT',
                files=files,
                targets=[target],
                roleArn=self.ota_role_arn
            )

            print("OTA Update Status: %s" % ota_update)

        except Exception as e:
            print("Error creating OTA Job: %s" % e)
            sys.exit

    def DoUpdate(self):
        self.BuildFirmwareFileNames()
        self.CopyFirmwareFileToS3()
        self.GetLatestS3FileVersion()
        self.VerifySigningProfile()
        self.CreateOTAJob()
   
if __name__ == "__main__":
    ota = AWS_IoT_OTA()
    ota.DoUpdate()

requirements.txt

boto3==1.13.18
botocore==1.16.18