samba-tool dns: move dns_record_match to dnsserver.py

This function is used here and in tests, but the tests should not be
importing things from netcmd.dns, which is really supposed to be UI
code. So we move to a common place.

the only difference is the function raises DNSParseError instead of
CommandError, and netcmd.dns has to catch and wrap that.

Signed-off-by: Douglas Bagnall <douglas.bagnall@catalyst.net.nz>
Reviewed-by: Andrew Bartlett <abartlet@samba.org>

diff --git a/python/samba/dnsserver.py b/python/samba/dnsserver.py
index a9fcb7662e8184bc0771a93acbd731be38da8921..68634dbeb5f48213276091c2e9326f3ceeed8695 100644 (file)
--- a/python/samba/dnsserver.py
+++ b/python/samba/dnsserver.py
@@ -295,3 +295,94 @@ def flag_from_string(rec_type):
         return getattr(dnsp, 'DNS_TYPE_' + rtype)
     except AttributeError:
         raise DNSParseError('Unknown type of DNS record %s' % rec_type) from e
+
+
+def dns_name_equal(n1, n2):
+    """Match dns name (of type DNS_RPC_NAME)"""
+    return n1.str.rstrip('.').lower() == n2.str.rstrip('.').lower()
+
+
+def dns_record_match(dns_conn, server, zone, name, record_type, data):
+    """Find a dns record that matches the specified data"""
+
+    # The matching is not as precises as that offered by
+    # dsdb_dns.match_record, which, for example, compares IPv6 records
+    # semantically rather than as strings. However that function
+    # compares database DnssrvRpcRecord structures, not wire
+    # DNS_RPC_RECORD structures.
+    #
+    # While it would be possible, perhaps desirable, to wrap that
+    # function for use in samba-tool, there is value in having a
+    # separate implementation for tests, to avoid the circularity of
+    # asserting the function matches itself.
+
+    urec = record_from_string(record_type, data)
+
+    select_flags = dnsserver.DNS_RPC_VIEW_AUTHORITY_DATA
+
+    try:
+        buflen, res = dns_conn.DnssrvEnumRecords2(
+            dnsserver.DNS_CLIENT_VERSION_LONGHORN, 0, server, zone, name, None,
+            record_type, select_flags, None, None)
+    except WERRORError as e:
+        if e.args[0] == werror.WERR_DNS_ERROR_NAME_DOES_NOT_EXIST:
+            # Either the zone doesn't exist, or there were no records.
+            # We can't differentiate the two.
+            return None
+        raise e
+
+    if not res or res.count == 0:
+        return None
+
+    for rec in res.rec[0].records:
+        if rec.wType != record_type:
+            continue
+
+        found = False
+        if record_type == dnsp.DNS_TYPE_A:
+            if rec.data == urec.data:
+                found = True
+        elif record_type == dnsp.DNS_TYPE_AAAA:
+            if rec.data == urec.data:
+                found = True
+        elif record_type == dnsp.DNS_TYPE_PTR:
+            if dns_name_equal(rec.data, urec.data):
+                found = True
+        elif record_type == dnsp.DNS_TYPE_CNAME:
+            if dns_name_equal(rec.data, urec.data):
+                found = True
+        elif record_type == dnsp.DNS_TYPE_NS:
+            if dns_name_equal(rec.data, urec.data):
+                found = True
+        elif record_type == dnsp.DNS_TYPE_MX:
+            if dns_name_equal(rec.data.nameExchange, urec.data.nameExchange) and \
+               rec.data.wPreference == urec.data.wPreference:
+                found = True
+        elif record_type == dnsp.DNS_TYPE_SRV:
+            if rec.data.wPriority == urec.data.wPriority and \
+               rec.data.wWeight == urec.data.wWeight and \
+               rec.data.wPort == urec.data.wPort and \
+               dns_name_equal(rec.data.nameTarget, urec.data.nameTarget):
+                found = True
+        elif record_type == dnsp.DNS_TYPE_SOA:
+            if rec.data.dwSerialNo == urec.data.dwSerialNo and \
+               rec.data.dwRefresh == urec.data.dwRefresh and \
+               rec.data.dwRetry == urec.data.dwRetry and \
+               rec.data.dwExpire == urec.data.dwExpire and \
+               rec.data.dwMinimumTtl == urec.data.dwMinimumTtl and \
+               dns_name_equal(rec.data.NamePrimaryServer,
+                              urec.data.NamePrimaryServer) and \
+               dns_name_equal(rec.data.ZoneAdministratorEmail,
+                              urec.data.ZoneAdministratorEmail):
+                found = True
+        elif record_type == dnsp.DNS_TYPE_TXT:
+            if rec.data.count == urec.data.count:
+                found = True
+                for i in range(rec.data.count):
+                    found = found and \
+                            (rec.data.str[i].str == urec.data.str[i].str)
+
+        if found:
+            return rec
+
+    return None

