source4/dsdb/tests/python/deletetest.py

   1 #!/usr/bin/env python
   2 # -*- coding: utf-8 -*-
   3
   4 import optparse
   5 import sys
   6 import os
   7
   8 sys.path.append("bin/python")
   9 import samba
  10 samba.ensure_external_module("testtools", "testtools")
  11 samba.ensure_external_module("subunit", "subunit/python")
  12
  13 import samba.getopt as options
  14
  15 from samba.auth import system_session
  16 from ldb import SCOPE_BASE, LdbError
  17 from ldb import ERR_NO_SUCH_OBJECT, ERR_NOT_ALLOWED_ON_NON_LEAF
  18 from ldb import ERR_UNWILLING_TO_PERFORM
  19 from samba.samdb import SamDB
  20 from samba.tests import delete_force
  21
  22 from subunit.run import SubunitTestRunner
  23 import unittest
  24
  25 parser = optparse.OptionParser("deletetest.py [options] <host|file>")
  26 sambaopts = options.SambaOptions(parser)
  27 parser.add_option_group(sambaopts)
  28 parser.add_option_group(options.VersionOptions(parser))
  29 # use command line creds if available
  30 credopts = options.CredentialsOptions(parser)
  31 parser.add_option_group(credopts)
  32 opts, args = parser.parse_args()
  33
  34 if len(args) < 1:
  35     parser.print_usage()
  36     sys.exit(1)
  37
  38 host = args[0]
  39
  40 lp = sambaopts.get_loadparm()
  41 creds = credopts.get_credentials(lp)
  42
  43 class BasicDeleteTests(unittest.TestCase):
  44
  45
  46     def GUID_string(self, guid):
  47         return self.ldb.schema_format_value("objectGUID", guid)
  48
  49     def setUp(self):
  50         self.ldb = ldb
  51         self.base_dn = ldb.domain_dn()
  52         self.configuration_dn = ldb.get_config_basedn().get_linearized()
  53
  54     def search_guid(self, guid):
  55         print "SEARCH by GUID %s" % self.GUID_string(guid)
  56
  57         res = ldb.search(base="<GUID=%s>" % self.GUID_string(guid),
  58                          scope=SCOPE_BASE, controls=["show_deleted:1"])
  59         self.assertEquals(len(res), 1)
  60         return res[0]
  61
  62     def search_dn(self,dn):
  63         print "SEARCH by DN %s" % dn
  64
  65         res = ldb.search(expression="(objectClass=*)",
  66                          base=dn,
  67                          scope=SCOPE_BASE,
  68                          controls=["show_deleted:1"])
  69         self.assertEquals(len(res), 1)
  70         return res[0]
  71
  72     def del_attr_values(self, delObj):
  73         print "Checking attributes for %s" % delObj["dn"]
  74
  75         self.assertEquals(delObj["isDeleted"][0],"TRUE")
  76         self.assertTrue(not("objectCategory" in delObj))
  77         self.assertTrue(not("sAMAccountType" in delObj))
  78
  79     def preserved_attributes_list(self, liveObj, delObj):
  80         print "Checking for preserved attributes list"
  81
  82         preserved_list = ["nTSecurityDescriptor", "attributeID", "attributeSyntax", "dNReferenceUpdate", "dNSHostName",
  83         "flatName", "governsID", "groupType", "instanceType", "lDAPDisplayName", "legacyExchangeDN",
  84         "isDeleted", "isRecycled", "lastKnownParent", "msDS-LastKnownRDN", "mS-DS-CreatorSID",
  85         "mSMQOwnerID", "nCName", "objectClass", "distinguishedName", "objectGUID", "objectSid",
  86         "oMSyntax", "proxiedObjectName", "name", "replPropertyMetaData", "sAMAccountName",
  87         "securityIdentifier", "sIDHistory", "subClassOf", "systemFlags", "trustPartner", "trustDirection",
  88         "trustType", "trustAttributes", "userAccountControl", "uSNChanged", "uSNCreated", "whenCreated"]
  89
  90         for a in liveObj:
  91             if a in preserved_list:
  92                 self.assertTrue(a in delObj)
  93
  94     def check_rdn(self, liveObj, delObj, rdnName):
  95         print "Checking for correct rDN"
  96         rdn=liveObj[rdnName][0]
  97         rdn2=delObj[rdnName][0]
  98         name2=delObj[rdnName][0]
  99         guid=liveObj["objectGUID"][0]
 100         self.assertEquals(rdn2, rdn + "\nDEL:" + self.GUID_string(guid))
 101         self.assertEquals(name2, rdn + "\nDEL:" + self.GUID_string(guid))
 102
 103     def delete_deleted(self, ldb, dn):
 104         print "Testing the deletion of the already deleted dn %s" % dn
 105
 106         try:
 107             ldb.delete(dn)
 108             self.fail()
 109         except LdbError, (num, _):
 110             self.assertEquals(num, ERR_NO_SUCH_OBJECT)
 111
 112     def test_delete_protection(self):
 113         """Delete protection tests"""
 114
 115         print self.base_dn
 116
 117         delete_force(self.ldb, "cn=entry1,cn=ldaptestcontainer," + self.base_dn)
 118         delete_force(self.ldb, "cn=entry2,cn=ldaptestcontainer," + self.base_dn)
 119         delete_force(self.ldb, "cn=ldaptestcontainer," + self.base_dn)
 120
 121         ldb.add({
 122             "dn": "cn=ldaptestcontainer," + self.base_dn,
 123             "objectclass": "container"})
 124         ldb.add({
 125             "dn": "cn=entry1,cn=ldaptestcontainer," + self.base_dn,
 126             "objectclass": "container"})
 127         ldb.add({
 128             "dn": "cn=entry2,cn=ldaptestcontainer," + self.base_dn,
 129             "objectclass": "container"})
 130
 131         try:
 132             ldb.delete("cn=ldaptestcontainer," + self.base_dn)
 133             self.fail()
 134         except LdbError, (num, _):
 135             self.assertEquals(num, ERR_NOT_ALLOWED_ON_NON_LEAF)
 136
 137         ldb.delete("cn=ldaptestcontainer," + self.base_dn, ["tree_delete:1"])
 138
 139         try:
 140             res = ldb.search("cn=ldaptestcontainer," + self.base_dn,
 141                              scope=SCOPE_BASE, attrs=[])
 142             self.fail()
 143         except LdbError, (num, _):
 144             self.assertEquals(num, ERR_NO_SUCH_OBJECT)
 145         try:
 146             res = ldb.search("cn=entry1,cn=ldaptestcontainer," + self.base_dn,
 147                              scope=SCOPE_BASE, attrs=[])
 148             self.fail()
 149         except LdbError, (num, _):
 150             self.assertEquals(num, ERR_NO_SUCH_OBJECT)
 151         try:
 152             res = ldb.search("cn=entry2,cn=ldaptestcontainer," + self.base_dn,
 153                              scope=SCOPE_BASE, attrs=[])
 154             self.fail()
 155         except LdbError, (num, _):
 156             self.assertEquals(num, ERR_NO_SUCH_OBJECT)
 157
 158         delete_force(self.ldb, "cn=entry1,cn=ldaptestcontainer," + self.base_dn)
 159         delete_force(self.ldb, "cn=entry2,cn=ldaptestcontainer," + self.base_dn)
 160         delete_force(self.ldb, "cn=ldaptestcontainer," + self.base_dn)
 161
 162         # Performs some protected object delete testing
 163
 164         res = ldb.search(base="", expression="", scope=SCOPE_BASE,
 165                          attrs=["dsServiceName", "dNSHostName"])
 166         self.assertEquals(len(res), 1)
 167
 168         # Delete failing since DC's nTDSDSA object is protected
 169         try:
 170             ldb.delete(res[0]["dsServiceName"][0])
 171             self.fail()
 172         except LdbError, (num, _):
 173             self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
 174
 175         res = ldb.search(self.base_dn, attrs=["rIDSetReferences"],
 176                          expression="(&(objectClass=computer)(dNSHostName=" + res[0]["dNSHostName"][0] + "))")
 177         self.assertEquals(len(res), 1)
 178
 179         # Deletes failing since DC's rIDSet object is protected
 180         try:
 181             ldb.delete(res[0]["rIDSetReferences"][0])
 182             self.fail()
 183         except LdbError, (num, _):
 184             self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
 185         try:
 186             ldb.delete(res[0]["rIDSetReferences"][0], ["tree_delete:1"])
 187             self.fail()
 188         except LdbError, (num, _):
 189             self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
 190
 191         # Deletes failing since three main crossRef objects are protected
 192
 193         try:
 194             ldb.delete("cn=Enterprise Schema,cn=Partitions," + self.configuration_dn)
 195             self.fail()
 196         except LdbError, (num, _):
 197             self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
 198         try:
 199             ldb.delete("cn=Enterprise Schema,cn=Partitions," + self.configuration_dn, ["tree_delete:1"])
 200             self.fail()
 201         except LdbError, (num, _):
 202             self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
 203
 204         try:
 205             ldb.delete("cn=Enterprise Configuration,cn=Partitions," + self.configuration_dn)
 206             self.fail()
 207         except LdbError, (num, _):
 208             self.assertEquals(num, ERR_NOT_ALLOWED_ON_NON_LEAF)
 209         try:
 210             ldb.delete("cn=Enterprise Configuration,cn=Partitions," + self.configuration_dn, ["tree_delete:1"])
 211             self.fail()
 212         except LdbError, (num, _):
 213             self.assertEquals(num, ERR_NOT_ALLOWED_ON_NON_LEAF)
 214
 215         res = ldb.search("cn=Partitions," + self.configuration_dn, attrs=[],
 216                          expression="(nCName=%s)" % self.base_dn)
 217         self.assertEquals(len(res), 1)
 218
 219         try:
 220             ldb.delete(res[0].dn)
 221             self.fail()
 222         except LdbError, (num, _):
 223             self.assertEquals(num, ERR_NOT_ALLOWED_ON_NON_LEAF)
 224         try:
 225             ldb.delete(res[0].dn, ["tree_delete:1"])
 226             self.fail()
 227         except LdbError, (num, _):
 228             self.assertEquals(num, ERR_NOT_ALLOWED_ON_NON_LEAF)
 229
 230         # Delete failing since "SYSTEM_FLAG_DISALLOW_DELETE"
 231         try:
 232             ldb.delete("CN=Users," + self.base_dn)
 233             self.fail()
 234         except LdbError, (num, _):
 235             self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
 236
 237         # Tree-delete failing since "isCriticalSystemObject"
 238         try:
 239             ldb.delete("CN=Computers," + self.base_dn, ["tree_delete:1"])
 240             self.fail()
 241         except LdbError, (num, _):
 242             self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
 243
 244     def test_all(self):
 245         """Basic delete tests"""
 246
 247         print self.base_dn
 248
 249         usr1="cn=testuser,cn=users," + self.base_dn
 250         usr2="cn=testuser2,cn=users," + self.base_dn
 251         grp1="cn=testdelgroup1,cn=users," + self.base_dn
 252         sit1="cn=testsite1,cn=sites," + self.configuration_dn
 253         ss1="cn=NTDS Site Settings,cn=testsite1,cn=sites," + self.configuration_dn
 254         srv1="cn=Servers,cn=testsite1,cn=sites," + self.configuration_dn
 255         srv2="cn=TESTSRV,cn=Servers,cn=testsite1,cn=sites," + self.configuration_dn
 256
 257         delete_force(self.ldb, usr1)
 258         delete_force(self.ldb, usr2)
 259         delete_force(self.ldb, grp1)
 260         delete_force(self.ldb, ss1)
 261         delete_force(self.ldb, srv2)
 262         delete_force(self.ldb, srv1)
 263         delete_force(self.ldb, sit1)
 264
 265         ldb.add({
 266             "dn": usr1,
 267             "objectclass": "user",
 268             "description": "test user description",
 269             "samaccountname": "testuser"})
 270
 271         ldb.add({
 272             "dn": usr2,
 273             "objectclass": "user",
 274             "description": "test user 2 description",
 275             "samaccountname": "testuser2"})
 276
 277         ldb.add({
 278             "dn": grp1,
 279             "objectclass": "group",
 280             "description": "test group",
 281             "samaccountname": "testdelgroup1",
 282             "member": [ usr1, usr2 ],
 283             "isDeleted": "FALSE" })
 284
 285         ldb.add({
 286             "dn": sit1,
 287             "objectclass": "site" })
 288
 289         ldb.add({
 290             "dn": ss1,
 291             "objectclass": ["applicationSiteSettings", "nTDSSiteSettings"] })
 292
 293         ldb.add({
 294             "dn": srv1,
 295             "objectclass": "serversContainer" })
 296
 297         ldb.add({
 298             "dn": srv2,
 299             "objectClass": "server" })
 300
 301         objLive1 = self.search_dn(usr1)
 302         guid1=objLive1["objectGUID"][0]
 303
 304         objLive2 = self.search_dn(usr2)
 305         guid2=objLive2["objectGUID"][0]
 306
 307         objLive3 = self.search_dn(grp1)
 308         guid3=objLive3["objectGUID"][0]
 309
 310         objLive4 = self.search_dn(sit1)
 311         guid4=objLive4["objectGUID"][0]
 312
 313         objLive5 = self.search_dn(ss1)
 314         guid5=objLive5["objectGUID"][0]
 315
 316         objLive6 = self.search_dn(srv1)
 317         guid6=objLive6["objectGUID"][0]
 318
 319         objLive7 = self.search_dn(srv2)
 320         guid7=objLive7["objectGUID"][0]
 321
 322         ldb.delete(usr1)
 323         ldb.delete(usr2)
 324         ldb.delete(grp1)
 325         ldb.delete(srv1, ["tree_delete:1"])
 326         ldb.delete(sit1, ["tree_delete:1"])
 327
 328         objDeleted1 = self.search_guid(guid1)
 329         objDeleted2 = self.search_guid(guid2)
 330         objDeleted3 = self.search_guid(guid3)
 331         objDeleted4 = self.search_guid(guid4)
 332         objDeleted5 = self.search_guid(guid5)
 333         objDeleted6 = self.search_guid(guid6)
 334         objDeleted7 = self.search_guid(guid7)
 335
 336         self.del_attr_values(objDeleted1)
 337         self.del_attr_values(objDeleted2)
 338         self.del_attr_values(objDeleted3)
 339         self.del_attr_values(objDeleted4)
 340         self.del_attr_values(objDeleted5)
 341         self.del_attr_values(objDeleted6)
 342         self.del_attr_values(objDeleted7)
 343
 344         self.preserved_attributes_list(objLive1, objDeleted1)
 345         self.preserved_attributes_list(objLive2, objDeleted2)
 346         self.preserved_attributes_list(objLive3, objDeleted3)
 347         self.preserved_attributes_list(objLive4, objDeleted4)
 348         self.preserved_attributes_list(objLive5, objDeleted5)
 349         self.preserved_attributes_list(objLive6, objDeleted6)
 350         self.preserved_attributes_list(objLive7, objDeleted7)
 351
 352         self.check_rdn(objLive1, objDeleted1, "cn")
 353         self.check_rdn(objLive2, objDeleted2, "cn")
 354         self.check_rdn(objLive3, objDeleted3, "cn")
 355         self.check_rdn(objLive4, objDeleted4, "cn")
 356         self.check_rdn(objLive5, objDeleted5, "cn")
 357         self.check_rdn(objLive6, objDeleted6, "cn")
 358         self.check_rdn(objLive7, objDeleted7, "cn")
 359
 360         self.delete_deleted(ldb, usr1)
 361         self.delete_deleted(ldb, usr2)
 362         self.delete_deleted(ldb, grp1)
 363         self.delete_deleted(ldb, sit1)
 364         self.delete_deleted(ldb, ss1)
 365         self.delete_deleted(ldb, srv1)
 366         self.delete_deleted(ldb, srv2)
 367
 368         self.assertTrue("CN=Deleted Objects" in str(objDeleted1.dn))
 369         self.assertTrue("CN=Deleted Objects" in str(objDeleted2.dn))
 370         self.assertTrue("CN=Deleted Objects" in str(objDeleted3.dn))
 371         self.assertFalse("CN=Deleted Objects" in str(objDeleted4.dn))
 372         self.assertTrue("CN=Deleted Objects" in str(objDeleted5.dn))
 373         self.assertFalse("CN=Deleted Objects" in str(objDeleted6.dn))
 374         self.assertFalse("CN=Deleted Objects" in str(objDeleted7.dn))
 375
 376 if not "://" in host:
 377     if os.path.isfile(host):
 378         host = "tdb://%s" % host
 379     else:
 380         host = "ldap://%s" % host
 381
 382 ldb = SamDB(host, credentials=creds, session_info=system_session(), lp=lp)
 383
 384 runner = SubunitTestRunner()
 385 rc = 0
 386 if not runner.run(unittest.makeSuite(BasicDeleteTests)).wasSuccessful():
 387     rc = 1
 388
 389 sys.exit(rc)