fix(dns): remove legacy zones after migration

scripts/migrate_delegated_dns.py

@@ -2,8 +2,8 @@
 """Move legacy per-user deSEC zones into the parent zone's CNAME wildcard model.
 
 Run from the backend environment. The default is a read-only dry run; pass
-``--apply`` to copy records and cut over delegations. Child zones are retained
+``--apply`` to copy records, cut over delegations, export the old zonefiles,
-so their records remain available for rollback.
+and remove the child zones that would otherwise override the parent records.
 """
 
 import argparse
@@ -117,7 +117,7 @@ def _validate_owner(owner: str) -> None:
         raise ValueError(f'Unsupported legacy namespace owner: {owner!r}')
 
 
-def build_plan(api: DeSec) -> Tuple[List[str], List[Dict], Dict[Tuple[str, str], Dict]]:
+def build_plan(api: DeSec) -> Tuple[List[str], List[str], List[Dict], Dict[Tuple[str, str], Dict]]:
     zones_response = api.request('GET', f'{api.base_url}/domains/')
     if zones_response.status_code != 200:
         raise RuntimeError(f'Could not list deSEC zones ({zones_response.status_code})')
@@ -183,7 +183,7 @@ def build_plan(api: DeSec) -> Tuple[List[str], List[Dict], Dict[Tuple[str, str],
                     continue
                 raise RuntimeError(f'Conflicting parent RRset at {owner} {item["type"]}')
 
-    return sorted(namespaces), copies, parent_map
+    return sorted(namespaces), legacy_zones, copies, parent_map
 
 
 def _delete_rrset(api: DeSec, subname: str, record_type: str) -> None:
@@ -200,8 +200,34 @@ def _create_rrset(api: DeSec, payload: Dict) -> None:
         )
 
 
+def _backup_and_remove_legacy_zones(api: DeSec, legacy_zones: Iterable[str], backup_dir: str) -> None:
+    """Export old zones for rollback, then remove their competing authority."""
+    os.makedirs(backup_dir, mode=0o700, exist_ok=True)
+    zonefiles = {}
+    for zone in legacy_zones:
+        url = f'{api.base_url}/domains/{quote(zone, safe=".-")}/zonefile/'
+        response = api.request('GET', url)
+        if response.status_code != 200:
+            raise RuntimeError(f'Could not export legacy zone {zone} ({response.status_code})')
+        zonefiles[zone] = response.text
+
+    for zone, zonefile in zonefiles.items():
+        filename = os.path.join(backup_dir, f'{zone}.zone')
+        with open(filename, 'w', encoding='utf-8') as backup:
+            backup.write(zonefile)
+    print(f'Exported {len(zonefiles)} legacy zone backups to {backup_dir}')
+
+    for zone in legacy_zones:
+        url = f'{api.base_url}/domains/{quote(zone, safe=".-")}/'
+        response = api.request('DELETE', url)
+        if response.status_code not in {204, 404}:
+            raise RuntimeError(f'Could not remove legacy zone {zone} ({response.status_code})')
+        print(f'Removed legacy child zone {zone}')
+
+
 def apply_plan(api: DeSec, namespaces: Iterable[str], copies: List[Dict],
-               parent_map: Dict[Tuple[str, str], Dict]) -> None:
+               legacy_zones: Iterable[str], parent_map: Dict[Tuple[str, str], Dict],
+               backup_dir: str) -> None:
     # Copy child records and create wildcard records first; old delegations stay
     # authoritative until all non-destructive preparation has completed.
     for payload in copies:
@@ -246,6 +272,11 @@ def apply_plan(api: DeSec, namespaces: Iterable[str], copies: List[Dict],
                     })
             raise
 
+    # Hosted child zones continue to answer authoritatively even after their
+    # parent NS/DS records are removed. Back them up and remove them so the
+    # new parent-zone CNAMEs and wildcards are actually served.
+    _backup_and_remove_legacy_zones(api, legacy_zones, backup_dir)
+
     from database_service import database_connection
 
     with database_connection() as connection:
@@ -268,6 +299,10 @@ def apply_plan(api: DeSec, namespaces: Iterable[str], copies: List[Dict],
 def main() -> int:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument('--apply', action='store_true', help='Apply the reviewed DNS migration')
+    parser.add_argument(
+        '--backup-dir', default='/tmp/anchorhost-delegated-dns-backup',
+        help='Directory for exported legacy zonefiles before their zones are deleted',
+    )
     args = parser.parse_args()
     token = os.environ.get('DESEC_API_TOKEN')
     base_domain = os.environ.get('BASE_DOMAIN')
@@ -276,15 +311,15 @@ def main() -> int:
 
     api = DeSec(token, base_domain)
     try:
-        namespaces, copies, parent_map = build_plan(api)
+        namespaces, legacy_zones, copies, parent_map = build_plan(api)
         print(f'Namespaces to migrate: {len(namespaces)}')
         print(f'Legacy RRsets to copy: {len(copies)}')
         print('Namespaces:', ', '.join(namespaces))
         print('Record copies:', ', '.join(f'{item["type"]}:{item["subname"]}' for item in copies) or 'none')
         print('Mode:', 'APPLY' if args.apply else 'DRY RUN')
         if args.apply:
-            apply_plan(api, namespaces, copies, parent_map)
+            apply_plan(api, namespaces, copies, legacy_zones, parent_map, args.backup_dir)
-            print('DNS migration completed. Legacy child zones were retained for rollback.')
+            print('DNS migration completed. Legacy zonefiles are backed up; old child zones are removed.')
         else:
             print('No changes made. Run with --apply to perform the migration.')
     except Exception as error: