Fix validation for dynamic records with IPv4+IPv6 subnets

octodns/record/dynamic.py

@@ -3,6 +3,7 @@
 #
 
 import re
+from collections import defaultdict
 from logging import getLogger
 
 from .change import Update
@@ -220,7 +221,7 @@ class _DynamicMixin(object):
         reasons = []
         pools_seen = set()
 
-        subnets_seen = {}
+        subnets_seen = defaultdict(dict)
         geos_seen = {}
 
         if not isinstance(rules, (list, tuple)):
@@ -291,11 +292,21 @@ class _DynamicMixin(object):
                             # previous loop will log any invalid subnets, here we
                             # process only valid ones and skip invalid ones
                             pass
-                    # sort subnets from largest to smallest so that we can
+
+                    # subnets of type IPv4 and IPv6 can't be sorted together
+                    # separately sort them and then combine them
+                    networks_by_type = defaultdict(list)
+                    for network in networks:
+                        networks_by_type[network.__class__].append(network)
+                    sorted_networks = []
+                    for _, networks_of_type in networks_by_type.items():
+                        sorted_networks.extend(sorted(networks_of_type))
+
                     # detect rule that have needlessly targeted a more specific
                     # subnet along with a larger subnet that already contains it
-                    for subnet in sorted(networks):
-                        for seen, where in subnets_seen.items():
+                    for subnet in sorted_networks:
+                        subnets_seen_of_type = subnets_seen[subnet.__class__]
+                        for seen, where in subnets_seen_of_type.items():
                             if subnet == seen:
                                 reasons.append(
                                     f'rule {rule_num} targets subnet {subnet} which has previously been seen in rule {where}'
@@ -305,7 +316,7 @@ class _DynamicMixin(object):
                                     f'rule {rule_num} targets subnet {subnet} which is more specific than the previously seen {seen} in rule {where}'
                                 )
 
-                        subnets_seen[subnet] = rule_num
+                        subnets_seen_of_type[subnet] = rule_num
 
                 if not isinstance(geos, (list, tuple)):
                     reasons.append(f'rule {rule_num} geos must be a list')

tests/test_octodns_record_dynamic.py

@@ -1561,3 +1561,46 @@ class TestRecordDynamic(TestCase):
                 ]
             ),
         )
+
+    def test_dynamic_subnet_mixed_versions(self):
+        # mixed IPv4 and IPv6 subnets should not raise a validation error
+        Record.new(
+            self.zone,
+            'good',
+            {
+                'dynamic': {
+                    'pools': {
+                        'one': {'values': [{'value': '1.1.1.1'}]},
+                        'two': {'values': [{'value': '2.2.2.2'}]},
+                    },
+                    'rules': [
+                        {'subnets': ['10.1.0.0/16', '1::/66'], 'pool': 'one'},
+                        {'pool': 'two'},
+                    ],
+                },
+                'ttl': 60,
+                'type': 'A',
+                'values': ['2.2.2.2'],
+            },
+        )
+
+        Record.new(
+            self.zone,
+            'good',
+            {
+                'dynamic': {
+                    'pools': {
+                        'one': {'values': [{'value': '1.1.1.1'}]},
+                        'two': {'values': [{'value': '2.2.2.2'}]},
+                    },
+                    'rules': [
+                        {'subnets': ['10.1.0.0/16'], 'pool': 'one'},
+                        {'subnets': ['1::/66'], 'pool': 'two'},
+                        {'pool': 'two'},
+                    ],
+                },
+                'ttl': 60,
+                'type': 'A',
+                'values': ['2.2.2.2'],
+            },
+        )