Merging upstream version 2:4.20.2+dfsg.

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

6 files changed, 531 insertions, 101 deletions

diff --git a/python/samba/gp/gpclass.py b/python/samba/gp/gpclass.py
index 08be472..d86aace 100644
--- a/python/samba/gp/gpclass.py
+++ b/python/samba/gp/gpclass.py
@@ -805,9 +805,7 @@ def site_dn_for_machine(samdb, dc_hostname, lp, creds, hostname):
 
     samlogon_response = ndr_unpack(nbt.netlogon_samlogon_response,
                                    bytes(res.msgs[0]['Netlogon'][0]))
-    if samlogon_response.ntver not in [nbt.NETLOGON_NT_VERSION_5EX,
-                                       (nbt.NETLOGON_NT_VERSION_1
-                                        | nbt.NETLOGON_NT_VERSION_5EX)]:
+    if not (samlogon_response.ntver & nbt.NETLOGON_NT_VERSION_5EX):
         raise RuntimeError('site_dn_for_machine: Invalid NtVer in '
                            + 'netlogon_samlogon_response')
 
diff --git a/python/samba/tests/blackbox/misc_dfs_widelink.py b/python/samba/tests/blackbox/misc_dfs_widelink.py
new file mode 100644
index 0000000..7948590
--- /dev/null
+++ b/python/samba/tests/blackbox/misc_dfs_widelink.py
@@ -0,0 +1,86 @@
+# Blackbox tests for DFS (widelink)
+#
+# Copyright (C) Noel Power noel.power@suse.com
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+from samba.tests import BlackboxTestCase, BlackboxProcessError
+from samba.samba3 import param as s3param
+
+from samba.credentials import Credentials
+
+import os
+
+class DfsWidelinkBlockboxTestBase(BlackboxTestCase):
+
+    def setUp(self):
+        super().setUp()
+        self.lp = s3param.get_context()
+        self.server = os.environ["SERVER"]
+        self.user = os.environ["USER"]
+        self.passwd = os.environ["PASSWORD"]
+        self.creds = Credentials()
+        self.creds.guess(self.lp)
+        self.creds.set_username(self.user)
+        self.creds.set_password(self.passwd)
+        self.testdir = os.getenv("TESTDIR", "msdfs-share-wl")
+        self.share = os.getenv("SHARE", "msdfs-share-wl")
+        self.dirpath = os.path.join(os.environ["LOCAL_PATH"],self.testdir)
+        # allow a custom teardown function to be defined
+        self.cleanup = None
+        self.cleanup_args = []
+
+    def tearDown(self):
+        try:
+            if (self.cleanup):
+                self.cleanup(self.cleanup_args)
+        except Exception as e:
+            print("remote remove failed: %s" % str(e))
+
+    def build_test_cmd(self, cmd, args):
+        cmd = [cmd, "-U%s%%%s" % (self.user, self.passwd)]
+        cmd.extend(args)
+        return cmd
+
+    def test_ci_chdir(self):
+        parent_dir = "msdfs-src1"
+        dirs = [parent_dir, parent_dir.upper()]
+        # try as named dir first then try upper-cased version
+        for adir in dirs:
+            smbclient_args = self.build_test_cmd("smbclient", ["//%s/%s" % (self.server, self.share), "-c", "cd %s" % (adir)])
+            try:
+                out_str = self.check_output(smbclient_args)
+            except BlackboxProcessError as e:
+                print(str(e))
+                self.fail(str(e))
+
+    def test_nested_chdir(self):
+        parent_dir = "dfshop1"
+        child_dir = "dfshop2"
+        smbclient_args = self.build_test_cmd("smbclient", ["//%s/%s" % (self.server, self.share), "-c", "cd %s/%s" % (parent_dir,child_dir)])
+        try:
+            out_str = self.check_output(smbclient_args)
+        except BlackboxProcessError as e:
+            print(str(e))
+            self.fail(str(e))
+
+    def test_enumerate_dfs_link(self):
+        smbclient_args = self.build_test_cmd("smbclient", ["//%s/%s" % (self.server, self.share), "-c", "dir"])
+        try:
+            out_str = self.check_output(smbclient_args)
+        except BlackboxProcessError as e:
+            print(str(e))
+            self.fail(str(e))
+        out_str = out_str.decode()
+        self.assertIn("msdfs-src1", out_str)
diff --git a/python/samba/tests/dns_base.py b/python/samba/tests/dns_base.py
index d320a0e..43a62b1 100644
--- a/python/samba/tests/dns_base.py
+++ b/python/samba/tests/dns_base.py
@@ -20,6 +20,7 @@ from samba.tests import TestCaseInTempDir
 from samba.dcerpc import dns, dnsp
 from samba import gensec, tests
 from samba import credentials
+from samba import NTSTATUSError
 import struct
 import samba.ndr as ndr
 import random
