tests: Add corner-case test: fromServer points to dead server

The fromServer attribute is slightly unique, in that it's a DN (similar
to a one-way link), but it is also a mandatory attribute.

Currently, if fromServer gets a bad value (i.e. a dead server that has
been expunged), the DSDB rejects any attempts to modify the associated
nTDSConnection object (regardless of whether or not you're actually
changing the fromServer attribute).

This patch adds a test-case that demonstrates how the DB can get into
such a state.

BUG: https://bugzilla.samba.org/show_bug.cgi?id=13621

Signed-off-by: Tim Beale <timbeale@catalyst.net.nz>
Reviewed-by: Douglas Bagnall <douglas.bagnall@catalyst.net.nz>
Reviewed-by: Gary Lockyer <gary@catalyst.net.nz>

selftest/knownfail.d/attr_from_server

@@ -0,0 +1,5 @@
+# test currently fails because once we have a fromServer attribute that points
+# to a non-existent object, the extended_dn DSDB module then suppresses that
+# attribute, which means the object is missing a mandatory attribute, thus
+# invalidating the schema 
+^samba4.tests.attr_from_server.python\(ad_dc_ntvfs\).__main__.FromServerAttrTest.test_dangling_server_attr\(ad_dc_ntvfs:local\)

source4/dsdb/tests/python/attr_from_server.py

