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