source4/dsdb/tests/python/acl.py

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