@@ -76,6 +77,24 @@ class DNSTest(TestCaseInTempDir):
         self.assertEqual(p_opcode, opcode, "Expected OPCODE %s, got %s" %
                           (opcode, p_opcode))
 
+    def assert_dns_flags_equals(self, packet, flags):
+        "Helper function to check opcode"
+        p_flags = packet.operation & (~(dns.DNS_OPCODE|dns.DNS_RCODE))
+        self.assertEqual(p_flags, flags, "Expected FLAGS %02x, got %02x" %
+                          (flags, p_flags))
+
+    def assert_echoed_dns_error(self, request, response, response_p, rcode):
+
+        request_p = ndr.ndr_pack(request)
+
+        self.assertEqual(response.id, request.id)
+        self.assert_dns_rcode_equals(response, rcode)
+        self.assert_dns_opcode_equals(response, request.operation & dns.DNS_OPCODE)
+        self.assert_dns_flags_equals(response,
+            (request.operation | dns.DNS_FLAG_REPLY) & (~(dns.DNS_OPCODE|dns.DNS_RCODE)))
+        self.assertEqual(len(response_p), len(request_p))
+        self.assertEqual(response_p[4:], request_p[4:])
+
     def make_name_packet(self, opcode, qid=None):
         "Helper creating a dns.name_packet"
         p = dns.name_packet()
@@ -112,6 +131,8 @@ class DNSTest(TestCaseInTempDir):
         return self.creds.get_realm().lower()
 
     def dns_transaction_udp(self, packet, host,
+                            allow_remaining=False,
+                            allow_truncated=False,
                             dump=False, timeout=None):
         "send a DNS query and read the reply"
         s = None
@@ -128,8 +149,22 @@ class DNSTest(TestCaseInTempDir):
             recv_packet = s.recv(2048, 0)
             if dump:
                 print(self.hexdump(recv_packet))
-            response = ndr.ndr_unpack(dns.name_packet, recv_packet)
+            if allow_truncated:
+                # with allow_remaining
+                # we add some zero bytes
+                # in order to also parse truncated
+                # responses
+                recv_packet_p = recv_packet + 32*b"\x00"
+                allow_remaining = True
+            else:
+                recv_packet_p = recv_packet
+            response = ndr.ndr_unpack(dns.name_packet, recv_packet_p,
+                                      allow_remaining=allow_remaining)
             return (response, recv_packet)
+        except RuntimeError as re:
+            if s is not None:
+                s.close()
+            raise AssertionError(re)
         finally:
             if s is not None:
                 s.close()
@@ -151,11 +186,26 @@ class DNSTest(TestCaseInTempDir):
             tcp_packet += send_packet
             s.sendall(tcp_packet)
 
-            recv_packet = s.recv(0xffff + 2, 0)
+            recv_packet = b''
+            length = None
+            for i in range(0, 2 + 0xffff):
+                if len(recv_packet) >= 2:
+                    length, = struct.unpack('!H', recv_packet[0:2])
+                    remaining = 2 + length
+                else:
+                    remaining = 2 + 12
+                remaining -= len(recv_packet)
+                if remaining == 0:
+                    break
+                recv_packet += s.recv(remaining, 0)
             if dump:
                 print(self.hexdump(recv_packet))
             response = ndr.ndr_unpack(dns.name_packet, recv_packet[2:])
 
+        except RuntimeError as re:
+            if s is not None:
+                s.close()
+            raise AssertionError(re)
         finally:
             if s is not None:
                 s.close()
@@ -217,18 +267,41 @@ class DNSTKeyTest(DNSTest):
         self.creds.set_username(tests.env_get_var_value('USERNAME'))
         self.creds.set_password(tests.env_get_var_value('PASSWORD'))
         self.creds.set_kerberos_state(credentials.MUST_USE_KERBEROS)
+
+        self.unpriv_creds = None
+
         self.newrecname = "tkeytsig.%s" % self.get_dns_domain()
 
-    def tkey_trans(self, creds=None):
+    def get_unpriv_creds(self):
+        if self.unpriv_creds is not None:
+            return self.unpriv_creds
+
+        self.unpriv_creds = credentials.Credentials()
+        self.unpriv_creds.guess(self.lp_ctx)
+        self.unpriv_creds.set_username(tests.env_get_var_value('USERNAME_UNPRIV'))
+        self.unpriv_creds.set_password(tests.env_get_var_value('PASSWORD_UNPRIV'))
+        self.unpriv_creds.set_kerberos_state(credentials.MUST_USE_KERBEROS)
+
+        return self.unpriv_creds
+
+    def tkey_trans(self, creds=None, algorithm_name="gss-tsig",
+                   tkey_req_in_answers=False,
+                   expected_rcode=dns.DNS_RCODE_OK):
         "Do a TKEY transaction and establish a gensec context"
 
         if creds is None:
             creds = self.creds
 
