Enforce CNAME constrains for DNS commands

RFC 1912 states that no record (besides PTR) is allowed to coexist
with any other record type. When BIND detects this situation, it
refuses to load such records.

Enforce the constrain for dnsrecord-mod and dnsrecord-add commands.

https://fedorahosted.org/freeipa/ticket/2601

2 files changed, 98 insertions, 24 deletions

diff --git a/ipalib/plugins/dns.py b/ipalib/plugins/dns.py
index 85781491..6727d052 100644
--- a/ipalib/plugins/dns.py
+++ b/ipalib/plugins/dns.py
@@ -2155,6 +2155,26 @@ class dnsrecord(LDAPObject):
                 processed.append(rrparam.name)
                 yield rrparam
 
+    def check_record_type_collisions(self, old_entry, entry_attrs):
+        # Test that only allowed combination of record types was created
+        attrs = set(attr for attr in entry_attrs if attr in _record_attributes
+                        and entry_attrs[attr])
+        attrs.update(attr for attr in old_entry if attr not in entry_attrs)
+        try:
+            attrs.remove('cnamerecord')
+        except KeyError:
+            rec_has_cname = False
+        else:
+            rec_has_cname = True
+        # CNAME and PTR record combination is allowed
+        attrs.discard('ptrrecord')
+        rec_has_other_types = True if attrs else False
+
+        if rec_has_cname and rec_has_other_types:
+            raise errors.ValidationError(name='cnamerecord',
+                      error=_('CNAME record is not allowed to coexist with any other '
+                              'records except PTR'))
+
 api.register(dnsrecord)
 
 
@@ -2297,11 +2317,16 @@ class dnsrecord_add(LDAPCreate):
         # new attributes only and not for all attributes in the LDAP entry
         setattr(context, 'dnsrecord_precallback_attrs', precallback_attrs)
 
+        # We always want to retrieve all DNS record attributes to test for
+        # record type collisions (#2601)
         try:
             (dn_, old_entry) = ldap.get_entry(
-                        dn, entry_attrs.keys(),
+                        dn, _record_attributes,
                         normalize=self.obj.normalize_dn)
-            for attr in old_entry.keys():
+        except errors.NotFound:
+            pass
+        else:
+            for attr in entry_attrs:
                 if attr not in _record_attributes:
                     continue
                 if entry_attrs[attr] is None:
@@ -2310,9 +2335,9 @@ class dnsrecord_add(LDAPCreate):
                     vals = [entry_attrs[attr]]
                 else:
                     vals = list(entry_attrs[attr])
-                entry_attrs[attr] = list(set(old_entry[attr] + vals))
-        except errors.NotFound:
-            pass
+                entry_attrs[attr] = list(set(old_entry.get(attr, []) + vals))
+
+            self.obj.check_record_type_collisions(old_entry, entry_attrs)
         return dn
 
     def exc_callback(self, keys, options, exc, call_func, *call_args, **call_kwargs):
@@ -2386,14 +2411,16 @@ class dnsrecord_mod(LDAPUpdate):
         # Run pre_callback validators
         self.obj.run_precallback_validators(dn, entry_attrs, *keys, **options)
 
-        if len(updated_attrs):
-            try:
-                (dn_, old_entry) = ldap.get_entry(
-                            dn, updated_attrs.keys(),
-                            normalize=self.obj.normalize_dn)
-            except errors.NotFound:
+        # current entry is needed in case of per-dns-record-part updates and
+        # for record type collision check
+        try:
+            (dn_, old_entry) = ldap.get_entry(dn, _record_attributes,
+                                              normalize=self.obj.normalize_dn)
+        except errors.NotFound:
+            if updated_attrs:
                 self.obj.handle_not_found(*keys)
 
+        if updated_attrs:
             for attr in updated_attrs:
                 param = self.params[attr]
                 old_dnsvalue, new_parts = updated_attrs[attr]
@@ -2411,6 +2438,7 @@ class dnsrecord_mod(LDAPUpdate):
                 new_dnsvalue = [param._convert_scalar(modified_parts)]
                 entry_attrs[attr] = list(set(old_entry[attr] + new_dnsvalue))
 
+        self.obj.check_record_type_collisions(old_entry, entry_attrs)
         return dn
 
     def execute(self, *keys, **options):
diff --git a/tests/test_xmlrpc/test_dns_plugin.py b/tests/test_xmlrpc/test_dns_plugin.py
index d121b2f0..9c9b2239 100644
--- a/tests/test_xmlrpc/test_dns_plugin.py
+++ b/tests/test_xmlrpc/test_dns_plugin.py
@@ -48,6 +48,8 @@ dnsrev1 = u'80'
 dnsrev1_dn = DN(('idnsname',dnsrev1), revdnszone1_dn)
 dnsrev2 = u'81'
 dnsrev2_dn = DN(('idnsname',dnsrev2), revdnszone1_dn)
+dnsrescname = u'testcnamerec'
+dnsrescname_dn = DN(('idnsname',dnsrescname), dnszone1_dn)
 
 class test_dns(Declarative):
 
