c45399ab605da791a785973c204eee425019d0a2
[kai/samba.git] / source4 / dsdb / tests / python / acl.py
1 #!/usr/bin/env python
2 # -*- coding: utf-8 -*-
3 # This is unit with tests for LDAP access checks
4
5 import optparse
6 import sys
7 import base64
8 import re
9
10 sys.path.append("bin/python")
11 import samba
12 samba.ensure_external_module("testtools", "testtools")
13 samba.ensure_external_module("subunit", "subunit/python")
14
15 import samba.getopt as options
16
17 from ldb import (
18     SCOPE_BASE, SCOPE_SUBTREE, LdbError, ERR_NO_SUCH_OBJECT,
19     ERR_UNWILLING_TO_PERFORM, ERR_INSUFFICIENT_ACCESS_RIGHTS)
20 from ldb import ERR_CONSTRAINT_VIOLATION
21 from ldb import ERR_OPERATIONS_ERROR
22 from ldb import Message, MessageElement, Dn
23 from ldb import FLAG_MOD_REPLACE, FLAG_MOD_DELETE
24 from samba.ndr import ndr_pack, ndr_unpack
25 from samba.dcerpc import security
26
27 from samba.auth import system_session
28 from samba import gensec
29 from samba.samdb import SamDB
30 from samba.credentials import Credentials
31 import samba.tests
32 from subunit.run import SubunitTestRunner
33 import unittest
34
35 parser = optparse.OptionParser("acl.py [options] <host>")
36 sambaopts = options.SambaOptions(parser)
37 parser.add_option_group(sambaopts)
38 parser.add_option_group(options.VersionOptions(parser))
39
40 # use command line creds if available
41 credopts = options.CredentialsOptions(parser)
42 parser.add_option_group(credopts)
43 opts, args = parser.parse_args()
44
45 if len(args) < 1:
46     parser.print_usage()
47     sys.exit(1)
48
49 host = args[0]
50
51 lp = sambaopts.get_loadparm()
52 creds = credopts.get_credentials(lp)
53 creds.set_gensec_features(creds.get_gensec_features() | gensec.FEATURE_SEAL)
54
55 #
56 # Tests start here
57 #
58
59 class AclTests(samba.tests.TestCase):
60
61     def delete_force(self, ldb, dn):
62         try:
63             ldb.delete(dn)
64         except LdbError, (num, _):
65             self.assertEquals(num, ERR_NO_SUCH_OBJECT)
66
67     def find_domain_sid(self, ldb):
68         res = ldb.search(base=self.base_dn, expression="(objectClass=*)", scope=SCOPE_BASE)
69         return ndr_unpack(security.dom_sid,res[0]["objectSid"][0])
70
71     def setUp(self):
72         super(AclTests, self).setUp()
73         self.ldb_admin = ldb
74         self.base_dn = ldb.domain_dn()
75         self.domain_sid = self.find_domain_sid(self.ldb_admin)
76         self.user_pass = "samba123@"
77         self.configuration_dn = self.ldb_admin.get_config_basedn().get_linearized()
78         print "baseDN: %s" % self.base_dn
79
80     def get_user_dn(self, name):
81         return "CN=%s,CN=Users,%s" % (name, self.base_dn)
82
83     def modify_desc(self, object_dn, desc):
84         """ Modify security descriptor using either SDDL string
85             or security.descriptor object
86         """
87         assert(isinstance(desc, str) or isinstance(desc, security.descriptor))
88         mod = """
89 dn: """ + object_dn + """
90 changetype: modify
91 replace: nTSecurityDescriptor
92 """
93         if isinstance(desc, str):
94             mod += "nTSecurityDescriptor: %s" % desc
95         elif isinstance(desc, security.descriptor):
96             mod += "nTSecurityDescriptor:: %s" % base64.b64encode(ndr_pack(desc))
97         self.ldb_admin.modify_ldif(mod)
98
99     def read_desc(self, object_dn):
100         res = self.ldb_admin.search(object_dn, SCOPE_BASE, None, ["nTSecurityDescriptor"])
101         desc = res[0]["nTSecurityDescriptor"][0]
102         return ndr_unpack(security.descriptor, desc)
103
104     def get_ldb_connection(self, target_username, target_password):
105         creds_tmp = Credentials()
106         creds_tmp.set_username(target_username)
107         creds_tmp.set_password(target_password)
108         creds_tmp.set_domain(creds.get_domain())
109         creds_tmp.set_realm(creds.get_realm())
110         creds_tmp.set_workstation(creds.get_workstation())
111         creds_tmp.set_gensec_features(creds_tmp.get_gensec_features()
112                                       | gensec.FEATURE_SEAL)
113         ldb_target = SamDB(url=host, credentials=creds_tmp, lp=lp)
114         return ldb_target
115
116     def get_object_sid(self, object_dn):
117         res = self.ldb_admin.search(object_dn)
118         return ndr_unpack(security.dom_sid, res[0]["objectSid"][0])
119
120     def dacl_add_ace(self, object_dn, ace):
121         desc = self.read_desc(object_dn)
122         desc_sddl = desc.as_sddl(self.domain_sid)
123         if ace in desc_sddl:
124             return
125         if desc_sddl.find("(") >= 0:
126             desc_sddl = desc_sddl[:desc_sddl.index("(")] + ace + desc_sddl[desc_sddl.index("("):]
127         else:
128             desc_sddl = desc_sddl + ace
129         self.modify_desc(object_dn, desc_sddl)
130
131     def get_desc_sddl(self, object_dn):
132         """ Return object nTSecutiryDescriptor in SDDL format
133         """
134         desc = self.read_desc(object_dn)
135         return desc.as_sddl(self.domain_sid)
136
137     # Test if we have any additional groups for users than default ones
138     def assert_user_no_group_member(self, username):
139         res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" % self.get_user_dn(username))
140         try:
141             self.assertEqual(res[0]["memberOf"][0], "")
142         except KeyError:
143             pass
144         else:
145             self.fail()
146
147 #tests on ldap add operations
148 class AclAddTests(AclTests):
149
150     def setUp(self):
151         super(AclAddTests, self).setUp()
152         # Domain admin that will be creator of OU parent-child structure
153         self.usr_admin_owner = "acl_add_user1"
154         # Second domain admin that will not be creator of OU parent-child structure
155         self.usr_admin_not_owner = "acl_add_user2"
156         # Regular user
157         self.regular_user = "acl_add_user3"
158         self.test_user1 = "test_add_user1"
159         self.test_group1 = "test_add_group1"
160         self.ou1 = "OU=test_add_ou1"
161         self.ou2 = "OU=test_add_ou2,%s" % self.ou1
162         self.ldb_admin.newuser(self.usr_admin_owner, self.user_pass)
163         self.ldb_admin.newuser(self.usr_admin_not_owner, self.user_pass)
164         self.ldb_admin.newuser(self.regular_user, self.user_pass)
165
166         # add admins to the Domain Admins group
167         self.ldb_admin.add_remove_group_members("Domain Admins", self.usr_admin_owner,
168                        add_members_operation=True)
169         self.ldb_admin.add_remove_group_members("Domain Admins", self.usr_admin_not_owner,
170                        add_members_operation=True)
171
172         self.ldb_owner = self.get_ldb_connection(self.usr_admin_owner, self.user_pass)
173         self.ldb_notowner = self.get_ldb_connection(self.usr_admin_not_owner, self.user_pass)
174         self.ldb_user = self.get_ldb_connection(self.regular_user, self.user_pass)
175
176     def tearDown(self):
177         super(AclAddTests, self).tearDown()
178         self.delete_force(self.ldb_admin, "CN=%s,%s,%s" %
179                           (self.test_user1, self.ou2, self.base_dn))
180         self.delete_force(self.ldb_admin, "CN=%s,%s,%s" %
181                           (self.test_group1, self.ou2, self.base_dn))
182         self.delete_force(self.ldb_admin, "%s,%s" % (self.ou2, self.base_dn))
183         self.delete_force(self.ldb_admin, "%s,%s" % (self.ou1, self.base_dn))
184         self.delete_force(self.ldb_admin, self.get_user_dn(self.usr_admin_owner))
185         self.delete_force(self.ldb_admin, self.get_user_dn(self.usr_admin_not_owner))
186         self.delete_force(self.ldb_admin, self.get_user_dn(self.regular_user))
187
188     # Make sure top OU is deleted (and so everything under it)
189     def assert_top_ou_deleted(self):
190         res = self.ldb_admin.search(self.base_dn,
191             expression="(distinguishedName=%s,%s)" % (
192                 "OU=test_add_ou1", self.base_dn))
193         self.assertEqual(res, [])
194
195     def test_add_u1(self):
196         """Testing OU with the rights of Doman Admin not creator of the OU """
197         self.assert_top_ou_deleted()
198         # Change descriptor for top level OU
199         self.ldb_owner.create_ou("OU=test_add_ou1," + self.base_dn)
200         self.ldb_owner.create_ou("OU=test_add_ou2,OU=test_add_ou1," + self.base_dn)
201         user_sid = self.get_object_sid(self.get_user_dn(self.usr_admin_not_owner))
202         mod = "(D;CI;WPCC;;;%s)" % str(user_sid)
203         self.dacl_add_ace("OU=test_add_ou1," + self.base_dn, mod)
204         # Test user and group creation with another domain admin's credentials
205         self.ldb_notowner.newuser(self.test_user1, self.user_pass, userou=self.ou2)
206         self.ldb_notowner.newgroup("test_add_group1", groupou="OU=test_add_ou2,OU=test_add_ou1",
207                                    grouptype=4)
208         # Make sure we HAVE created the two objects -- user and group
209         # !!! We should not be able to do that, but however beacuse of ACE ordering our inherited Deny ACE
210         # !!! comes after explicit (A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA) that comes from somewhere
211         res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s,%s)" % ("CN=test_add_user1,OU=test_add_ou2,OU=test_add_ou1", self.base_dn))
212         self.assertTrue(len(res) > 0)
213         res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s,%s)" % ("CN=test_add_group1,OU=test_add_ou2,OU=test_add_ou1", self.base_dn))
214         self.assertTrue(len(res) > 0)
215
216     def test_add_u2(self):
217         """Testing OU with the regular user that has no rights granted over the OU """
218         self.assert_top_ou_deleted()
219         # Create a parent-child OU structure with domain admin credentials
220         self.ldb_owner.create_ou("OU=test_add_ou1," + self.base_dn)
221         self.ldb_owner.create_ou("OU=test_add_ou2,OU=test_add_ou1," + self.base_dn)
222         # Test user and group creation with regular user credentials
223         try:
224             self.ldb_user.newuser(self.test_user1, self.user_pass, userou=self.ou2)
225             self.ldb_user.newgroup("test_add_group1", groupou="OU=test_add_ou2,OU=test_add_ou1",
226                                    grouptype=4)
227         except LdbError, (num, _):
228             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
229         else:
230             self.fail()
231         # Make sure we HAVEN'T created any of two objects -- user or group
232         res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s,%s)" % ("CN=test_add_user1,OU=test_add_ou2,OU=test_add_ou1", self.base_dn))
233         self.assertEqual(res, [])
234         res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s,%s)" % ("CN=test_add_group1,OU=test_add_ou2,OU=test_add_ou1", self.base_dn))
235         self.assertEqual(res, [])
236
237     def test_add_u3(self):
238         """Testing OU with the rights of regular user granted the right 'Create User child objects' """
239         self.assert_top_ou_deleted()
240         # Change descriptor for top level OU
241         self.ldb_owner.create_ou("OU=test_add_ou1," + self.base_dn)
242         user_sid = self.get_object_sid(self.get_user_dn(self.regular_user))
243         mod = "(OA;CI;CC;bf967aba-0de6-11d0-a285-00aa003049e2;;%s)" % str(user_sid)
244         self.dacl_add_ace("OU=test_add_ou1," + self.base_dn, mod)
245         self.ldb_owner.create_ou("OU=test_add_ou2,OU=test_add_ou1," + self.base_dn)
246         # Test user and group creation with granted user only to one of the objects
247         self.ldb_user.newuser(self.test_user1, self.user_pass, userou=self.ou2, setpassword=False)
248         try:
249             self.ldb_user.newgroup("test_add_group1", groupou="OU=test_add_ou2,OU=test_add_ou1",
250                                    grouptype=4)
251         except LdbError, (num, _):
252             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
253         else:
254             self.fail()
255         # Make sure we HAVE created the one of two objects -- user
256         res = self.ldb_admin.search(self.base_dn,
257                 expression="(distinguishedName=%s,%s)" %
258                 ("CN=test_add_user1,OU=test_add_ou2,OU=test_add_ou1",
259                     self.base_dn))
260         self.assertNotEqual(len(res), 0)
261         res = self.ldb_admin.search(self.base_dn,
262                 expression="(distinguishedName=%s,%s)" %
263                 ("CN=test_add_group1,OU=test_add_ou2,OU=test_add_ou1",
264                     self.base_dn) )
265         self.assertEqual(res, [])
266
267     def test_add_u4(self):
268         """ 4 Testing OU with the rights of Doman Admin creator of the OU"""
269         self.assert_top_ou_deleted()
270         self.ldb_owner.create_ou("OU=test_add_ou1," + self.base_dn)
271         self.ldb_owner.create_ou("OU=test_add_ou2,OU=test_add_ou1," + self.base_dn)
272         self.ldb_owner.newuser(self.test_user1, self.user_pass, userou=self.ou2)
273         self.ldb_owner.newgroup("test_add_group1", groupou="OU=test_add_ou2,OU=test_add_ou1",
274                                  grouptype=4)
275         # Make sure we have successfully created the two objects -- user and group
276         res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s,%s)" % ("CN=test_add_user1,OU=test_add_ou2,OU=test_add_ou1", self.base_dn))
277         self.assertTrue(len(res) > 0)
278         res = self.ldb_admin.search(self.base_dn,
279                 expression="(distinguishedName=%s,%s)" % ("CN=test_add_group1,OU=test_add_ou2,OU=test_add_ou1", self.base_dn))
280         self.assertTrue(len(res) > 0)
281
282 #tests on ldap modify operations
283 class AclModifyTests(AclTests):
284
285     def setUp(self):
286         super(AclModifyTests, self).setUp()
287         self.user_with_wp = "acl_mod_user1"
288         self.user_with_sm = "acl_mod_user2"
289         self.user_with_group_sm = "acl_mod_user3"
290         self.ldb_admin.newuser(self.user_with_wp, self.user_pass)
291         self.ldb_admin.newuser(self.user_with_sm, self.user_pass)
292         self.ldb_admin.newuser(self.user_with_group_sm, self.user_pass)
293         self.ldb_user = self.get_ldb_connection(self.user_with_wp, self.user_pass)
294         self.ldb_user2 = self.get_ldb_connection(self.user_with_sm, self.user_pass)
295         self.ldb_user3 = self.get_ldb_connection(self.user_with_group_sm, self.user_pass)
296         self.user_sid = self.get_object_sid( self.get_user_dn(self.user_with_wp))
297         self.ldb_admin.newgroup("test_modify_group2", grouptype=4)
298         self.ldb_admin.newgroup("test_modify_group3", grouptype=4)
299         self.ldb_admin.newuser("test_modify_user2", self.user_pass)
300
301     def tearDown(self):
302         super(AclModifyTests, self).tearDown()
303         self.delete_force(self.ldb_admin, self.get_user_dn("test_modify_user1"))
304         self.delete_force(self.ldb_admin, "CN=test_modify_group1,CN=Users," + self.base_dn)
305         self.delete_force(self.ldb_admin, "CN=test_modify_group2,CN=Users," + self.base_dn)
306         self.delete_force(self.ldb_admin, "CN=test_modify_group3,CN=Users," + self.base_dn)
307         self.delete_force(self.ldb_admin, "OU=test_modify_ou1," + self.base_dn)
308         self.delete_force(self.ldb_admin, self.get_user_dn(self.user_with_wp))
309         self.delete_force(self.ldb_admin, self.get_user_dn(self.user_with_sm))
310         self.delete_force(self.ldb_admin, self.get_user_dn(self.user_with_group_sm))
311         self.delete_force(self.ldb_admin, self.get_user_dn("test_modify_user2"))
312
313     def test_modify_u1(self):
314         """5 Modify one attribute if you have DS_WRITE_PROPERTY for it"""
315         mod = "(OA;;WP;bf967953-0de6-11d0-a285-00aa003049e2;;%s)" % str(self.user_sid)
316         # First test object -- User
317         print "Testing modify on User object"
318         self.ldb_admin.newuser("test_modify_user1", self.user_pass)
319         self.dacl_add_ace(self.get_user_dn("test_modify_user1"), mod)
320         ldif = """
321 dn: """ + self.get_user_dn("test_modify_user1") + """
322 changetype: modify
323 replace: displayName
324 displayName: test_changed"""
325         self.ldb_user.modify_ldif(ldif)
326         res = self.ldb_admin.search(self.base_dn,
327                 expression="(distinguishedName=%s)" % self.get_user_dn("test_modify_user1"))
328         self.assertEqual(res[0]["displayName"][0], "test_changed")
329         # Second test object -- Group
330         print "Testing modify on Group object"
331         self.ldb_admin.newgroup("test_modify_group1", grouptype=4)
332         self.dacl_add_ace("CN=test_modify_group1,CN=Users," + self.base_dn, mod)
333         ldif = """
334 dn: CN=test_modify_group1,CN=Users,""" + self.base_dn + """
335 changetype: modify
336 replace: displayName
337 displayName: test_changed"""
338         self.ldb_user.modify_ldif(ldif)
339         res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" % str("CN=test_modify_group1,CN=Users," + self.base_dn))
340         self.assertEqual(res[0]["displayName"][0], "test_changed")
341         # Third test object -- Organizational Unit
342         print "Testing modify on OU object"
343         #self.delete_force(self.ldb_admin, "OU=test_modify_ou1," + self.base_dn)
344         self.ldb_admin.create_ou("OU=test_modify_ou1," + self.base_dn)
345         self.dacl_add_ace("OU=test_modify_ou1," + self.base_dn, mod)
346         ldif = """
347 dn: OU=test_modify_ou1,""" + self.base_dn + """
348 changetype: modify
349 replace: displayName
350 displayName: test_changed"""
351         self.ldb_user.modify_ldif(ldif)
352         res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" % str("OU=test_modify_ou1," + self.base_dn))
353         self.assertEqual(res[0]["displayName"][0], "test_changed")
354
355     def test_modify_u2(self):
356         """6 Modify two attributes as you have DS_WRITE_PROPERTY granted only for one of them"""
357         mod = "(OA;;WP;bf967953-0de6-11d0-a285-00aa003049e2;;%s)" % str(self.user_sid)
358         # First test object -- User
359         print "Testing modify on User object"
360         #self.delete_force(self.ldb_admin, self.get_user_dn("test_modify_user1"))
361         self.ldb_admin.newuser("test_modify_user1", self.user_pass)
362         self.dacl_add_ace(self.get_user_dn("test_modify_user1"), mod)
363         # Modify on attribute you have rights for
364         ldif = """
365 dn: """ + self.get_user_dn("test_modify_user1") + """
366 changetype: modify
367 replace: displayName
368 displayName: test_changed"""
369         self.ldb_user.modify_ldif(ldif)
370         res = self.ldb_admin.search(self.base_dn,
371                 expression="(distinguishedName=%s)" %
372                 self.get_user_dn("test_modify_user1"))
373         self.assertEqual(res[0]["displayName"][0], "test_changed")
374         # Modify on attribute you do not have rights for granted
375         ldif = """
376 dn: """ + self.get_user_dn("test_modify_user1") + """
377 changetype: modify
378 replace: url
379 url: www.samba.org"""
380         try:
381             self.ldb_user.modify_ldif(ldif)
382         except LdbError, (num, _):
383             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
384         else:
385             # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
386             self.fail()
387         # Second test object -- Group
388         print "Testing modify on Group object"
389         self.ldb_admin.newgroup("test_modify_group1", grouptype=4)
390         self.dacl_add_ace("CN=test_modify_group1,CN=Users," + self.base_dn, mod)
391         ldif = """
392 dn: CN=test_modify_group1,CN=Users,""" + self.base_dn + """
393 changetype: modify
394 replace: displayName
395 displayName: test_changed"""
396         self.ldb_user.modify_ldif(ldif)
397         res = self.ldb_admin.search(self.base_dn,
398                 expression="(distinguishedName=%s)" %
399                 str("CN=test_modify_group1,CN=Users," + self.base_dn))
400         self.assertEqual(res[0]["displayName"][0], "test_changed")
401         # Modify on attribute you do not have rights for granted
402         ldif = """
403 dn: CN=test_modify_group1,CN=Users,""" + self.base_dn + """
404 changetype: modify
405 replace: url
406 url: www.samba.org"""
407         try:
408             self.ldb_user.modify_ldif(ldif)
409         except LdbError, (num, _):
410             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
411         else:
412             # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
413             self.fail()
414         # Second test object -- Organizational Unit
415         print "Testing modify on OU object"
416         self.ldb_admin.create_ou("OU=test_modify_ou1," + self.base_dn)
417         self.dacl_add_ace("OU=test_modify_ou1," + self.base_dn, mod)
418         ldif = """
419 dn: OU=test_modify_ou1,""" + self.base_dn + """
420 changetype: modify
421 replace: displayName
422 displayName: test_changed"""
423         self.ldb_user.modify_ldif(ldif)
424         res = self.ldb_admin.search(self.base_dn,
425                 expression="(distinguishedName=%s)" % str("OU=test_modify_ou1,"
426                     + self.base_dn))
427         self.assertEqual(res[0]["displayName"][0], "test_changed")
428         # Modify on attribute you do not have rights for granted
429         ldif = """
430 dn: OU=test_modify_ou1,""" + self.base_dn + """
431 changetype: modify
432 replace: url
433 url: www.samba.org"""
434         try:
435             self.ldb_user.modify_ldif(ldif)
436         except LdbError, (num, _):
437             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
438         else:
439             # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
440             self.fail()
441
442     def test_modify_u3(self):
443         """7 Modify one attribute as you have no what so ever rights granted"""
444         # First test object -- User
445         print "Testing modify on User object"
446         self.ldb_admin.newuser("test_modify_user1", self.user_pass)
447         # Modify on attribute you do not have rights for granted
448         ldif = """
449 dn: """ + self.get_user_dn("test_modify_user1") + """
450 changetype: modify
451 replace: url
452 url: www.samba.org"""
453         try:
454             self.ldb_user.modify_ldif(ldif)
455         except LdbError, (num, _):
456             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
457         else:
458             # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
459             self.fail()
460
461         # Second test object -- Group
462         print "Testing modify on Group object"
463         self.ldb_admin.newgroup("test_modify_group1", grouptype=4)
464         # Modify on attribute you do not have rights for granted
465         ldif = """
466 dn: CN=test_modify_group1,CN=Users,""" + self.base_dn + """
467 changetype: modify
468 replace: url
469 url: www.samba.org"""
470         try:
471             self.ldb_user.modify_ldif(ldif)
472         except LdbError, (num, _):
473             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
474         else:
475             # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
476             self.fail()
477
478         # Second test object -- Organizational Unit
479         print "Testing modify on OU object"
480         #self.delete_force(self.ldb_admin, "OU=test_modify_ou1," + self.base_dn)
481         self.ldb_admin.create_ou("OU=test_modify_ou1," + self.base_dn)
482         # Modify on attribute you do not have rights for granted
483         ldif = """
484 dn: OU=test_modify_ou1,""" + self.base_dn + """
485 changetype: modify
486 replace: url
487 url: www.samba.org"""
488         try:
489             self.ldb_user.modify_ldif(ldif)
490         except LdbError, (num, _):
491             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
492         else:
493             # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
494             self.fail()
495
496
497     def test_modify_u4(self):
498         """11 Grant WP to PRINCIPAL_SELF and test modify"""
499         ldif = """
500 dn: """ + self.get_user_dn(self.user_with_wp) + """
501 changetype: modify
502 add: adminDescription
503 adminDescription: blah blah blah"""
504         try:
505             self.ldb_user.modify_ldif(ldif)
506         except LdbError, (num, _):
507             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
508         else:
509             # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
510             self.fail()
511
512         mod = "(OA;;WP;bf967919-0de6-11d0-a285-00aa003049e2;;PS)"
513         self.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
514         # Modify on attribute you have rights for
515         self.ldb_user.modify_ldif(ldif)
516         res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" \
517                                     % self.get_user_dn(self.user_with_wp), attrs=["adminDescription"] )
518         self.assertEqual(res[0]["adminDescription"][0], "blah blah blah")
519
520     def test_modify_u5(self):
521         """12 test self membership"""
522         ldif = """
523 dn: CN=test_modify_group2,CN=Users,""" + self.base_dn + """
524 changetype: modify
525 add: Member
526 Member: """ +  self.get_user_dn(self.user_with_sm)
527 #the user has no rights granted, this should fail
528         try:
529             self.ldb_user2.modify_ldif(ldif)
530         except LdbError, (num, _):
531             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
532         else:
533             # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
534             self.fail()
535
536 #grant self-membership, should be able to add himself
537         user_sid = self.get_object_sid(self.get_user_dn(self.user_with_sm))
538         mod = "(OA;;SW;bf9679c0-0de6-11d0-a285-00aa003049e2;;%s)" % str(user_sid)
539         self.dacl_add_ace("CN=test_modify_group2,CN=Users," + self.base_dn, mod)
540         self.ldb_user2.modify_ldif(ldif)
541         res = self.ldb_admin.search( self.base_dn, expression="(distinguishedName=%s)" \
542                                     % ("CN=test_modify_group2,CN=Users," + self.base_dn), attrs=["Member"])
543         self.assertEqual(res[0]["Member"][0], self.get_user_dn(self.user_with_sm))
544 #but not other users
545         ldif = """
546 dn: CN=test_modify_group2,CN=Users,""" + self.base_dn + """
547 changetype: modify
548 add: Member
549 Member: CN=test_modify_user2,CN=Users,""" + self.base_dn
550         try:
551             self.ldb_user2.modify_ldif(ldif)
552         except LdbError, (num, _):
553             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
554         else:
555             self.fail()
556
557     def test_modify_u6(self):
558         """13 test self membership"""
559         ldif = """
560 dn: CN=test_modify_group2,CN=Users,""" + self.base_dn + """
561 changetype: modify
562 add: Member
563 Member: """ +  self.get_user_dn(self.user_with_sm) + """
564 Member: CN=test_modify_user2,CN=Users,""" + self.base_dn
565
566 #grant self-membership, should be able to add himself  but not others at the same time
567         user_sid = self.get_object_sid(self.get_user_dn(self.user_with_sm))
568         mod = "(OA;;SW;bf9679c0-0de6-11d0-a285-00aa003049e2;;%s)" % str(user_sid)
569         self.dacl_add_ace("CN=test_modify_group2,CN=Users," + self.base_dn, mod)
570         try:
571             self.ldb_user2.modify_ldif(ldif)
572         except LdbError, (num, _):
573             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
574         else:
575             self.fail()
576
577     def test_modify_u7(self):
578         """13 User with WP modifying Member"""
579 #a second user is given write property permission
580         user_sid = self.get_object_sid(self.get_user_dn(self.user_with_wp))
581         mod = "(A;;WP;;;%s)" % str(user_sid)
582         self.dacl_add_ace("CN=test_modify_group2,CN=Users," + self.base_dn, mod)
583         ldif = """
584 dn: CN=test_modify_group2,CN=Users,""" + self.base_dn + """
585 changetype: modify
586 add: Member
587 Member: """ +  self.get_user_dn(self.user_with_wp)
588         self.ldb_user.modify_ldif(ldif)
589         res = self.ldb_admin.search( self.base_dn, expression="(distinguishedName=%s)" \
590                                     % ("CN=test_modify_group2,CN=Users," + self.base_dn), attrs=["Member"])
591         self.assertEqual(res[0]["Member"][0], self.get_user_dn(self.user_with_wp))
592         ldif = """
593 dn: CN=test_modify_group2,CN=Users,""" + self.base_dn + """
594 changetype: modify
595 delete: Member"""
596         self.ldb_user.modify_ldif(ldif)
597         ldif = """
598 dn: CN=test_modify_group2,CN=Users,""" + self.base_dn + """
599 changetype: modify
600 add: Member
601 Member: CN=test_modify_user2,CN=Users,""" + self.base_dn
602         self.ldb_user.modify_ldif(ldif)
603         res = self.ldb_admin.search( self.base_dn, expression="(distinguishedName=%s)" \
604                                     % ("CN=test_modify_group2,CN=Users," + self.base_dn), attrs=["Member"])
605         self.assertEqual(res[0]["Member"][0], "CN=test_modify_user2,CN=Users," + self.base_dn)
606
607 #enable these when we have search implemented
608 class AclSearchTests(AclTests):
609
610     def setUp(self):
611         super(AclSearchTests, self).setUp()
612         self.u1 = "search_u1"
613         self.u2 = "search_u2"
614         self.u3 = "search_u3"
615         self.group1 = "group1"
616         self.creds_tmp = Credentials()
617         self.creds_tmp.set_username("")
618         self.creds_tmp.set_password("")
619         self.creds_tmp.set_domain(creds.get_domain())
620         self.creds_tmp.set_realm(creds.get_realm())
621         self.creds_tmp.set_workstation(creds.get_workstation())
622         self.anonymous = SamDB(url=host, credentials=self.creds_tmp, lp=lp)
623         self.dsheuristics = self.ldb_admin.get_dsheuristics()
624         self.ldb_admin.newuser(self.u1, self.user_pass)
625         self.ldb_admin.newuser(self.u2, self.user_pass)
626         self.ldb_admin.newuser(self.u3, self.user_pass)
627         self.ldb_admin.newgroup(self.group1, grouptype=-2147483646)
628         self.ldb_admin.add_remove_group_members(self.group1, self.u2,
629                                                 add_members_operation=True)
630         self.ldb_user = self.get_ldb_connection(self.u1, self.user_pass)
631         self.ldb_user2 = self.get_ldb_connection(self.u2, self.user_pass)
632         self.ldb_user3 = self.get_ldb_connection(self.u3, self.user_pass)
633         self.full_list = [Dn(self.ldb_admin,  "OU=ou2,OU=ou1," + self.base_dn),
634                           Dn(self.ldb_admin,  "OU=ou1," + self.base_dn),
635                           Dn(self.ldb_admin,  "OU=ou3,OU=ou2,OU=ou1," + self.base_dn),
636                           Dn(self.ldb_admin,  "OU=ou4,OU=ou2,OU=ou1," + self.base_dn),
637                           Dn(self.ldb_admin,  "OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn),
638                           Dn(self.ldb_admin,  "OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn)]
639         self.user_sid = self.get_object_sid(self.get_user_dn(self.u1))
640         self.group_sid = self.get_object_sid(self.get_user_dn(self.group1))
641
642     def create_clean_ou(self, object_dn):
643         """ Base repeating setup for unittests to follow """
644         res = self.ldb_admin.search(base=self.base_dn, scope=SCOPE_SUBTREE, \
645                 expression="distinguishedName=%s" % object_dn)
646         # Make sure top testing OU has been deleted before starting the test
647         self.assertEqual(res, [])
648         self.ldb_admin.create_ou(object_dn)
649         desc_sddl = self.get_desc_sddl(object_dn)
650         # Make sure there are inheritable ACEs initially
651         self.assertTrue("CI" in desc_sddl or "OI" in desc_sddl)
652         # Find and remove all inherit ACEs
653         res = re.findall("\(.*?\)", desc_sddl)
654         res = [x for x in res if ("CI" in x) or ("OI" in x)]
655         for x in res:
656             desc_sddl = desc_sddl.replace(x, "")
657         # Add flag 'protected' in both DACL and SACL so no inherit ACEs
658         # can propagate from above
659         # remove SACL, we are not interested
660         desc_sddl = desc_sddl.replace(":AI", ":AIP")
661         self.modify_desc(object_dn, desc_sddl)
662         # Verify all inheritable ACEs are gone
663         desc_sddl = self.get_desc_sddl(object_dn)
664         self.assertFalse("CI" in desc_sddl)
665         self.assertFalse("OI" in desc_sddl)
666
667     def tearDown(self):
668         super(AclSearchTests, self).tearDown()
669         self.delete_force(self.ldb_admin, "OU=test_search_ou2,OU=test_search_ou1," + self.base_dn)
670         self.delete_force(self.ldb_admin, "OU=test_search_ou1," + self.base_dn)
671         self.delete_force(self.ldb_admin, "OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn)
672         self.delete_force(self.ldb_admin, "OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn)
673         self.delete_force(self.ldb_admin, "OU=ou4,OU=ou2,OU=ou1," + self.base_dn)
674         self.delete_force(self.ldb_admin, "OU=ou3,OU=ou2,OU=ou1," + self.base_dn)
675         self.delete_force(self.ldb_admin, "OU=ou2,OU=ou1," + self.base_dn)
676         self.delete_force(self.ldb_admin, "OU=ou1," + self.base_dn)
677         self.delete_force(self.ldb_admin, self.get_user_dn("search_u1"))
678         self.delete_force(self.ldb_admin, self.get_user_dn("search_u2"))
679         self.delete_force(self.ldb_admin, self.get_user_dn("search_u3"))
680         self.delete_force(self.ldb_admin, self.get_user_dn("group1"))
681
682     def test_search_anonymous1(self):
683         """Verify access of rootDSE with the correct request"""
684         res = self.anonymous.search("", expression="(objectClass=*)", scope=SCOPE_BASE)
685         self.assertEquals(len(res), 1)
686         #verify some of the attributes
687         #dont care about values
688         self.assertTrue("ldapServiceName" in res[0])
689         self.assertTrue("namingContexts" in res[0])
690         self.assertTrue("isSynchronized" in res[0])
691         self.assertTrue("dsServiceName" in res[0])
692         self.assertTrue("supportedSASLMechanisms" in res[0])
693         self.assertTrue("isGlobalCatalogReady" in res[0])
694         self.assertTrue("domainControllerFunctionality" in res[0])
695         self.assertTrue("serverName" in res[0])
696
697     def test_search_anonymous2(self):
698         """Make sure we cannot access anything else"""
699         try:
700             res = self.anonymous.search("", expression="(objectClass=*)", scope=SCOPE_SUBTREE)
701         except LdbError, (num, _):
702             self.assertEquals(num, ERR_OPERATIONS_ERROR)
703         else:
704             self.fail()
705         try:
706             res = self.anonymous.search(self.base_dn, expression="(objectClass=*)", scope=SCOPE_SUBTREE)
707         except LdbError, (num, _):
708             self.assertEquals(num, ERR_OPERATIONS_ERROR)
709         else:
710             self.fail()
711         try:
712             res = self.anonymous.search("CN=Configuration," + self.base_dn, expression="(objectClass=*)",
713                                         scope=SCOPE_SUBTREE)
714         except LdbError, (num, _):
715             self.assertEquals(num, ERR_OPERATIONS_ERROR)
716         else:
717             self.fail()
718
719     def test_search_anonymous3(self):
720         """Set dsHeuristics and repeat"""
721         self.ldb_admin.set_dsheuristics("0000002")
722         self.ldb_admin.create_ou("OU=test_search_ou1," + self.base_dn)
723         mod = "(A;CI;LC;;;AN)"
724         self.dacl_add_ace("OU=test_search_ou1," + self.base_dn, mod)
725         self.ldb_admin.create_ou("OU=test_search_ou2,OU=test_search_ou1," + self.base_dn)
726         res = self.anonymous.search("OU=test_search_ou2,OU=test_search_ou1," + self.base_dn,
727                                     expression="(objectClass=*)", scope=SCOPE_SUBTREE)
728         self.assertEquals(len(res), 1)
729         self.assertTrue("dn" in res[0])
730         self.assertTrue(res[0]["dn"] == Dn(self.ldb_admin,
731                                            "OU=test_search_ou2,OU=test_search_ou1," + self.base_dn))
732         res = self.anonymous.search("CN=Configuration," + self.base_dn, expression="(objectClass=*)",
733                                     scope=SCOPE_SUBTREE)
734         self.assertEquals(len(res), 1)
735         self.assertTrue("dn" in res[0])
736         self.assertTrue(res[0]["dn"] == Dn(self.ldb_admin, self.configuration_dn))
737         self.ldb_admin.set_dsheuristics(self.dsheuristics)
738
739     def test_search1(self):
740         """Make sure users can see us if given LC to user and group"""
741         self.create_clean_ou("OU=ou1," + self.base_dn)
742         mod = "(A;;LC;;;%s)(A;;LC;;;%s)" % (str(self.user_sid), str(self.group_sid))
743         self.dacl_add_ace("OU=ou1," + self.base_dn, mod)
744         self.ldb_admin.create_ou("OU=ou2,OU=ou1," + self.base_dn,
745                                  "D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)" + mod)
746         self.ldb_admin.create_ou("OU=ou3,OU=ou2,OU=ou1," + self.base_dn,
747                                  "D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)" + mod)
748         self.ldb_admin.create_ou("OU=ou4,OU=ou2,OU=ou1," + self.base_dn,
749                                  "D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)" + mod)
750         self.ldb_admin.create_ou("OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn,
751                                  "D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)" + mod)
752         self.ldb_admin.create_ou("OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn,
753                                  "D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)" + mod)
754
755         #regular users must see only ou1 and ou2
756         res = self.ldb_user3.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
757                                     scope=SCOPE_SUBTREE)
758         self.assertEquals(len(res), 2)
759         ok_list = [Dn(self.ldb_admin,  "OU=ou2,OU=ou1," + self.base_dn),
760                    Dn(self.ldb_admin,  "OU=ou1," + self.base_dn)]
761
762         res_list = [ x["dn"] for x in res if x["dn"] in ok_list ]
763         self.assertEquals(sorted(res_list), sorted(ok_list))
764
765         #these users should see all ous
766         res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
767                                     scope=SCOPE_SUBTREE)
768         self.assertEquals(len(res), 6)
769         res_list = [ x["dn"] for x in res if x["dn"] in self.full_list ]
770         self.assertEquals(sorted(res_list), sorted(self.full_list))
771
772         res = self.ldb_user2.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
773                                     scope=SCOPE_SUBTREE)
774         self.assertEquals(len(res), 6)
775         res_list = [ x["dn"] for x in res if x["dn"] in self.full_list ]
776         self.assertEquals(sorted(res_list), sorted(self.full_list))
777
778     def test_search2(self):
779         """Make sure users can't see us if access is explicitly denied"""
780         self.create_clean_ou("OU=ou1," + self.base_dn)
781         self.ldb_admin.create_ou("OU=ou2,OU=ou1," + self.base_dn)
782         self.ldb_admin.create_ou("OU=ou3,OU=ou2,OU=ou1," + self.base_dn)
783         self.ldb_admin.create_ou("OU=ou4,OU=ou2,OU=ou1," + self.base_dn)
784         self.ldb_admin.create_ou("OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn)
785         self.ldb_admin.create_ou("OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn)
786         mod = "(D;;LC;;;%s)(D;;LC;;;%s)" % (str(self.user_sid), str(self.group_sid)) 
787         self.dacl_add_ace("OU=ou2,OU=ou1," + self.base_dn, mod)
788         res = self.ldb_user3.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
789                                     scope=SCOPE_SUBTREE)
790         #this user should see all ous
791         res_list = [ x["dn"] for x in res if x["dn"] in self.full_list ]
792         self.assertEquals(sorted(res_list), sorted(self.full_list))
793
794         #these users should see ou1, 2, 5 and 6 but not 3 and 4
795         res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
796                                     scope=SCOPE_SUBTREE)
797         ok_list = [Dn(self.ldb_admin,  "OU=ou2,OU=ou1," + self.base_dn),
798                    Dn(self.ldb_admin,  "OU=ou1," + self.base_dn),
799                    Dn(self.ldb_admin,  "OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn),
800                    Dn(self.ldb_admin,  "OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn)]
801         res_list = [ x["dn"] for x in res if x["dn"] in ok_list ]
802         self.assertEquals(sorted(res_list), sorted(ok_list))
803
804         res = self.ldb_user2.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
805                                     scope=SCOPE_SUBTREE)
806         self.assertEquals(len(res), 4)
807         res_list = [ x["dn"] for x in res if x["dn"] in ok_list ]
808         self.assertEquals(sorted(res_list), sorted(ok_list))
809
810     def test_search3(self):
811         """Make sure users can't see ous if access is explicitly denied - 2"""
812         self.create_clean_ou("OU=ou1," + self.base_dn)
813         mod = "(A;CI;LC;;;%s)(A;CI;LC;;;%s)" % (str(self.user_sid), str(self.group_sid))
814         self.dacl_add_ace("OU=ou1," + self.base_dn, mod)
815         self.ldb_admin.create_ou("OU=ou2,OU=ou1," + self.base_dn,
816                                  "D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)")
817         self.ldb_admin.create_ou("OU=ou3,OU=ou2,OU=ou1," + self.base_dn,
818                                  "D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)")
819         self.ldb_admin.create_ou("OU=ou4,OU=ou2,OU=ou1," + self.base_dn,
820                                  "D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)")
821         self.ldb_admin.create_ou("OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn,
822                                  "D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)")
823         self.ldb_admin.create_ou("OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn,
824                                  "D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)")
825
826         print "Testing correct behavior on nonaccessible search base"
827         try:
828              self.ldb_user3.search("OU=ou3,OU=ou2,OU=ou1," + self.base_dn, expression="(objectClass=*)",
829                                    scope=SCOPE_BASE)
830         except LdbError, (num, _):
831             self.assertEquals(num, ERR_NO_SUCH_OBJECT)
832         else:
833             self.fail()
834
835         mod = "(D;;LC;;;%s)(D;;LC;;;%s)" % (str(self.user_sid), str(self.group_sid))
836         self.dacl_add_ace("OU=ou2,OU=ou1," + self.base_dn, mod)
837
838         ok_list = [Dn(self.ldb_admin,  "OU=ou2,OU=ou1," + self.base_dn),
839                    Dn(self.ldb_admin,  "OU=ou1," + self.base_dn)]
840
841         res = self.ldb_user3.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
842                                     scope=SCOPE_SUBTREE)
843         res_list = [ x["dn"] for x in res if x["dn"] in ok_list ]
844         self.assertEquals(sorted(res_list), sorted(ok_list))
845
846         ok_list = [Dn(self.ldb_admin,  "OU=ou2,OU=ou1," + self.base_dn),
847                    Dn(self.ldb_admin,  "OU=ou1," + self.base_dn),
848                    Dn(self.ldb_admin,  "OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn),
849                    Dn(self.ldb_admin,  "OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn)]
850
851         #should not see ou3 and ou4, but should see ou5 and ou6
852         res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
853                                     scope=SCOPE_SUBTREE)
854         self.assertEquals(len(res), 4)
855         res_list = [ x["dn"] for x in res if x["dn"] in ok_list ]
856         self.assertEquals(sorted(res_list), sorted(ok_list))
857
858         res = self.ldb_user2.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
859                                     scope=SCOPE_SUBTREE)
860         self.assertEquals(len(res), 4)
861         res_list = [ x["dn"] for x in res if x["dn"] in ok_list ]
862         self.assertEquals(sorted(res_list), sorted(ok_list))
863
864     def test_search4(self):
865         """There is no difference in visibility if the user is also creator"""
866         self.create_clean_ou("OU=ou1," + self.base_dn)
867         mod = "(A;CI;CC;;;%s)" % (str(self.user_sid))
868         self.dacl_add_ace("OU=ou1," + self.base_dn, mod)
869         self.ldb_user.create_ou("OU=ou2,OU=ou1," + self.base_dn,
870                                 "D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)")
871         self.ldb_user.create_ou("OU=ou3,OU=ou2,OU=ou1," + self.base_dn,
872                                 "D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)")
873         self.ldb_user.create_ou("OU=ou4,OU=ou2,OU=ou1," + self.base_dn,
874                                 "D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)")
875         self.ldb_user.create_ou("OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn,
876                                 "D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)")
877         self.ldb_user.create_ou("OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn,
878                                 "D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)")
879
880         ok_list = [Dn(self.ldb_admin,  "OU=ou2,OU=ou1," + self.base_dn),
881                    Dn(self.ldb_admin,  "OU=ou1," + self.base_dn)]
882         res = self.ldb_user3.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
883                                     scope=SCOPE_SUBTREE)
884         self.assertEquals(len(res), 2)
885         res_list = [ x["dn"] for x in res if x["dn"] in ok_list ]
886         self.assertEquals(sorted(res_list), sorted(ok_list))
887
888         res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
889                                     scope=SCOPE_SUBTREE)
890         self.assertEquals(len(res), 2)
891         res_list = [ x["dn"] for x in res if x["dn"] in ok_list ]
892         self.assertEquals(sorted(res_list), sorted(ok_list))
893
894     def test_search5(self):
895         """Make sure users can see only attributes they are allowed to see"""
896         self.create_clean_ou("OU=ou1," + self.base_dn)
897         mod = "(A;CI;LC;;;%s)" % (str(self.user_sid))
898         self.dacl_add_ace("OU=ou1," + self.base_dn, mod)
899         self.ldb_admin.create_ou("OU=ou2,OU=ou1," + self.base_dn,
900                                  "D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)" + mod)
901         # assert user can only see dn
902         res = self.ldb_user.search("OU=ou2,OU=ou1," + self.base_dn, expression="(objectClass=*)",
903                                     scope=SCOPE_SUBTREE)
904         ok_list = ['dn']
905         self.assertEquals(len(res), 1)
906         res_list = res[0].keys()
907         self.assertEquals(res_list, ok_list)
908
909         res = self.ldb_user.search("OU=ou2,OU=ou1," + self.base_dn, expression="(objectClass=*)",
910                                     scope=SCOPE_BASE, attrs=["ou"])
911
912         self.assertEquals(len(res), 1)
913         res_list = res[0].keys()
914         self.assertEquals(res_list, ok_list)
915
916         #give read property on ou and assert user can only see dn and ou
917         mod = "(OA;;RP;bf9679f0-0de6-11d0-a285-00aa003049e2;;%s)" % (str(self.user_sid))
918         self.dacl_add_ace("OU=ou1," + self.base_dn, mod)
919         self.dacl_add_ace("OU=ou2,OU=ou1," + self.base_dn, mod)
920         res = self.ldb_user.search("OU=ou2,OU=ou1," + self.base_dn, expression="(objectClass=*)",
921                                     scope=SCOPE_SUBTREE)
922         ok_list = ['dn', 'ou']
923         self.assertEquals(len(res), 1)
924         res_list = res[0].keys()
925         self.assertEquals(sorted(res_list), sorted(ok_list))
926
927         #give read property on Public Information and assert user can see ou and other members
928         mod = "(OA;;RP;e48d0154-bcf8-11d1-8702-00c04fb96050;;%s)" % (str(self.user_sid))
929         self.dacl_add_ace("OU=ou1," + self.base_dn, mod)
930         self.dacl_add_ace("OU=ou2,OU=ou1," + self.base_dn, mod)
931         res = self.ldb_user.search("OU=ou2,OU=ou1," + self.base_dn, expression="(objectClass=*)",
932                                     scope=SCOPE_SUBTREE)
933
934         ok_list = ['dn', 'objectClass', 'ou', 'distinguishedName', 'name', 'objectGUID', 'objectCategory']
935         res_list = res[0].keys()
936         self.assertEquals(sorted(res_list), sorted(ok_list))
937
938     def test_search6(self):
939         """If an attribute that cannot be read is used in a filter, it is as if the attribute does not exist"""
940         self.create_clean_ou("OU=ou1," + self.base_dn)
941         mod = "(A;CI;LCCC;;;%s)" % (str(self.user_sid))
942         self.dacl_add_ace("OU=ou1," + self.base_dn, mod)
943         self.ldb_admin.create_ou("OU=ou2,OU=ou1," + self.base_dn,
944                                  "D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)" + mod)
945         self.ldb_user.create_ou("OU=ou3,OU=ou2,OU=ou1," + self.base_dn,
946                                 "D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)")
947
948         res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(ou=ou3)",
949                                     scope=SCOPE_SUBTREE)
950         #nothing should be returned as ou is not accessible
951         self.assertEquals(res, [])
952
953         #give read property on ou and assert user can only see dn and ou
954         mod = "(OA;;RP;bf9679f0-0de6-11d0-a285-00aa003049e2;;%s)" % (str(self.user_sid))
955         self.dacl_add_ace("OU=ou3,OU=ou2,OU=ou1," + self.base_dn, mod)
956         res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(ou=ou3)",
957                                     scope=SCOPE_SUBTREE)
958         self.assertEquals(len(res), 1)
959         ok_list = ['dn', 'ou']
960         res_list = res[0].keys()
961         self.assertEquals(sorted(res_list), sorted(ok_list))
962
963         #give read property on Public Information and assert user can see ou and other members
964         mod = "(OA;;RP;e48d0154-bcf8-11d1-8702-00c04fb96050;;%s)" % (str(self.user_sid))
965         self.dacl_add_ace("OU=ou2,OU=ou1," + self.base_dn, mod)
966         res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(ou=ou2)",
967                                    scope=SCOPE_SUBTREE)
968         self.assertEquals(len(res), 1)
969         ok_list = ['dn', 'objectClass', 'ou', 'distinguishedName', 'name', 'objectGUID', 'objectCategory']
970         res_list = res[0].keys()
971         self.assertEquals(sorted(res_list), sorted(ok_list))
972
973 #tests on ldap delete operations
974 class AclDeleteTests(AclTests):
975
976     def setUp(self):
977         super(AclDeleteTests, self).setUp()
978         self.regular_user = "acl_delete_user1"
979         # Create regular user
980         self.ldb_admin.newuser(self.regular_user, self.user_pass)
981         self.ldb_user = self.get_ldb_connection(self.regular_user, self.user_pass)
982
983     def tearDown(self):
984         super(AclDeleteTests, self).tearDown()
985         self.delete_force(self.ldb_admin, self.get_user_dn("test_delete_user1"))
986         self.delete_force(self.ldb_admin, self.get_user_dn(self.regular_user))
987
988     def test_delete_u1(self):
989         """User is prohibited by default to delete another User object"""
990         # Create user that we try to delete
991         self.ldb_admin.newuser("test_delete_user1", self.user_pass)
992         # Here delete User object should ALWAYS through exception
993         try:
994             self.ldb_user.delete(self.get_user_dn("test_delete_user1"))
995         except LdbError, (num, _):
996             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
997         else:
998             self.fail()
999
1000     def test_delete_u2(self):
1001         """User's group has RIGHT_DELETE to another User object"""
1002         user_dn = self.get_user_dn("test_delete_user1")
1003         # Create user that we try to delete
1004         self.ldb_admin.newuser("test_delete_user1", self.user_pass)
1005         mod = "(A;;SD;;;AU)"
1006         self.dacl_add_ace(user_dn, mod)
1007         # Try to delete User object
1008         self.ldb_user.delete(user_dn)
1009         res = self.ldb_admin.search(self.base_dn,
1010                 expression="(distinguishedName=%s)" % user_dn)
1011         self.assertEqual(res, [])
1012
1013     def test_delete_u3(self):
1014         """User indentified by SID has RIGHT_DELETE to another User object"""
1015         user_dn = self.get_user_dn("test_delete_user1")
1016         # Create user that we try to delete
1017         self.ldb_admin.newuser("test_delete_user1", self.user_pass)
1018         mod = "(A;;SD;;;%s)" % self.get_object_sid(self.get_user_dn(self.regular_user))
1019         self.dacl_add_ace(user_dn, mod)
1020         # Try to delete User object
1021         self.ldb_user.delete(user_dn)
1022         res = self.ldb_admin.search(self.base_dn,
1023                 expression="(distinguishedName=%s)" % user_dn)
1024         self.assertEqual(res, [])
1025
1026 #tests on ldap rename operations
1027 class AclRenameTests(AclTests):
1028
1029     def setUp(self):
1030         super(AclRenameTests, self).setUp()
1031         self.regular_user = "acl_rename_user1"
1032         self.ou1 = "OU=test_rename_ou1"
1033         self.ou2 = "OU=test_rename_ou2"
1034         self.ou3 = "OU=test_rename_ou3,%s" % self.ou2
1035         self.testuser1 = "test_rename_user1"
1036         self.testuser2 = "test_rename_user2"
1037         self.testuser3 = "test_rename_user3"
1038         self.testuser4 = "test_rename_user4"
1039         self.testuser5 = "test_rename_user5"
1040         # Create regular user
1041         self.ldb_admin.newuser(self.regular_user, self.user_pass)
1042         self.ldb_user = self.get_ldb_connection(self.regular_user, self.user_pass)
1043
1044     def tearDown(self):
1045         super(AclRenameTests, self).tearDown()
1046         # Rename OU3
1047         self.delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser1, self.ou3, self.base_dn))
1048         self.delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser2, self.ou3, self.base_dn))
1049         self.delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser5, self.ou3, self.base_dn))
1050         self.delete_force(self.ldb_admin, "%s,%s" % (self.ou3, self.base_dn))
1051         # Rename OU2
1052         self.delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser1, self.ou2, self.base_dn))
1053         self.delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser2, self.ou2, self.base_dn))
1054         self.delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser5, self.ou2, self.base_dn))
1055         self.delete_force(self.ldb_admin, "%s,%s" % (self.ou2, self.base_dn))
1056         # Rename OU1
1057         self.delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser1, self.ou1, self.base_dn))
1058         self.delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser2, self.ou1, self.base_dn))
1059         self.delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser5, self.ou1, self.base_dn))
1060         self.delete_force(self.ldb_admin, "OU=test_rename_ou3,%s,%s" % (self.ou1, self.base_dn))
1061         self.delete_force(self.ldb_admin, "%s,%s" % (self.ou1, self.base_dn))
1062         self.delete_force(self.ldb_admin, self.get_user_dn(self.regular_user))
1063
1064     def test_rename_u1(self):
1065         """Regular user fails to rename 'User object' within single OU"""
1066         # Create OU structure
1067         self.ldb_admin.create_ou("OU=test_rename_ou1," + self.base_dn)
1068         self.ldb_admin.newuser(self.testuser1, self.user_pass, userou=self.ou1)
1069         try:
1070             self.ldb_user.rename("CN=%s,%s,%s" % (self.testuser1, self.ou1, self.base_dn), \
1071                                      "CN=%s,%s,%s" % (self.testuser5, self.ou1, self.base_dn))
1072         except LdbError, (num, _):
1073             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1074         else:
1075             self.fail()
1076
1077     def test_rename_u2(self):
1078         """Grant WRITE_PROPERTY to AU so regular user can rename 'User object' within single OU"""
1079         ou_dn = "OU=test_rename_ou1," + self.base_dn
1080         user_dn = "CN=test_rename_user1," + ou_dn
1081         rename_user_dn = "CN=test_rename_user5," + ou_dn
1082         # Create OU structure
1083         self.ldb_admin.create_ou(ou_dn)
1084         self.ldb_admin.newuser(self.testuser1, self.user_pass, userou=self.ou1)
1085         mod = "(A;;WP;;;AU)"
1086         self.dacl_add_ace(user_dn, mod)
1087         # Rename 'User object' having WP to AU
1088         self.ldb_user.rename(user_dn, rename_user_dn)
1089         res = self.ldb_admin.search(self.base_dn,
1090                 expression="(distinguishedName=%s)" % user_dn)
1091         self.assertEqual(res, [])
1092         res = self.ldb_admin.search(self.base_dn,
1093                 expression="(distinguishedName=%s)" % rename_user_dn)
1094         self.assertNotEqual(res, [])
1095
1096     def test_rename_u3(self):
1097         """Test rename with rights granted to 'User object' SID"""
1098         ou_dn = "OU=test_rename_ou1," + self.base_dn
1099         user_dn = "CN=test_rename_user1," + ou_dn
1100         rename_user_dn = "CN=test_rename_user5," + ou_dn
1101         # Create OU structure
1102         self.ldb_admin.create_ou(ou_dn)
1103         self.ldb_admin.newuser(self.testuser1, self.user_pass, userou=self.ou1)
1104         sid = self.get_object_sid(self.get_user_dn(self.regular_user))
1105         mod = "(A;;WP;;;%s)" % str(sid)
1106         self.dacl_add_ace(user_dn, mod)
1107         # Rename 'User object' having WP to AU
1108         self.ldb_user.rename(user_dn, rename_user_dn)
1109         res = self.ldb_admin.search(self.base_dn,
1110                 expression="(distinguishedName=%s)" % user_dn)
1111         self.assertEqual(res, [])
1112         res = self.ldb_admin.search(self.base_dn,
1113                 expression="(distinguishedName=%s)" % rename_user_dn)
1114         self.assertNotEqual(res, [])
1115
1116     def test_rename_u4(self):
1117         """Rename 'User object' cross OU with WP, SD and CC right granted on reg. user to AU"""
1118         ou1_dn = "OU=test_rename_ou1," + self.base_dn
1119         ou2_dn = "OU=test_rename_ou2," + self.base_dn
1120         user_dn = "CN=test_rename_user2," + ou1_dn
1121         rename_user_dn = "CN=test_rename_user5," + ou2_dn
1122         # Create OU structure
1123         self.ldb_admin.create_ou(ou1_dn)
1124         self.ldb_admin.create_ou(ou2_dn)
1125         self.ldb_admin.newuser(self.testuser2, self.user_pass, userou=self.ou1)
1126         mod = "(A;;WPSD;;;AU)"
1127         self.dacl_add_ace(user_dn, mod)
1128         mod = "(A;;CC;;;AU)"
1129         self.dacl_add_ace(ou2_dn, mod)
1130         # Rename 'User object' having SD and CC to AU
1131         self.ldb_user.rename(user_dn, rename_user_dn)
1132         res = self.ldb_admin.search(self.base_dn,
1133                 expression="(distinguishedName=%s)" % user_dn)
1134         self.assertEqual(res, [])
1135         res = self.ldb_admin.search(self.base_dn,
1136                 expression="(distinguishedName=%s)" % rename_user_dn)
1137         self.assertNotEqual(res, [])
1138
1139     def test_rename_u5(self):
1140         """Test rename with rights granted to 'User object' SID"""
1141         ou1_dn = "OU=test_rename_ou1," + self.base_dn
1142         ou2_dn = "OU=test_rename_ou2," + self.base_dn
1143         user_dn = "CN=test_rename_user2," + ou1_dn
1144         rename_user_dn = "CN=test_rename_user5," + ou2_dn
1145         # Create OU structure
1146         self.ldb_admin.create_ou(ou1_dn)
1147         self.ldb_admin.create_ou(ou2_dn)
1148         self.ldb_admin.newuser(self.testuser2, self.user_pass, userou=self.ou1)
1149         sid = self.get_object_sid(self.get_user_dn(self.regular_user))
1150         mod = "(A;;WPSD;;;%s)" % str(sid)
1151         self.dacl_add_ace(user_dn, mod)
1152         mod = "(A;;CC;;;%s)" % str(sid)
1153         self.dacl_add_ace(ou2_dn, mod)
1154         # Rename 'User object' having SD and CC to AU
1155         self.ldb_user.rename(user_dn, rename_user_dn)
1156         res = self.ldb_admin.search(self.base_dn,
1157                 expression="(distinguishedName=%s)" % user_dn)
1158         self.assertEqual(res, [])
1159         res = self.ldb_admin.search(self.base_dn,
1160                 expression="(distinguishedName=%s)" % rename_user_dn)
1161         self.assertNotEqual(res, [])
1162
1163     def test_rename_u6(self):
1164         """Rename 'User object' cross OU with WP, DC and CC right granted on OU & user to AU"""
1165         ou1_dn = "OU=test_rename_ou1," + self.base_dn
1166         ou2_dn = "OU=test_rename_ou2," + self.base_dn
1167         user_dn = "CN=test_rename_user2," + ou1_dn
1168         rename_user_dn = "CN=test_rename_user2," + ou2_dn
1169         # Create OU structure
1170         self.ldb_admin.create_ou(ou1_dn)
1171         self.ldb_admin.create_ou(ou2_dn)
1172         #mod = "(A;CI;DCWP;;;AU)"
1173         mod = "(A;;DC;;;AU)"
1174         self.dacl_add_ace(ou1_dn, mod)
1175         mod = "(A;;CC;;;AU)"
1176         self.dacl_add_ace(ou2_dn, mod)
1177         self.ldb_admin.newuser(self.testuser2, self.user_pass, userou=self.ou1)
1178         mod = "(A;;WP;;;AU)"
1179         self.dacl_add_ace(user_dn, mod)
1180         # Rename 'User object' having SD and CC to AU
1181         self.ldb_user.rename(user_dn, rename_user_dn)
1182         res = self.ldb_admin.search(self.base_dn,
1183                 expression="(distinguishedName=%s)" % user_dn)
1184         self.assertEqual(res, [])
1185         res = self.ldb_admin.search(self.base_dn,
1186                 expression="(distinguishedName=%s)" % rename_user_dn)
1187         self.assertNotEqual(res, [])
1188
1189     def test_rename_u7(self):
1190         """Rename 'User object' cross OU (second level) with WP, DC and CC right granted on OU to AU"""
1191         ou1_dn = "OU=test_rename_ou1," + self.base_dn
1192         ou2_dn = "OU=test_rename_ou2," + self.base_dn
1193         ou3_dn = "OU=test_rename_ou3," + ou2_dn
1194         user_dn = "CN=test_rename_user2," + ou1_dn
1195         rename_user_dn = "CN=test_rename_user5," + ou3_dn
1196         # Create OU structure
1197         self.ldb_admin.create_ou(ou1_dn)
1198         self.ldb_admin.create_ou(ou2_dn)
1199         self.ldb_admin.create_ou(ou3_dn)
1200         mod = "(A;CI;WPDC;;;AU)"
1201         self.dacl_add_ace(ou1_dn, mod)
1202         mod = "(A;;CC;;;AU)"
1203         self.dacl_add_ace(ou3_dn, mod)
1204         self.ldb_admin.newuser(self.testuser2, self.user_pass, userou=self.ou1)
1205         # Rename 'User object' having SD and CC to AU
1206         self.ldb_user.rename(user_dn, rename_user_dn)
1207         res = self.ldb_admin.search(self.base_dn,
1208                 expression="(distinguishedName=%s)" % user_dn)
1209         self.assertEqual(res, [])
1210         res = self.ldb_admin.search(self.base_dn,
1211                 expression="(distinguishedName=%s)" % rename_user_dn)
1212         self.assertNotEqual(res, [])
1213
1214     def test_rename_u8(self):
1215         """Test rename on an object with and without modify access on the RDN attribute"""
1216         ou1_dn = "OU=test_rename_ou1," + self.base_dn
1217         ou2_dn = "OU=test_rename_ou2," + ou1_dn
1218         ou3_dn = "OU=test_rename_ou3," + ou1_dn
1219         # Create OU structure
1220         self.ldb_admin.create_ou(ou1_dn)
1221         self.ldb_admin.create_ou(ou2_dn)
1222         sid = self.get_object_sid(self.get_user_dn(self.regular_user))
1223         mod = "(OA;;WP;bf967a0e-0de6-11d0-a285-00aa003049e2;;%s)" % str(sid)
1224         self.dacl_add_ace(ou2_dn, mod)
1225         mod = "(OD;;WP;bf9679f0-0de6-11d0-a285-00aa003049e2;;%s)" % str(sid)
1226         self.dacl_add_ace(ou2_dn, mod)
1227         try:
1228             self.ldb_user.rename(ou2_dn, ou3_dn)
1229         except LdbError, (num, _):
1230             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1231         else:
1232             # This rename operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
1233             self.fail()
1234         sid = self.get_object_sid(self.get_user_dn(self.regular_user))
1235         mod = "(A;;WP;bf9679f0-0de6-11d0-a285-00aa003049e2;;%s)" % str(sid)
1236         self.dacl_add_ace(ou2_dn, mod)
1237         self.ldb_user.rename(ou2_dn, ou3_dn)
1238         res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" % ou2_dn)
1239         self.assertEqual(res, [])
1240         res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" % ou3_dn)
1241         self.assertNotEqual(res, [])
1242
1243 #tests on Control Access Rights
1244 class AclCARTests(AclTests):
1245
1246     def setUp(self):
1247         super(AclCARTests, self).setUp()
1248         self.user_with_wp = "acl_car_user1"
1249         self.user_with_pc = "acl_car_user2"
1250         self.ldb_admin.newuser(self.user_with_wp, self.user_pass)
1251         self.ldb_admin.newuser(self.user_with_pc, self.user_pass)
1252         self.ldb_user = self.get_ldb_connection(self.user_with_wp, self.user_pass)
1253         self.ldb_user2 = self.get_ldb_connection(self.user_with_pc, self.user_pass)
1254
1255         res = self.ldb_admin.search("CN=Directory Service, CN=Windows NT, CN=Services, "
1256                  + self.configuration_dn, scope=SCOPE_BASE, attrs=["dSHeuristics"])
1257         if "dSHeuristics" in res[0]:
1258             self.dsheuristics = res[0]["dSHeuristics"][0]
1259         else:
1260             self.dsheuristics = None
1261
1262         self.minPwdAge = self.ldb_admin.get_minPwdAge()
1263
1264         # Set the "dSHeuristics" to have the tests run against Windows Server
1265         self.ldb_admin.set_dsheuristics("000000001")
1266         # Set minPwdAge to 0
1267         self.ldb_admin.set_minPwdAge("0")
1268
1269     def tearDown(self):
1270         super(AclCARTests, self).tearDown()
1271         #restore original values
1272         self.ldb_admin.set_dsheuristics(self.dsheuristics)
1273         self.ldb_admin.set_minPwdAge(self.minPwdAge)
1274         self.delete_force(self.ldb_admin, self.get_user_dn(self.user_with_wp))
1275         self.delete_force(self.ldb_admin, self.get_user_dn(self.user_with_pc))
1276
1277     def test_change_password1(self):
1278         """Try a password change operation without any CARs given"""
1279         #users have change password by default - remove for negative testing
1280         desc = self.read_desc(self.get_user_dn(self.user_with_wp))
1281         sddl = desc.as_sddl(self.domain_sid)
1282         sddl = sddl.replace("(OA;;CR;ab721a53-1e2f-11d0-9819-00aa0040529b;;WD)", "")
1283         sddl = sddl.replace("(OA;;CR;ab721a53-1e2f-11d0-9819-00aa0040529b;;PS)", "")
1284         self.modify_desc(self.get_user_dn(self.user_with_wp), sddl)
1285         try:
1286             self.ldb_user.modify_ldif("""
1287 dn: """ + self.get_user_dn(self.user_with_wp) + """
1288 changetype: modify
1289 delete: unicodePwd
1290 unicodePwd:: """ + base64.b64encode("\"samba123@\"".encode('utf-16-le')) + """
1291 add: unicodePwd
1292 unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS2\"".encode('utf-16-le')) + """
1293 """)
1294         except LdbError, (num, _):
1295             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
1296         else:
1297             # for some reason we get constraint violation instead of insufficient access error
1298             self.fail()
1299
1300     def test_change_password2(self):
1301         """Make sure WP has no influence"""
1302         desc = self.read_desc(self.get_user_dn(self.user_with_wp))
1303         sddl = desc.as_sddl(self.domain_sid)
1304         sddl = sddl.replace("(OA;;CR;ab721a53-1e2f-11d0-9819-00aa0040529b;;WD)", "")
1305         sddl = sddl.replace("(OA;;CR;ab721a53-1e2f-11d0-9819-00aa0040529b;;PS)", "")
1306         self.modify_desc(self.get_user_dn(self.user_with_wp), sddl)
1307         mod = "(A;;WP;;;PS)"
1308         self.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
1309         desc = self.read_desc(self.get_user_dn(self.user_with_wp))
1310         sddl = desc.as_sddl(self.domain_sid)
1311         try:
1312             self.ldb_user.modify_ldif("""
1313 dn: """ + self.get_user_dn(self.user_with_wp) + """
1314 changetype: modify
1315 delete: unicodePwd
1316 unicodePwd:: """ + base64.b64encode("\"samba123@\"".encode('utf-16-le')) + """
1317 add: unicodePwd
1318 unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS2\"".encode('utf-16-le')) + """
1319 """)
1320         except LdbError, (num, _):
1321             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
1322         else:
1323             # for some reason we get constraint violation instead of insufficient access error
1324             self.fail()
1325
1326     def test_change_password3(self):
1327         """Make sure WP has no influence"""
1328         mod = "(D;;WP;;;PS)"
1329         self.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
1330         desc = self.read_desc(self.get_user_dn(self.user_with_wp))
1331         sddl = desc.as_sddl(self.domain_sid)
1332         self.ldb_user.modify_ldif("""
1333 dn: """ + self.get_user_dn(self.user_with_wp) + """
1334 changetype: modify
1335 delete: unicodePwd
1336 unicodePwd:: """ + base64.b64encode("\"samba123@\"".encode('utf-16-le')) + """
1337 add: unicodePwd
1338 unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS2\"".encode('utf-16-le')) + """
1339 """)
1340
1341     def test_change_password5(self):
1342         """Make sure rights have no influence on dBCSPwd"""
1343         desc = self.read_desc(self.get_user_dn(self.user_with_wp))
1344         sddl = desc.as_sddl(self.domain_sid)
1345         sddl = sddl.replace("(OA;;CR;ab721a53-1e2f-11d0-9819-00aa0040529b;;WD)", "")
1346         sddl = sddl.replace("(OA;;CR;ab721a53-1e2f-11d0-9819-00aa0040529b;;PS)", "")
1347         self.modify_desc(self.get_user_dn(self.user_with_wp), sddl)
1348         mod = "(D;;WP;;;PS)"
1349         self.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
1350         try:
1351             self.ldb_user.modify_ldif("""
1352 dn: """ + self.get_user_dn(self.user_with_wp) + """
1353 changetype: modify
1354 delete: dBCSPwd
1355 dBCSPwd: XXXXXXXXXXXXXXXX
1356 add: dBCSPwd
1357 dBCSPwd: YYYYYYYYYYYYYYYY
1358 """)
1359         except LdbError, (num, _):
1360             self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
1361         else:
1362             self.fail()
1363
1364     def test_change_password6(self):
1365         """Test uneven delete/adds"""
1366         try:
1367             self.ldb_user.modify_ldif("""
1368 dn: """ + self.get_user_dn(self.user_with_wp) + """
1369 changetype: modify
1370 delete: userPassword
1371 userPassword: thatsAcomplPASS1
1372 delete: userPassword
1373 userPassword: thatsAcomplPASS1
1374 add: userPassword
1375 userPassword: thatsAcomplPASS2
1376 """)
1377         except LdbError, (num, _):
1378             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1379         else:
1380             self.fail()
1381         mod = "(OA;;CR;00299570-246d-11d0-a768-00aa006e0529;;PS)"
1382         self.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
1383         try:
1384             self.ldb_user.modify_ldif("""
1385 dn: """ + self.get_user_dn(self.user_with_wp) + """
1386 changetype: modify
1387 delete: userPassword
1388 userPassword: thatsAcomplPASS1
1389 delete: userPassword
1390 userPassword: thatsAcomplPASS1
1391 add: userPassword
1392 userPassword: thatsAcomplPASS2
1393 """)
1394             # This fails on Windows 2000 domain level with constraint violation
1395         except LdbError, (num, _):
1396             self.assertTrue(num == ERR_CONSTRAINT_VIOLATION or
1397                             num == ERR_UNWILLING_TO_PERFORM)
1398         else:
1399             self.fail()
1400
1401
1402     def test_change_password7(self):
1403         """Try a password change operation without any CARs given"""
1404         #users have change password by default - remove for negative testing
1405         desc = self.read_desc(self.get_user_dn(self.user_with_wp))
1406         sddl = desc.as_sddl(self.domain_sid)
1407         self.modify_desc(self.get_user_dn(self.user_with_wp), sddl)
1408         #first change our own password
1409         self.ldb_user2.modify_ldif("""
1410 dn: """ + self.get_user_dn(self.user_with_pc) + """
1411 changetype: modify
1412 delete: unicodePwd
1413 unicodePwd:: """ + base64.b64encode("\"samba123@\"".encode('utf-16-le')) + """
1414 add: unicodePwd
1415 unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS1\"".encode('utf-16-le')) + """
1416 """)
1417         #then someone else's
1418         self.ldb_user2.modify_ldif("""
1419 dn: """ + self.get_user_dn(self.user_with_wp) + """
1420 changetype: modify
1421 delete: unicodePwd
1422 unicodePwd:: """ + base64.b64encode("\"samba123@\"".encode('utf-16-le')) + """
1423 add: unicodePwd
1424 unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS2\"".encode('utf-16-le')) + """
1425 """)
1426
1427     def test_reset_password1(self):
1428         """Try a user password reset operation (unicodePwd) before and after granting CAR"""
1429         try:
1430             self.ldb_user.modify_ldif("""
1431 dn: """ + self.get_user_dn(self.user_with_wp) + """
1432 changetype: modify
1433 replace: unicodePwd
1434 unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS1\"".encode('utf-16-le')) + """
1435 """)
1436         except LdbError, (num, _):
1437             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1438         else:
1439             self.fail()
1440         mod = "(OA;;CR;00299570-246d-11d0-a768-00aa006e0529;;PS)"
1441         self.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
1442         self.ldb_user.modify_ldif("""
1443 dn: """ + self.get_user_dn(self.user_with_wp) + """
1444 changetype: modify
1445 replace: unicodePwd
1446 unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS1\"".encode('utf-16-le')) + """
1447 """)
1448
1449     def test_reset_password2(self):
1450         """Try a user password reset operation (userPassword) before and after granting CAR"""
1451         try:
1452             self.ldb_user.modify_ldif("""
1453 dn: """ + self.get_user_dn(self.user_with_wp) + """
1454 changetype: modify
1455 replace: userPassword
1456 userPassword: thatsAcomplPASS1
1457 """)
1458         except LdbError, (num, _):
1459             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1460         else:
1461             self.fail()
1462         mod = "(OA;;CR;00299570-246d-11d0-a768-00aa006e0529;;PS)"
1463         self.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
1464         try:
1465             self.ldb_user.modify_ldif("""
1466 dn: """ + self.get_user_dn(self.user_with_wp) + """
1467 changetype: modify
1468 replace: userPassword
1469 userPassword: thatsAcomplPASS1
1470 """)
1471             # This fails on Windows 2000 domain level with constraint violation
1472         except LdbError, (num, _):
1473             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
1474
1475     def test_reset_password3(self):
1476         """Grant WP and see what happens (unicodePwd)"""
1477         mod = "(A;;WP;;;PS)"
1478         self.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
1479         try:
1480             self.ldb_user.modify_ldif("""
1481 dn: """ + self.get_user_dn(self.user_with_wp) + """
1482 changetype: modify
1483 replace: unicodePwd
1484 unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS1\"".encode('utf-16-le')) + """
1485 """)
1486         except LdbError, (num, _):
1487             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1488         else:
1489             self.fail()
1490
1491     def test_reset_password4(self):
1492         """Grant WP and see what happens (userPassword)"""
1493         mod = "(A;;WP;;;PS)"
1494         self.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
1495         try:
1496             self.ldb_user.modify_ldif("""
1497 dn: """ + self.get_user_dn(self.user_with_wp) + """
1498 changetype: modify
1499 replace: userPassword
1500 userPassword: thatsAcomplPASS1
1501 """)
1502         except LdbError, (num, _):
1503             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1504         else:
1505             self.fail()
1506
1507     def test_reset_password5(self):
1508         """Explicitly deny WP but grant CAR (unicodePwd)"""
1509         mod = "(D;;WP;;;PS)(OA;;CR;00299570-246d-11d0-a768-00aa006e0529;;PS)"
1510         self.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
1511         self.ldb_user.modify_ldif("""
1512 dn: """ + self.get_user_dn(self.user_with_wp) + """
1513 changetype: modify
1514 replace: unicodePwd
1515 unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS1\"".encode('utf-16-le')) + """
1516 """)
1517
1518     def test_reset_password6(self):
1519         """Explicitly deny WP but grant CAR (userPassword)"""
1520         mod = "(D;;WP;;;PS)(OA;;CR;00299570-246d-11d0-a768-00aa006e0529;;PS)"
1521         self.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
1522         try:
1523             self.ldb_user.modify_ldif("""
1524 dn: """ + self.get_user_dn(self.user_with_wp) + """
1525 changetype: modify
1526 replace: userPassword
1527 userPassword: thatsAcomplPASS1
1528 """)
1529             # This fails on Windows 2000 domain level with constraint violation
1530         except LdbError, (num, _):
1531             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
1532
1533 class AclExtendedTests(AclTests):
1534
1535     def setUp(self):
1536         super(AclExtendedTests, self).setUp()
1537         #regular user, will be the creator
1538         self.u1 = "ext_u1"
1539         #regular user
1540         self.u2 = "ext_u2"
1541         #admin user
1542         self.u3 = "ext_u3"
1543         self.ldb_admin.newuser(self.u1, self.user_pass)
1544         self.ldb_admin.newuser(self.u2, self.user_pass)
1545         self.ldb_admin.newuser(self.u3, self.user_pass)
1546         self.ldb_admin.add_remove_group_members("Domain Admins", self.u3,
1547                                                 add_members_operation=True)
1548         self.ldb_user1 = self.get_ldb_connection(self.u1, self.user_pass)
1549         self.ldb_user2 = self.get_ldb_connection(self.u2, self.user_pass)
1550         self.ldb_user3 = self.get_ldb_connection(self.u3, self.user_pass)
1551         self.user_sid1 = self.get_object_sid(self.get_user_dn(self.u1))
1552         self.user_sid2 = self.get_object_sid(self.get_user_dn(self.u2))
1553
1554     def tearDown(self):
1555         super(AclExtendedTests, self).tearDown()
1556         self.delete_force(self.ldb_admin, self.get_user_dn(self.u1))
1557         self.delete_force(self.ldb_admin, self.get_user_dn(self.u2))
1558         self.delete_force(self.ldb_admin, self.get_user_dn(self.u3))
1559         self.delete_force(self.ldb_admin, "CN=ext_group1,OU=ext_ou1," + self.base_dn)
1560         self.delete_force(self.ldb_admin, "ou=ext_ou1," + self.base_dn)
1561
1562     def test_ntSecurityDescriptor(self):
1563         #create empty ou
1564         self.ldb_admin.create_ou("ou=ext_ou1," + self.base_dn)
1565         #give u1 Create children access
1566         mod = "(A;;CC;;;%s)" % str(self.user_sid1)
1567         self.dacl_add_ace("OU=ext_ou1," + self.base_dn, mod)
1568         mod = "(A;;LC;;;%s)" % str(self.user_sid2)
1569         self.dacl_add_ace("OU=ext_ou1," + self.base_dn, mod)
1570         #create a group under that, grant RP to u2
1571         self.ldb_user1.newgroup("ext_group1", groupou="OU=ext_ou1", grouptype=4)
1572         mod = "(A;;RP;;;%s)" % str(self.user_sid2)
1573         self.dacl_add_ace("CN=ext_group1,OU=ext_ou1," + self.base_dn, mod)
1574         #u2 must not read the descriptor
1575         res = self.ldb_user2.search("CN=ext_group1,OU=ext_ou1," + self.base_dn,
1576                                     SCOPE_BASE, None, ["nTSecurityDescriptor"])
1577         self.assertNotEqual(res,[])
1578         self.assertFalse("nTSecurityDescriptor" in res[0].keys())
1579         #grant RC to u2 - still no access
1580         mod = "(A;;RC;;;%s)" % str(self.user_sid2)
1581         self.dacl_add_ace("CN=ext_group1,OU=ext_ou1," + self.base_dn, mod)
1582         res = self.ldb_user2.search("CN=ext_group1,OU=ext_ou1," + self.base_dn,
1583                                     SCOPE_BASE, None, ["nTSecurityDescriptor"])
1584         self.assertNotEqual(res,[])
1585         self.assertFalse("nTSecurityDescriptor" in res[0].keys())
1586         #u3 is member of administrators group, should be able to read sd
1587         res = self.ldb_user3.search("CN=ext_group1,OU=ext_ou1," + self.base_dn,
1588                                     SCOPE_BASE, None, ["nTSecurityDescriptor"])
1589         self.assertEqual(len(res),1)
1590         self.assertTrue("nTSecurityDescriptor" in res[0].keys())
1591
1592 # Important unit running information
1593
1594 if not "://" in host:
1595     host = "ldap://%s" % host
1596 ldb = SamDB(host, credentials=creds, session_info=system_session(), lp=lp)
1597
1598 runner = SubunitTestRunner()
1599 rc = 0
1600 if not runner.run(unittest.makeSuite(AclAddTests)).wasSuccessful():
1601     rc = 1
1602 if not runner.run(unittest.makeSuite(AclModifyTests)).wasSuccessful():
1603     rc = 1
1604 if not runner.run(unittest.makeSuite(AclDeleteTests)).wasSuccessful():
1605     rc = 1
1606 if not runner.run(unittest.makeSuite(AclRenameTests)).wasSuccessful():
1607     rc = 1
1608 if not runner.run(unittest.makeSuite(AclCARTests)).wasSuccessful():
1609     rc = 1
1610 if not runner.run(unittest.makeSuite(AclSearchTests)).wasSuccessful():
1611     rc = 1
1612 if not runner.run(unittest.makeSuite(AclExtendedTests)).wasSuccessful():
1613     rc = 1
1614
1615
1616 sys.exit(rc)