Add a trailing dot to all CNAME, MX, NS and PTR records, in all zones, on Route 53

route53_trailing_dot.py

from copy import deepcopy
from pprint import pprint

import boto3

client = boto3.client('route53')


def get_values(record):
    values = record.get("ResourceRecords")
    if values:
        return [d["Value"] for d in values]
    return []


def get_records(zone):
    paginator = client.get_paginator("list_resource_record_sets")
    for page in paginator.paginate(HostedZoneId=zone["Id"]):
        for record in page["ResourceRecordSets"]:
            yield record


def get_invalid_records(zone):
    for record in get_records(zone):
        if record['Type'] in ("CNAME", "MX", "NS", "PTR"):
            values = get_values(record)
            if any(not value.endswith(".") for value in values):
                yield record


def get_zones(only=None):
    paginator = client.get_paginator("list_hosted_zones")
    for page in paginator.paginate():
        for zone in page["HostedZones"]:
            if only and not zone["Name"] in only:
                continue
            yield zone


def fix_record(record):
    record = deepcopy(record)
    for index, value in enumerate(get_values(record)):
        if not value.endswith("."):
            record["ResourceRecords"][index]["Value"] = value + "."
    return record


def fix_zone(zone, pretend=False):
    for invalid_record in get_invalid_records(zone):
        changes = []
        changes.append({
            "Action": "DELETE",
            "ResourceRecordSet": invalid_record
        })
        valid_record = fix_record(invalid_record)
        changes.append({
            "Action": "CREATE",
            "ResourceRecordSet": valid_record
        })

        change_batch = {
            "Comment": "add trailing dots",
            "Changes": changes
        }
        pprint(change_batch)

        if not changes:
            continue

        if not pretend:
            client.change_resource_record_sets(
                HostedZoneId=zone["Id"],
                ChangeBatch=change_batch
            )
            time.sleep(0.5)


def main():
    for zone in get_zones():
        fix_zone(zone)


if __name__ == '__main__':
    main()