Merge "Ensure CNAME RRSets only have one record"

designate/central/service.py

@@ -423,6 +423,17 @@ class Service(service.RPCService, service.Service):
                     child_domain['name']
                 raise exceptions.InvalidRecordSetLocation(msg)
 
+    def _is_valid_recordset_records(self, recordset):
+        """
+        Check to make sure that the records in the recordset
+        follow the rules, and won't blow up on the nameserver.
+        """
+        if hasattr(recordset, 'records'):
+            if len(recordset.records) > 1 and recordset.type == 'CNAME':
+                raise exceptions.BadRequest(
+                    'CNAME recordsets may not have more than 1 record'
+                )
+
     def _is_blacklisted_domain_name(self, context, domain_name):
         """
         Ensures the provided domain_name is not blacklisted.
@@ -1204,6 +1215,7 @@ class Service(service.RPCService, service.Service):
                                            recordset.type)
         self._is_valid_recordset_placement_subdomain(
             context, domain, recordset.name)
+        self._is_valid_recordset_records(recordset)
 
         if recordset.obj_attr_is_set('records') and len(recordset.records) > 0:
             if increment_serial:
@@ -1328,6 +1340,7 @@ class Service(service.RPCService, service.Service):
                                            recordset.type, recordset.id)
         self._is_valid_recordset_placement_subdomain(
             context, domain, recordset.name)
+        self._is_valid_recordset_records(recordset)
 
         # Ensure TTL is above the minimum
         ttl = changes.get('ttl', None)

designate/tests/unit/test_central/test_basic.py

@@ -626,6 +626,19 @@ class CentralDomainTestCase(CentralBasic):
                 'bar.example.org.'
             )
 
+    def test_is_valid_recordset_records(self):
+        recordset = RoObject(
+            records=[
+                'ww1.example.com.',
+                'ww2.example.com.'
+            ],
+            type='CNAME'
+        )
+        with testtools.ExpectedException(exceptions.BadRequest):
+            self.service._is_valid_recordset_records(
+                recordset
+            )
+
     def test__is_superdomain(self):
         self.service.storage.find_domains = Mock()
         self.service._is_superdomain(self.context, 'example.org.', '1')