-        self.key_name = "%s.%s" % (uuid.uuid4(), self.get_dns_domain())
+        mech = 'spnego'
+
+        tkey = {}
+        tkey['name'] = "%s.%s" % (uuid.uuid4(), self.get_dns_domain())
+        tkey['creds'] = creds
+        tkey['mech'] = mech
+        tkey['algorithm'] = algorithm_name
 
         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
-        q = self.make_name_question(self.key_name,
+        q = self.make_name_question(tkey['name'],
                                     dns.DNS_QTYPE_TKEY,
                                     dns.DNS_QCLASS_IN)
         questions = []
@@ -236,30 +309,30 @@ class DNSTKeyTest(DNSTest):
         self.finish_name_packet(p, questions)
 
         r = dns.res_rec()
-        r.name = self.key_name
+        r.name = tkey['name']
         r.rr_type = dns.DNS_QTYPE_TKEY
         r.rr_class = dns.DNS_QCLASS_IN
         r.ttl = 0
         r.length = 0xffff
         rdata = dns.tkey_record()
-        rdata.algorithm = "gss-tsig"
+        rdata.algorithm = algorithm_name
         rdata.inception = int(time.time())
         rdata.expiration = int(time.time()) + 60 * 60
         rdata.mode = dns.DNS_TKEY_MODE_GSSAPI
         rdata.error = 0
         rdata.other_size = 0
 
-        self.g = gensec.Security.start_client(self.settings)
-        self.g.set_credentials(creds)
-        self.g.set_target_service("dns")
-        self.g.set_target_hostname(self.server)
-        self.g.want_feature(gensec.FEATURE_SIGN)
-        self.g.start_mech_by_name("spnego")
+        tkey['gensec'] = gensec.Security.start_client(self.settings)
+        tkey['gensec'].set_credentials(creds)
+        tkey['gensec'].set_target_service("dns")
+        tkey['gensec'].set_target_hostname(self.server)
+        tkey['gensec'].want_feature(gensec.FEATURE_SIGN)
+        tkey['gensec'].start_mech_by_name(tkey['mech'])
 
         finished = False
         client_to_server = b""
 
-        (finished, server_to_client) = self.g.update(client_to_server)
+        (finished, server_to_client) = tkey['gensec'].update(client_to_server)
         self.assertFalse(finished)
 
         data = [x if isinstance(x, int) else ord(x) for x in list(server_to_client)]
@@ -268,56 +341,76 @@ class DNSTKeyTest(DNSTest):
         r.rdata = rdata
 
         additional = [r]
-        p.arcount = 1
-        p.additional = additional
+        if tkey_req_in_answers:
+            p.ancount = 1
+            p.answers = additional
+        else:
+            p.arcount = 1
+            p.additional = additional
 
         (response, response_packet) =\
             self.dns_transaction_tcp(p, self.server_ip)
+        if expected_rcode != dns.DNS_RCODE_OK:
+            self.assert_echoed_dns_error(p, response, response_packet, expected_rcode)
+            return
         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
 
         tkey_record = response.answers[0].rdata
         server_to_client = bytes(tkey_record.key_data)
-        (finished, client_to_server) = self.g.update(server_to_client)
+        (finished, client_to_server) = tkey['gensec'].update(server_to_client)
         self.assertTrue(finished)
 
+        self.tkey = tkey
+
         self.verify_packet(response, response_packet)
 
     def verify_packet(self, response, response_packet, request_mac=b""):
+        self.assertEqual(response.arcount, 1)
         self.assertEqual(response.additional[0].rr_type, dns.DNS_QTYPE_TSIG)
 
+        if self.tkey['algorithm'] == "gss-tsig":
+            gss_tsig = True
+        else:
+            gss_tsig = False
+
+        request_mac_len = b""
+        if len(request_mac) > 0 and gss_tsig:
+            request_mac_len = struct.pack('!H', len(request_mac))
+
         tsig_record = response.additional[0].rdata
         mac = bytes(tsig_record.mac)
 
+        self.assertEqual(tsig_record.original_id, response.id)
+        self.assertEqual(tsig_record.mac_size, len(mac))
+
         # Cut off tsig record from dns response packet for MAC verification
         # and reset additional record count.
-        key_name_len = len(self.key_name) + 2
-        tsig_record_len = len(ndr.ndr_pack(tsig_record)) + key_name_len + 10
-
-        # convert str/bytes to a list (of string char or int)
-        # so it can be modified
-        response_packet_list = [x if isinstance(x, int) else ord(x) for x in response_packet]
-        del response_packet_list[-tsig_record_len:]
-        response_packet_list[11] = 0
-
-        # convert modified list (of string char or int) to str/bytes
-        response_packet_wo_tsig = bytes(response_packet_list)
+        response_copy = ndr.ndr_deepcopy(response)
+        response_copy.arcount = 0
+        response_packet_wo_tsig = ndr.ndr_pack(response_copy)
 
         fake_tsig = dns.fake_tsig_rec()
-        fake_tsig.name = self.key_name
+        fake_tsig.name = self.tkey['name']
         fake_tsig.rr_class = dns.DNS_QCLASS_ANY
         fake_tsig.ttl = 0
         fake_tsig.time_prefix = tsig_record.time_prefix
         fake_tsig.time = tsig_record.time
         fake_tsig.algorithm_name = tsig_record.algorithm_name
         fake_tsig.fudge = tsig_record.fudge
-        fake_tsig.error = 0
-        fake_tsig.other_size = 0
+        fake_tsig.error = tsig_record.error
+        fake_tsig.other_size = tsig_record.other_size
+        fake_tsig.other_data = tsig_record.other_data
         fake_tsig_packet = ndr.ndr_pack(fake_tsig)
 
-        data = request_mac + response_packet_wo_tsig + fake_tsig_packet
-        self.g.check_packet(data, data, mac)
+        data = request_mac_len + request_mac + response_packet_wo_tsig + fake_tsig_packet
+        try:
+            self.tkey['gensec'].check_packet(data, data, mac)
+        except NTSTATUSError as nt:
+            raise AssertionError(nt)
 
-    def sign_packet(self, packet, key_name):
+    def sign_packet(self, packet, key_name,
+                    algorithm_name="gss-tsig",
+                    bad_sig=False):
         "Sign a packet, calculate a MAC and add TSIG record"
         packet_data = ndr.ndr_pack(packet)
 
@@ -327,18 +420,35 @@ class DNSTKeyTest(DNSTest):
         fake_tsig.ttl = 0
         fake_tsig.time_prefix = 0
         fake_tsig.time = int(time.time())
-        fake_tsig.algorithm_name = "gss-tsig"
+        fake_tsig.algorithm_name = algorithm_name
         fake_tsig.fudge = 300
         fake_tsig.error = 0
         fake_tsig.other_size = 0
         fake_tsig_packet = ndr.ndr_pack(fake_tsig)
 
         data = packet_data + fake_tsig_packet
-        mac = self.g.sign_packet(data, data)
+        mac = self.tkey['gensec'].sign_packet(data, data)
         mac_list = [x if isinstance(x, int) else ord(x) for x in list(mac)]
+        if bad_sig:
+            if len(mac) > 8:
+                mac_list[-8] = mac_list[-8] ^ 0xff
+            if len(mac) > 7:
+                mac_list[-7] = ord('b')
+            if len(mac) > 6:
+                mac_list[-6] = ord('a')
+            if len(mac) > 5:
+                mac_list[-5] = ord('d')
+            if len(mac) > 4:
+                mac_list[-4] = ord('m')
+            if len(mac) > 3:
+                mac_list[-3] = ord('a')
+            if len(mac) > 2:
+                mac_list[-2] = ord('c')
+            if len(mac) > 1:
+                mac_list[-1] = mac_list[-1] ^ 0xff
 
         rdata = dns.tsig_record()
-        rdata.algorithm_name = "gss-tsig"
+        rdata.algorithm_name = algorithm_name
         rdata.time_prefix = 0
         rdata.time = fake_tsig.time
         rdata.fudge = 300
@@ -363,33 +473,10 @@ class DNSTKeyTest(DNSTest):
         return mac
 
     def bad_sign_packet(self, packet, key_name):
-        """Add bad signature for a packet by bitflipping
-        the final byte in the MAC"""
-
-        mac_list = [x if isinstance(x, int) else ord(x) for x in list("badmac")]
-
-        rdata = dns.tsig_record()
-        rdata.algorithm_name = "gss-tsig"
-        rdata.time_prefix = 0
-        rdata.time = int(time.time())
-        rdata.fudge = 300
-        rdata.original_id = packet.id
-        rdata.error = 0
-        rdata.other_size = 0
-        rdata.mac = mac_list
-        rdata.mac_size = len(mac_list)
+        """Add bad signature for a packet by
+        bitflipping and hardcoding bytes at the end of the MAC"""
 
-        r = dns.res_rec()
-        r.name = key_name
-        r.rr_type = dns.DNS_QTYPE_TSIG
-        r.rr_class = dns.DNS_QCLASS_ANY
-        r.ttl = 0
-        r.length = 0xffff
-        r.rdata = rdata
-
-        additional = [r]
-        packet.additional = additional
-        packet.arcount = 1
+        return self.sign_packet(packet, key_name, bad_sig=True)
 
     def search_record(self, name):
         p = self.make_name_packet(dns.DNS_OPCODE_QUERY)
diff --git a/python/samba/tests/dns_tkey.py b/python/samba/tests/dns_tkey.py
index 69af14d..f8417ea 100644
--- a/python/samba/tests/dns_tkey.py
+++ b/python/samba/tests/dns_tkey.py
@@ -19,6 +19,7 @@
 import sys
 import optparse
 import samba.getopt as options
+import samba.ndr as ndr
 from samba.dcerpc import dns
 from samba.tests.subunitrun import SubunitOptions, TestProgram
 from samba.tests.dns_base import DNSTKeyTest
@@ -55,17 +56,34 @@ class TestDNSUpdates(DNSTKeyTest):
         self.server_ip = server_ip
         super().setUp()
 
-    def test_tkey(self):
-        "test DNS TKEY handshake"
+    def test_tkey_gss_tsig(self):
+        "test DNS TKEY handshake with gss-tsig"
 
         self.tkey_trans()
 
+    def test_tkey_gss_microsoft_com(self):
+        "test DNS TKEY handshake with gss.microsoft.com"
+
+        self.tkey_trans(algorithm_name="gss.microsoft.com")
+
+    def test_tkey_invalid_gss_TSIG(self):
+        "test DNS TKEY handshake with invalid gss-TSIG"
+
+        self.tkey_trans(algorithm_name="gss-TSIG",
+                        expected_rcode=dns.DNS_RCODE_REFUSED)
+
+    def test_tkey_invalid_gss_MICROSOFT_com(self):
+        "test DNS TKEY handshake with invalid gss.MICROSOFT.com"
+
+        self.tkey_trans(algorithm_name="gss.MICROSOFT.com",
+                        expected_rcode=dns.DNS_RCODE_REFUSED)
+
     def test_update_wo_tsig(self):
         "test DNS update without TSIG record"
 
         p = self.make_update_request()
         (response, response_p) = self.dns_transaction_udp(p, self.server_ip)
-        self.assert_dns_rcode_equals(response, dns.DNS_RCODE_REFUSED)
+        self.assert_echoed_dns_error(p, response, response_p, dns.DNS_RCODE_REFUSED)
 
         rcode = self.search_record(self.newrecname)
         self.assert_rcode_equals(rcode, dns.DNS_RCODE_NXDOMAIN)
@@ -78,10 +96,7 @@ class TestDNSUpdates(DNSTKeyTest):
         p = self.make_update_request()
         self.sign_packet(p, "badkey")
         (response, response_p) = self.dns_transaction_udp(p, self.server_ip)
-        self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NOTAUTH)
-        tsig_record = response.additional[0].rdata
-        self.assertEqual(tsig_record.error, dns.DNS_RCODE_BADKEY)
-        self.assertEqual(tsig_record.mac_size, 0)
+        self.assert_echoed_dns_error(p, response, response_p, dns.DNS_RCODE_REFUSED)
 
         rcode = self.search_record(self.newrecname)
         self.assert_rcode_equals(rcode, dns.DNS_RCODE_NXDOMAIN)
