encode zone names in url paths

CHANGELOG.md

@@ -1,3 +1,8 @@
+## v0.0.? - 2024-??-?? - ???
+
+* Support for fully managing zones with special characters in their names, e.g.
+  128/26.2.0.192.in-addr.arpa. added.
+
 ## v0.0.6 - 2024-03-08 - Get port type straight
 
 * DS Record support added

octodns_powerdns/__init__.py

@@ -4,6 +4,7 @@
 
 import logging
 from operator import itemgetter
+from urllib.parse import quote_plus
 
 from requests import HTTPError, Session
 
@@ -27,6 +28,15 @@
 __version__ = __VERSION__ = '0.0.6'
 
 
+def _encode_zone_name(name):
+    # Powerdns uses a special encoding for URLs. Instead of "%2F" for a slash,
+    # the slash must be encoded with "=2F". (This must be done in version 4.7.3
+    # from Debian, from version >= 4.8 Powerdns accepts “%2F” and “=2F” as path
+    # argument. The output of "/api/v1/servers/localhost/zones" still shows the
+    # zone URL with "=2F")
+    return quote_plus(name).replace('%', '=')
+
+
 def _escape_unescaped_semicolons(value):
     pieces = value.split(';')
     if len(pieces) == 1:
@@ -453,10 +463,10 @@ def populate(self, zone, target=False, lenient=False):
             target,
             lenient,
         )
-
+        encoded_name = _encode_zone_name(zone.name)
         resp = None
         try:
-            resp = self._get(f'zones/{zone.name}')
+            resp = self._get(f'zones/{encoded_name}')
             self.log.debug('populate:   loaded')
         except HTTPError as e:
             error = self._get_error(e)
@@ -674,6 +684,7 @@ def _get_error(self, http_error):
     def _apply(self, plan):
         desired = plan.desired
         changes = plan.changes
+        encoded_name = _encode_zone_name(desired.name)
         self.log.debug(
             '_apply: zone=%s, len(changes)=%d', desired.name, len(changes)
         )
@@ -691,7 +702,7 @@ def _apply(self, plan):
         self.log.debug('_apply:   sending change request')
 
         try:
-            self._patch(f'zones/{desired.name}', data={'rrsets': mods})
+            self._patch(f'zones/{encoded_name}', data={'rrsets': mods})
             self.log.debug('_apply:   patched')
         except HTTPError as e:
             error = self._get_error(e)
@@ -736,7 +747,7 @@ def _apply(self, plan):
             self.log.debug('_apply:   created')
 
         if self.notify:
-            self._request_notify(desired.name)
+            self._request_notify(encoded_name)
 
         self.log.debug('_apply:   complete')
 

tests/test_octodns_provider_powerdns.py

@@ -19,6 +19,7 @@
 from octodns_powerdns import (
     PowerDnsBaseProvider,
     PowerDnsProvider,
+    _encode_zone_name,
     _escape_unescaped_semicolons,
 )
 from octodns_powerdns.record import PowerDnsLuaRecord, _PowerDnsLuaValue
@@ -792,3 +793,11 @@ def test_lua_validate(self):
         # list w/a bad value
         got = _PowerDnsLuaValue.validate([val, {}], PowerDnsLuaRecord._type)
         self.assertEqual(['missing type', 'missing script'], got)
+
+    def test_encode_zone_name(self):
+        for expected, value in (
+            ('unit.tests.', 'unit.tests.'),
+            ('another_one.unit.tests.', 'another_one.unit.tests.'),
+            ('128=2F26.2.0.192.in-addr.arpa.', '128/26.2.0.192.in-addr.arpa.'),
+        ):
+            self.assertEqual(expected, _encode_zone_name(value))