PEP8: fix E202: whitespace before ')'
[samba.git] / 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 # Copyright (C) Jelmer Vernooij <jelmer@samba.org> 2014
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 from __future__ import print_function
22 import optparse
23 import sys
24 sys.path.insert(0, "bin/python")
25 import samba
26 from samba.tests.subunitrun import TestProgram, SubunitOptions
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, security
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, DONT_USE_KERBEROS
41 import samba.tests
42 from samba.tests import delete_force
43
44 parser = optparse.OptionParser("dirsync.py [options] <host>")
45 sambaopts = options.SambaOptions(parser)
46 parser.add_option_group(sambaopts)
47 parser.add_option_group(options.VersionOptions(parser))
48
49 # use command line creds if available
50 credopts = options.CredentialsOptions(parser)
51 parser.add_option_group(credopts)
52 subunitopts = SubunitOptions(parser)
53 parser.add_option_group(subunitopts)
54 opts, args = parser.parse_args()
55
56 if len(args) < 1:
57     parser.print_usage()
58     sys.exit(1)
59
60 host = args.pop()
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 = SamDB(ldapshost, credentials=creds, session_info=system_session(lp), lp=lp)
81         self.base_dn = self.ldb_admin.domain_dn()
82         self.domain_sid = security.dom_sid(self.ldb_admin.get_domain_sid())
83         self.user_pass = samba.generate_random_password(12, 16)
84         self.configuration_dn = self.ldb_admin.get_config_basedn().get_linearized()
85         self.sd_utils = sd_utils.SDUtils(self.ldb_admin)
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         creds_tmp.set_kerberos_state(DONT_USE_KERBEROS) # kinit is too expensive to use in a tight loop
102         ldb_target = SamDB(url=ldaphost, credentials=creds_tmp, lp=lp)
103         return ldb_target
104
105
106 #tests on ldap add operations
107 class SimpleDirsyncTests(DirsyncBaseTests):
108
109     def setUp(self):
110         super(SimpleDirsyncTests, self).setUp()
111         # Regular user
112         self.dirsync_user = "test_dirsync_user"
113         self.simple_user = "test_simple_user"
114         self.admin_user = "test_admin_user"
115         self.ouname = None
116
117         self.ldb_admin.newuser(self.dirsync_user, self.user_pass)
118         self.ldb_admin.newuser(self.simple_user, self.user_pass)
119         self.ldb_admin.newuser(self.admin_user, self.user_pass)
120         self.desc_sddl = self.sd_utils.get_sd_as_sddl(self.base_dn)
121
122         user_sid = self.sd_utils.get_object_sid(self.get_user_dn(self.dirsync_user))
123         mod = "(OA;;CR;%s;;%s)" % (security.GUID_DRS_GET_CHANGES,
124                                    str(user_sid))
125         self.sd_utils.dacl_add_ace(self.base_dn, mod)
126
127         # add admins to the Domain Admins group
128         self.ldb_admin.add_remove_group_members("Domain Admins", [self.admin_user],
129                                                 add_members_operation=True)
130
131     def tearDown(self):
132         super(SimpleDirsyncTests, self).tearDown()
133         delete_force(self.ldb_admin, self.get_user_dn(self.dirsync_user))
134         delete_force(self.ldb_admin, self.get_user_dn(self.simple_user))
135         delete_force(self.ldb_admin, self.get_user_dn(self.admin_user))
136         if self.ouname:
137             delete_force(self.ldb_admin, self.ouname)
138         self.sd_utils.modify_sd_on_dn(self.base_dn, self.desc_sddl)
139         try:
140             self.ldb_admin.deletegroup("testgroup")
141         except Exception:
142             pass
143
144     #def test_dirsync_errors(self):
145
146     def test_dirsync_supported(self):
147         """Test the basic of the dirsync is supported"""
148         self.ldb_dirsync = self.get_ldb_connection(self.dirsync_user, self.user_pass)
149         self.ldb_simple = self.get_ldb_connection(self.simple_user, self.user_pass)
150         res = self.ldb_admin.search(self.base_dn, expression="samaccountname=*", controls=["dirsync:1:0:1"])
151         res = self.ldb_dirsync.search(self.base_dn, expression="samaccountname=*", controls=["dirsync:1:0:1"])
152         try:
153             self.ldb_simple.search(self.base_dn,
154                                    expression="samaccountname=*",
155                                    controls=["dirsync:1:0:1"])
156         except LdbError as l:
157             self.assertTrue(str(l).find("LDAP_INSUFFICIENT_ACCESS_RIGHTS") != -1)
158
159     def test_parentGUID_referrals(self):
160         res2 = self.ldb_admin.search(self.base_dn, scope=SCOPE_BASE, attrs=["objectGUID"])
161
162         res = self.ldb_admin.search(self.base_dn,
163                                     expression="name=Configuration",
164                                     controls=["dirsync:1:0:1"])
165         self.assertEqual(res2[0].get("objectGUID"), res[0].get("parentGUID"))
166
167     def test_ok_not_rootdc(self):
168         """Test if it's ok to do dirsync on another NC that is not the root DC"""
169         self.ldb_admin.search(self.ldb_admin.get_config_basedn(),
170                               expression="samaccountname=*",
171                               controls=["dirsync:1:0:1"])
172
173     def test_dirsync_errors(self):
174         """Test if dirsync returns the correct LDAP errors in case of pb"""
175         self.ldb_simple = self.get_ldb_connection(self.simple_user, self.user_pass)
176         self.ldb_dirsync = self.get_ldb_connection(self.dirsync_user, self.user_pass)
177         try:
178             self.ldb_simple.search(self.base_dn,
179                                    expression="samaccountname=*",
180                                    controls=["dirsync:1:0:1"])
181         except LdbError as l:
182             print(l)
183             self.assertTrue(str(l).find("LDAP_INSUFFICIENT_ACCESS_RIGHTS") != -1)
184
185         try:
186             self.ldb_simple.search("CN=Users,%s" % self.base_dn,
187                                    expression="samaccountname=*",
188                                    controls=["dirsync:1:0:1"])
189         except LdbError as l:
190             print(l)
191             self.assertTrue(str(l).find("LDAP_INSUFFICIENT_ACCESS_RIGHTS") != -1)
192
193         try:
194             self.ldb_simple.search("CN=Users,%s" % self.base_dn,
195                                    expression="samaccountname=*",
196                                    controls=["dirsync:1:1:1"])
197         except LdbError as l:
198             print(l)
199             self.assertTrue(str(l).find("LDAP_UNWILLING_TO_PERFORM") != -1)
200
201         try:
202             self.ldb_dirsync.search("CN=Users,%s" % self.base_dn,
203                                     expression="samaccountname=*",
204                                     controls=["dirsync:1:0:1"])
205         except LdbError as l:
206             print(l)
207             self.assertTrue(str(l).find("LDAP_INSUFFICIENT_ACCESS_RIGHTS") != -1)
208
209         try:
210             self.ldb_admin.search("CN=Users,%s" % self.base_dn,
211                                   expression="samaccountname=*",
212                                   controls=["dirsync:1:0:1"])
213         except LdbError as l:
214             print(l)
215             self.assertTrue(str(l).find("LDAP_INSUFFICIENT_ACCESS_RIGHTS") != -1)
216
217         try:
218             self.ldb_admin.search("CN=Users,%s" % self.base_dn,
219                                   expression="samaccountname=*",
220                                   controls=["dirsync:1:1:1"])
221         except LdbError as l:
222             print(l)
223             self.assertTrue(str(l).find("LDAP_UNWILLING_TO_PERFORM") != -1)
224
225     def test_dirsync_attributes(self):
226         """Check behavior with some attributes """
227         res = self.ldb_admin.search(self.base_dn,
228                                     expression="samaccountname=*",
229                                     controls=["dirsync:1:0:1"])
230         # Check that nTSecurityDescriptor is returned as it's the case when doing dirsync
231         self.assertTrue(res.msgs[0].get("ntsecuritydescriptor") != None)
232         # Check that non replicated attributes are not returned
233         self.assertTrue(res.msgs[0].get("badPwdCount") == None)
234         # Check that non forward link are not returned
235         self.assertTrue(res.msgs[0].get("memberof") == None)
236
237         # Asking for instanceType will return also objectGUID
238         res = self.ldb_admin.search(self.base_dn,
239                                     expression="samaccountname=Administrator",
240                                     attrs=["instanceType"],
241                                     controls=["dirsync:1:0:1"])
242         self.assertTrue(res.msgs[0].get("objectGUID") != None)
243         self.assertTrue(res.msgs[0].get("instanceType") != None)
244
245         # We don't return an entry if asked for objectGUID
246         res = self.ldb_admin.search(self.base_dn,
247                                     expression="(distinguishedName=%s)" % str(self.base_dn),
248                                     attrs=["objectGUID"],
249                                     controls=["dirsync:1:0:1"])
250         self.assertEquals(len(res.msgs), 0)
251
252         # a request on the root of a NC didn't return parentGUID
253         res = self.ldb_admin.search(self.base_dn,
254                                     expression="(distinguishedName=%s)" % str(self.base_dn),
255                                     attrs=["name"],
256                                     controls=["dirsync:1:0:1"])
257         self.assertTrue(res.msgs[0].get("objectGUID") != None)
258         self.assertTrue(res.msgs[0].get("name") != None)
259         self.assertTrue(res.msgs[0].get("parentGUID") == None)
260         self.assertTrue(res.msgs[0].get("instanceType") != None)
261
262         # Asking for name will return also objectGUID and parentGUID
263         # and instanceType and of course name
264         res = self.ldb_admin.search(self.base_dn,
265                                     expression="samaccountname=Administrator",
266                                     attrs=["name"],
267                                     controls=["dirsync:1:0:1"])
268         self.assertTrue(res.msgs[0].get("objectGUID") != None)
269         self.assertTrue(res.msgs[0].get("name") != None)
270         self.assertTrue(res.msgs[0].get("parentGUID") != None)
271         self.assertTrue(res.msgs[0].get("instanceType") != None)
272
273         # Asking for dn will not return not only DN but more like if attrs=*
274         # parentGUID should be returned
275         res = self.ldb_admin.search(self.base_dn,
276                                     expression="samaccountname=Administrator",
277                                     attrs=["dn"],
278                                     controls=["dirsync:1:0:1"])
279         count = len(res.msgs[0])
280         res2 = self.ldb_admin.search(self.base_dn,
281                                      expression="samaccountname=Administrator",
282                                      controls=["dirsync:1:0:1"])
283         count2 = len(res2.msgs[0])
284         self.assertEqual(count, count2)
285
286         # Asking for cn will return nothing on objects that have CN as RDN
287         res = self.ldb_admin.search(self.base_dn,
288                                     expression="samaccountname=Administrator",
289                                     attrs=["cn"],
290                                     controls=["dirsync:1:0:1"])
291         self.assertEqual(len(res.msgs), 0)
292         # Asking for parentGUID will return nothing too
293         res = self.ldb_admin.search(self.base_dn,
294                                     expression="samaccountname=Administrator",
295                                     attrs=["parentGUID"],
296                                     controls=["dirsync:1:0:1"])
297         self.assertEqual(len(res.msgs), 0)
298         ouname="OU=testou,%s" % self.base_dn
299         self.ouname = ouname
300         self.ldb_admin.create_ou(ouname)
301         delta = Message()
302         delta.dn = Dn(self.ldb_admin, str(ouname))
303         delta["cn"] = MessageElement("test ou",
304                                      FLAG_MOD_ADD,
305                                      "cn")
306         self.ldb_admin.modify(delta)
307         res = self.ldb_admin.search(self.base_dn,
308                                     expression="name=testou",
309                                     attrs=["cn"],
310                                     controls=["dirsync:1:0:1"])
311
312         self.assertEqual(len(res.msgs), 1)
313         self.assertEqual(len(res.msgs[0]), 3)
314         delete_force(self.ldb_admin, ouname)
315
316     def test_dirsync_with_controls(self):
317         """Check that dirsync return correct informations when dealing with the NC"""
318         res = self.ldb_admin.search(self.base_dn,
319                                     expression="(distinguishedName=%s)" % str(self.base_dn),
320                                     attrs=["name"],
321                                     controls=["dirsync:1:0:10000", "extended_dn:1", "show_deleted:1"])
322
323     def test_dirsync_basenc(self):
324         """Check that dirsync return correct informations when dealing with the NC"""
325         res = self.ldb_admin.search(self.base_dn,
326                                     expression="(distinguishedName=%s)" % str(self.base_dn),
327                                     attrs=["name"],
328                                     controls=["dirsync:1:0:10000"])
329         self.assertEqual(len(res.msgs), 1)
330         self.assertEqual(len(res.msgs[0]), 3)
331
332         res = self.ldb_admin.search(self.base_dn,
333                                     expression="(distinguishedName=%s)" % str(self.base_dn),
334                                     attrs=["ntSecurityDescriptor"],
335                                     controls=["dirsync:1:0:10000"])
336         self.assertEqual(len(res.msgs), 1)
337         self.assertEqual(len(res.msgs[0]), 3)
338
339     def test_dirsync_othernc(self):
340         """Check that dirsync return information for entries that are normaly referrals (ie. other NCs)"""
341         res = self.ldb_admin.search(self.base_dn,
342                                     expression="(objectclass=configuration)",
343                                     attrs=["name"],
344                                     controls=["dirsync:1:0:10000"])
345         self.assertEqual(len(res.msgs), 1)
346         self.assertEqual(len(res.msgs[0]), 4)
347
348         res = self.ldb_admin.search(self.base_dn,
349                                     expression="(objectclass=configuration)",
350                                     attrs=["ntSecurityDescriptor"],
351                                     controls=["dirsync:1:0:10000"])
352         self.assertEqual(len(res.msgs), 1)
353         self.assertEqual(len(res.msgs[0]), 3)
354
355         res = self.ldb_admin.search(self.base_dn,
356                                     expression="(objectclass=domaindns)",
357                                     attrs=["ntSecurityDescriptor"],
358                                     controls=["dirsync:1:0:10000"])
359         nb = len(res.msgs)
360
361         # only sub nc returns a result when asked for objectGUID
362         res = self.ldb_admin.search(self.base_dn,
363                                     expression="(objectclass=domaindns)",
364                                     attrs=["objectGUID"],
365                                     controls=["dirsync:1:0:0"])
366         self.assertEqual(len(res.msgs), nb - 1)
367         if nb > 1:
368             self.assertTrue(res.msgs[0].get("objectGUID") != None)
369         else:
370             res = self.ldb_admin.search(self.base_dn,
371                                         expression="(objectclass=configuration)",
372                                         attrs=["objectGUID"],
373                                         controls=["dirsync:1:0:0"])
374
375
376     def test_dirsync_send_delta(self):
377         """Check that dirsync return correct delta when sending the last cookie"""
378         res = self.ldb_admin.search(self.base_dn,
379                                     expression="(&(samaccountname=test*)(!(isDeleted=*)))",
380                                     controls=["dirsync:1:0:10000"])
381         ctl = str(res.controls[0]).split(":")
382         ctl[1] = "1"
383         ctl[2] = "0"
384         ctl[3] = "10000"
385         control = str(":".join(ctl))
386         res = self.ldb_admin.search(self.base_dn,
387                                     expression="(&(samaccountname=test*)(!(isDeleted=*)))",
388                                     controls=[control])
389         self.assertEqual(len(res), 0)
390
391         res = self.ldb_admin.search(self.base_dn,
392                                     expression="(&(objectClass=organizationalUnit)(!(isDeleted=*)))",
393                                     controls=["dirsync:1:0:100000"])
394
395         ctl = str(res.controls[0]).split(":")
396         ctl[1] = "1"
397         ctl[2] = "0"
398         ctl[3] = "10000"
399         control2 = str(":".join(ctl))
400
401         # Let's create an OU
402         ouname="OU=testou2,%s" % self.base_dn
403         self.ouname = ouname
404         self.ldb_admin.create_ou(ouname)
405         res = self.ldb_admin.search(self.base_dn,
406                                     expression="(&(objectClass=organizationalUnit)(!(isDeleted=*)))",
407                                     controls=[control2])
408         self.assertEqual(len(res), 1)
409         ctl = str(res.controls[0]).split(":")
410         ctl[1] = "1"
411         ctl[2] = "0"
412         ctl[3] = "10000"
413         control3 = str(":".join(ctl))
414
415         delta = Message()
416         delta.dn = Dn(self.ldb_admin, str(ouname))
417
418         delta["cn"] = MessageElement("test ou",
419                                      FLAG_MOD_ADD,
420                                      "cn")
421         self.ldb_admin.modify(delta)
422         res = self.ldb_admin.search(self.base_dn,
423                                     expression="(&(objectClass=organizationalUnit)(!(isDeleted=*)))",
424                                     controls=[control3])
425
426         self.assertEqual(len(res.msgs), 1)
427         # 3 attributes: instanceType, cn and objectGUID
428         self.assertEqual(len(res.msgs[0]), 3)
429
430         delta = Message()
431         delta.dn = Dn(self.ldb_admin, str(ouname))
432         delta["cn"] = MessageElement([],
433                                      FLAG_MOD_DELETE,
434                                      "cn")
435         self.ldb_admin.modify(delta)
436         res = self.ldb_admin.search(self.base_dn,
437                                     expression="(&(objectClass=organizationalUnit)(!(isDeleted=*)))",
438                                     controls=[control3])
439
440         self.assertEqual(len(res.msgs), 1)
441         # So we won't have much attribute returned but instanceType and GUID
442         # are.
443         # 3 attributes: instanceType and objectGUID and cn but empty
444         self.assertEqual(len(res.msgs[0]), 3)
445         ouname = "OU=newouname,%s" % self.base_dn
446         self.ldb_admin.rename(str(res[0].dn), str(Dn(self.ldb_admin, ouname)))
447         self.ouname = ouname
448         ctl = str(res.controls[0]).split(":")
449         ctl[1] = "1"
450         ctl[2] = "0"
451         ctl[3] = "10000"
452         control4 = str(":".join(ctl))
453         res = self.ldb_admin.search(self.base_dn,
454                                     expression="(&(objectClass=organizationalUnit)(!(isDeleted=*)))",
455                                     controls=[control3])
456
457         self.assertTrue(res[0].get("parentGUID") != None)
458         self.assertTrue(res[0].get("name") != None)
459         delete_force(self.ldb_admin, ouname)
460
461     def test_dirsync_linkedattributes(self):
462         """Check that dirsync returnd deleted objects too"""
463         # Let's search for members
464         self.ldb_simple = self.get_ldb_connection(self.simple_user, self.user_pass)
465         res = self.ldb_simple.search(self.base_dn,
466                                      expression="(name=Administrators)",
467                                      controls=["dirsync:1:1:1"])
468
469         self.assertTrue(len(res[0].get("member")) > 0)
470         size = len(res[0].get("member"))
471
472         ctl = str(res.controls[0]).split(":")
473         ctl[1] = "1"
474         ctl[2] = "1"
475         ctl[3] = "10000"
476         control1 = str(":".join(ctl))
477         self.ldb_admin.add_remove_group_members("Administrators", [self.simple_user],
478                                                 add_members_operation=True)
479
480         res = self.ldb_simple.search(self.base_dn,
481                                      expression="(name=Administrators)",
482                                      controls=[control1])
483
484         self.assertEqual(len(res[0].get("member")), size + 1)
485         ctl = str(res.controls[0]).split(":")
486         ctl[1] = "1"
487         ctl[2] = "1"
488         ctl[3] = "10000"
489         control1 = str(":".join(ctl))
490
491         # remove the user from the group
492         self.ldb_admin.add_remove_group_members("Administrators", [self.simple_user],
493                                                 add_members_operation=False)
494
495         res = self.ldb_simple.search(self.base_dn,
496                                      expression="(name=Administrators)",
497                                      controls=[control1])
498
499         self.assertEqual(len(res[0].get("member")), size)
500
501         self.ldb_admin.newgroup("testgroup")
502         self.ldb_admin.add_remove_group_members("testgroup", [self.simple_user],
503                                                 add_members_operation=True)
504
505         res = self.ldb_admin.search(self.base_dn,
506                                     expression="(name=testgroup)",
507                                     controls=["dirsync:1:0:1"])
508
509         self.assertEqual(len(res[0].get("member")), 1)
510         self.assertTrue(res[0].get("member") != "")
511
512         ctl = str(res.controls[0]).split(":")
513         ctl[1] = "1"
514         ctl[2] = "0"
515         ctl[3] = "1"
516         control1 = str(":".join(ctl))
517
518         # Check that reasking the same question but with an updated cookie
519         # didn't return any results.
520         print(control1)
521         res = self.ldb_admin.search(self.base_dn,
522                                     expression="(name=testgroup)",
523                                     controls=[control1])
524         self.assertEqual(len(res), 0)
525
526         ctl = str(res.controls[0]).split(":")
527         ctl[1] = "1"
528         ctl[2] = "1"
529         ctl[3] = "10000"
530         control1 = str(":".join(ctl))
531
532         self.ldb_admin.add_remove_group_members("testgroup", [self.simple_user],
533                                                 add_members_operation=False)
534
535         res = self.ldb_admin.search(self.base_dn,
536                                     expression="(name=testgroup)",
537                                     attrs=["member"],
538                                     controls=[control1])
539
540         self.ldb_admin.deletegroup("testgroup")
541         self.assertEqual(len(res[0].get("member")), 0)
542
543
544
545     def test_dirsync_deleted_items(self):
546         """Check that dirsync returnd deleted objects too"""
547         # Let's create an OU
548         ouname="OU=testou3,%s" % self.base_dn
549         self.ouname = ouname
550         self.ldb_admin.create_ou(ouname)
551         res = self.ldb_admin.search(self.base_dn,
552                                     expression="(&(objectClass=organizationalUnit)(!(isDeleted=*)))",
553                                     controls=["dirsync:1:0:1"])
554         guid = None
555         for e in res:
556             if str(e["name"]) == "testou3":
557                 guid = str(ndr_unpack(misc.GUID,e.get("objectGUID")[0]))
558
559         ctl = str(res.controls[0]).split(":")
560         ctl[1] = "1"
561         ctl[2] = "0"
562         ctl[3] = "10000"
563         control1 = str(":".join(ctl))
564
565         # So now delete the object and check that
566         # we can see the object but deleted when admin
567         delete_force(self.ldb_admin, ouname)
568
569         res = self.ldb_admin.search(self.base_dn,
570                                     expression="(objectClass=organizationalUnit)",
571                                     controls=[control1])
572         self.assertEqual(len(res), 1)
573         guid2 = str(ndr_unpack(misc.GUID,res[0].get("objectGUID")[0]))
574         self.assertEqual(guid2, guid)
575         self.assertTrue(res[0].get("isDeleted"))
576         self.assertTrue(res[0].get("name") != None)
577
578     def test_cookie_from_others(self):
579         res = self.ldb_admin.search(self.base_dn,
580                                     expression="(&(objectClass=organizationalUnit)(!(isDeleted=*)))",
581                                     controls=["dirsync:1:0:1"])
582         ctl = str(res.controls[0]).split(":")
583         cookie = ndr_unpack(drsblobs.ldapControlDirSyncCookie, base64.b64decode(str(ctl[4])))
584         cookie.blob.guid1 = misc.GUID("128a99bf-abcd-1234-abcd-1fb625e530db")
585         controls=["dirsync:1:0:0:%s" % base64.b64encode(ndr_pack(cookie)).decode('utf8')]
586         res = self.ldb_admin.search(self.base_dn,
587                                     expression="(&(objectClass=organizationalUnit)(!(isDeleted=*)))",
588                                     controls=controls)
589
590
591 class ExtendedDirsyncTests(SimpleDirsyncTests):
592
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 if not getattr(opts, "listtests", False):
701     lp = sambaopts.get_loadparm()
702     samba.tests.cmdline_credentials = credopts.get_credentials(lp)
703
704
705 TestProgram(module=__name__, opts=subunitopts)