@@ -92,23 +107,149 @@ class TestDNSUpdates(DNSTKeyTest):
         self.tkey_trans()
 
         p = self.make_update_request()
-        self.bad_sign_packet(p, self.key_name)
+        self.bad_sign_packet(p, self.tkey['name'])
         (response, response_p) = self.dns_transaction_udp(p, self.server_ip)
-        self.assert_dns_rcode_equals(response, dns.DNS_RCODE_NOTAUTH)
-        tsig_record = response.additional[0].rdata
-        self.assertEqual(tsig_record.error, dns.DNS_RCODE_BADSIG)
-        self.assertEqual(tsig_record.mac_size, 0)
+        self.assert_echoed_dns_error(p, response, response_p, dns.DNS_RCODE_REFUSED)
 
         rcode = self.search_record(self.newrecname)
         self.assert_rcode_equals(rcode, dns.DNS_RCODE_NXDOMAIN)
 
-    def test_update_tsig(self):
-        "test DNS update with correct TSIG record"
+    def test_update_tsig_bad_algorithm(self):
+        "test DNS update with a TSIG record with a bad algorithm"
 
         self.tkey_trans()
 
+        algorithm_name = "gss-TSIG"
         p = self.make_update_request()
-        mac = self.sign_packet(p, self.key_name)
+        mac = self.sign_packet(p, self.tkey['name'],
+                               algorithm_name=algorithm_name)
+        (response, response_p) = self.dns_transaction_udp(p, self.server_ip)
+        self.assert_echoed_dns_error(p, response, response_p, dns.DNS_RCODE_REFUSED)
+
+        rcode = self.search_record(self.newrecname)
+        self.assert_rcode_equals(rcode, dns.DNS_RCODE_NXDOMAIN)
+
+    def test_update_tsig_changed_algorithm1(self):
+        "test DNS update with a TSIG record with a changed algorithm"
+
+        algorithm_name = "gss-tsig"
+        self.tkey_trans(algorithm_name=algorithm_name)
+
+        # Now delete the record, it's most likely
+        # a no-op as it should not be there if the test
+        # runs the first time
+        p = self.make_update_request(delete=True)
+        mac = self.sign_packet(p, self.tkey['name'], algorithm_name=algorithm_name)
+        (response, response_p) = self.dns_transaction_udp(p, self.server_ip)
+        self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+        self.verify_packet(response, response_p, mac)
+
+        # Now do an update with the algorithm_name
+        # changed in the requests TSIG message.
+        p = self.make_update_request()
+        algorithm_name = "gss.microsoft.com"
+        mac = self.sign_packet(p, self.tkey['name'],
+                               algorithm_name=algorithm_name)
+        algorithm_name = "gss-tsig"
+        (response, response_p) = self.dns_transaction_udp(p, self.server_ip,
+                                                          allow_remaining=True)
+        self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+        self.verify_packet(response, response_p, mac)
+
+        # Check the record is around
+        rcode = self.search_record(self.newrecname)
+        self.assert_rcode_equals(rcode, dns.DNS_RCODE_OK)
+
+        # Now delete the record, with the original
+        # algorithm_name used in the tkey exchange
+        p = self.make_update_request(delete=True)
+        mac = self.sign_packet(p, self.tkey['name'], algorithm_name=algorithm_name)
+        (response, response_p) = self.dns_transaction_udp(p, self.server_ip)
+        self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+        self.verify_packet(response, response_p, mac)
+
+        rcode = self.search_record(self.newrecname)
+        self.assert_rcode_equals(rcode, dns.DNS_RCODE_NXDOMAIN)
+
+    def test_update_tsig_changed_algorithm2(self):
+        "test DNS update with a TSIG record with a changed algorithm"
+
+        algorithm_name = "gss.microsoft.com"
+        self.tkey_trans(algorithm_name=algorithm_name)
+
+        # Now delete the record, it's most likely
+        # a no-op as it should not be there if the test
+        # runs the first time
+        p = self.make_update_request(delete=True)
+        mac = self.sign_packet(p, self.tkey['name'], algorithm_name=algorithm_name)
+        (response, response_p) = self.dns_transaction_udp(p, self.server_ip)
+        self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+        self.verify_packet(response, response_p, mac)
+
+        # Now do an update with the algorithm_name
+        # changed in the requests TSIG message.
+        p = self.make_update_request()
+        algorithm_name = "gss-tsig"
+        mac = self.sign_packet(p, self.tkey['name'],
+                               algorithm_name=algorithm_name)
+        algorithm_name = "gss.microsoft.com"
+        (response, response_p) = self.dns_transaction_udp(p, self.server_ip,
+                                                          allow_truncated=True)
+        self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+        response_p_pack = ndr.ndr_pack(response)
+        if len(response_p_pack) == len(response_p):
+            self.verify_packet(response, response_p, mac)
+        else:
+            pass # Windows bug
+
+        # Check the record is around
+        rcode = self.search_record(self.newrecname)
+        self.assert_rcode_equals(rcode, dns.DNS_RCODE_OK)
+
+        # Now delete the record, with the original
+        # algorithm_name used in the tkey exchange
+        p = self.make_update_request(delete=True)
+        mac = self.sign_packet(p, self.tkey['name'], algorithm_name=algorithm_name)
+        (response, response_p) = self.dns_transaction_udp(p, self.server_ip)
+        self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+        self.verify_packet(response, response_p, mac)
+
+        rcode = self.search_record(self.newrecname)
+        self.assert_rcode_equals(rcode, dns.DNS_RCODE_NXDOMAIN)
+
+    def test_update_gss_tsig_tkey_req_additional(self):
+        "test DNS update with correct gss-tsig record tkey req in additional"
+
+        self.tkey_trans()
+
+        p = self.make_update_request()
+        mac = self.sign_packet(p, self.tkey['name'])
+        (response, response_p) = self.dns_transaction_udp(p, self.server_ip)
+        self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+        self.verify_packet(response, response_p, mac)
+
+        # Check the record is around
+        rcode = self.search_record(self.newrecname)
+        self.assert_rcode_equals(rcode, dns.DNS_RCODE_OK)
+
+        # Now delete the record
+        p = self.make_update_request(delete=True)
+        mac = self.sign_packet(p, self.tkey['name'])
+        (response, response_p) = self.dns_transaction_udp(p, self.server_ip)
+        self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+        self.verify_packet(response, response_p, mac)
+
+        # check it's gone
+        rcode = self.search_record(self.newrecname)
+        self.assert_rcode_equals(rcode, dns.DNS_RCODE_NXDOMAIN)
+
+    def test_update_gss_tsig_tkey_req_answers(self):
+        "test DNS update with correct gss-tsig record tsig req in answers"
+
+        self.tkey_trans(tkey_req_in_answers=True)
+
+        p = self.make_update_request()
+        mac = self.sign_packet(p, self.tkey['name'])
         (response, response_p) = self.dns_transaction_udp(p, self.server_ip)
         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
         self.verify_packet(response, response_p, mac)
