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