IdnaError exception handling, ensure validation happens on encoded names

octodns/idna.py

@@ -4,13 +4,18 @@
 
 from collections.abc import MutableMapping
 
-from idna import decode as _decode, encode as _encode
+from idna import IDNAError as _IDNAError, decode as _decode, encode as _encode
 
 # Providers will need to to make calls to these at the appropriate points,
 # generally right before they pass names off to api calls. For an example of
 # usage see https://github.com/octodns/octodns-ns1/pull/20
 
 
+class IdnaError(Exception):
+    def __init__(self, idna_error):
+        super().__init__(str(idna_error))
+
+
 def idna_encode(name):
     # Based on https://github.com/psf/requests/pull/3695/files
     # #diff-0debbb2447ce5debf2872cb0e17b18babe3566e9d9900739e8581b355bd513f7R39
@@ -20,21 +25,27 @@ def idna_encode(name):
         # No utf8 chars, just use as-is
         return name
     except UnicodeEncodeError:
+        try:
             if name.startswith('*'):
                 # idna.encode doesn't like the *
                 name = _encode(name[2:]).decode('utf-8')
                 return f'*.{name}'
             return _encode(name).decode('utf-8')
+        except _IDNAError as e:
+            raise IdnaError(e)
 
 
 def idna_decode(name):
     pieces = name.lower().split('.')
     if any(p.startswith('xn--') for p in pieces):
+        try:
             # it's idna
             if name.startswith('*'):
                 # idna.decode doesn't like the *
                 return f'*.{_decode(name[2:])}'
             return _decode(name)
+        except _IDNAError as e:
+            raise IdnaError(e)
     # not idna, just return as-is
     return name
 

octodns/record/__init__.py

@@ -16,7 +16,7 @@ import re
 from fqdn import FQDN
 
 from ..equality import EqualityTupleMixin
-from ..idna import idna_decode, idna_encode
+from ..idna import IdnaError, idna_decode, idna_encode
 from .geo import GeoCodes
 
 
@@ -105,7 +105,13 @@ class Record(EqualityTupleMixin):
 
     @classmethod
     def new(cls, zone, name, data, source=None, lenient=False):
+        reasons = []
+        try:
             name = idna_encode(str(name))
+        except IdnaError as e:
+            # convert the error into a reason
+            reasons.append(str(e))
+            name = str(name)
         fqdn = f'{name}.{zone.name}' if name else zone.name
         try:
             _type = data['type']
@@ -115,7 +121,7 @@ class Record(EqualityTupleMixin):
             _class = cls._CLASSES[_type]
         except KeyError:
             raise Exception(f'Unknown record type: "{_type}"')
-        reasons = _class.validate(name, fqdn, data)
+        reasons.extend(_class.validate(name, fqdn, data))
         try:
             lenient |= data['octodns']['lenient']
         except KeyError:
@@ -145,6 +151,7 @@ class Record(EqualityTupleMixin):
                     f'invalid label, "{label}" is too long at {n}'
                     ' chars, max is 63'
                 )
+        # TODO: look at the idna lib for a lot more potential validations...
         try:
             ttl = int(data['ttl'])
             if ttl < 0:
@@ -191,6 +198,8 @@ class Record(EqualityTupleMixin):
 
     @property
     def fqdn(self):
+        # TODO: these should be calculated and set in __init__ rather than on
+        # each use
         if self.name:
             return f'{self.name}.{self.zone.name}'
         return self.zone.name

tests/test_octodns_idna.py

@@ -11,7 +11,7 @@ from __future__ import (
 
 from unittest import TestCase
 
-from octodns.idna import IdnaDict, idna_decode, idna_encode
+from octodns.idna import IdnaDict, IdnaError, idna_decode, idna_encode
 
 
 class TestIdna(TestCase):
@@ -67,6 +67,15 @@ class TestIdna(TestCase):
             'xn--zajzyk-y4a.pl.', idna_encode(idna_encode('zajęzyk.pl.'))
         )
 
+    def test_exception_translation(self):
+        with self.assertRaises(IdnaError) as ctx:
+            idna_encode('déjà..vu.')
+        self.assertEqual('Empty Label', str(ctx.exception))
+
+        with self.assertRaises(IdnaError) as ctx:
+            idna_decode('xn--djvu-1na6c..com.')
+        self.assertEqual('Empty Label', str(ctx.exception))
+
 
 class TestIdnaDict(TestCase):
     plain = 'testing.tests.'

tests/test_octodns_record.py

@@ -1889,6 +1889,51 @@ class TestRecordValidation(TestCase):
             self.zone, name, {'ttl': 300, 'type': 'A', 'value': '1.2.3.4'}
         )
 
+        # make sure we're validating with encoded fqdns
+        utf8 = 'déjà-vu'
+        padding = ('.' + ('x' * 57)) * 4
+        utf8_name = f'{utf8}{padding}'
+        # make sure our test is valid here, we're under 253 chars long as utf8
+        self.assertEqual(251, len(f'{utf8_name}.{self.zone.name}'))
+        with self.assertRaises(ValidationError) as ctx:
+            Record.new(
+                self.zone,
+                utf8_name,
+                {'ttl': 300, 'type': 'A', 'value': '1.2.3.4'},
+            )
+        reason = ctx.exception.reasons[0]
+        self.assertTrue(reason.startswith('invalid fqdn, "déjà-vu'))
+        self.assertTrue(
+            reason.endswith(
+                '.unit.tests." is too long at 259' ' chars, max is 253'
+            )
+        )
+
+        # same, but with ascii version of things
+        plain = 'deja-vu'
+        plain_name = f'{plain}{padding}'
+        self.assertEqual(251, len(f'{plain_name}.{self.zone.name}'))
+        Record.new(
+            self.zone, plain_name, {'ttl': 300, 'type': 'A', 'value': '1.2.3.4'}
+        )
+
+        # check that we're validating encoded labels
+        padding = 'x' * (60 - len(utf8))
+        utf8_name = f'{utf8}{padding}'
+        # make sure the test is valid, we're at 63 chars
+        self.assertEqual(60, len(utf8_name))
+        with self.assertRaises(ValidationError) as ctx:
+            Record.new(
+                self.zone,
+                utf8_name,
+                {'ttl': 300, 'type': 'A', 'value': '1.2.3.4'},
+            )
+        reason = ctx.exception.reasons[0]
+        # Unfortunately this is a translated IDNAError so we don't have much
+        # control over the exact message :-/ (doesn't give context like octoDNS
+        # does)
+        self.assertEqual('Label too long', reason)
+
         # no ttl
         with self.assertRaises(ValidationError) as ctx:
             Record.new(self.zone, '', {'type': 'A', 'value': '1.2.3.4'})