diff --git a/python/samba/netcmd/dns.py b/python/samba/netcmd/dns.py
index 11ca90b1f3eca53bd65b82dda88e5c84b19d45b4..4fcd752783136fd3ed82798435bfc3c85e138062 100644 (file)
--- a/python/samba/netcmd/dns.py
+++ b/python/samba/netcmd/dns.py
@@ -41,6 +41,7 @@ from samba.netcmd import (
 from samba.dcerpc import dnsp, dnsserver
 
 from samba.dnsserver import record_from_string, DNSParseError, flag_from_string
+from samba.dnsserver import dns_record_match
 
 
 def dns_connect(server, lp, creds):
@@ -394,85 +395,6 @@ def data_to_dns_record(record_type, data):
     return rec
 
 
-# Match dns name (of type DNS_RPC_NAME)
-def dns_name_equal(n1, n2):
-    return n1.str.rstrip('.').lower() == n2.str.rstrip('.').lower()
-
-
-# Match a dns record with specified data
-def dns_record_match(dns_conn, server, zone, name, record_type, data):
-    urec = data_to_dns_record(record_type, data)
-
-    select_flags = dnsserver.DNS_RPC_VIEW_AUTHORITY_DATA
-
-    try:
-        buflen, res = dns_conn.DnssrvEnumRecords2(
-            dnsserver.DNS_CLIENT_VERSION_LONGHORN, 0, server, zone, name, None,
-            record_type, select_flags, None, None)
-    except WERRORError as e:
-        if e.args[0] == werror.WERR_DNS_ERROR_NAME_DOES_NOT_EXIST:
-            # Either the zone doesn't exist, or there were no records.
-            # We can't differentiate the two.
-            return None
-        raise e
-
-    if not res or res.count == 0:
-        return None
-
-    for rec in res.rec[0].records:
-        if rec.wType != record_type:
-            continue
-
-        found = False
-        if record_type == dnsp.DNS_TYPE_A:
-            if rec.data == urec.data:
-                found = True
-        elif record_type == dnsp.DNS_TYPE_AAAA:
-            if rec.data == urec.data:
-                found = True
-        elif record_type == dnsp.DNS_TYPE_PTR:
-            if dns_name_equal(rec.data, urec.data):
-                found = True
-        elif record_type == dnsp.DNS_TYPE_CNAME:
-            if dns_name_equal(rec.data, urec.data):
-                found = True
-        elif record_type == dnsp.DNS_TYPE_NS:
-            if dns_name_equal(rec.data, urec.data):
-                found = True
-        elif record_type == dnsp.DNS_TYPE_MX:
-            if dns_name_equal(rec.data.nameExchange, urec.data.nameExchange) and \
-               rec.data.wPreference == urec.data.wPreference:
-                found = True
-        elif record_type == dnsp.DNS_TYPE_SRV:
-            if rec.data.wPriority == urec.data.wPriority and \
-               rec.data.wWeight == urec.data.wWeight and \
-               rec.data.wPort == urec.data.wPort and \
-               dns_name_equal(rec.data.nameTarget, urec.data.nameTarget):
-                found = True
-        elif record_type == dnsp.DNS_TYPE_SOA:
-            if rec.data.dwSerialNo == urec.data.dwSerialNo and \
-               rec.data.dwRefresh == urec.data.dwRefresh and \
-               rec.data.dwRetry == urec.data.dwRetry and \
-               rec.data.dwExpire == urec.data.dwExpire and \
-               rec.data.dwMinimumTtl == urec.data.dwMinimumTtl and \
-               dns_name_equal(rec.data.NamePrimaryServer,
-                              urec.data.NamePrimaryServer) and \
-               dns_name_equal(rec.data.ZoneAdministratorEmail,
-                              urec.data.ZoneAdministratorEmail):
-                found = True
-        elif record_type == dnsp.DNS_TYPE_TXT:
-            if rec.data.count == urec.data.count:
-                found = True
-                for i in range(rec.data.count):
-                    found = found and \
-                            (rec.data.str[i].str == urec.data.str[i].str)
-
-        if found:
-            return rec
-
-    return None
-
-
 class cmd_serverinfo(Command):
     """Query for Server information."""
 
@@ -1016,8 +938,12 @@ class cmd_update_record(Command):
         self.creds = credopts.get_credentials(self.lp)
         dns_conn = dns_connect(server, self.lp, self.creds)
 
-        rec_match = dns_record_match(dns_conn, server, zone, name, record_type,
-                                     olddata)
+        try:
+            rec_match = dns_record_match(dns_conn, server, zone, name, record_type,
+                                         olddata)
+        except DNSParseError as e:
+            raise CommandError(*e.args) from None
+
         if not rec_match:
             raise CommandError('Record or zone does not exist.')
 

diff --git a/python/samba/tests/dns.py b/python/samba/tests/dns.py
index 1a05634725a5741b2fade34afc9eb8bef0a56ef3..36d6f77decf4262db7aaad6eccfc5a1a839dacf4 100644 (file)
--- a/python/samba/tests/dns.py
+++ b/python/samba/tests/dns.py
@@ -29,8 +29,7 @@ import samba.ndr as ndr
 from samba import credentials
 from samba.dcerpc import dns, dnsp, dnsserver
 from samba.dnsserver import TXTRecord
-from samba.dnsserver import record_from_string
-from samba.netcmd.dns import dns_record_match
+from samba.dnsserver import record_from_string, dns_record_match
 from samba.tests.subunitrun import SubunitOptions, TestProgram
 from samba import werror, WERRORError
 from samba.tests.dns_base import DNSTest