@@ -119,7 +260,66 @@ class TestDNSUpdates(DNSTKeyTest):
 
         # Now delete the record
         p = self.make_update_request(delete=True)
-        mac = self.sign_packet(p, self.key_name)
+        mac = self.sign_packet(p, self.tkey['name'])
+        (response, response_p) = self.dns_transaction_udp(p, self.server_ip)
+        self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+        self.verify_packet(response, response_p, mac)
+
+        # check it's gone
+        rcode = self.search_record(self.newrecname)
+        self.assert_rcode_equals(rcode, dns.DNS_RCODE_NXDOMAIN)
+
+    def test_update_gss_microsoft_com_tkey_req_additional(self):
+        "test DNS update with correct gss.microsoft.com record tsig req in additional"
+
+        algorithm_name = "gss.microsoft.com"
+        self.tkey_trans(algorithm_name=algorithm_name)
+
+        p = self.make_update_request()
+        mac = self.sign_packet(p, self.tkey['name'],
+                               algorithm_name=algorithm_name)
+        (response, response_p) = self.dns_transaction_udp(p, self.server_ip)
+        self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+        self.verify_packet(response, response_p, mac)
+
+        # Check the record is around
+        rcode = self.search_record(self.newrecname)
+        self.assert_rcode_equals(rcode, dns.DNS_RCODE_OK)
+
+        # Now delete the record
+        p = self.make_update_request(delete=True)
+        mac = self.sign_packet(p, self.tkey['name'],
+                               algorithm_name=algorithm_name)
+        (response, response_p) = self.dns_transaction_udp(p, self.server_ip)
+        self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+        self.verify_packet(response, response_p, mac)
+
+        # check it's gone
+        rcode = self.search_record(self.newrecname)
+        self.assert_rcode_equals(rcode, dns.DNS_RCODE_NXDOMAIN)
+
+    def test_update_gss_microsoft_com_tkey_req_answers(self):
+        "test DNS update with correct gss.microsoft.com record tsig req in answers"
+
+        algorithm_name = "gss.microsoft.com"
+        self.tkey_trans(algorithm_name=algorithm_name,
+                        tkey_req_in_answers=True)
+
+        p = self.make_update_request()
+        mac = self.sign_packet(p, self.tkey['name'],
+                               algorithm_name=algorithm_name)
+        (response, response_p) = self.dns_transaction_udp(p, self.server_ip)
+        self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+        self.verify_packet(response, response_p, mac)
+
+        # Check the record is around
+        rcode = self.search_record(self.newrecname)
+        self.assert_rcode_equals(rcode, dns.DNS_RCODE_OK)
+
+        # Now delete the record
+        p = self.make_update_request(delete=True)
+        mac = self.sign_packet(p, self.tkey['name'],
+                               algorithm_name=algorithm_name)
         (response, response_p) = self.dns_transaction_udp(p, self.server_ip)
         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
         self.verify_packet(response, response_p, mac)
