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