diff options
Diffstat (limited to 'source4/dsdb/tests/python/attr_from_server.py')
| -rw-r--r-- | source4/dsdb/tests/python/attr_from_server.py | 149 |
1 files changed, 149 insertions, 0 deletions
diff --git a/source4/dsdb/tests/python/attr_from_server.py b/source4/dsdb/tests/python/attr_from_server.py new file mode 100644 index 0000000..aca356b --- /dev/null +++ b/source4/dsdb/tests/python/attr_from_server.py @@ -0,0 +1,149 @@ +# -*- 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/>. +# +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) |