@@ -131,35 +331,28 @@ class TestDNSUpdates(DNSTKeyTest):
     def test_update_tsig_windows(self):
         "test DNS update with correct TSIG record (follow Windows pattern)"
 
-        newrecname = "win" + self.newrecname
+        p = self.make_update_request()
+
         rr_class = dns.DNS_QCLASS_IN
         ttl = 1200
 
-        p = self.make_name_packet(dns.DNS_OPCODE_UPDATE)
-        q = self.make_name_question(self.get_dns_domain(),
-                                    dns.DNS_QTYPE_SOA,
-                                    dns.DNS_QCLASS_IN)
-        questions = []
-        questions.append(q)
-        self.finish_name_packet(p, questions)
-
         updates = []
         r = dns.res_rec()
-        r.name = newrecname
+        r.name = self.newrecname
         r.rr_type = dns.DNS_QTYPE_A
         r.rr_class = dns.DNS_QCLASS_ANY
         r.ttl = 0
         r.length = 0
         updates.append(r)
         r = dns.res_rec()
-        r.name = newrecname
+        r.name = self.newrecname
         r.rr_type = dns.DNS_QTYPE_AAAA
         r.rr_class = dns.DNS_QCLASS_ANY
         r.ttl = 0
         r.length = 0
         updates.append(r)
         r = dns.res_rec()
