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