AWS ACM Certificates

import boto3
import csv
import argparse


def get_top_domain(domain_name):
    """Extracts the top-level domain from a given domain name,
    handling various TLDs including .com.au.
    """

    try:
        # Split the domain name by dots
        parts = domain_name.split(".")

        # Handle special cases like .com.au, .co.uk, etc.
        if len(parts) >= 3 and parts[-2] in ["com", "co", "org", "net"]:
            return ".".join(
                parts[-3:]
            )  # Get the last three parts (e.g., example.com.au)
        elif len(parts) >= 2:  # Other cases (e.g., example.com, example.net)
            return ".".join(parts[-2:])  # Get the last two parts
        else:
            return None  # Invalid domain format
    except Exception as e:
        print(f"Error extracting top domain from {domain_name}: {e}")
        return None


def get_hosted_zone_id(domain_name):
    """Retrieves the Hosted Zone ID for a given domain name,
    handling wildcards by looking for the parent domain's zone.
    """

    route53_client = boto3.client("route53")

    try:
        print(f"Looking up Hosted Zone ID for {domain_name}")
        response = route53_client.list_hosted_zones_by_name(
            DNSName=domain_name, MaxItems="1"
        )
        if response["HostedZones"]:
            return response["HostedZones"][0]["Id"]
        else:
            return None
    except Exception as e:
        print(f"Error getting Hosted Zone ID for {domain_name}: {e}")
        return None


def update_cname_records(hosted_zone_id, cname_name, cname_value):
    """Updates a CNAME record in Route 53."""

    route53_client = boto3.client("route53")

    try:
        response = route53_client.change_resource_record_sets(
            HostedZoneId=hosted_zone_id,
            ChangeBatch={
                "Changes": [
                    {
                        "Action": "UPSERT",  # Use UPSERT to create or update
                        "ResourceRecordSet": {
                            "Name": cname_name,
                            "Type": "CNAME",
                            "TTL": 300,
                            "ResourceRecords": [{"Value": cname_value}],
                        },
                    }
                ]
            },
        )
        print(f"CNAME record updated successfully: {cname_name} -> {cname_value}")
    except Exception as e:
        print(f"Error updating CNAME record: {e}")


def main():
    """Reads CNAME records from a CSV file and updates Route 53."""

    parser = argparse.ArgumentParser(
        description="Update Route 53 CNAME records from a CSV file."
    )
    parser.add_argument(
        "csv_file", help="Path to the CSV file containing CNAME records."
    )
    args = parser.parse_args()

    with open(args.csv_file, "r") as csvfile:
        reader = csv.DictReader(csvfile)
        for row in reader:
            domain_name = row["Domain name"]
            cname_name = row["CNAME name"]
            record_type = row["Type"]
            cname_value = row["CNAME value"]

            print("")

            if record_type == "CNAME":
                hosted_zone_id = get_hosted_zone_id(get_top_domain(domain_name))
                if hosted_zone_id:
                    print(
                        f"Updating CNAME record for {domain_name} in hosted zone {hosted_zone_id}: {cname_name} -> {cname_value}"
                    )
                    update_cname_records(hosted_zone_id, cname_name, cname_value)
                else:
                    print(
                        f"Error: Could not find Hosted Zone ID for {domain_name}. Skipping record."
                    )
            else:
                print(
                    f"Skipping non-CNAME record for {domain_name}: {cname_name} ({record_type})"
                )


if __name__ == "__main__":
    main()

dig CNAME <cname_name>. +short