-        r.name = newrecname
+        r.name = self.newrecname
         r.rr_type = dns.DNS_QTYPE_A
         r.rr_class = rr_class
         r.ttl = ttl
@@ -171,7 +364,7 @@ class TestDNSUpdates(DNSTKeyTest):
 
         prereqs = []
         r = dns.res_rec()
-        r.name = newrecname
+        r.name = self.newrecname
         r.rr_type = dns.DNS_QTYPE_CNAME
         r.rr_class = dns.DNS_QCLASS_NONE
         r.ttl = 0
@@ -181,21 +374,87 @@ class TestDNSUpdates(DNSTKeyTest):
         p.answers = prereqs
 
         (response, response_p) = self.dns_transaction_udp(p, self.server_ip)
-        self.assert_dns_rcode_equals(response, dns.DNS_RCODE_REFUSED)
+        self.assert_echoed_dns_error(p, response, response_p, dns.DNS_RCODE_REFUSED)
 
         self.tkey_trans()
-        mac = self.sign_packet(p, self.key_name)
+        mac = self.sign_packet(p, self.tkey['name'])
         (response, response_p) = self.dns_transaction_udp(p, self.server_ip)
         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
         self.verify_packet(response, response_p, mac)
 
         # Check the record is around
-        rcode = self.search_record(newrecname)
+        rcode = self.search_record(self.newrecname)
         self.assert_rcode_equals(rcode, dns.DNS_RCODE_OK)
 
         # Now delete the record
