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 samba.auth import system_session
  36 from ldb import SCOPE_BASE, SCOPE_SUBTREE
  37 from samba.samdb import SamDB
  38
  39 import samba.tests
  40
  41
  42 class DrsDeleteObjectTestCase(samba.tests.TestCase):
  43
  44     # RootDSE msg for DC1
  45     info_dc1 = None
  46     ldb_dc1 = None
  47     # RootDSE msg for DC1
  48     info_dc2 = None
  49     ldb_dc2 = None
  50
  51     def setUp(self):
  52         super(DrsDeleteObjectTestCase, self).setUp()
  53
  54         # connect to DCs singleton
  55         if self.ldb_dc1 is None:
  56             DrsDeleteObjectTestCase.dc1 = get_env_var("DC1")
  57             DrsDeleteObjectTestCase.ldb_dc1 = connect_samdb(self.dc1)
  58         if self.ldb_dc2 is None:
  59             DrsDeleteObjectTestCase.dc2 = get_env_var("DC2")
  60             DrsDeleteObjectTestCase.ldb_dc2 = connect_samdb(self.dc2)
  61
  62         # fetch rootDSEs
  63         if self.info_dc1 is None:
  64             ldb = self.ldb_dc1
  65             res = ldb.search(base="", expression="", scope=SCOPE_BASE, attrs=["*"])
  66             self.assertEquals(len(res), 1)
  67             DrsDeleteObjectTestCase.info_dc1 = res[0]
  68         if self.info_dc2 is None:
  69             ldb = self.ldb_dc2
  70             res = ldb.search(base="", expression="", scope=SCOPE_BASE, attrs=["*"])
  71             self.assertEquals(len(res), 1)
  72             DrsDeleteObjectTestCase.info_dc2 = res[0]
  73
  74         # cache some of RootDSE props
  75         self.schema_dn = self.info_dc1["schemaNamingContext"][0]
  76         self.domain_dn = self.info_dc1["defaultNamingContext"][0]
  77         self.config_dn = self.info_dc1["configurationNamingContext"][0]
  78         self.forest_level = int(self.info_dc1["forestFunctionality"][0])
  79
  80         # we will need DCs DNS names for 'net drs' command
  81         self.dnsname_dc1 = self.info_dc1["dnsHostName"][0]
  82         self.dnsname_dc2 = self.info_dc2["dnsHostName"][0]
  83         pass
  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         pass
 129
 130     def _net_drs_replicate(self, DC, fromDC):
 131         # find out where is net command
 132         net_cmd = os.path.abspath("./bin/net")
 133         # make command line credentials string
 134         creds = samba.tests.cmdline_credentials
 135         cmd_line_auth = "-U%s/%s%%%s" % (creds.get_domain(),
 136                                          creds.get_username(), creds.get_password())
 137         # bin/net drs replicate <Dest_DC_NAME> <Src_DC_NAME> <Naming Context>
 138         cmd_line = "%s drs replicate %s %s %s %s" % (net_cmd, DC, fromDC,
 139                                                      self.domain_dn, cmd_line_auth)
 140         ret = os.system(cmd_line)
 141         self.assertEquals(ret, 0, "Replicating %s from %s has failed!" % (DC, fromDC))
 142         pass
 143
 144
 145     def test_NetReplicateCmd(self):
 146         """Triggers replication from DC1 to DC2
 147            and vice versa so both DCs are synchronized
 148            before test_ReplicateDeteleteObject test"""
 149         # replicate Domain NC on DC2 from DC1
 150         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1)
 151         # replicate Domain NC on DC1 from DC2
 152         self._net_drs_replicate(DC=self.dnsname_dc1, fromDC=self.dnsname_dc2)
 153         pass
 154
 155     def test_ReplicateDeteleteObject(self):
 156         """Verifies how a deleted-object is replicated between two DCs.
 157            This test should verify that:
 158             - deleted-object is replicated properly
 159            TODO: We should verify that after replication,
 160                  object's state to conform to a deleted-object state
 161                  or tombstone -object, depending on DC's features
 162                  It will also be great if check replPropertyMetaData."""
 163         # work-out unique username to test with
 164         username = self._make_username()
 165
 166         # create user on DC1
 167         self.ldb_dc1.newuser(username=username, password="P@sswOrd!")
 168         ldb_res = self.ldb_dc1.search(base=self.domain_dn,
 169                                       scope=SCOPE_SUBTREE,
 170                                       expression="(samAccountName=%s)" % username)
 171         self.assertEquals(len(ldb_res), 1)
 172         user_orig = ldb_res[0]
 173         user_dn   = ldb_res[0]["dn"]
 174
 175         # check user info on DC1
 176         print "Testing for %s with GUID %s" % (username, self._GUID_string(user_orig["objectGUID"][0]))
 177         self._check_user(sam_ldb=self.ldb_dc1, user_orig=user_orig, is_deleted=False)
 178
 179         # trigger replication from DC1 to DC2
 180         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1)
 181
 182         # delete user on DC1
 183         self.ldb_dc1.delete(user_dn)
 184         # check user info on DC1 - should be deleted
 185         self._check_user(sam_ldb=self.ldb_dc1, user_orig=user_orig, is_deleted=True)
 186         # check user info on DC2 - should be valid user
 187         self._check_user(sam_ldb=self.ldb_dc2, user_orig=user_orig, is_deleted=False)
 188
 189         # trigger replication from DC2 to DC1
 190         # to check if deleted object gets restored
 191         self._net_drs_replicate(DC=self.dnsname_dc1, fromDC=self.dnsname_dc2)
 192         # check user info on DC1 - should be deleted
 193         self._check_user(sam_ldb=self.ldb_dc1, user_orig=user_orig, is_deleted=True)
 194         # check user info on DC2 - should be valid user
 195         try:
 196             self._check_user(sam_ldb=self.ldb_dc2, user_orig=user_orig, is_deleted=False)
 197         except self.failureException:
 198             print ("Checking for not isDeleted user on %s failed, "
 199                    "probably because a replication took place. "
 200                    "Ideally we should block automatic replications during this test, "
 201                    "but until then, just ignore the error" % self.dnsname_dc2)
 202
 203         # trigger replication from DC1 to DC2
 204         # to check if deleted object is replicated
 205         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1)
 206         # check user info on DC1 - should be deleted
 207         self._check_user(sam_ldb=self.ldb_dc1, user_orig=user_orig, is_deleted=True)
 208         # check user info on DC2 - should be deleted
 209         self._check_user(sam_ldb=self.ldb_dc2, user_orig=user_orig, is_deleted=True)
 210         pass
 211
 212
 213
 214 ########################################################################################
 215 def get_env_var(var_name):
 216     if not var_name in os.environ.keys():
 217         raise AssertionError("Please supply %s in environment" % var_name)
 218     return os.environ[var_name]
 219
 220 def connect_samdb(samdb_url):
 221     ldb_options = []
 222     if not "://" in samdb_url:
 223         if os.path.isfile(samdb_url):
 224             samdb_url = "tdb://%s" % samdb_url
 225         else:
 226             samdb_url = "ldap://%s" % samdb_url
 227             # user 'paged_search' module when connecting remotely
 228             ldb_options = ["modules:paged_searches"]
 229
 230     return SamDB(url=samdb_url,
 231                  lp=samba.tests.env_loadparm(),
 232                  session_info=system_session(),
 233                  credentials=samba.tests.cmdline_credentials,
 234                  options=ldb_options)
 235