求Python实现AWS IoT自动注册树莓派：生成Thing、证书并下载

I’ve converted your Node.js AWS IoT thing creation logic into a Python script tailored for Raspberry Pi auto-registration. This script handles creating an IoT Thing, Policy, Certificate, linking them together, and saving the necessary certificate files locally for your device to use.

Prerequisites #

• Install the AWS SDK for Python:

  pip install boto3
  

• Configure AWS credentials on your Raspberry Pi (best practice: use the ~/.aws/credentials file or environment variables instead of hardcoding keys). You can set this up with aws configure if you have the AWS CLI installed, or manually create the file.

Complete Python Script #

import boto3
import uuid
import os

# Initialize AWS IoT client
region = 'us-east-1'  # Replace with your AWS region
iot_client = boto3.client('iot', region_name=region)

# Configuration
device_type = "MySmartIoTDevice"
registration_topic = "registration"
cert_save_dir = "./iot_certs"  # Directory to save certificates

# Create directory for certs if it doesn't exist
os.makedirs(cert_save_dir, exist_ok=True)

def generate_unique_ids():
    """Generate unique identifiers for the device"""
    serial_number = f"SN-{uuid.uuid4().hex[:15].upper()}"
    client_id = f"ID-{uuid.uuid4().hex[:12].upper()}"
    activation_code = f"AC-{uuid.uuid4().hex[:20].upper()}"
    thing_name = f"myThing-{uuid.uuid4().hex[:8]}"  # Unique thing name
    return serial_number, client_id, activation_code, thing_name

def create_thing(thing_name):
    """Create AWS IoT Thing"""
    try:
        response = iot_client.create_thing(thingName=thing_name)
        print(f"Created IoT Thing: {response['thingName']}")
        return response
    except Exception as e:
        print(f"Error creating thing: {str(e)}")
        raise

def create_iot_policy(thing_name, client_id):
    """Create IoT Policy for the device"""
    policy_name = f"{thing_name}-policy"
    policy_document = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Effect": "Allow",
                "Action": [
                    "iot:Connect",
                    "iot:Publish",
                    "iot:Subscribe",
                    "iot:Receive"
                ],
                "Resource": [
                    f"arn:aws:iot:{region}:*:client/{client_id}",
                    f"arn:aws:iot:{region}:*:topic/{registration_topic}",
                    f"arn:aws:iot:{region}:*:topicfilter/device/{thing_name}/#"
                ]
            }
        ]
    }
    
    try:
        response = iot_client.create_policy(
            policyName=policy_name,
            policyDocument=str(policy_document).replace("'", '"')  # Convert dict to valid JSON
        )
        print(f"Created Policy: {response['policyName']}")
        return policy_name
    except Exception as e:
        print(f"Error creating policy: {str(e)}")
        raise

def generate_and_attach_certificate(thing_name, policy_name):
    """Generate certificate/keys and attach to thing and policy"""
    try:
        # Create keys and certificate
        cert_response = iot_client.create_keys_and_certificate(setAsActive=True)
        
        # Attach policy to certificate
        iot_client.attach_policy(
            policyName=policy_name,
            target=cert_response['certificateArn']
        )
        
        # Attach certificate to thing
        iot_client.attach_thing_principal(
            thingName=thing_name,
            principal=cert_response['certificateArn']
        )
        
        print(f"Generated and attached certificate: {cert_response['certificateId']}")
        return cert_response
    except Exception as e:
        print(f"Error handling certificate: {str(e)}")
        raise

def save_certificates(cert_response, save_dir):
    """Save certificate, private key, public key to files"""
    files = [
        ("certificate.pem.crt", cert_response['certificatePem']),
        ("private.pem.key", cert_response['keyPair']['PrivateKey']),
        ("public.pem.key", cert_response['keyPair']['PublicKey'])
    ]
    
    for filename, content in files:
        file_path = os.path.join(save_dir, filename)
        with open(file_path, 'w') as f:
            f.write(content)
        print(f"Saved {filename} to {file_path}")

def publish_registration_data(serial_number, client_id, thing_name, endpoint):
    """Publish registration data to MQTT topic"""
    # Initialize IoT Data client with endpoint
    iot_data_client = boto3.client('iot-data', region_name=region, endpoint_url=f"https://{endpoint}")
    
    registration_data = {
        "serialNumber": serial_number,
        "clientId": client_id,
        "device": thing_name,
        "endpoint": endpoint,
        "type": device_type
        # Add any additional fields you need here
    }
    
    try:
        iot_data_client.publish(
            topic=registration_topic,
            payload=str(registration_data).replace("'", '"'),
            qos=1
        )
        print(f"Published registration data to topic {registration_topic}")
    except Exception as e:
        print(f"Error publishing registration data: {str(e)}")
        raise

if __name__ == "__main__":
    try:
        # Generate unique IDs
        serial_number, client_id, activation_code, thing_name = generate_unique_ids()
        
        # Get IoT data endpoint
        endpoint = iot_client.describe_endpoint(endpointType='iot:Data-ATS')['endpointAddress']
        
        # Create Thing
        create_thing(thing_name)
        
        # Create Policy
        policy_name = create_iot_policy(thing_name, client_id)
        
        # Generate and attach certificate
        cert_response = generate_and_attach_certificate(thing_name, policy_name)
        
        # Save certificates locally
        save_certificates(cert_response, cert_save_dir)
        
        # Publish registration data
        publish_registration_data(serial_number, client_id, thing_name, endpoint)
        
        print("\nRegistration completed successfully!")
        print(f"Thing Name: {thing_name}")
        print(f"Client ID: {client_id}")
        print(f"Certificates saved to: {cert_save_dir}")
        
    except Exception as e:
        print(f"Registration failed: {str(e)}")

Key Notes #

• Credentials: Never hardcode your AWS access keys in the script. Use the AWS credentials file or environment variables for security.
• Policy Restrictions: The policy in the script is restricted to the device's client ID and specific topics. Adjust the resource ARNs to match your use case for tighter security.
• Root CA: You’ll need the Amazon Root CA certificate to connect your Raspberry Pi to AWS IoT. Save it in the same iot_certs directory (you can retrieve this from the AWS IoT Core console documentation).
• Scalability: If you need to register multiple devices, wrap the main logic in a loop like your original Node.js code—just be mindful of AWS IoT service quotas.

内容的提问来源于stack exchange，提问作者Parthiban Soundram