rename

local-app/python-tools/route53-migration/delete_zone.py

@@ -0,0 +1,395 @@
+#!/usr/bin/env python3
+"""
+delete_zone.py v1.0.0
+
+Remove all deletable resource records from a Route 53 hosted zone,
+then optionally delete the zone itself.
+
+SOA and apex NS records are skipped — AWS requires them to be present
+and will reject attempts to delete them.
+
+Safety backups of all records and zone metadata are always written
+before any destructive action is taken.
+
+Usage:
+  # Dry run — show what would be deleted, write backups
+  python delete_zone.py --zone-id Z1234567890ABC --profile source-account --dry-run
+
+  # Empty the zone (delete records, keep zone)
+  python delete_zone.py --zone-id Z1234567890ABC --profile source-account
+
+  # Empty and delete the zone
+  python delete_zone.py --zone-id Z1234567890ABC --profile source-account --delete-zone
+
+  # Non-interactive (no confirmation prompts)
+  python delete_zone.py --zone-id Z1234567890ABC --profile source-account --delete-zone --yes
+"""
+
+import argparse
+import boto3
+import json
+import os
+import sys
+import time
+from datetime import datetime, timezone
+from typing import Optional
+
+VERSION = "1.0.0"
+
+SKIP_TYPES  = {'SOA', 'NS'}
+BATCH_SIZE  = 100
+
+
+def get_client(profile: Optional[str], region: str, service: str = 'route53'):
+    session = boto3.Session(profile_name=profile, region_name=region)
+    return session.client(service)
+
+
+def get_account_id(profile: Optional[str]) -> str:
+    try:
+        session = boto3.Session(profile_name=profile)
+        return session.client('sts').get_caller_identity()['Account']
+    except Exception as e:
+        return f"UNKNOWN ({e})"
+
+
+def get_zone_details(client, zone_id: str) -> dict:
+    try:
+        response = client.get_hosted_zone(Id=zone_id)
+    except client.exceptions.NoSuchHostedZone:
+        print(f"ERROR: Hosted zone '{zone_id}' not found.", file=sys.stderr)
+        sys.exit(1)
+    except Exception as e:
+        print(f"ERROR fetching zone details: {e}", file=sys.stderr)
+        sys.exit(1)
+
+    return {
+        'HostedZone':    response['HostedZone'],
+        'DelegationSet': response.get('DelegationSet', {}),
+        'VPCs':          response.get('VPCs', []),
+    }
+
+
+def get_all_records(client, zone_id: str) -> list:
+    records = []
+    paginator = client.get_paginator('list_resource_record_sets')
+    for page in paginator.paginate(HostedZoneId=zone_id):
+        records.extend(page['ResourceRecordSets'])
+    return records
+
+
+def get_tags(client, zone_id: str) -> dict:
+    try:
+        response = client.list_tags_for_resource(
+            ResourceType='hostedzone',
+            ResourceId=zone_id
+        )
+        return {tag['Key']: tag['Value'] for tag in response['ResourceTagSet']['Tags']}
+    except Exception as e:
+        print(f"WARNING: Could not fetch tags: {e}", file=sys.stderr)
+        return {}
+
+
+# ---------------------------------------------------------------------------
+# Safety backups
+# ---------------------------------------------------------------------------
+
+def backup_filename(zone_id: str, suffix: str, output_dir: str) -> str:
+    ts = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')
+    return os.path.join(output_dir, f"{zone_id}_{suffix}_{ts}.json")
+
+
+def write_records_backup(records: list, zone_id: str, output_dir: str) -> str:
+    path = backup_filename(zone_id, 'records', output_dir)
+    data = {
+        'SchemaVersion':     '1.0',
+        'ToolVersion':       VERSION,
+        'HostedZoneId':      zone_id,
+        'RecordCount':       len(records),
+        'ExportedAt':        datetime.now(timezone.utc).isoformat(),
+        'ResourceRecordSets': records,
+    }
+    with open(path, 'w') as f:
+        json.dump(data, f, indent=2, default=str)
+    print(f"  Records backup : {path}")
+    return path
+
+
+def write_zone_backup(details: dict, tags: dict, account_id: str,
+                      zone_id: str, output_dir: str) -> str:
+    path = backup_filename(zone_id, 'zone', output_dir)
+    hz     = details['HostedZone']
+    config = hz.get('Config', {})
+    delegation = details.get('DelegationSet', {})
+
+    data = {
+        'SchemaVersion': '1.0',
+        'ToolVersion':   VERSION,
+        'AccountId':     account_id,
+        'ExportedAt':    datetime.now(timezone.utc).isoformat(),
+        'HostedZone': {
+            'Id':              zone_id,
+            'Name':            hz.get('Name', ''),
+            'Comment':         config.get('Comment', ''),
+            'PrivateZone':     config.get('PrivateZone', False),
+            'CallerReference': hz.get('CallerReference', ''),
+            'ResourceRecordSetCount': hz.get('ResourceRecordSetCount', 'N/A'),
+        },
+        'DelegationSet': {
+            'Id':              delegation.get('Id', ''),
+            'CallerReference': delegation.get('CallerReference', ''),
+            'NameServers':     delegation.get('NameServers', []),
+        },
+        'AssociatedVPCs': [
+            {'VPCId': v.get('VPCId', ''), 'VPCRegion': v.get('VPCRegion', '')}
+            for v in details.get('VPCs', [])
+        ],
+        'Tags': tags,
+    }
+    with open(path, 'w') as f:
+        json.dump(data, f, indent=2, default=str)
+    print(f"  Zone backup    : {path}")
+    return path
+
+
+# ---------------------------------------------------------------------------
+# Record deletion
+# ---------------------------------------------------------------------------
+
+def partition_records(records: list, zone_name: str) -> tuple[list, list]:
+    """
+    Split records into deletable and skipped (SOA + apex NS).
+    zone_name should be the fully-qualified zone apex (e.g. 'example.com.').
+    """
+    deletable = []
+    skipped   = []
+
+    for record in records:
+        rtype = record['Type']
+        rname = record['Name']
+
+        # Always skip SOA
+        if rtype == 'SOA':
+            skipped.append(record)
+            continue
+
+        # Skip apex NS (AWS-managed); non-apex NS delegations are deletable
+        if rtype == 'NS' and rname == zone_name:
+            skipped.append(record)
+            continue
+
+        deletable.append(record)
+
+    return deletable, skipped
+
+
+def build_delete_batch(records: list) -> dict:
+    return {
+        'Changes': [
+            {'Action': 'DELETE', 'ResourceRecordSet': r}
+            for r in records
+        ]
+    }
+
+
+def wait_for_change(client, change_id: str, timeout: int = 120):
+    print(f"    Waiting for change {change_id}", end='', flush=True)
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        status = client.get_change(Id=change_id)['ChangeInfo']['Status']
+        if status == 'INSYNC':
+            print(" done.")
+            return
+        print('.', end='', flush=True)
+        time.sleep(5)
+    print(f"\n    WARNING: Change {change_id} did not reach INSYNC within {timeout}s.")
+
+
+def delete_records(client, zone_id: str, records: list, dry_run: bool) -> dict:
+    results = {'succeeded': [], 'failed': []}
+    total   = len(records)
+
+    if total == 0:
+        print("\n  No deletable records found.")
+        return results
+
+    print(f"\n  Deleting {total} records in batches of {BATCH_SIZE}...")
+
+    for i in range(0, total, BATCH_SIZE):
+        chunk     = records[i:i + BATCH_SIZE]
+        batch_num = (i // BATCH_SIZE) + 1
+        print(f"\n  Batch {batch_num} ({len(chunk)} records):")
+        for r in chunk:
+            print(f"    DELETE  {r['Type']:<8} {r['Name']}")
+
+        if dry_run:
+            print(f"    [DRY RUN] Skipping actual deletion.")
+            results['succeeded'].extend([r['Name'] for r in chunk])
+            continue
+
+        try:
+            batch     = build_delete_batch(chunk)
+            response  = client.change_resource_record_sets(
+                HostedZoneId=zone_id,
+                ChangeBatch=batch
+            )
+            change_id = response['ChangeInfo']['Id'].split('/')[-1]
+            wait_for_change(client, change_id)
+            results['succeeded'].extend([r['Name'] for r in chunk])
+        except Exception as e:
+            print(f"    ERROR in batch {batch_num}: {e}", file=sys.stderr)
+            results['failed'].extend([r['Name'] for r in chunk])
+
+    return results
+
+
+# ---------------------------------------------------------------------------
+# Zone deletion
+# ---------------------------------------------------------------------------
+
+def delete_hosted_zone(client, zone_id: str, dry_run: bool) -> bool:
+    if dry_run:
+        print(f"\n  [DRY RUN] Would delete hosted zone {zone_id}.")
+        return True
+    try:
+        client.delete_hosted_zone(Id=zone_id)
+        print(f"\n  Hosted zone {zone_id} deleted successfully.")
+        return True
+    except client.exceptions.HostedZoneNotEmpty as e:
+        print(f"\n  ERROR: Zone is not empty — cannot delete: {e}", file=sys.stderr)
+        return False
+    except Exception as e:
+        print(f"\n  ERROR deleting zone: {e}", file=sys.stderr)
+        return False
+
+
+# ---------------------------------------------------------------------------
+# Confirmation prompt
+# ---------------------------------------------------------------------------
+
+def confirm(prompt: str) -> bool:
+    while True:
+        answer = input(prompt + " [yes/no]: ").strip().lower()
+        if answer == 'yes':
+            return True
+        elif answer == 'no':
+            return False
+        print("  Please type 'yes' or 'no'.")
+
+
+# ---------------------------------------------------------------------------
+# Main
+# ---------------------------------------------------------------------------
+
+def main():
+    parser = argparse.ArgumentParser(
+        description=f'Empty a Route 53 hosted zone and optionally delete it. v{VERSION}'
+    )
+    parser.add_argument('--zone-id',     required=True, help='Hosted Zone ID (e.g., Z1234567890ABC)')
+    parser.add_argument('--profile',     help='AWS profile name')
+    parser.add_argument('--region',      default='us-east-1', help='AWS region (default: us-east-1)')
+    parser.add_argument('--delete-zone', action='store_true',
+                        help='Delete the hosted zone after emptying it')
+    parser.add_argument('--dry-run',     action='store_true',
+                        help='Show what would happen without making any changes')
+    parser.add_argument('--yes',         action='store_true',
+                        help='Skip confirmation prompts')
+    parser.add_argument('--output-dir',  default='.',
+                        help='Directory for backup JSON files (default: current directory)')
+    args = parser.parse_args()
+
+    zone_id    = args.zone_id.split('/')[-1]
+    client     = get_client(args.profile, args.region)
+    account_id = get_account_id(args.profile)
+    details    = get_zone_details(client, zone_id)
+    tags       = get_tags(client, zone_id)
+    records    = get_all_records(client, zone_id)
+
+    hz        = details['HostedZone']
+    zone_name = hz.get('Name', '')
+    is_private = hz.get('Config', {}).get('PrivateZone', False)
+
+    print(f"\ndelete_zone.py v{VERSION}")
+    print("=" * 60)
+    print("ZONE DELETION PLAN")
+    print("=" * 60)
+    print(f"\n{'Zone ID':<22} {zone_id}")
+    print(f"{'Zone Name':<22} {zone_name}")
+    print(f"{'Account ID':<22} {account_id}")
+    print(f"{'Private Zone':<22} {is_private}")
+    print(f"{'Total Records':<22} {len(records)}")
+
+    deletable, skipped = partition_records(records, zone_name)
+    print(f"{'Deletable Records':<22} {len(deletable)}")
+    print(f"{'Skipped (SOA/NS)':<22} {len(skipped)}")
+    print(f"{'Delete Zone':<22} {args.delete_zone}")
+    print(f"{'Dry Run':<22} {args.dry_run}")
+
+    # Always write safety backups before touching anything
+    print("\n--- Safety Backups " + "-" * 40)
+    if args.dry_run:
+        print("  [DRY RUN] Backups would be written here:")
+        print(f"  Records backup : {backup_filename(zone_id, 'records', args.output_dir)}")
+        print(f"  Zone backup    : {backup_filename(zone_id, 'zone', args.output_dir)}")
+    else:
+        os.makedirs(args.output_dir, exist_ok=True)
+        write_records_backup(records, zone_id, args.output_dir)
+        write_zone_backup(details, tags, account_id, zone_id, args.output_dir)
+
+    # Confirm record deletion
+    print("\n--- Record Deletion " + "-" * 40)
+    if deletable:
+        for r in sorted(deletable, key=lambda x: (x['Name'], x['Type'])):
+            print(f"  DELETE  {r['Type']:<8} {r['Name']}")
+    else:
+        print("  Nothing to delete.")
+
+    if skipped:
+        print(f"\n  Skipping (AWS-managed):")
+        for r in skipped:
+            print(f"  KEEP    {r['Type']:<8} {r['Name']}")
+
+    if deletable:
+        if not args.dry_run and not args.yes:
+            if not confirm(f"\nDelete {len(deletable)} records from zone {zone_id}?"):
+                print("Aborted.")
+                sys.exit(0)
+
+        delete_results = delete_records(client, zone_id, deletable, args.dry_run)
+    else:
+        delete_results = {'succeeded': [], 'failed': []}
+
+    # Confirm zone deletion
+    if args.delete_zone:
+        print("\n--- Zone Deletion " + "-" * 42)
+        if delete_results['failed']:
+            print("  WARNING: Some record deletions failed. Skipping zone deletion to avoid data loss.")
+        else:
+            if not args.dry_run and not args.yes:
+                if not confirm(f"Permanently delete hosted zone {zone_id} ({zone_name})?"):
+                    print("  Zone deletion skipped.")
+                else:
+                    delete_hosted_zone(client, zone_id, args.dry_run)
+            else:
+                delete_hosted_zone(client, zone_id, args.dry_run)
+
+    # Summary
+    print("\n--- Summary " + "-" * 47)
+    if args.dry_run:
+        print(f"  Would delete records : {len(deletable)}")
+        print(f"  Would keep (SOA/NS)  : {len(skipped)}")
+        if args.delete_zone:
+            print(f"  Would delete zone    : {zone_id}")
+    else:
+        print(f"  Records succeeded    : {len(delete_results['succeeded'])}")
+        print(f"  Records failed       : {len(delete_results['failed'])}")
+        print(f"  Kept (SOA/NS)        : {len(skipped)}")
+        if delete_results['failed']:
+            print(f"\n  Failed records:")
+            for name in delete_results['failed']:
+                print(f"    {name}")
+    print()
+
+
+if __name__ == '__main__':
+    main()