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