@@ -746,30 +748,64 @@ class test_dns(Declarative):
         ),
 
         dict(
-            desc='Try to add invalid CNAME record %r using dnsrecord_add' % (dnsres1),
-            command=('dnsrecord_add', [dnszone1, dnsres1], {'cnamerecord': u'-.example.com' }),
+            desc='Try to add CNAME record to %r using dnsrecord_add' % (dnsres1),
+            command=('dnsrecord_add', [dnszone1, dnsres1], {'cnamerecord': u'foo-1.example.com.'}),
+            expected=errors.ValidationError(name='cnamerecord',
+                error=u'CNAME record is not allowed to coexist with any other records except PTR'),
+        ),
+
+        dict(
+            desc='Try to add invalid CNAME record %r using dnsrecord_add' % (dnsrescname),
+            command=('dnsrecord_add', [dnszone1, dnsrescname], {'cnamerecord': u'-.example.com'}),
             expected=errors.ValidationError(name='hostname',
                 error=u'invalid domain-name: only letters, numbers, and - ' +
                     u'are allowed. DNS label may not start or end with -'),
         ),
 
         dict(
-            desc='Add CNAME record to %r using dnsrecord_add' % (dnsres1),
-            command=('dnsrecord_add', [dnszone1, dnsres1], {'cnamerecord': u'foo-1.example.com.' }),
+            desc='Add CNAME record to %r using dnsrecord_add' % (dnsrescname),
+            command=('dnsrecord_add', [dnszone1, dnsrescname], {'cnamerecord': u'foo-1.example.com.'}),
             expected={
-                'value': dnsres1,
+                'value': dnsrescname,
                 'summary': None,
                 'result': {
                     'objectclass': objectclasses.dnsrecord,
-                    'dn': unicode(dnsres1_dn),
-                    'idnsname': [dnsres1],
-                    'arecord': [u'10.10.0.1'],
+                    'dn': unicode(dnsrescname_dn),
+                    'idnsname': [dnsrescname],
                     'cnamerecord': [u'foo-1.example.com.'],
                 },
             },
         ),
 
         dict(
+            desc='Try to add other record to CNAME record %r using dnsrecord_add' % (dnsrescname),
+            command=('dnsrecord_add', [dnszone1, dnsrescname], {'arecord': u'10.0.0.1'}),
+            expected=errors.ValidationError(name='cnamerecord',
+                error=u'CNAME record is not allowed to coexist with any other records except PTR'),
+        ),
+
+        dict(
+            desc='Try to add other record to CNAME record %r using dnsrecord_mod' % (dnsrescname),
+            command=('dnsrecord_mod', [dnszone1, dnsrescname], {'arecord': u'10.0.0.1'}),
+            expected=errors.ValidationError(name='cnamerecord',
+                error=u'CNAME record is not allowed to coexist with any other records except PTR'),
+        ),
+
+        dict(
+            desc='Add A record and delete CNAME record in %r with dnsrecord_mod' % (dnsrescname),
+            command=('dnsrecord_mod', [dnszone1, dnsrescname], {'arecord': u'10.0.0.1',
+                                                                'cnamerecord': None}),
+            expected={
+                'value': dnsrescname,
+                'summary': None,
+                'result': {
+                    'idnsname': [dnsrescname],
+                    'arecord': [u'10.0.0.1'],
+                },
+            },
+        ),
+
+        dict(
             desc='Try to add invalid KX record %r using dnsrecord_add' % (dnsres1),
             command=('dnsrecord_add', [dnszone1, dnsres1], {'kxrecord': u'foo-1.example.com' }),
             expected=errors.ValidationError(name='kx_rec',
@@ -788,7 +824,6 @@ class test_dns(Declarative):
                     'dn': unicode(dnsres1_dn),
                     'idnsname': [dnsres1],
                     'arecord': [u'10.10.0.1'],
-                    'cnamerecord': [u'foo-1.example.com.'],
                     'kxrecord': [u'1 foo-1'],
                 },
             },
@@ -805,7 +840,6 @@ class test_dns(Declarative):
                     'dn': unicode(dnsres1_dn),
                     'idnsname': [dnsres1],
                     'arecord': [u'10.10.0.1'],
-                    'cnamerecord': [u'foo-1.example.com.'],
                     'kxrecord': [u'1 foo-1'],
                     'txtrecord': [u'foo bar'],
                 },
@@ -825,7 +859,6 @@ class test_dns(Declarative):
                     'dn': unicode(dnsres1_dn),
                     'idnsname': [dnsres1],
                     'arecord': [u'10.10.0.1'],
-                    'cnamerecord': [u'foo-1.example.com.'],
                     'kxrecord': [u'1 foo-1'],
                     'txtrecord': [u'foo bar'],
                     'nsecrecord': [dnszone1 + u' TXT A'],
@@ -857,7 +890,6 @@ class test_dns(Declarative):
                     'dn': unicode(dnsres1_dn),
                     'idnsname': [dnsres1],
                     'arecord': [u'10.10.0.1'],
-                    'cnamerecord': [u'foo-1.example.com.'],
                     'kxrecord': [u'1 foo-1'],
                     'txtrecord': [u'foo bar'],
                     'nsecrecord': [dnszone1 + u' TXT A'],
@@ -882,7 +914,6 @@ class test_dns(Declarative):
                 'result': {
                     'idnsname': [dnsres1_renamed],
                     'arecord': [u'10.10.0.1'],
-                    'cnamerecord': [u'foo-1.example.com.'],
                     'kxrecord': [u'1 foo-1'],
                     'txtrecord': [u'foo bar'],
                     'nsecrecord': [dnszone1 + u' TXT A'],
@@ -976,6 +1007,21 @@ class test_dns(Declarative):
             },
         ),
 
+        dict(
+            desc='Test that CNAME/PTR record type combination in record %r is allowed' % (dnsrev1),
+            command=('dnsrecord_add', [revdnszone1, dnsrev1], {'cnamerecord': u'foo-1.example.com.' }),
+            expected={
+                'value': dnsrev1,
+                'summary': None,
+                'result': {
+                    'objectclass': objectclasses.dnsrecord,
+                    'dn': unicode(dnsrev1_dn),
+                    'idnsname': [dnsrev1],
+                    'ptrrecord': [u'foo-1.example.com.'],
+                    'cnamerecord': [u'foo-1.example.com.'],
+                },
+            },
+        ),
 
         dict(
             desc='Update global DNS settings',