Script to migrate a Route53-hosted domain to Webflow's new records (for easier updating of multiple domains)

migrate_webflow_route53_dns.py

#!/usr/bin/env python3
"""
Webflow DNS Migration Script for Route 53

This script migrates Webflow domains from old hosting to new hosting by:
1. Verifying current A and CNAME records match expected old values
2. Updating root domain A record to new value (198.202.211.1)
3. Removing extra A record if present
4. Updating www CNAME record to new value (cdn.webflow.com)

Usage:
    python migrate_webflow_dns.py <domain>

Example:
    python migrate_webflow_dns.py example.com
"""

import sys
import boto3
from botocore.exceptions import ClientError
import argparse


class WebflowDNSMigrator:
    def __init__(self):
        self.route53 = boto3.client('route53')

        # Old Webflow values to verify
        self.OLD_A_VALUES = ['75.2.70.75', '99.83.190.102']
        self.OLD_CNAME_VALUE = 'proxy-ssl.webflow.com'

        # New Webflow values to set
        self.NEW_A_VALUE = '198.202.211.1'
        self.NEW_CNAME_VALUE = 'cdn.webflow.com'

    def get_hosted_zone_id(self, domain):
        """Find the Route 53 hosted zone ID for the given domain."""
        try:
            response = self.route53.list_hosted_zones()
            for zone in response['HostedZones']:
                zone_name = zone['Name'].rstrip('.')
                if zone_name == domain:
                    return zone['Id'].replace('/hostedzone/', '')

            print(f"❌ No hosted zone found for domain: {domain}")
            return None

        except ClientError as e:
            print(f"❌ Error finding hosted zone: {e}")
            return None

    def get_current_records(self, hosted_zone_id, domain):
        """Get current A and CNAME records for the domain."""
        try:
            response = self.route53.list_resource_record_sets(HostedZoneId=hosted_zone_id)

            a_records = []
            cname_record = None

            for record in response['ResourceRecordSets']:
                record_name = record['Name'].rstrip('.')

                # Root domain A records
                if record_name == domain and record['Type'] == 'A':
                    a_records.append(record)

                # www subdomain CNAME record
                elif record_name == f'www.{domain}' and record['Type'] == 'CNAME':
                    cname_record = record

            return a_records, cname_record

        except ClientError as e:
            print(f"❌ Error retrieving records: {e}")
            return [], None

    def verify_old_records(self, a_records, cname_record, domain):
        """Verify that current records match expected old Webflow values."""
        print(f"\n🔍 Verifying current DNS records for {domain}...")

        # Check A records
        a_values = []
        for record in a_records:
            if 'ResourceRecords' in record:
                for rr in record['ResourceRecords']:
                    a_values.append(rr['Value'])

        print(f"📍 Current A records: {a_values}")

        old_a_found = any(value in self.OLD_A_VALUES for value in a_values)
        if not old_a_found:
            print(f"⚠️  Warning: No old Webflow A records found. Expected: {self.OLD_A_VALUES}")
            response = input("Continue anyway? (y/N): ")
            if response.lower() != 'y':
                return False
        else:
            print("✅ Found old Webflow A record(s)")

        # Check CNAME record
        if cname_record and 'ResourceRecords' in cname_record:
            cname_value = cname_record['ResourceRecords'][0]['Value'].rstrip('.')
            print(f"📍 Current www CNAME: {cname_value}")

            if cname_value != self.OLD_CNAME_VALUE:
                print(f"⚠️  Warning: CNAME doesn't match expected old value. Expected: {self.OLD_CNAME_VALUE}")
                response = input("Continue anyway? (y/N): ")
                if response.lower() != 'y':
                    return False
            else:
                print("✅ Found old Webflow CNAME record")
        else:
            print("⚠️  Warning: No www CNAME record found")
            response = input("Continue anyway? (y/N): ")
            if response.lower() != 'y':
                return False

        return True

    def update_a_records(self, hosted_zone_id, domain, a_records):
        """Update A records: keep one with new value, remove others."""
        changes = []

        # Remove all existing A records
        for record in a_records:
            changes.append({
                'Action': 'DELETE',
                'ResourceRecordSet': record
            })

        # Add new A record
        changes.append({
            'Action': 'CREATE',
            'ResourceRecordSet': {
                'Name': domain,
                'Type': 'A',
                'TTL': 300,
                'ResourceRecords': [{'Value': self.NEW_A_VALUE}]
            }
        })

        return self.execute_changes(hosted_zone_id, changes, "A records")

    def update_cname_record(self, hosted_zone_id, domain, cname_record):
        """Update www CNAME record to new value."""
        changes = []

        # Remove existing CNAME record
        if cname_record:
            changes.append({
                'Action': 'DELETE',
                'ResourceRecordSet': cname_record
            })

        # Add new CNAME record
        changes.append({
            'Action': 'CREATE',
            'ResourceRecordSet': {
                'Name': f'www.{domain}',
                'Type': 'CNAME',
                'TTL': 300,
                'ResourceRecords': [{'Value': self.NEW_CNAME_VALUE}]
            }
        })

        return self.execute_changes(hosted_zone_id, changes, "CNAME record")

    def execute_changes(self, hosted_zone_id, changes, record_type):
        """Execute DNS record changes."""
        try:
            response = self.route53.change_resource_record_sets(
                HostedZoneId=hosted_zone_id,
                ChangeBatch={'Changes': changes}
            )

            change_id = response['ChangeInfo']['Id']
            print(f"✅ {record_type} update submitted. Change ID: {change_id}")
            return True

        except ClientError as e:
            print(f"❌ Error updating {record_type}: {e}")
            return False

    def migrate_domain(self, domain):
        """Perform complete DNS migration for a domain."""
        print(f"🚀 Starting Webflow DNS migration for: {domain}")

        # Get hosted zone
        hosted_zone_id = self.get_hosted_zone_id(domain)
        if not hosted_zone_id:
            return False

        print(f"✅ Found hosted zone: {hosted_zone_id}")

        # Get current records
        a_records, cname_record = self.get_current_records(hosted_zone_id, domain)

        if not a_records:
            print(f"❌ No A records found for {domain}")
            return False

        # Verify old values
        if not self.verify_old_records(a_records, cname_record, domain):
            print("❌ Migration cancelled")
            return False

        # Confirm migration
        print(f"\n📋 Migration plan for {domain}:")
        print(f"   • Update A record to: {self.NEW_A_VALUE}")
        print(f"   • Update www CNAME to: {self.NEW_CNAME_VALUE}")

        # Show what A records will be removed
        if len(a_records) > 1:
            print(f"   • Remove {len(a_records) - 1} extra A record(s):")
            for i, record in enumerate(a_records):
                if 'ResourceRecords' in record:
                    values = [rr['Value'] for rr in record['ResourceRecords']]
                    print(f"     - A record {i+1}: {', '.join(values)}")
        else:
            print(f"   • Replace existing A record")

        confirm = input("\n🤔 Proceed with migration? (y/N): ")
        if confirm.lower() != 'y':
            print("❌ Migration cancelled")
            return False

        # Execute migration
        print(f"\n🔄 Migrating DNS records...")

        success = True
        success &= self.update_a_records(hosted_zone_id, domain, a_records)
        success &= self.update_cname_record(hosted_zone_id, domain, cname_record)

        if success:
            print(f"\n🎉 DNS migration completed successfully for {domain}!")
            print(f"📝 Records updated:")
            print(f"   • {domain} A → {self.NEW_A_VALUE}")
            print(f"   • www.{domain} CNAME → {self.NEW_CNAME_VALUE}")
            print(f"\n⏰ Changes may take a few minutes to propagate worldwide.")
        else:
            print(f"\n❌ Migration failed for {domain}")

        return success


def main():
    parser = argparse.ArgumentParser(description='Migrate Webflow DNS records in Route 53')
    parser.add_argument('domain', help='Root domain to migrate (e.g., example.com)')
    args = parser.parse_args()

    migrator = WebflowDNSMigrator()
    success = migrator.migrate_domain(args.domain)

    sys.exit(0 if success else 1)


if __name__ == '__main__':
    main()