source4/torture/drs/python/delete_object.py

   1 #!/usr/bin/env python
   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 import sys
  30 import time
  31 import os
  32
  33 sys.path.append("bin/python")
  34
  35 from ldb import (
  36     SCOPE_BASE,
  37     SCOPE_SUBTREE
  38     )
  39
  40 import samba.tests
  41
  42
  43 class DrsDeleteObjectTestCase(samba.tests.TestCase):
  44
  45     # RootDSE msg for DC1
  46     info_dc1 = None
  47     ldb_dc1 = None
  48     # RootDSE msg for DC1
  49     info_dc2 = None
  50     ldb_dc2 = None
  51
  52     def setUp(self):
  53         super(DrsDeleteObjectTestCase, self).setUp()
  54
  55         # connect to DCs singleton
  56         if self.ldb_dc1 is None:
  57             DrsDeleteObjectTestCase.dc1 = samba.tests.env_get_var_value("DC1")
  58             DrsDeleteObjectTestCase.ldb_dc1 = samba.tests.connect_samdb(self.dc1, ldap_only=True)
  59         if self.ldb_dc2 is None:
  60             DrsDeleteObjectTestCase.dc2 = samba.tests.env_get_var_value("DC2")
  61             DrsDeleteObjectTestCase.ldb_dc2 = samba.tests.connect_samdb(self.dc2, ldap_only=True)
  62
  63         # fetch rootDSEs
  64         if self.info_dc1 is None:
  65             ldb = self.ldb_dc1
  66             res = ldb.search(base="", expression="", scope=SCOPE_BASE, attrs=["*"])
  67             self.assertEquals(len(res), 1)
  68             DrsDeleteObjectTestCase.info_dc1 = res[0]
  69         if self.info_dc2 is None:
  70             ldb = self.ldb_dc2
  71             res = ldb.search(base="", expression="", scope=SCOPE_BASE, attrs=["*"])
  72             self.assertEquals(len(res), 1)
  73             DrsDeleteObjectTestCase.info_dc2 = res[0]
  74
  75         # cache some of RootDSE props
  76         self.schema_dn = self.info_dc1["schemaNamingContext"][0]
  77         self.domain_dn = self.info_dc1["defaultNamingContext"][0]
  78         self.config_dn = self.info_dc1["configurationNamingContext"][0]
  79         self.forest_level = int(self.info_dc1["forestFunctionality"][0])
  80
  81         # we will need DCs DNS names for 'samba-tool drs' command
  82         self.dnsname_dc1 = self.info_dc1["dnsHostName"][0]
  83         self.dnsname_dc2 = self.info_dc2["dnsHostName"][0]
  84
  85     def tearDown(self):
  86         super(DrsDeleteObjectTestCase, self).tearDown()
  87
  88     def _GUID_string(self, guid):
  89         return self.ldb_dc1.schema_format_value("objectGUID", guid)
  90
  91     def _deleted_objects_dn(self, sam_ldb):
  92         wkdn = "<WKGUID=18E2EA80684F11D2B9AA00C04F79F805,%s>" % self.domain_dn
  93         res = sam_ldb.search(base=wkdn,
  94                              scope=SCOPE_BASE,
  95                              controls=["show_deleted:1"])
  96         self.assertEquals(len(res), 1)
  97         return str(res[0]["dn"])
  98
  99     def _make_username(self):
 100         return "DrsDelObjUser_" + time.strftime("%s", time.gmtime())
 101
 102     def _check_user(self, sam_ldb, user_orig, is_deleted):
 103         # search the user by guid as it may be deleted
 104         guid_str = self._GUID_string(user_orig["objectGUID"][0])
 105         expression = "(objectGUID=%s)" % guid_str
 106         res = sam_ldb.search(base=self.domain_dn,
 107                              expression=expression,
 108                              controls=["show_deleted:1"])
 109         self.assertEquals(len(res), 1)
 110         user_cur = res[0]
 111         # Deleted Object base DN
 112         dodn = self._deleted_objects_dn(sam_ldb)
 113         # now check properties of the user
 114         name_orig = user_orig["cn"][0]
 115         name_cur  = user_cur["cn"][0]
 116         if is_deleted:
 117             self.assertEquals(user_cur["isDeleted"][0],"TRUE")
 118             self.assertTrue(not("objectCategory" in user_cur))
 119             self.assertTrue(not("sAMAccountType" in user_cur))
 120             self.assertTrue(dodn in str(user_cur["dn"]),
 121                             "User %s is deleted but it is not located under %s!" % (name_orig, dodn))
 122             self.assertEquals(name_cur, name_orig + "\nDEL:" + guid_str)
 123         else:
 124             self.assertTrue(not("isDeleted" in user_cur))
 125             self.assertEquals(name_cur, name_orig)
 126             self.assertEquals(user_orig["dn"], user_cur["dn"])
 127             self.assertTrue(dodn not in str(user_cur["dn"]))
 128
 129     def _net_drs_replicate(self, DC, fromDC):
 130         # find out where is net command
 131         samba_tool_cmd = os.path.abspath("./bin/samba-tool")
 132         # make command line credentials string
 133         creds = samba.tests.cmdline_credentials
 134         cmd_line_auth = "-U%s/%s%%%s" % (creds.get_domain(),
 135                                          creds.get_username(), creds.get_password())
 136         # bin/samba-tool drs replicate <Dest_DC_NAME> <Src_DC_NAME> <Naming Context>
 137         cmd_line = "%s drs replicate %s %s %s %s" % (samba_tool_cmd, DC, fromDC,
 138                                                      self.domain_dn, cmd_line_auth)
 139         ret = os.system(cmd_line)
 140         self.assertEquals(ret, 0, "Replicating %s from %s has failed!" % (DC, fromDC))
 141
 142     def test_NetReplicateCmd(self):
 143         """Triggers replication from DC1 to DC2
 144            and vice versa so both DCs are synchronized
 145            before test_ReplicateDeteleteObject test"""
 146         # replicate Domain NC on DC2 from DC1
 147         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1)
 148         # replicate Domain NC on DC1 from DC2
 149         self._net_drs_replicate(DC=self.dnsname_dc1, fromDC=self.dnsname_dc2)
 150
 151     def test_ReplicateDeteleteObject(self):
 152         """Verifies how a deleted-object is replicated between two DCs.
 153            This test should verify that:
 154             - deleted-object is replicated properly
 155            TODO: We should verify that after replication,
 156                  object's state to conform to a deleted-object state
 157                  or tombstone -object, depending on DC's features
 158                  It will also be great if check replPropertyMetaData."""
 159         # work-out unique username to test with
 160         username = self._make_username()
 161
 162         # create user on DC1
 163         self.ldb_dc1.newuser(username=username, password="P@sswOrd!")
 164         ldb_res = self.ldb_dc1.search(base=self.domain_dn,
 165                                       scope=SCOPE_SUBTREE,
 166                                       expression="(samAccountName=%s)" % username)
 167         self.assertEquals(len(ldb_res), 1)
 168         user_orig = ldb_res[0]
 169         user_dn   = ldb_res[0]["dn"]
 170
 171         # check user info on DC1
 172         print "Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0]))
 173         self._check_user(sam_ldb=self.ldb_dc1, user_orig=user_orig, is_deleted=False)
 174
 175         # trigger replication from DC1 to DC2
 176         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1)
 177
 178         # delete user on DC1
 179         self.ldb_dc1.delete(user_dn)
 180         # check user info on DC1 - should be deleted
 181         self._check_user(sam_ldb=self.ldb_dc1, user_orig=user_orig, is_deleted=True)
 182         # check user info on DC2 - should be valid user
 183         try:
 184             self._check_user(sam_ldb=self.ldb_dc2, user_orig=user_orig, is_deleted=False)
 185         except self.failureException:
 186             print ("Checking for not isDeleted user on %s failed, "
 187                    "probably because a replication took place. "
 188                    "Ideally we should block automatic replications during this test, "
 189                    "but until then, just ignore the error" % self.dnsname_dc2)
 190
 191         # trigger replication from DC2 to DC1
 192         # to check if deleted object gets restored
 193         self._net_drs_replicate(DC=self.dnsname_dc1, fromDC=self.dnsname_dc2)
 194         # check user info on DC1 - should be deleted
 195         self._check_user(sam_ldb=self.ldb_dc1, user_orig=user_orig, is_deleted=True)
 196         # check user info on DC2 - should be valid user
 197         try:
 198             self._check_user(sam_ldb=self.ldb_dc2, user_orig=user_orig, is_deleted=False)
 199         except self.failureException:
 200             print ("Checking for not isDeleted user on %s failed, "
 201                    "probably because a replication took place. "
 202                    "Ideally we should block automatic replications during this test, "
 203                    "but until then, just ignore the error" % self.dnsname_dc2)
 204
 205         # trigger replication from DC1 to DC2
 206         # to check if deleted object is replicated
 207         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1)
 208         # check user info on DC1 - should be deleted
 209         self._check_user(sam_ldb=self.ldb_dc1, user_orig=user_orig, is_deleted=True)
 210         # check user info on DC2 - should be deleted
 211         self._check_user(sam_ldb=self.ldb_dc2, user_orig=user_orig, is_deleted=True)
 212