source4/dsdb/tests/python/dirsync.py

   1 #!/usr/bin/env python
   2 #
   3 # Unit tests for dirsync control
   4 # Copyright (C) Matthieu Patou <mat@matws.net> 2011
   5 #
   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 import optparse
  22 import sys
  23 sys.path.insert(0, "bin/python")
  24 import samba
  25 samba.ensure_external_module("testtools", "testtools")
  26 samba.ensure_external_module("subunit", "subunit/python")
  27
  28 import samba.getopt as options
  29 import base64
  30
  31 from ldb import LdbError, SCOPE_BASE
  32 from ldb import Message, MessageElement, Dn
  33 from ldb import FLAG_MOD_ADD, FLAG_MOD_DELETE
  34 from samba.dcerpc import security, misc, drsblobs
  35 from samba.ndr import ndr_unpack, ndr_pack
  36
  37 from samba.auth import system_session
  38 from samba import gensec, sd_utils
  39 from samba.samdb import SamDB
  40 from samba.credentials import Credentials
  41 import samba.tests
  42 from samba.tests import delete_force
  43 from subunit.run import SubunitTestRunner
  44 import unittest
  45
  46 parser = optparse.OptionParser("dirsync.py [options] <host>")
  47 sambaopts = options.SambaOptions(parser)
  48 parser.add_option_group(sambaopts)
  49 parser.add_option_group(options.VersionOptions(parser))
  50
  51 # use command line creds if available
  52 credopts = options.CredentialsOptions(parser)
  53 parser.add_option_group(credopts)
  54 opts, args = parser.parse_args()
  55
  56 if len(args) < 1:
  57     parser.print_usage()
  58     sys.exit(1)
  59
  60 host = args[0]
  61 if not "://" in host:
  62     ldaphost = "ldap://%s" % host
  63     ldapshost = "ldaps://%s" % host
  64 else:
  65     ldaphost = host
  66     start = host.rindex("://")
  67     host = host.lstrip(start+3)
  68
  69 lp = sambaopts.get_loadparm()
  70 creds = credopts.get_credentials(lp)
  71
  72 #
  73 # Tests start here
  74 #
  75
  76 class DirsyncBaseTests(samba.tests.TestCase):
  77
  78     def setUp(self):
  79         super(DirsyncBaseTests, self).setUp()
  80         self.ldb_admin = ldb
  81         self.base_dn = ldb.domain_dn()
  82         self.domain_sid = security.dom_sid(ldb.get_domain_sid())
  83         self.user_pass = "samba123@AAA"
  84         self.configuration_dn = self.ldb_admin.get_config_basedn().get_linearized()
  85         self.sd_utils = sd_utils.SDUtils(ldb)
  86         #used for anonymous login
  87         print "baseDN: %s" % self.base_dn
  88
  89     def get_user_dn(self, name):
  90         return "CN=%s,CN=Users,%s" % (name, self.base_dn)
  91
  92     def get_ldb_connection(self, target_username, target_password):
  93         creds_tmp = Credentials()
  94         creds_tmp.set_username(target_username)
  95         creds_tmp.set_password(target_password)
  96         creds_tmp.set_domain(creds.get_domain())
  97         creds_tmp.set_realm(creds.get_realm())
  98         creds_tmp.set_workstation(creds.get_workstation())
  99         creds_tmp.set_gensec_features(creds_tmp.get_gensec_features()
 100                                       | gensec.FEATURE_SEAL)
 101         ldb_target = SamDB(url=ldaphost, credentials=creds_tmp, lp=lp)
 102         return ldb_target
 103
 104
 105 #tests on ldap add operations
 106 class SimpleDirsyncTests(DirsyncBaseTests):
 107
 108     def setUp(self):
 109         super(SimpleDirsyncTests, self).setUp()
 110         # Regular user
 111         self.dirsync_user = "test_dirsync_user"
 112         self.simple_user = "test_simple_user"
 113         self.admin_user = "test_admin_user"
 114         self.ouname = None
 115
 116         self.ldb_admin.newuser(self.dirsync_user, self.user_pass)
 117         self.ldb_admin.newuser(self.simple_user, self.user_pass)
 118         self.ldb_admin.newuser(self.admin_user, self.user_pass)
 119         self.desc_sddl = self.sd_utils.get_sd_as_sddl(self.base_dn)
 120
 121         user_sid = self.sd_utils.get_object_sid(self.get_user_dn(self.dirsync_user))
 122         mod = "(A;;CR;1131f6aa-9c07-11d1-f79f-00c04fc2dcd2;;%s)" % str(user_sid)
 123         self.sd_utils.dacl_add_ace(self.base_dn, mod)
 124
 125         # add admins to the Domain Admins group
 126         self.ldb_admin.add_remove_group_members("Domain Admins", self.admin_user,
 127                        add_members_operation=True)
 128
 129     def tearDown(self):
 130         super(SimpleDirsyncTests, self).tearDown()
 131         delete_force(self.ldb_admin, self.get_user_dn(self.dirsync_user))
 132         delete_force(self.ldb_admin, self.get_user_dn(self.simple_user))
 133         delete_force(self.ldb_admin, self.get_user_dn(self.admin_user))
 134         if self.ouname:
 135             delete_force(self.ldb_admin, self.ouname)
 136         self.sd_utils.modify_sd_on_dn(self.base_dn, self.desc_sddl)
 137         try:
 138             self.ldb_admin.deletegroup("testgroup")
 139         except Exception:
 140             pass
 141
 142     #def test_dirsync_errors(self):
 143
 144
 145     def test_dirsync_supported(self):
 146         """Test the basic of the dirsync is supported"""
 147         self.ldb_dirsync = self.get_ldb_connection(self.dirsync_user, self.user_pass)
 148         self.ldb_simple = self.get_ldb_connection(self.simple_user, self.user_pass)
 149         res = self.ldb_admin.search(self.base_dn, expression="samaccountname=*", controls=["dirsync:1:0:1"])
 150         res = self.ldb_dirsync.search(self.base_dn, expression="samaccountname=*", controls=["dirsync:1:0:1"])
 151         try:
 152             self.ldb_simple.search(self.base_dn,
 153                 expression="samaccountname=*",
 154                 controls=["dirsync:1:0:1"])
 155         except LdbError,l:
 156            self.assertTrue(str(l).find("LDAP_INSUFFICIENT_ACCESS_RIGHTS") != -1)
 157
 158     def test_parentGUID_referrals(self):
 159         res2 = self.ldb_admin.search(self.base_dn, scope=SCOPE_BASE, attrs=["objectGUID"])
 160
 161         res = self.ldb_admin.search(self.base_dn,
 162                                         expression="name=Configuration",
 163                                         controls=["dirsync:1:0:1"])
 164         self.assertEqual(res2[0].get("objectGUID"), res[0].get("parentGUID"))
 165
 166     def test_ok_not_rootdc(self):
 167         """Test if it's ok to do dirsync on another NC that is not the root DC"""
 168         self.ldb_admin.search(self.ldb_admin.get_config_basedn(),
 169                                     expression="samaccountname=*",
 170                                     controls=["dirsync:1:0:1"])
 171
 172     def test_dirsync_errors(self):
 173         """Test if dirsync returns the correct LDAP errors in case of pb"""
 174         self.ldb_simple = self.get_ldb_connection(self.simple_user, self.user_pass)
 175         self.ldb_dirsync = self.get_ldb_connection(self.dirsync_user, self.user_pass)
 176         try:
 177             self.ldb_simple.search(self.base_dn,
 178                 expression="samaccountname=*",
 179                 controls=["dirsync:1:0:1"])
 180         except LdbError,l:
 181             print l
 182             self.assertTrue(str(l).find("LDAP_INSUFFICIENT_ACCESS_RIGHTS") != -1)
 183
 184         try:
 185             self.ldb_simple.search("CN=Users,%s" % self.base_dn,
 186                 expression="samaccountname=*",
 187                 controls=["dirsync:1:0:1"])
 188         except LdbError,l:
 189             print l
 190             self.assertTrue(str(l).find("LDAP_INSUFFICIENT_ACCESS_RIGHTS") != -1)
 191
 192         try:
 193             self.ldb_simple.search("CN=Users,%s" % self.base_dn,
 194                 expression="samaccountname=*",
 195                 controls=["dirsync:1:1:1"])
 196         except LdbError,l:
 197             print l
 198             self.assertTrue(str(l).find("LDAP_UNWILLING_TO_PERFORM") != -1)
 199
 200         try:
 201             self.ldb_dirsync.search("CN=Users,%s" % self.base_dn,
 202                 expression="samaccountname=*",
 203                 controls=["dirsync:1:0:1"])
 204         except LdbError,l:
 205             print l
 206             self.assertTrue(str(l).find("LDAP_INSUFFICIENT_ACCESS_RIGHTS") != -1)
 207
 208         try:
 209             self.ldb_admin.search("CN=Users,%s" % self.base_dn,
 210                 expression="samaccountname=*",
 211                 controls=["dirsync:1:0:1"])
 212         except LdbError,l:
 213             print l
 214             self.assertTrue(str(l).find("LDAP_INSUFFICIENT_ACCESS_RIGHTS") != -1)
 215
 216         try:
 217             self.ldb_admin.search("CN=Users,%s" % self.base_dn,
 218                 expression="samaccountname=*",
 219                 controls=["dirsync:1:1:1"])
 220         except LdbError,l:
 221             print l
 222             self.assertTrue(str(l).find("LDAP_UNWILLING_TO_PERFORM") != -1)
 223
 224
 225
 226
 227     def test_dirsync_attributes(self):
 228         """Check behavior with some attributes """
 229         res = self.ldb_admin.search(self.base_dn,
 230                                     expression="samaccountname=*",
 231                                     controls=["dirsync:1:0:1"])
 232         # Check that nTSecurityDescriptor is returned as it's the case when doing dirsync
 233         self.assertTrue(res.msgs[0].get("ntsecuritydescriptor") != None)
 234         # Check that non replicated attributes are not returned
 235         self.assertTrue(res.msgs[0].get("badPwdCount") == None)
 236         # Check that non forward link are not returned
 237         self.assertTrue(res.msgs[0].get("memberof") == None)
 238
 239         # Asking for instanceType will return also objectGUID
 240         res = self.ldb_admin.search(self.base_dn,
 241                                     expression="samaccountname=Administrator",
 242                                     attrs=["instanceType"],
 243                                     controls=["dirsync:1:0:1"])
 244         self.assertTrue(res.msgs[0].get("objectGUID") != None)
 245         self.assertTrue(res.msgs[0].get("instanceType") != None)
 246
 247         # We don't return an entry if asked for objectGUID
 248         res = self.ldb_admin.search(self.base_dn,
 249                                     expression="dn=%s" % self.base_dn,
 250                                     attrs=["objectGUID"],
 251                                     controls=["dirsync:1:0:1"])
 252         self.assertEquals(len(res.msgs), 0)
 253
 254         # a request on the root of a NC didn't return parentGUID
 255         res = self.ldb_admin.search(self.base_dn,
 256                                     expression="dn=%s" % self.base_dn,
 257                                     attrs=["name"],
 258                                     controls=["dirsync:1:0:1"])
 259         self.assertTrue(res.msgs[0].get("objectGUID") != None)
 260         self.assertTrue(res.msgs[0].get("name") != None)
 261         self.assertTrue(res.msgs[0].get("parentGUID") == None)
 262         self.assertTrue(res.msgs[0].get("instanceType") != None)
 263
 264          # Asking for name will return also objectGUID and parentGUID
 265         # and instanceType and of course name
 266         res = self.ldb_admin.search(self.base_dn,
 267                                     expression="samaccountname=Administrator",
 268                                     attrs=["name"],
 269                                     controls=["dirsync:1:0:1"])
 270         self.assertTrue(res.msgs[0].get("objectGUID") != None)
 271         self.assertTrue(res.msgs[0].get("name") != None)
 272         self.assertTrue(res.msgs[0].get("parentGUID") != None)
 273         self.assertTrue(res.msgs[0].get("instanceType") != None)
 274
 275         # Asking for dn will not return not only DN but more like if attrs=*
 276         # parentGUID should be returned
 277         res = self.ldb_admin.search(self.base_dn,
 278                                     expression="samaccountname=Administrator",
 279                                     attrs=["dn"],
 280                                     controls=["dirsync:1:0:1"])
 281         count = len(res.msgs[0])
 282         res2 = self.ldb_admin.search(self.base_dn,
 283                                     expression="samaccountname=Administrator",
 284                                     controls=["dirsync:1:0:1"])
 285         count2 = len(res2.msgs[0])
 286         self.assertEqual(count, count2)
 287
 288         # Asking for cn will return nothing on objects that have CN as RDN
 289         res = self.ldb_admin.search(self.base_dn,
 290                                     expression="samaccountname=Administrator",
 291                                     attrs=["cn"],
 292                                     controls=["dirsync:1:0:1"])
 293         self.assertEqual(len(res.msgs), 0)
 294         # Asking for parentGUID will return nothing too
 295         res = self.ldb_admin.search(self.base_dn,
 296                                     expression="samaccountname=Administrator",
 297                                     attrs=["parentGUID"],
 298                                     controls=["dirsync:1:0:1"])
 299         self.assertEqual(len(res.msgs), 0)
 300         ouname="OU=testou,%s" % self.base_dn
 301         self.ouname = ouname
 302         self.ldb_admin.create_ou(ouname)
 303         delta = Message()
 304         delta.dn = Dn(self.ldb_admin, str(ouname))
 305         delta["cn"] = MessageElement("test ou",
 306                                         FLAG_MOD_ADD,
 307                                         "cn" )
 308         self.ldb_admin.modify(delta)
 309         res = self.ldb_admin.search(self.base_dn,
 310                                     expression="name=testou",
 311                                     attrs=["cn"],
 312                                     controls=["dirsync:1:0:1"])
 313
 314         self.assertEqual(len(res.msgs), 1)
 315         self.assertEqual(len(res.msgs[0]), 3)
 316         delete_force(self.ldb_admin, ouname)
 317
 318     def test_dirsync_with_controls(self):
 319         """Check that dirsync return correct informations when dealing with the NC"""
 320         res = self.ldb_admin.search(self.base_dn,
 321                                     expression="(dn=%s)" % str(self.base_dn),
 322                                     attrs=["name"],
 323                                     controls=["dirsync:1:0:10000", "extended_dn:1", "show_deleted:1"])
 324
 325     def test_dirsync_basenc(self):
 326         """Check that dirsync return correct informations when dealing with the NC"""
 327         res = self.ldb_admin.search(self.base_dn,
 328                                     expression="(dn=%s)" % str(self.base_dn),
 329                                     attrs=["name"],
 330                                     controls=["dirsync:1:0:10000"])
 331         self.assertEqual(len(res.msgs), 1)
 332         self.assertEqual(len(res.msgs[0]), 3)
 333
 334         res = self.ldb_admin.search(self.base_dn,
 335                                     expression="(dn=%s)" % str(self.base_dn),
 336                                     attrs=["ntSecurityDescriptor"],
 337                                     controls=["dirsync:1:0:10000"])
 338         self.assertEqual(len(res.msgs), 1)
 339         self.assertEqual(len(res.msgs[0]), 3)
 340
 341     def test_dirsync_othernc(self):
 342         """Check that dirsync return information for entries that are normaly referrals (ie. other NCs)"""
 343         res = self.ldb_admin.search(self.base_dn,
 344                                     expression="(objectclass=configuration)",
 345                                     attrs=["name"],
 346                                     controls=["dirsync:1:0:10000"])
 347         self.assertEqual(len(res.msgs), 1)
 348         self.assertEqual(len(res.msgs[0]), 4)
 349
 350         res = self.ldb_admin.search(self.base_dn,
 351                                     expression="(objectclass=configuration)",
 352                                     attrs=["ntSecurityDescriptor"],
 353                                     controls=["dirsync:1:0:10000"])
 354         self.assertEqual(len(res.msgs), 1)
 355         self.assertEqual(len(res.msgs[0]), 3)
 356
 357         res = self.ldb_admin.search(self.base_dn,
 358                                     expression="(objectclass=domaindns)",
 359                                     attrs=["ntSecurityDescriptor"],
 360                                     controls=["dirsync:1:0:10000"])
 361         nb = len(res.msgs)
 362
 363         # only sub nc returns a result when asked for objectGUID
 364         res = self.ldb_admin.search(self.base_dn,
 365                                     expression="(objectclass=domaindns)",
 366                                     attrs=["objectGUID"],
 367                                     controls=["dirsync:1:0:0"])
 368         self.assertEqual(len(res.msgs), nb - 1)
 369         if nb > 1:
 370             self.assertTrue(res.msgs[0].get("objectGUID") != None)
 371         else:
 372             res = self.ldb_admin.search(self.base_dn,
 373                                         expression="(objectclass=configuration)",
 374                                         attrs=["objectGUID"],
 375                                         controls=["dirsync:1:0:0"])
 376
 377
 378     def test_dirsync_send_delta(self):
 379         """Check that dirsync return correct delta when sending the last cookie"""
 380         res = self.ldb_admin.search(self.base_dn,
 381                                     expression="(&(samaccountname=test*)(!(isDeleted=*)))",
 382                                     controls=["dirsync:1:0:10000"])
 383         ctl = str(res.controls[0]).split(":")
 384         ctl[1] = "1"
 385         ctl[2] = "0"
 386         ctl[3] = "10000"
 387         control = str(":".join(ctl))
 388         res = self.ldb_admin.search(self.base_dn,
 389                                     expression="(&(samaccountname=test*)(!(isDeleted=*)))",
 390                                     controls=[control])
 391         self.assertEqual(len(res), 0)
 392
 393         res = self.ldb_admin.search(self.base_dn,
 394                                     expression="(&(objectClass=organizationalUnit)(!(isDeleted=*)))",
 395                                     controls=["dirsync:1:0:100000"])
 396
 397         ctl = str(res.controls[0]).split(":")
 398         ctl[1] = "1"
 399         ctl[2] = "0"
 400         ctl[3] = "10000"
 401         control2 = str(":".join(ctl))
 402
 403         # Let's create an OU
 404         ouname="OU=testou2,%s" % self.base_dn
 405         self.ouname = ouname
 406         self.ldb_admin.create_ou(ouname)
 407         res = self.ldb_admin.search(self.base_dn,
 408                                     expression="(&(objectClass=organizationalUnit)(!(isDeleted=*)))",
 409                                     controls=[control2])
 410         self.assertEqual(len(res), 1)
 411         ctl = str(res.controls[0]).split(":")
 412         ctl[1] = "1"
 413         ctl[2] = "0"
 414         ctl[3] = "10000"
 415         control3 = str(":".join(ctl))
 416
 417         delta = Message()
 418         delta.dn = Dn(self.ldb_admin, str(ouname))
 419
 420         delta["cn"] = MessageElement("test ou",
 421                                         FLAG_MOD_ADD,
 422                                         "cn" )
 423         self.ldb_admin.modify(delta)
 424         res = self.ldb_admin.search(self.base_dn,
 425                                     expression="(&(objectClass=organizationalUnit)(!(isDeleted=*)))",
 426                                     controls=[control3])
 427
 428         self.assertEqual(len(res.msgs), 1)
 429         # 3 attributes: instanceType, cn and objectGUID
 430         self.assertEqual(len(res.msgs[0]), 3)
 431
 432         delta = Message()
 433         delta.dn = Dn(self.ldb_admin, str(ouname))
 434         delta["cn"] = MessageElement([],
 435                                         FLAG_MOD_DELETE,
 436                                         "cn" )
 437         self.ldb_admin.modify(delta)
 438         res = self.ldb_admin.search(self.base_dn,
 439                                     expression="(&(objectClass=organizationalUnit)(!(isDeleted=*)))",
 440                                     controls=[control3])
 441
 442         self.assertEqual(len(res.msgs), 1)
 443         # So we won't have much attribute returned but instanceType and GUID
 444         # are.
 445         # 3 attributes: instanceType and objectGUID and cn but empty
 446         self.assertEqual(len(res.msgs[0]), 3)
 447         ouname = "OU=newouname,%s" % self.base_dn
 448         self.ldb_admin.rename(str(res[0].dn), str(Dn(self.ldb_admin, ouname)))
 449         self.ouname = ouname
 450         ctl = str(res.controls[0]).split(":")
 451         ctl[1] = "1"
 452         ctl[2] = "0"
 453         ctl[3] = "10000"
 454         control4 = str(":".join(ctl))
 455         res = self.ldb_admin.search(self.base_dn,
 456                                     expression="(&(objectClass=organizationalUnit)(!(isDeleted=*)))",
 457                                     controls=[control3])
 458
 459         self.assertTrue(res[0].get("parentGUID") != None)
 460         self.assertTrue(res[0].get("name") != None)
 461         delete_force(self.ldb_admin, ouname)
 462
 463     def test_dirsync_linkedattributes(self):
 464         """Check that dirsync returnd deleted objects too"""
 465         # Let's search for members
 466         self.ldb_simple = self.get_ldb_connection(self.simple_user, self.user_pass)
 467         res = self.ldb_simple.search(self.base_dn,
 468                                     expression="(name=Administrators)",
 469                                     controls=["dirsync:1:1:1"])
 470
 471         self.assertTrue(len(res[0].get("member")) > 0)
 472         size = len(res[0].get("member"))
 473
 474         ctl = str(res.controls[0]).split(":")
 475         ctl[1] = "1"
 476         ctl[2] = "1"
 477         ctl[3] = "10000"
 478         control1 = str(":".join(ctl))
 479         self.ldb_admin.add_remove_group_members("Administrators", self.simple_user,
 480                        add_members_operation=True)
 481
 482         res = self.ldb_simple.search(self.base_dn,
 483                                     expression="(name=Administrators)",
 484                                     controls=[control1])
 485
 486         self.assertEqual(len(res[0].get("member")), size + 1)
 487         ctl = str(res.controls[0]).split(":")
 488         ctl[1] = "1"
 489         ctl[2] = "1"
 490         ctl[3] = "10000"
 491         control1 = str(":".join(ctl))
 492
 493         # remove the user from the group
 494         self.ldb_admin.add_remove_group_members("Administrators", self.simple_user,
 495                        add_members_operation=False)
 496
 497         res = self.ldb_simple.search(self.base_dn,
 498                                     expression="(name=Administrators)",
 499                                     controls=[control1])
 500
 501         self.assertEqual(len(res[0].get("member")), size )
 502
 503         self.ldb_admin.newgroup("testgroup")
 504         self.ldb_admin.add_remove_group_members("testgroup", self.simple_user,
 505                        add_members_operation=True)
 506
 507         res = self.ldb_admin.search(self.base_dn,
 508                                     expression="(name=testgroup)",
 509                                     controls=["dirsync:1:0:1"])
 510
 511         self.assertEqual(len(res[0].get("member")), 1)
 512         self.assertTrue(res[0].get("member") != "" )
 513
 514         ctl = str(res.controls[0]).split(":")
 515         ctl[1] = "1"
 516         ctl[2] = "0"
 517         ctl[3] = "1"
 518         control1 = str(":".join(ctl))
 519
 520         # Check that reasking the same question but with an updated cookie
 521         # didn't return any results.
 522         print control1
 523         res = self.ldb_admin.search(self.base_dn,
 524                                     expression="(name=testgroup)",
 525                                     controls=[control1])
 526         self.assertEqual(len(res), 0)
 527
 528         ctl = str(res.controls[0]).split(":")
 529         ctl[1] = "1"
 530         ctl[2] = "1"
 531         ctl[3] = "10000"
 532         control1 = str(":".join(ctl))
 533
 534         self.ldb_admin.add_remove_group_members("testgroup", self.simple_user,
 535                        add_members_operation=False)
 536
 537         res = self.ldb_admin.search(self.base_dn,
 538                                     expression="(name=testgroup)",
 539                                     attrs=["member"],
 540                                     controls=[control1])
 541
 542         self.ldb_admin.deletegroup("testgroup")
 543         self.assertEqual(len(res[0].get("member")), 0)
 544
 545
 546
 547     def test_dirsync_deleted_items(self):
 548         """Check that dirsync returnd deleted objects too"""
 549         # Let's create an OU
 550         ouname="OU=testou3,%s" % self.base_dn
 551         self.ouname = ouname
 552         self.ldb_admin.create_ou(ouname)
 553         res = self.ldb_admin.search(self.base_dn,
 554                                     expression="(&(objectClass=organizationalUnit)(!(isDeleted=*)))",
 555                                     controls=["dirsync:1:0:1"])
 556         guid = None
 557         for e in res:
 558             if str(e["name"]) == "testou3":
 559                 guid = str(ndr_unpack(misc.GUID,e.get("objectGUID")[0]))
 560
 561         ctl = str(res.controls[0]).split(":")
 562         ctl[1] = "1"
 563         ctl[2] = "0"
 564         ctl[3] = "10000"
 565         control1 = str(":".join(ctl))
 566
 567         # So now delete the object and check that
 568         # we can see the object but deleted when admin
 569         delete_force(self.ldb_admin, ouname)
 570
 571         res = self.ldb_admin.search(self.base_dn,
 572                                     expression="(objectClass=organizationalUnit)",
 573                                     controls=[control1])
 574         self.assertEqual(len(res), 1)
 575         guid2 = str(ndr_unpack(misc.GUID,res[0].get("objectGUID")[0]))
 576         self.assertEqual(guid2, guid)
 577         self.assertTrue(res[0].get("isDeleted"))
 578         self.assertTrue(res[0].get("name") != None)
 579
 580     def test_cookie_from_others(self):
 581         res = self.ldb_admin.search(self.base_dn,
 582                                     expression="(&(objectClass=organizationalUnit)(!(isDeleted=*)))",
 583                                     controls=["dirsync:1:0:1"])
 584         ctl = str(res.controls[0]).split(":")
 585         cookie = ndr_unpack(drsblobs.ldapControlDirSyncCookie, base64.b64decode(str(ctl[4])))
 586         cookie.blob.guid1 = misc.GUID("128a99bf-abcd-1234-abcd-1fb625e530db")
 587         controls=["dirsync:1:0:0:%s" % base64.b64encode(ndr_pack(cookie))]
 588         res = self.ldb_admin.search(self.base_dn,
 589                                     expression="(&(objectClass=organizationalUnit)(!(isDeleted=*)))",
 590                                     controls=controls)
 591
 592 class ExtendedDirsyncTests(SimpleDirsyncTests):
 593     def test_dirsync_linkedattributes(self):
 594         flag_incr_linked = 2147483648
 595         self.ldb_simple = self.get_ldb_connection(self.simple_user, self.user_pass)
 596         res = self.ldb_admin.search(self.base_dn,
 597                                     attrs=["member"],
 598                                     expression="(name=Administrators)",
 599                                     controls=["dirsync:1:%d:1" % flag_incr_linked])
 600
 601         self.assertTrue(res[0].get("member;range=1-1") != None )
 602         self.assertTrue(len(res[0].get("member;range=1-1")) > 0)
 603         size = len(res[0].get("member;range=1-1"))
 604
 605         ctl = str(res.controls[0]).split(":")
 606         ctl[1] = "1"
 607         ctl[2] = "%d" % flag_incr_linked
 608         ctl[3] = "10000"
 609         control1 = str(":".join(ctl))
 610         self.ldb_admin.add_remove_group_members("Administrators", self.simple_user,
 611                        add_members_operation=True)
 612         self.ldb_admin.add_remove_group_members("Administrators", self.dirsync_user,
 613                        add_members_operation=True)
 614
 615
 616         res = self.ldb_admin.search(self.base_dn,
 617                                     expression="(name=Administrators)",
 618                                     controls=[control1])
 619
 620         self.assertEqual(len(res[0].get("member;range=1-1")), 2)
 621         ctl = str(res.controls[0]).split(":")
 622         ctl[1] = "1"
 623         ctl[2] = "%d" % flag_incr_linked
 624         ctl[3] = "10000"
 625         control1 = str(":".join(ctl))
 626
 627         # remove the user from the group
 628         self.ldb_admin.add_remove_group_members("Administrators", self.simple_user,
 629                        add_members_operation=False)
 630
 631         res = self.ldb_admin.search(self.base_dn,
 632                                     expression="(name=Administrators)",
 633                                     controls=[control1])
 634
 635         self.assertEqual(res[0].get("member;range=1-1"), None )
 636         self.assertEqual(len(res[0].get("member;range=0-0")), 1)
 637
 638         ctl = str(res.controls[0]).split(":")
 639         ctl[1] = "1"
 640         ctl[2] = "%d" % flag_incr_linked
 641         ctl[3] = "10000"
 642         control2 = str(":".join(ctl))
 643
 644         self.ldb_admin.add_remove_group_members("Administrators", self.dirsync_user,
 645                        add_members_operation=False)
 646
 647         res = self.ldb_admin.search(self.base_dn,
 648                                     expression="(name=Administrators)",
 649                                     controls=[control2])
 650
 651         self.assertEqual(res[0].get("member;range=1-1"), None )
 652         self.assertEqual(len(res[0].get("member;range=0-0")), 1)
 653
 654         res = self.ldb_admin.search(self.base_dn,
 655                                     expression="(name=Administrators)",
 656                                     controls=[control1])
 657
 658         self.assertEqual(res[0].get("member;range=1-1"), None )
 659         self.assertEqual(len(res[0].get("member;range=0-0")), 2)
 660
 661     def test_dirsync_deleted_items(self):
 662         """Check that dirsync returnd deleted objects too"""
 663         # Let's create an OU
 664         self.ldb_simple = self.get_ldb_connection(self.simple_user, self.user_pass)
 665         ouname="OU=testou3,%s" % self.base_dn
 666         self.ouname = ouname
 667         self.ldb_admin.create_ou(ouname)
 668
 669         # Specify LDAP_DIRSYNC_OBJECT_SECURITY
 670         res = self.ldb_simple.search(self.base_dn,
 671                                     expression="(&(objectClass=organizationalUnit)(!(isDeleted=*)))",
 672                                     controls=["dirsync:1:1:1"])
 673
 674         guid = None
 675         for e in res:
 676             if str(e["name"]) == "testou3":
 677                 guid = str(ndr_unpack(misc.GUID,e.get("objectGUID")[0]))
 678
 679         self.assertTrue(guid != None)
 680         ctl = str(res.controls[0]).split(":")
 681         ctl[1] = "1"
 682         ctl[2] = "1"
 683         ctl[3] = "10000"
 684         control1 = str(":".join(ctl))
 685
 686         # So now delete the object and check that
 687         # we can see the object but deleted when admin
 688         # we just see the objectGUID when simple user
 689         delete_force(self.ldb_admin, ouname)
 690
 691         res = self.ldb_simple.search(self.base_dn,
 692                                     expression="(objectClass=organizationalUnit)",
 693                                     controls=[control1])
 694         self.assertEqual(len(res), 1)
 695         guid2 = str(ndr_unpack(misc.GUID,res[0].get("objectGUID")[0]))
 696         self.assertEqual(guid2, guid)
 697         self.assertEqual(str(res[0].dn), "")
 698
 699
 700 ldb = SamDB(ldapshost, credentials=creds, session_info=system_session(lp), lp=lp)
 701
 702 runner = SubunitTestRunner()
 703 rc = 0
 704 #
 705 if not runner.run(unittest.makeSuite(SimpleDirsyncTests)).wasSuccessful():
 706     rc = 1
 707 if not runner.run(unittest.makeSuite(ExtendedDirsyncTests)).wasSuccessful():
 708     rc = 1
 709
 710 sys.exit(rc)