import boto3 from botocore.exceptions import ClientError import time import os # A script to launch a basic EC2 instance, create necessary prerequisites, # and then clean up all resources using Boto3. # --- Configuration --- REGION = "us-east-1" INSTANCE_TYPE = "t2.micro" KEY_PAIR_NAME = "MyBoto3KeyPair" SECURITY_GROUP_NAME = "MyBoto3SecurityGroup" KEY_FILE_PATH = f"{KEY_PAIR_NAME}.pem" ec2_client = boto3.client('ec2', region_name=REGION) ec2_resource = boto3.resource('ec2', region_name=REGION) def get_latest_ami_id(): """Dynamically finds the latest Amazon Linux 2 AMI ID.""" print("--- Finding latest Amazon Linux 2 AMI ID ---") try: response = ec2_client.describe_images( Owners=['amazon'], Filters=[ {'Name': 'name', 'Values': ['amzn2-ami-hvm-*-x86_64-gp2']}, {'Name': 'state', 'Values': ['available']} ] ) images = sorted(response['Images'], key=lambda x: x['CreationDate'], reverse=True) if images: ami_id = images[0]['ImageId'] print(f"Found AMI ID: {ami_id}") return ami_id else: raise Exception("Could not find a suitable AMI.") except ClientError as e: print(f"Error finding AMI: {e}") raise def create_key_pair(): """Creates an EC2 Key Pair and saves the private key locally.""" print(f"\n--- Creating EC2 Key Pair: {KEY_PAIR_NAME} ---") try: key_pair = ec2_client.create_key_pair(KeyName=KEY_PAIR_NAME) with open(KEY_FILE_PATH, 'w') as f: f.write(key_pair['KeyMaterial']) os.chmod(KEY_FILE_PATH, 0o400) # Set read-only permissions print(f"Key pair '{KEY_PAIR_NAME}.pem' created. Remember to keep it secure.") return key_pair['KeyPairId'] except ClientError as e: if e.response['Error']['Code'] == 'InvalidKeyPair.Duplicate': print(f"Key pair '{KEY_PAIR_NAME}' already exists. Skipping creation.") return None # Indicate that it wasn't newly created else: print(f"Error creating key pair: {e}") raise def create_security_group(vpc_id): """Creates a Security Group allowing SSH and HTTP access.""" print(f"\n--- Creating Security Group: {SECURITY_GROUP_NAME} ---") try: sg = ec2_resource.create_security_group( GroupName=SECURITY_GROUP_NAME, Description="Allow SSH and HTTP access", VpcId=vpc_id ) sg.authorize_ingress( IpPermissions=[ {'IpProtocol': 'tcp', 'FromPort': 22, 'ToPort': 22, 'IpRanges': [{'CidrIp': '0.0.0.0/0'}]}, {'IpProtocol': 'tcp', 'FromPort': 80, 'ToPort': 80, 'IpRanges': [{'CidrIp': '0.0.0.0/0'}]} ] ) print(f"Security Group '{SECURITY_GROUP_NAME}' created with ID: {sg.id}") return sg.id except ClientError as e: if e.response['Error']['Code'] == 'InvalidGroup.Duplicate': print(f"Security Group '{SECURITY_GROUP_NAME}' already exists. Fetching ID.") response = ec2_client.describe_security_groups(GroupNames=[SECURITY_GROUP_NAME]) return response['SecurityGroups'][0]['GroupId'] else: print(f"Error creating security group: {e}") raise def launch_ec2_instance(ami_id, key_pair_name, sg_id, subnet_id): """Launches an EC2 instance.""" print(f"\n--- Launching EC2 Instance ---") try: instance = ec2_resource.create_instances( ImageId=ami_id, MinCount=1, MaxCount=1, InstanceType=INSTANCE_TYPE, KeyName=key_pair_name, SecurityGroupIds=[sg_id], SubnetId=subnet_id, TagSpecifications=[{'ResourceType': 'instance', 'Tags': [{'Key': 'Name', 'Value': 'MyBoto3Instance'}]}] )[0] print(f"Instance '{instance.id}' launched. Waiting for it to be running...") instance.wait_until_running() instance.reload() # Reload attributes to get public IP print("Instance is running.") return instance except ClientError as e: print(f"Error launching instance: {e}") raise def cleanup_resources(instance_id, sg_id, key_pair_name): """Cleans up all created resources.""" print(f"\n--- Cleaning up resources ---") # Terminate instance if instance_id: print(f"Terminating instance '{instance_id}'...") ec2_client.terminate_instances(InstanceIds=[instance_id]) ec2_client.get_waiter('instance_terminated').wait(InstanceIds=[instance_id]) print("Instance terminated.") # Delete Security Group if sg_id: print(f"Deleting Security Group '{sg_id}'...") try: ec2_client.delete_security_group(GroupId=sg_id) print("Security Group deleted.") except ClientError as e: if e.response['Error']['Code'] == 'DependencyViolation': print(f"Security Group '{sg_id}' is still in use. Retrying deletion after a short delay.") time.sleep(10) ec2_client.delete_security_group(GroupId=sg_id) print("Security Group deleted.") else: print(f"Error deleting security group: {e}") # Delete Key Pair if key_pair_name: print(f"Deleting Key Pair '{key_pair_name}'...") ec2_client.delete_key_pair(KeyName=key_pair_name) print("Key Pair deleted.") # Delete local key file if os.path.exists(KEY_FILE_PATH): os.remove(KEY_FILE_PATH) print(f"Local key file '{KEY_FILE_PATH}' deleted.") def main(): instance_id = None sg_id = None key_pair_name = KEY_PAIR_NAME # Use the global name for cleanup try: ami_id = get_latest_ami_id() # Get default VPC ID vpc_id = ec2_client.describe_vpcs(Filters=[{'Name': 'is-default', 'Values': ['true']}])['Vpcs'][0]['VpcId'] print(f"Using Default VPC ID: {vpc_id}") # Get a default subnet ID in the default VPC subnet_id = ec2_client.describe_subnets(Filters=[{'Name': 'vpc-id', 'Values': [vpc_id]}, {'Name': 'default-for-az', 'Values': ['true']}])['Subnets'][0]['SubnetId'] print(f"Using Default Subnet ID: {subnet_id}") create_key_pair() # This function handles if key exists sg_id = create_security_group(vpc_id) instance = launch_ec2_instance(ami_id, KEY_PAIR_NAME, sg_id, subnet_id) instance_id = instance.id print("\n--- EC2 Instance Launched Successfully! ---") print(f"Instance ID: {instance_id}") print(f"Public IP Address: {instance.public_ip_address}") print(f"You can SSH into your instance using: ssh -i {KEY_FILE_PATH} ec2-user@{instance.public_ip_address}") print(f"Or access a web server on http://{instance.public_ip_address}") input("Press Enter to terminate the instance and clean up resources...") except ClientError as e: print(f"An AWS client error occurred: {e}") except Exception as e: print(f"An unexpected error occurred: {e}") finally: cleanup_resources(instance_id, sg_id, key_pair_name) print("\n--- EC2 instance demonstration and cleanup complete ---") if __name__ == "__main__": main()