@@ -0,0 +1,150 @@
+# -*- coding: utf-8 -*-
+#
+# Tests a corner-case involving the fromServer attribute, which is slightly
+# unique: it's an Object(DS-DN) (like a one-way link), but it is also a
+# mandatory attribute (for nTDSConnection). The corner-case is that the
+# fromServer can potentially end up pointing to a non-existent object.
+# This can happen with other one-way links, but these other one-way links
+# are not mandatory attributes.
+#
+# Copyright (C) Andrew Bartlett 2018
+#
+# 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 __future__ import print_function
+import optparse
+import sys
+sys.path.insert(0, "bin/python")
+import samba
+import os
+import time
+import ldb
+import samba.tests
+from samba.tests.subunitrun import TestProgram, SubunitOptions
+from samba.dcerpc import misc
+from samba.provision import DEFAULTSITE
+
+# note we must connect to the local ldb file on disk, in order to
+# add system-only nTDSDSA objects
+parser = optparse.OptionParser("attr_from_server.py <LDB-filepath>")
+subunitopts = SubunitOptions(parser)
+parser.add_option_group(subunitopts)
+opts, args = parser.parse_args()
+
+if len(args) < 1:
+    parser.print_usage()
+    sys.exit(1)
+
+ldb_path = args[0]
+
+
+class FromServerAttrTest(samba.tests.TestCase):
+    def setUp(self):
+        super(FromServerAttrTest, self).setUp()
+        self.ldb = samba.tests.connect_samdb(ldb_path)
+
+    def tearDown(self):
+        super(FromServerAttrTest, self).tearDown()
+
+    def set_attribute(self, dn, attr, value, operation=ldb.FLAG_MOD_ADD):
+        """Modifies an attribute for an object"""
+        m = ldb.Message()
+        m.dn = ldb.Dn(self.ldb, dn)
+        m[attr] = ldb.MessageElement(value, operation, attr)
+        self.ldb.modify(m)
+
+    def get_object_guid(self, dn):
+        res = self.ldb.search(base=dn, attrs=["objectGUID"],
+                              scope=ldb.SCOPE_BASE)
+        self.assertTrue(len(res) == 1)
+        return str(misc.GUID(res[0]['objectGUID'][0]))
+
+    def test_dangling_server_attr(self):
+        """
+        Tests a scenario where an object has a fromServer attribute that points
+        to an object that no longer exists.
+        """
+
+        # add a temporary server and its associated NTDS Settings object
+        config_dn = self.ldb.get_config_basedn()
+        sites_dn = "CN=Sites,{0}".format(config_dn)
+        servers_dn = "CN=Servers,CN={0},{1}".format(DEFAULTSITE, sites_dn)
+        tmp_server = "CN=TMPSERVER,{0}".format(servers_dn)
+        self.ldb.add({"dn": tmp_server, "objectclass": "server"})
+        server_guid = self.get_object_guid(tmp_server)
+        tmp_ntds_settings = "CN=NTDS Settings,{0}".format(tmp_server)
+        self.ldb.add({"dn": tmp_ntds_settings, "objectClass": "nTDSDSA"},
+                     ["relax:0"])
+
+        # add an NTDS connection under the testenv DC that points to the tmp DC
+        testenv_dc = "CN={0},{1}".format(os.environ["SERVER"], servers_dn)
+        ntds_conn = "CN=Test-NTDS-Conn,CN=NTDS Settings,{0}".format(testenv_dc)
+        ldif = """
+dn: {dn}
+objectClass: nTDSConnection
+fromServer: CN=NTDS Settings,{fromServer}
+options: 1
+enabledConnection: TRUE
+""".format(dn=ntds_conn, fromServer=tmp_server)
+        self.ldb.add_ldif(ldif)
+        self.addCleanup(self.ldb.delete, ntds_conn)
+
+        # sanity-check we can modify the NTDS Connection object
+        self.set_attribute(ntds_conn, 'description', 'Test-value')
+
+        # sanity-check we can't modify the fromServer to point to a bad DN
+        try:
+            bad_dn = "CN=NTDS Settings,CN=BAD-DC,{0}".format(servers_dn)
+            self.set_attribute(ntds_conn, 'fromServer', bad_dn,
+                               operation=ldb.FLAG_MOD_REPLACE)
+            self.fail("Successfully set fromServer to bad DN")
+        except ldb.LdbError as err:
+            enum = err.args[0]
+            self.assertEqual(enum, ldb.ERR_CONSTRAINT_VIOLATION)
+
+        # delete the tmp server, i.e. pretend we demoted it
+        self.ldb.delete(tmp_server, ["tree_delete:1"])
+
+        # check we can still see the deleted server object
+        search_expr = '(objectGUID={0})'.format(server_guid)
+        res = self.ldb.search(config_dn, scope=ldb.SCOPE_SUBTREE,
+                              expression=search_expr,
+                              controls=["show_deleted:1"])
+        self.assertTrue(len(res) == 1, "Could not find deleted server entry")
+
+        # now pretend some time has passed and the deleted server object
+        # has been tombstone-expunged from the DB
+        time.sleep(1)
+        current_time = int(time.time())
+        self.ldb.garbage_collect_tombstones([str(config_dn)], current_time,
+                                            tombstone_lifetime=0)
+
+        # repeat the search to sanity-check the deleted object is really gone
+        res = self.ldb.search(config_dn, scope=ldb.SCOPE_SUBTREE,
+                              expression=search_expr,
+                              controls=["show_deleted:1"])
+        self.assertTrue(len(res) == 0, "Did not expunge deleted server")
+
+        # the nTDSConnection now has a (mandatory) fromServer attribute that
+        # points to an object that no longer exists. Now try to modify an
+        # unrelated attribute on the nTDSConnection
+        try:
+            self.set_attribute(ntds_conn, 'description', 'Test-value-2',
+                               operation=ldb.FLAG_MOD_REPLACE)
+        except ldb.LdbError as err:
+            print(err)
+            self.fail("Could not modify NTDS connection")
+
+
+TestProgram(module=__name__, opts=subunitopts)

source4/selftest/tests.py

@@ -147,6 +147,11 @@ if os.path.exists(os.path.join(samba4bindir, "ldbtest")):
 else:
     skiptestsuite("ldb.base", "Using system LDB, ldbtest not available")
 
+plantestsuite_loadlist("samba4.tests.attr_from_server.python(ad_dc_ntvfs)",
+                       "ad_dc_ntvfs:local",
+                       [python, os.path.join(samba4srcdir, "dsdb/tests/python/attr_from_server.py"),
+                        '$PREFIX_ABS/ad_dc_ntvfs/private/sam.ldb', '$LOADLIST', '$LISTOPT'])
+
 # Tests for RPC
 
 # add tests to this list as they start passing, so we test