Merge remote-tracking branch 'origin/master' into update-readme-post-extraction

CHANGELOG.md

@@ -44,6 +44,11 @@
   records created or updated using this version and matching the said edge-case
   will not be read/parsed correctly by the older version and will show a diff.
 
+#### Stuff
+
+* Additional FQDN validation to ALIAS/CNAME value, MX exchange, SRV target and
+  tests of the functionality.
+
 ## v0.9.14 - 2021-10-10 - A new supports system
 
 #### Noteworthy changes

octodns/record/__init__.py

@@ -1091,7 +1091,10 @@ class MxValue(EqualityTupleMixin):
             exchange = None
             try:
                 exchange = value.get('exchange', None) or value['value']
-                if not exchange.endswith('.'):
+                if not FQDN(str(exchange), allow_underscores=True).is_valid:
+                    reasons.append(f'Invalid MX exchange "{exchange}" is not '
+                                   'a valid FQDN.')
+                elif not exchange.endswith('.'):
                     reasons.append(f'MX value "{exchange}" missing trailing .')
             except KeyError:
                 reasons.append('missing exchange')
@@ -1225,7 +1228,10 @@ class _NsValue(object):
             data = (data,)
         reasons = []
         for value in data:
-            if not value.endswith('.'):
+            if not FQDN(str(value), allow_underscores=True).is_valid:
+                reasons.append(f'Invalid NS value "{value}" is not '
+                               'a valid FQDN.')
+            elif not value.endswith('.'):
                 reasons.append(f'NS value "{value}" missing trailing .')
         return reasons
 
@@ -1413,9 +1419,13 @@ class SrvValue(EqualityTupleMixin):
             except ValueError:
                 reasons.append(f'invalid port "{value["port"]}"')
             try:
-                if not value['target'].endswith('.'):
-                    reasons.append(f'SRV value "{value["target"]}" missing '
-                                   'trailing .')
+                target = value['target']
+                if not target.endswith('.'):
+                    reasons.append(f'SRV value "{target}" missing trailing .')
+                if target != '.' and \
+                   not FQDN(str(target), allow_underscores=True).is_valid:
+                    reasons.append(f'Invalid SRV target "{target}" is not '
+                                   'a valid FQDN.')
             except KeyError:
                 reasons.append('missing target')
         return reasons

tests/test_octodns_record.py

@@ -2281,6 +2281,36 @@ class TestRecordValidation(TestCase):
         self.assertEqual(['CNAME value "foo.bar.com" missing trailing .'],
                          ctx.exception.reasons)
 
+        # doesn't allow urls
+        with self.assertRaises(ValidationError) as ctx:
+            Record.new(self.zone, 'www', {
+                'type': 'CNAME',
+                'ttl': 600,
+                'value': 'https://google.com',
+            })
+        self.assertEqual(['CNAME value "https://google.com" is not a valid '
+                          'FQDN'], ctx.exception.reasons)
+
+        # doesn't allow urls with paths
+        with self.assertRaises(ValidationError) as ctx:
+            Record.new(self.zone, 'www', {
+                'type': 'CNAME',
+                'ttl': 600,
+                'value': 'https://google.com/a/b/c',
+            })
+        self.assertEqual(['CNAME value "https://google.com/a/b/c" is not a '
+                          'valid FQDN'], ctx.exception.reasons)
+
+        # doesn't allow paths
+        with self.assertRaises(ValidationError) as ctx:
+            Record.new(self.zone, 'www', {
+                'type': 'CNAME',
+                'ttl': 600,
+                'value': 'google.com/some/path',
+            })
+        self.assertEqual(['CNAME value "google.com/some/path" is not a valid '
+                          'FQDN'], ctx.exception.reasons)
+
     def test_DNAME(self):
         # A valid DNAME record.
         Record.new(self.zone, 'sub', {
@@ -2674,6 +2704,19 @@ class TestRecordValidation(TestCase):
         self.assertEqual(['MX value "foo.bar.com" missing trailing .'],
                          ctx.exception.reasons)
 
+        # exchange must be a valid FQDN
+        with self.assertRaises(ValidationError) as ctx:
+            Record.new(self.zone, '', {
+                'type': 'MX',
+                'ttl': 600,
+                'value': {
+                    'preference': 10,
+                    'exchange': '100 foo.bar.com.'
+                }
+            })
+        self.assertEqual(['Invalid MX exchange "100 foo.bar.com." is not a '
+                          'valid FQDN.'], ctx.exception.reasons)
+
     def test_NXPTR(self):
         # doesn't blow up
         Record.new(self.zone, '', {
@@ -2772,6 +2815,16 @@ class TestRecordValidation(TestCase):
         self.assertEqual(['NS value "foo.bar" missing trailing .'],
                          ctx.exception.reasons)
 
+        # exchange must be a valid FQDN
+        with self.assertRaises(ValidationError) as ctx:
+            Record.new(self.zone, '', {
+                'type': 'NS',
+                'ttl': 600,
+                'value': '100 foo.bar.com.'
+            })
+        self.assertEqual(['Invalid NS value "100 foo.bar.com." is not a '
+                          'valid FQDN.'], ctx.exception.reasons)
+
     def test_PTR(self):
         # doesn't blow up (name & zone here don't make any sense, but not
         # important)
@@ -3089,6 +3142,21 @@ class TestRecordValidation(TestCase):
         self.assertEqual(['SRV value "foo.bar.baz" missing trailing .'],
                          ctx.exception.reasons)
 
+        # target must be a valid FQDN
+        with self.assertRaises(ValidationError) as ctx:
+            Record.new(self.zone, '_srv._tcp', {
+                'type': 'SRV',
+                'ttl': 600,
+                'value': {
+                    'priority': 1,
+                    'weight': 2,
+                    'port': 3,
+                    'target': '100 foo.bar.com.'
+                }
+            })
+        self.assertEqual(['Invalid SRV target "100 foo.bar.com." is not a '
+                          'valid FQDN.'], ctx.exception.reasons)
+
     def test_TXT(self):
         # doesn't blow up (name & zone here don't make any sense, but not
         # important)