diff --git a/designate/central/service.py b/designate/central/service.py index a6a0df95..845b86e5 100644 --- a/designate/central/service.py +++ b/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) diff --git a/designate/tests/unit/test_central/test_basic.py b/designate/tests/unit/test_central/test_basic.py index 77068ae6..42c90b90 100644 --- a/designate/tests/unit/test_central/test_basic.py +++ b/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')