ec2_launch_instance.py

#!/usr/bin/python3

"""
ec2_launch_instance.py
Launches an EC2 instance with optional key pair and security group creation.
"""

import argparse
import boto3
from botocore.exceptions import ClientError


def parse_args():
    parser = argparse.ArgumentParser(description="Launch an EC2 instance.")
    parser.add_argument("--region", required=True, help="AWS region")
    parser.add_argument("--ami-id", required=True, help="AMI ID")
    parser.add_argument("--instance-type", default="t3.micro", help="Instance type")
    parser.add_argument("--subnet-id", required=True, help="Subnet ID")
    parser.add_argument("--instance-name", default="ec2-instance", help="EC2 Name tag")
    parser.add_argument("--key-pair-name", help="Create/use this key pair")
    parser.add_argument("--security-group-name", help="Create/use this security group")
    parser.add_argument("--vpc-id", help="VPC ID required when creating a security group")
    parser.add_argument(
        "--ingress-cidrs",
        default="0.0.0.0/0",
        help="Comma-separated CIDRs for ingress rules (default: 0.0.0.0/0)",
    )
    parser.add_argument(
        "--ingress-ports",
        default="22,80,443",
        help="Comma-separated TCP ports for ingress rules (default: 22,80,443)",
    )
    return parser.parse_args()


def ensure_key_pair(ec2_client, key_pair_name):
    if not key_pair_name:
        return None

    try:
        ec2_client.describe_key_pairs(KeyNames=[key_pair_name])
        print(f"Key pair exists: {key_pair_name}")
        return key_pair_name
    except ClientError as error:
        if "InvalidKeyPair.NotFound" not in str(error):
            raise

    key_pair = ec2_client.create_key_pair(KeyName=key_pair_name)
    key_file = f"{key_pair_name}.pem"
    with open(key_file, "w", encoding="utf-8") as file_handle:
        file_handle.write(key_pair["KeyMaterial"])
    print(f"Created key pair and saved private key: {key_file}")
    return key_pair_name


def ensure_security_group(ec2_client, security_group_name, vpc_id, ports, cidrs):
    if not security_group_name:
        return None

    if not vpc_id:
        raise SystemExit("--vpc-id is required when --security-group-name is provided")

    try:
        response = ec2_client.describe_security_groups(
            Filters=[
                {"Name": "group-name", "Values": [security_group_name]},
                {"Name": "vpc-id", "Values": [vpc_id]},
            ]
        )
        groups = response.get("SecurityGroups", [])
        if groups:
            group_id = groups[0]["GroupId"]
            print(f"Security group exists: {group_id}")
            return group_id
    except ClientError as error:
        raise SystemExit(f"Failed to check security group: {error}")

    try:
        created = ec2_client.create_security_group(
            GroupName=security_group_name,
            Description="Managed by ec2_launch_instance.py",
            VpcId=vpc_id,
        )
        group_id = created["GroupId"]
        print(f"Created security group: {group_id}")

        permissions = []
        for port in ports:
            permissions.append(
                {
                    "IpProtocol": "tcp",
                    "FromPort": port,
                    "ToPort": port,
                    "IpRanges": [{"CidrIp": cidr} for cidr in cidrs],
                }
            )

        ec2_client.authorize_security_group_ingress(GroupId=group_id, IpPermissions=permissions)
        print(f"Added ingress rules for ports {ports} and CIDRs {cidrs}")
        return group_id
    except ClientError as error:
        raise SystemExit(f"Failed to create/configure security group: {error}")


def launch_instance(ec2_client, args, key_pair_name, security_group_id):
    kwargs = {
        "ImageId": args.ami_id,
        "InstanceType": args.instance_type,
        "MinCount": 1,
        "MaxCount": 1,
        "SubnetId": args.subnet_id,
        "TagSpecifications": [
            {
                "ResourceType": "instance",
                "Tags": [{"Key": "Name", "Value": args.instance_name}],
            }
        ],
    }

    if key_pair_name:
        kwargs["KeyName"] = key_pair_name
    if security_group_id:
        kwargs["SecurityGroupIds"] = [security_group_id]

    response = ec2_client.run_instances(**kwargs)
    instance_id = response["Instances"][0]["InstanceId"]
    print(f"Launched instance: {instance_id}")


def main():
    args = parse_args()
    ports = [int(port.strip()) for port in args.ingress_ports.split(",") if port.strip()]
    cidrs = [cidr.strip() for cidr in args.ingress_cidrs.split(",") if cidr.strip()]

    ec2_client = boto3.client("ec2", region_name=args.region)

    key_pair_name = ensure_key_pair(ec2_client, args.key_pair_name)
    security_group_id = ensure_security_group(
        ec2_client,
        args.security_group_name,
        args.vpc_id,
        ports,
        cidrs,
    )

    launch_instance(ec2_client, args, key_pair_name, security_group_id)


if __name__ == "__main__":
    main()