Merge pull request #51 from octodns/backwards-compat-ds-fields

Support both old and new DS field names
octodns · Sep 25, 2023 · 61355e6 · 61355e6
2 parents bdbc4b7 + b45cb62
commit 61355e6
Show file tree

Hide file tree

Showing 2 changed files with 116 additions and 14 deletions.
diff --git a/octodns_powerdns/__init__.py b/octodns_powerdns/__init__.py
@@ -11,6 +11,7 @@
 from octodns.provider import ProviderException
 from octodns.provider.base import BaseProvider
 from octodns.record import Record
+from octodns.record.ds import DsValue
 
 from .record import PowerDnsLuaRecord
 
@@ -63,6 +64,10 @@ class PowerDnsBaseProvider(BaseProvider):
     }
     POWERDNS_LEGACY_MODES_OF_OPERATION = {'native', 'master', 'slave'}
 
+    # TODO: once we require octoDNS 2.0 this backwards compatibility code can go
+    # away
+    OLD_DS_FIELDS = hasattr(DsValue, 'flags')
+
     def __init__(
         self,
         id,
@@ -174,17 +179,25 @@ def _data_for_TLSA(self, rrset):
     def _data_for_DS(self, rrset):
         values = []
         for record in rrset['records']:
-            (flags, protocol, algorithm, public_key) = record['content'].split(
+            (key_tag, algorithm, digest_type, digest) = record['content'].split(
                 ' ', 3
             )
-            values.append(
-                {
-                    'flags': flags,
-                    'protocol': protocol,
+            if self.OLD_DS_FIELDS:
+                value = {
+                    'flags': key_tag,
+                    'protocol': algorithm,
+                    'algorithm': digest_type,
+                    'public_key': digest,
+                }
+            else:
+                value = {
+                    'key_tag': key_tag,
                     'algorithm': algorithm,
-                    'public_key': public_key,
+                    'digest_type': digest_type,
+                    'digest': digest,
                 }
-            )
+            values.append(value)
+
         return {'type': rrset['type'], 'values': values, 'ttl': rrset['ttl']}
 
     def _data_for_CAA(self, rrset):
@@ -484,13 +497,16 @@ def _records_for_TLSA(self, record):
         ], record._type
 
     def _records_for_DS(self, record):
-        return [
-            {
-                'content': f'{v.flags} {v.protocol} {v.algorithm} {v.public_key}',
-                'disabled': False,
-            }
-            for v in record.values
-        ], record._type
+        data = []
+        for v in record.values:
+            if self.OLD_DS_FIELDS:
+                content = f'{v.flags} {v.protocol} {v.algorithm} {v.public_key}'
+            else:
+                content = (
+                    f'{v.key_tag} {v.algorithm} {v.digest_type} {v.digest}'
+                )
+            data.append({'content': content, 'disabled': False})
+        return data, record._type
 
     def _records_for_CAA(self, record):
         return [

diff --git a/tests/test_octodns_provider_powerdns.py b/tests/test_octodns_provider_powerdns.py
@@ -542,6 +542,92 @@ def test_list_zones(self):
             )
             self.assertEqual(['alpha.com.', 'zeta.net.'], provider.list_zones())
 
+    def test_data_for_DS_compat(self):
+        provider = PowerDnsProvider('test', 'non.existent', 'api-key')
+
+        rrset = {
+            'records': [{'content': 'one two three four'}],
+            'ttl': 42,
+            'type': 'DS',
+        }
+
+        # old
+        provider.OLD_DS_FIELDS = True
+        value = provider._data_for_DS(rrset)['values'][0]
+        self.assertEqual(
+            {
+                'algorithm': 'three',
+                'flags': 'one',
+                'protocol': 'two',
+                'public_key': 'four',
+            },
+            value,
+        )
+
+        # new
+        provider.OLD_DS_FIELDS = False
+        value = provider._data_for_DS(rrset)['values'][0]
+        self.assertEqual(
+            {
+                'algorithm': 'two',
+                'digest': 'four',
+                'digest_type': 'three',
+                'key_tag': 'one',
+            },
+            value,
+        )
+
+    def test_records_for_DS_compat(self):
+        provider = PowerDnsProvider('test', 'non.existent', 'api-key')
+
+        class DummyRecord:
+            _type = 'DS'
+
+            def __init__(self, value):
+                self.values = [value]
+
+        class OldFields:
+            flags = 'flags'
+            protocol = 'protocol'
+            algorithm = 'algorithm'
+            public_key = 'public_key'
+
+        old_fields = OldFields()
+
+        class NewFields:
+            key_tag = 'key_tag'
+            algorithm = 'algorithm'
+            digest_type = 'digest_type'
+            digest = 'digest'
+
+        new_fields = NewFields()
+
+        # old
+        provider.OLD_DS_FIELDS = True
+        data = provider._records_for_DS(DummyRecord(old_fields))[0]
+        self.assertEqual(
+            [
+                {
+                    'content': 'flags protocol algorithm public_key',
+                    'disabled': False,
+                }
+            ],
+            data,
+        )
+
+        # new
+        provider.OLD_DS_FIELDS = False
+        data = provider._records_for_DS(DummyRecord(new_fields))[0]
+        self.assertEqual(
+            [
+                {
+                    'content': 'key_tag algorithm digest_type digest',
+                    'disabled': False,
+                }
+            ],
+            data,
+        )
+
 
 class TestPowerDnsLuaRecord(TestCase):
     def test_basics(self):