Added escaping semicolon for TXT records

CHANGELOG.md

@@ -1,3 +1,10 @@
+## v0.99.2 - 2024-03-14 - Add escaping semicolon for TXT records
+
+#### Changes
+
+* Add escaping semicolon for TXT records in `SelectelProvider` and `SelectelProviderLegacy`
+
+
 ## v0.99.1 - 2024-02-01 - Fix project structure for distribution
 
 #### Changes

octodns_selectel/escaping_semicolon.py

@@ -0,0 +1,6 @@
+def escape_semicolon(s):
+    return s.replace(';', '\\;')
+
+
+def unescape_semicolon(s):
+    return s.replace('\\;', ';')

octodns_selectel/v1/provider.py

@@ -9,6 +9,10 @@
 from octodns.provider.base import BaseProvider
 from octodns.record import Record, Update
 
+from octodns_selectel.escaping_semicolon import (
+    escape_semicolon,
+    unescape_semicolon,
+)
 from octodns_selectel.version import __version__ as provider_version
 
 
@@ -149,6 +153,15 @@ def _params_for_single(self, record):
             'type': record._type,
         }
 
+    def _params_for_TXT(self, record):
+        for value in record.values:
+            yield {
+                'content': unescape_semicolon(value),
+                'name': record.fqdn,
+                'ttl': max(self.MIN_TTL, record.ttl),
+                'type': record._type,
+            }
+
     def _params_for_MX(self, record):
         for value in record.values:
             yield {
@@ -185,7 +198,6 @@ def _params_for_SSHFP(self, record):
     _params_for_A = _params_for_multiple
     _params_for_AAAA = _params_for_multiple
     _params_for_NS = _params_for_multiple
-    _params_for_TXT = _params_for_multiple
 
     _params_for_CNAME = _params_for_single
     _params_for_ALIAS = _params_for_single
@@ -231,7 +243,7 @@ def _data_for_TXT(self, _type, records):
         return {
             'ttl': records[0]['ttl'],
             'type': _type,
-            'values': [r['content'] for r in records],
+            'values': [escape_semicolon(r['content']) for r in records],
         }
 
     def _data_for_SRV(self, _type, records):

octodns_selectel/v2/mappings.py

@@ -1,5 +1,10 @@
 from string import Template
 
+from octodns_selectel.escaping_semicolon import (
+    escape_semicolon,
+    unescape_semicolon,
+)
+
 from .exceptions import SelectelException
 
 
@@ -17,7 +22,8 @@ def to_selectel_rrset(record):
         rrset_records = [{'content': record.value}]
     elif record._type == "TXT":
         rrset_records = [
-            dict(content=f'\"{value}\"') for value in record.values
+            dict(content=f'\"{unescape_semicolon(value)}\"')
+            for value in record.values
         ]
     elif record._type == "MX":
         rrset_records = list(
@@ -76,7 +82,10 @@ def to_octodns_record_data(rrset):
         key_for_record_values = "value"
         record_values = rrset["records"][0]["content"]
     elif rrset_type == "TXT":
-        record_values = [r['content'].strip('"\'') for r in rrset["records"]]
+        record_values = [
+            escape_semicolon(r['content']).strip('"\'')
+            for r in rrset["records"]
+        ]
     elif rrset_type == "MX":
         for record in rrset["records"]:
             preference, exchange = record["content"].split(" ")

tests/v1/test_provider_octodns_selectel_v1.py

@@ -259,6 +259,28 @@ class TestSelectelProvider(TestCase):
         )
     )
 
+    # TXT with semicolon
+    api_record.append(
+        {
+            'type': 'TXT',
+            'ttl': 300,
+            'content': 'v=DKIM1; k=rsa; p=some-key',
+            'name': 'dkim.unit.tests',
+            'id': 18,
+        }
+    )
+    expected.add(
+        Record.new(
+            zone,
+            'dkim',
+            {
+                'ttl': 200,
+                'type': 'TXT',
+                'value': 'v=DKIM1\\; k=rsa\\; p=some-key',
+            },
+        )
+    )
+
     # SSHFP
     api_record.append(
         {
@@ -378,8 +400,8 @@ def test_apply(self, fake_http):
             zone.add_record(record)
 
         plan = provider.plan(zone)
-        self.assertEqual(10, len(plan.changes))
-        self.assertEqual(10, provider.apply(plan))
+        self.assertEqual(11, len(plan.changes))
+        self.assertEqual(11, provider.apply(plan))
 
     @requests_mock.Mocker()
     def test_domain_list(self, fake_http):
@@ -450,8 +472,8 @@ def test_not_exist_domain(self, fake_http):
             zone.add_record(record)
 
         plan = provider.plan(zone)
-        self.assertEqual(10, len(plan.changes))
-        self.assertEqual(10, provider.apply(plan))
+        self.assertEqual(11, len(plan.changes))
+        self.assertEqual(11, provider.apply(plan))
 
     @requests_mock.Mocker()
     def test_delete_no_exist_record(self, fake_http):
@@ -516,8 +538,8 @@ def test_change_record(self, fake_http):
             zone.add_record(record)
 
         plan = provider.plan(zone)
-        self.assertEqual(10, len(plan.changes))
-        self.assertEqual(10, provider.apply(plan))
+        self.assertEqual(11, len(plan.changes))
+        self.assertEqual(11, provider.apply(plan))
 
     @requests_mock.Mocker()
     def test_include_change_returns_false(self, fake_http):

tests/v2/test_selectel_mappings.py

@@ -269,6 +269,8 @@ def test_mapping_record_aaaa(self):
 
     def test_mapping_record_txt(self):
         txt_list = ["\"Buzz\"", "\"Fizz\""]
+        unescaping_dkim_value = "\"v=DKIM1; k=rsa; p=some-key\""
+        escaping_dkim_value = "\"v=DKIM1\\; k=rsa\\; p=some-key\""
         test_pairs = (
             PairTest(
                 TxtRecord(
@@ -307,6 +309,19 @@ def test_mapping_record_txt(self):
                     ],
                 ),
             ),
+            PairTest(
+                TxtRecord(
+                    self.zone,
+                    "dkim",
+                    dict(ttl=self.ttl, value=escaping_dkim_value),
+                ),
+                dict(
+                    name=f"dkim.{self.zone.name}",
+                    ttl=self.ttl,
+                    type="TXT",
+                    records=[dict(content=unescaping_dkim_value)],
+                ),
+            ),
         )
         self._assert_mapping_common(test_pairs)
         self._assert_mapping_values(test_pairs)