source4/torture/drs/python/delete_object.py

   1 #!/usr/bin/env python3
   2 # -*- coding: utf-8 -*-
   3 #
   4 # Unix SMB/CIFS implementation.
   5 # Copyright (C) Kamen Mazdrashki <kamenim@samba.org> 2010
   6 #
   7 # This program is free software; you can redistribute it and/or modify
   8 # it under the terms of the GNU General Public License as published by
   9 # the Free Software Foundation; either version 3 of the License, or
  10 # (at your option) any later version.
  11 #
  12 # This program is distributed in the hope that it will be useful,
  13 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15 # GNU General Public License for more details.
  16 #
  17 # You should have received a copy of the GNU General Public License
  18 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  19 #
  20
  21 #
  22 # Usage:
  23 #  export DC1=dc1_dns_name
  24 #  export DC2=dc2_dns_name
  25 #  export SUBUNITRUN=$samba4srcdir/scripting/bin/subunitrun
  26 #  PYTHONPATH="$PYTHONPATH:$samba4srcdir/torture/drs/python" $SUBUNITRUN delete_object -U"$DOMAIN/$DC_USERNAME"%"$DC_PASSWORD"
  27 #
  28
  29 from __future__ import print_function
  30 import time
  31
  32
  33 from ldb import (
  34     SCOPE_SUBTREE,
  35 )
  36
  37 import drs_base
  38 import ldb
  39
  40
  41 class DrsDeleteObjectTestCase(drs_base.DrsBaseTestCase):
  42
  43     def setUp(self):
  44         super(DrsDeleteObjectTestCase, self).setUp()
  45         # disable automatic replication temporary
  46         self._disable_all_repl(self.dnsname_dc1)
  47         self._disable_all_repl(self.dnsname_dc2)
  48         # make sure DCs are synchronized before the test
  49         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True)
  50         self._net_drs_replicate(DC=self.dnsname_dc1, fromDC=self.dnsname_dc2, forced=True)
  51
  52     def tearDown(self):
  53         self._enable_all_repl(self.dnsname_dc1)
  54         self._enable_all_repl(self.dnsname_dc2)
  55         super(DrsDeleteObjectTestCase, self).tearDown()
  56
  57     def _make_username(self):
  58         return "DrsDelObjUser_" + time.strftime("%s", time.gmtime())
  59
  60     # now also used to check the group
  61     def _check_obj(self, sam_ldb, obj_orig, is_deleted):
  62         # search the user by guid as it may be deleted
  63         guid_str = self._GUID_string(obj_orig["objectGUID"][0])
  64         expression = "(objectGUID=%s)" % guid_str
  65         res = sam_ldb.search(base=self.domain_dn,
  66                              expression=expression,
  67                              controls=["show_deleted:1"])
  68         self.assertEquals(len(res), 1)
  69         user_cur = res[0]
  70         # Deleted Object base DN
  71         dodn = self._deleted_objects_dn(sam_ldb)
  72         # now check properties of the user
  73         cn_orig = str(obj_orig["cn"][0])
  74         cn_cur  = str(user_cur["cn"][0])
  75         name_orig = str(obj_orig["name"][0])
  76         name_cur  = str(user_cur["name"][0])
  77         if is_deleted:
  78             self.assertEquals(str(user_cur["isDeleted"][0]), "TRUE")
  79             self.assertFalse("objectCategory" in user_cur)
  80             self.assertFalse("sAMAccountType" in user_cur)
  81             self.assertFalse("description" in user_cur)
  82             self.assertFalse("memberOf" in user_cur)
  83             self.assertFalse("member" in user_cur)
  84             self.assertTrue(dodn in str(user_cur["dn"]),
  85                             "User %s is deleted but it is not located under %s (found at %s)!" % (name_orig, dodn, user_cur["dn"]))
  86             self.assertEquals(name_cur, name_orig + "\nDEL:" + guid_str)
  87             self.assertEquals(name_cur, user_cur.dn.get_rdn_value())
  88             self.assertEquals(cn_cur, cn_orig + "\nDEL:" + guid_str)
  89             self.assertEquals(name_cur, cn_cur)
  90         else:
  91             self.assertFalse("isDeleted" in user_cur)
  92             self.assertEquals(name_cur, name_orig)
  93             self.assertEquals(name_cur, user_cur.dn.get_rdn_value())
  94             self.assertEquals(cn_cur, cn_orig)
  95             self.assertEquals(name_cur, cn_cur)
  96             self.assertEquals(obj_orig["dn"], user_cur["dn"])
  97             self.assertTrue(dodn not in str(user_cur["dn"]))
  98         return user_cur
  99
 100     def test_ReplicateDeletedObject1(self):
 101         """Verifies how a deleted-object is replicated between two DCs.
 102            This test should verify that:
 103             - deleted-object is replicated properly
 104             - We verify that after replication,
 105               object's state to conform to a tombstone-object state
 106             - This test replicates the object modifications to
 107               the server with the user deleted first
 108
 109            TODO:  It will also be great if check replPropertyMetaData.
 110            TODO:  Check for deleted-object state, depending on DC's features
 111                   when recycle-bin is enabled
 112            """
 113         # work-out unique username to test with
 114         username = self._make_username()
 115
 116         # create user on DC1
 117         self.ldb_dc1.newuser(username=username, password="P@sswOrd!")
 118         ldb_res = self.ldb_dc1.search(base=self.domain_dn,
 119                                       scope=SCOPE_SUBTREE,
 120                                       expression="(samAccountName=%s)" % username)
 121         self.assertEquals(len(ldb_res), 1)
 122         user_orig = ldb_res[0]
 123         user_dn   = ldb_res[0]["dn"]
 124
 125         # check user info on DC1
 126         print("Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0])))
 127         self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=user_orig, is_deleted=False)
 128
 129         # trigger replication from DC1 to DC2
 130         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True)
 131
 132         # delete user on DC1
 133         self.ldb_dc1.delete(user_dn)
 134         # check user info on DC1 - should be deleted
 135         self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=user_orig, is_deleted=True)
 136         # check user info on DC2 - should be valid user
 137         user_cur = self._check_obj(sam_ldb=self.ldb_dc2, obj_orig=user_orig, is_deleted=False)
 138
 139         # The user should not have a description or memberOf yet
 140         self.assertFalse("description" in user_cur)
 141         self.assertFalse("memberOf" in user_cur)
 142
 143         self.ldb_dc2.newgroup("group_%s" % username)
 144
 145         self.ldb_dc2.newgroup("group2_%s" % username)
 146
 147         ldb_res = self.ldb_dc2.search(base=self.domain_dn,
 148                                       scope=SCOPE_SUBTREE,
 149                                       expression="(samAccountName=group_%s)" % username)
 150         self.assertTrue(len(ldb_res) == 1)
 151         self.assertTrue("sAMAccountName" in ldb_res[0])
 152         group_orig = ldb_res[0]
 153         group_dn = ldb_res[0]["dn"]
 154
 155         # modify user on DC2 to have a description and be a member of the group
 156         m = ldb.Message()
 157         m.dn = user_dn
 158         m["description"] = ldb.MessageElement("a description",
 159                                               ldb.FLAG_MOD_ADD, "description")
 160         self.ldb_dc2.modify(m)
 161         m = ldb.Message()
 162         m.dn = group_dn
 163         m["member"] = ldb.MessageElement(str(user_dn),
 164                                          ldb.FLAG_MOD_ADD, "member")
 165         self.ldb_dc2.modify(m)
 166
 167         ldb_res = self.ldb_dc2.search(base=self.domain_dn,
 168                                       scope=SCOPE_SUBTREE,
 169                                       expression="(samAccountName=group2_%s)" % username)
 170         self.assertTrue(len(ldb_res) == 1)
 171         self.assertTrue("sAMAccountName" in ldb_res[0])
 172         group2_dn = ldb_res[0]["dn"]
 173         group2_orig = ldb_res[0]
 174
 175         m = ldb.Message()
 176         m.dn = group2_dn
 177         m["member"] = ldb.MessageElement(str(group_dn),
 178                                          ldb.FLAG_MOD_ADD, "member")
 179         self.ldb_dc2.modify(m)
 180
 181         # check user info on DC2 - should be valid user
 182         user_cur = self._check_obj(sam_ldb=self.ldb_dc2, obj_orig=user_orig, is_deleted=False)
 183
 184         # The user should not have a description yet
 185         self.assertTrue("description" in user_cur)
 186         self.assertTrue("memberOf" in user_cur)
 187
 188         ldb_res = self.ldb_dc2.search(base=self.domain_dn,
 189                                       scope=SCOPE_SUBTREE,
 190                                       expression="(samAccountName=group_%s)" % username)
 191         self.assertTrue(len(ldb_res) == 1)
 192
 193         # This group is a member of another group
 194         self.assertTrue("memberOf" in ldb_res[0])
 195
 196         # The user was deleted on DC1, but check the modify we just did on DC2
 197         self.assertTrue("member" in ldb_res[0])
 198
 199         # trigger replication from DC2 to DC1
 200         # to check if deleted object gets restored
 201         self._net_drs_replicate(DC=self.dnsname_dc1, fromDC=self.dnsname_dc2, forced=True)
 202         # check user info on DC1 - should be deleted
 203         self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=user_orig, is_deleted=True)
 204         # check user info on DC2 - should be valid user
 205         self._check_obj(sam_ldb=self.ldb_dc2, obj_orig=user_orig, is_deleted=False)
 206
 207         ldb_res = self.ldb_dc1.search(base=self.domain_dn,
 208                                       scope=SCOPE_SUBTREE,
 209                                       expression="(samAccountName=group_%s)" % username)
 210         self.assertTrue(len(ldb_res) == 1)
 211
 212         # This group is a member of another group
 213         self.assertTrue("memberOf" in ldb_res[0])
 214
 215         # The user was deleted on DC1, but the modify we did on DC2, check it never replicated in
 216         self.assertFalse("member" in ldb_res[0])
 217
 218         # trigger replication from DC1 to DC2
 219         # to check if deleted object is replicated
 220         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True)
 221         # check user info on DC1 - should be deleted
 222         self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=user_orig, is_deleted=True)
 223         # check user info on DC2 - should be deleted
 224         self._check_obj(sam_ldb=self.ldb_dc2, obj_orig=user_orig, is_deleted=True)
 225
 226         # delete group on DC1
 227         self.ldb_dc1.delete(group_dn)
 228
 229         # trigger replication from DC1 to DC2
 230         # to check if deleted object is replicated
 231         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True)
 232
 233         # check group info on DC1 - should be deleted
 234         self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=group_orig, is_deleted=True)
 235         # check group info on DC2 - should be deleted
 236         self._check_obj(sam_ldb=self.ldb_dc2, obj_orig=group_orig, is_deleted=True)
 237
 238         ldb_res = self.ldb_dc2.search(base=self.domain_dn,
 239                                       scope=SCOPE_SUBTREE,
 240                                       expression="(samAccountName=group2_%s)" % username)
 241         self.assertTrue(len(ldb_res) == 1)
 242         self.assertFalse("member" in ldb_res[0])
 243
 244         # delete group on DC1
 245         self.ldb_dc1.delete(group2_dn)
 246
 247         # trigger replication from DC1 to DC2
 248         # to check if deleted object is replicated
 249         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True)
 250
 251         # check group info on DC1 - should be deleted
 252         self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=group2_orig, is_deleted=True)
 253         # check group info on DC2 - should be deleted
 254         self._check_obj(sam_ldb=self.ldb_dc2, obj_orig=group2_orig, is_deleted=True)
 255
 256     def test_ReplicateDeletedObject2(self):
 257         """Verifies how a deleted-object is replicated between two DCs.
 258            This test should verify that:
 259             - deleted-object is replicated properly
 260             - We verify that after replication,
 261               object's state to conform to a tombstone-object state
 262             - This test replicates the delete to the server with the
 263               object modifications first
 264
 265            TODO:  It will also be great if check replPropertyMetaData.
 266            TODO:  Check for deleted-object state, depending on DC's features
 267                   when recycle-bin is enabled
 268            """
 269         # work-out unique username to test with
 270         username = self._make_username()
 271
 272         # create user on DC1
 273         self.ldb_dc1.newuser(username=username, password="P@sswOrd!")
 274         ldb_res = self.ldb_dc1.search(base=self.domain_dn,
 275                                       scope=SCOPE_SUBTREE,
 276                                       expression="(samAccountName=%s)" % username)
 277         self.assertEquals(len(ldb_res), 1)
 278         user_orig = ldb_res[0]
 279         user_dn   = ldb_res[0]["dn"]
 280
 281         # check user info on DC1
 282         print("Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0])))
 283         self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=user_orig, is_deleted=False)
 284
 285         # trigger replication from DC1 to DC2
 286         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True)
 287
 288         # delete user on DC1
 289         self.ldb_dc1.delete(user_dn)
 290         # check user info on DC1 - should be deleted
 291         self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=user_orig, is_deleted=True)
 292         # check user info on DC2 - should be valid user
 293         user_cur = self._check_obj(sam_ldb=self.ldb_dc2, obj_orig=user_orig, is_deleted=False)
 294
 295         # The user should not have a description or memberOf yet
 296         self.assertFalse("description" in user_cur)
 297         self.assertFalse("memberOf" in user_cur)
 298
 299         self.ldb_dc2.newgroup("group_%s" % username)
 300
 301         self.ldb_dc2.newgroup("group2_%s" % username)
 302
 303         ldb_res = self.ldb_dc2.search(base=self.domain_dn,
 304                                       scope=SCOPE_SUBTREE,
 305                                       expression="(samAccountName=group_%s)" % username)
 306         self.assertTrue(len(ldb_res) == 1)
 307         self.assertTrue("sAMAccountName" in ldb_res[0])
 308         group_orig = ldb_res[0]
 309         group_dn = ldb_res[0]["dn"]
 310
 311         # modify user on DC2 to have a description and be a member of the group
 312         m = ldb.Message()
 313         m.dn = user_dn
 314         m["description"] = ldb.MessageElement("a description",
 315                                               ldb.FLAG_MOD_ADD, "description")
 316         self.ldb_dc2.modify(m)
 317         m = ldb.Message()
 318         m.dn = group_dn
 319         m["member"] = ldb.MessageElement(str(user_dn),
 320                                          ldb.FLAG_MOD_ADD, "member")
 321         self.ldb_dc2.modify(m)
 322
 323         ldb_res = self.ldb_dc2.search(base=self.domain_dn,
 324                                       scope=SCOPE_SUBTREE,
 325                                       expression="(samAccountName=group2_%s)" % username)
 326         self.assertTrue(len(ldb_res) == 1)
 327         self.assertTrue("sAMAccountName" in ldb_res[0])
 328         group2_dn = ldb_res[0]["dn"]
 329         group2_orig = ldb_res[0]
 330
 331         m = ldb.Message()
 332         m.dn = group2_dn
 333         m["member"] = ldb.MessageElement(str(group_dn),
 334                                          ldb.FLAG_MOD_ADD, "member")
 335         self.ldb_dc2.modify(m)
 336
 337         # check user info on DC2 - should be valid user
 338         user_cur = self._check_obj(sam_ldb=self.ldb_dc2, obj_orig=user_orig, is_deleted=False)
 339
 340         # The user should not have a description yet
 341         self.assertTrue("description" in user_cur)
 342         self.assertTrue("memberOf" in user_cur)
 343
 344         # trigger replication from DC1 to DC2
 345         # to check if deleted object gets restored
 346         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True)
 347         # check user info on DC1 - should be deleted
 348         self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=user_orig, is_deleted=True)
 349         # check user info on DC2 - should be deleted
 350         self._check_obj(sam_ldb=self.ldb_dc2, obj_orig=user_orig, is_deleted=True)
 351
 352         ldb_res = self.ldb_dc2.search(base=self.domain_dn,
 353                                       scope=SCOPE_SUBTREE,
 354                                       expression="(samAccountName=group_%s)" % username)
 355         self.assertTrue(len(ldb_res) == 1)
 356         self.assertTrue("memberOf" in ldb_res[0])
 357         self.assertFalse("member" in ldb_res[0])
 358
 359         # trigger replication from DC2 to DC1
 360         # to check if deleted object is replicated
 361         self._net_drs_replicate(DC=self.dnsname_dc1, fromDC=self.dnsname_dc2, forced=True)
 362         # check user info on DC1 - should be deleted
 363         self._check_obj(sam_ldb=self.ldb_dc1, obj_orig=user_orig, is_deleted=True)
 364         # check user info on DC2 - should be deleted
 365         self._check_obj(sam_ldb=self.ldb_dc2, obj_orig=user_orig, is_deleted=True)
 366
 367         ldb_res = self.ldb_dc1.search(base=self.domain_dn,
 368                                       scope=SCOPE_SUBTREE,
 369                                       expression="(samAccountName=group_%s)" % username)
 370         self.assertTrue(len(ldb_res) == 1)
 371         self.assertTrue("memberOf" in ldb_res[0])
 372         self.assertFalse("member" in ldb_res[0])
 373
 374         # delete group on DC1
 375         self.ldb_dc1.delete(group_dn)
 376         self.ldb_dc1.delete(group2_dn)
 377
 378         # trigger replication from DC1 to DC2, for cleanup
 379         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1, forced=True)