boto3 aws iot - how can I check if certificate has attached things to it?

Question

in aws iot-

with boto3 (python) I want to check if a certificate has attached things to it. how can I do that? Thanks.

Answer 1

import boto3, time, csv
from concurrent.futures import ThreadPoolExecutor
from csv import writer
from pyjavaproperties import Properties

#########################################################################
# Properties reading from properties file
p = Properties()
p.load(open('prop.properties'))
# AWS Credentilas
aws_region='eu-west-1'
aws_access_key_id=p['aws_access_key_id']
aws_secret_access_key=p['aws_secret_access_key']
aws_session_token=p['aws_session_token']
#########################################################################
file_name1='QA-Ireland-Certificate-Attached-With-Thing.csv'
file_name2='QA-Ireland-Certificate-NOT-Attached-With-Thing.csv'
#########################################################################

client = boto3.client('iot',
region_name=aws_region,
aws_access_key_id= aws_access_key_id,  
aws_secret_access_key= aws_secret_access_key,
aws_session_token=aws_session_token)

##Function to Append List in CSV
def append_list_as_row(file_name, list_of_elem):
    with open(file_name, 'a') as write_obj:
        csv_writer = csv.writer(write_obj)
        csv_writer.writerow(list_of_elem)

results = []
attached = []
non_attached = []

def check_certificates(certificate):
    principal = client.list_principal_things(principal=certificate["certificateArn"])
    # print("%s - number of things: %d" % (certificate["certificateId"], len(principal["things"])))
    if len(principal["things"])==0:
        non_attached.append(certificate["certificateArn"])
        print("Total orphan certificates are: ", len(non_attached))
        list_var = []
        list_var.append('iot')
        list_var.append(certificate["certificateArn"])
        append_list_as_row(file_name2, list_var)
        #### add upir extra logic here.
    else:
        attached.append(certificate["certificateArn"])
        print("Total certificates attached are: ", len(attached))
        list_var = []
        list_var.append('iot')
        list_var.append(certificate["certificateArn"])
        append_list_as_row(file_name1, list_var)

def lambda_handler():
    paginator = client.get_paginator('list_certificates')
    page_iterator = paginator.paginate()
    for page in page_iterator:
        results.extend(page['certificates'])
        print("Total certificates scanned are: ", len(results))
    print("Total certificates scanned are: ", len(results))
    start_time = time.time()
    with ThreadPoolExecutor(max_workers=10) as executor:
        future = [executor.submit(check_certificates, item) for item in results]
    print("Certs have thing(s) attached: ", len(attached))
    print("Certs have NO thing(s) attached: ", len(non_attached))
    print("--- %s seconds using MultiThreading---" % (time.time() - start_time))
    
def main():
    lambda_handler()
if __name__== "__main__":
    main()