Extract Manager zones configuration, add checks for matching utf-u and idna zone names

octodns/manager.py

@@ -118,8 +118,8 @@ class Manager(object):
         with open(config_file, 'r') as fh:
             self.config = safe_load(fh, enforce_order=False)
 
-        # convert the zones portion of things into an IdnaDict
-        self.config['zones'] = IdnaDict(self.config['zones'])
+        zones = self.config['zones']
+        self.config['zones'] = self._config_zones(zones)
 
         manager_config = self.config.get('manager', {})
         self._executor = self._config_executor(manager_config, max_workers)
@@ -144,6 +144,24 @@ class Manager(object):
         )
         self.plan_outputs = self._config_plan_outputs(plan_outputs_config)
 
+    def _config_zones(self, zones):
+        # record the set of configured zones we have as they are
+        configured_zones = set([z.lower() for z in zones.keys()])
+        # walk the configured zones
+        for name in configured_zones:
+            if 'xn--' not in name:
+                continue
+            # this is an IDNA format zone name
+            decoded = idna_decode(name)
+            # do we also have a config for its utf-8
+            if decoded in configured_zones:
+                raise ManagerException(
+                    f'"{decoded}" configured both in utf-8 and idna "{name}"'
+                )
+
+        # convert the zones portion of things into an IdnaDict
+        return IdnaDict(zones)
+
     def _config_executor(self, manager_config, max_workers=None):
         max_workers = (
             manager_config.get('max_workers', 1)

tests/test_octodns_manager.py

@@ -13,6 +13,7 @@ from os import environ
 from os.path import dirname, isfile, join
 
 from octodns import __VERSION__
+from octodns.idna import IdnaDict, idna_encode
 from octodns.manager import (
     _AggregateTarget,
     MainThreadExecutor,
@@ -831,6 +832,41 @@ class TestManager(TestCase):
             set(), manager.configured_sub_zones('bar.foo.unit.tests.')
         )
 
+    def test_config_zones(self):
+        manager = Manager(get_config_filename('simple.yaml'))
+
+        # empty == empty
+        self.assertEqual({}, manager._config_zones({}))
+
+        # single ascii comes back as-is, but in a IdnaDict
+        zones = manager._config_zones({'unit.tests.': 42})
+        self.assertEqual({'unit.tests.': 42}, zones)
+        self.assertIsInstance(zones, IdnaDict)
+
+        # single utf-8 comes back idna encoded
+        self.assertEqual(
+            {idna_encode('Déjà.vu.'): 42},
+            dict(manager._config_zones({'Déjà.vu.': 42})),
+        )
+
+        # ascii and non-matching idna as ok
+        self.assertEqual(
+            {idna_encode('déjà.vu.'): 42, 'deja.vu.': 43},
+            dict(
+                manager._config_zones(
+                    {idna_encode('déjà.vu.'): 42, 'deja.vu.': 43}
+                )
+            ),
+        )
+
+        with self.assertRaises(ManagerException) as ctx:
+            # zone configured with both utf-8 and idna is an error
+            manager._config_zones({'Déjà.vu.': 42, idna_encode('Déjà.vu.'): 43})
+        self.assertEqual(
+            '"déjà.vu." configured both in utf-8 and idna "xn--dj-kia8a.vu."',
+            str(ctx.exception),
+        )
+
 
 class TestMainThreadExecutor(TestCase):
     def test_success(self):