+        delete_updates = []
+        r = dns.res_rec()
+        r.name = self.newrecname
+        r.rr_type = dns.DNS_QTYPE_A
+        r.rr_class = dns.DNS_QCLASS_NONE
+        r.ttl = 0
+        r.length = 0xffff
+        r.rdata = "10.1.45.64"
+        delete_updates.append(r)
+        p = self.make_update_request(delete=True)
+        p.nscount = len(delete_updates)
+        p.nsrecs = delete_updates
+        mac = self.sign_packet(p, self.tkey['name'])
+        (response, response_p) = self.dns_transaction_udp(p, self.server_ip)
+        self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+        self.verify_packet(response, response_p, mac)
+
+        # check it's gone
+        rcode = self.search_record(self.newrecname)
+        self.assert_rcode_equals(rcode, dns.DNS_RCODE_NXDOMAIN)
+
+    def test_update_tsig_record_access_denied(self):
+        """test DNS update with a TSIG record where the user does not have
+        permissions to change the record"""
+
+        self.tkey_trans()
+        adm_tkey = self.tkey
+
+        # First create the record as admin
+        p = self.make_update_request()
+        mac = self.sign_packet(p, self.tkey['name'])
+        (response, response_p) = self.dns_transaction_udp(p, self.server_ip)
+        self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+        self.verify_packet(response, response_p, mac)
+
+        # Check the record is around
+        rcode = self.search_record(self.newrecname)
+        self.assert_rcode_equals(rcode, dns.DNS_RCODE_OK)
+
+        # Now update the same values as normal user
+        # should work without error
+        self.tkey_trans(creds=self.get_unpriv_creds())
+        unpriv_tkey = self.tkey
+
+        p = self.make_update_request()
+        mac = self.sign_packet(p, self.tkey['name'])
+        (response, response_p) = self.dns_transaction_udp(p, self.server_ip)
+        self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
+        self.verify_packet(response, response_p, mac)
+
+        # Check the record is still around
+        rcode = self.search_record(self.newrecname)
+        self.assert_rcode_equals(rcode, dns.DNS_RCODE_OK)
+
+        # Now try to delete the record a normal user (should fail)
+        p = self.make_update_request(delete=True)
+        mac = self.sign_packet(p, self.tkey['name'])
+        (response, response_p) = self.dns_transaction_udp(p, self.server_ip)
+        self.assert_echoed_dns_error(p, response, response_p, dns.DNS_RCODE_REFUSED)
+
+        # Check the record is still around
+        rcode = self.search_record(self.newrecname)
+        self.assert_rcode_equals(rcode, dns.DNS_RCODE_OK)
+
+        # Now delete the record as admin
+        self.tkey = adm_tkey
         p = self.make_update_request(delete=True)
-        mac = self.sign_packet(p, self.key_name)
+        mac = self.sign_packet(p, self.tkey['name'])
         (response, response_p) = self.dns_transaction_udp(p, self.server_ip)
         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
         self.verify_packet(response, response_p, mac)
diff --git a/python/samba/tests/join.py b/python/samba/tests/join.py
index b47bc70..b04cb4a 100644
--- a/python/samba/tests/join.py
+++ b/python/samba/tests/join.py
@@ -156,7 +156,7 @@ class JoinTestCase(DNSTKeyTest):
         p.nscount = len(updates)
         p.nsrecs = updates
 
-        mac = self.sign_packet(p, self.key_name)
+        mac = self.sign_packet(p, self.tkey['name'])
         (response, response_p) = self.dns_transaction_udp(p, self.server_ip)
         self.assert_dns_rcode_equals(response, dns.DNS_RCODE_OK)
         self.verify_packet(response, response_p, mac)
diff --git a/python/samba/tests/ntacls.py b/python/samba/tests/ntacls.py
index 0b7963d..6e2adda 100644
--- a/python/samba/tests/ntacls.py
+++ b/python/samba/tests/ntacls.py
@@ -83,5 +83,5 @@ class NtaclsTests(TestCaseInTempDir):
         lp = LoadParm()
         open(self.tempf, 'w').write("empty")
         lp.set("posix:eadb", os.path.join(self.tempdir, "eadbtest.tdb"))
-        self.assertRaises(Exception, setntacl, lp, self.tempf, NTACL_SDDL,
+        self.assertRaises(PermissionError, setntacl, lp, self.tempf, NTACL_SDDL,
                           DOMAIN_SID, self.session_info, "native")