source4/dsdb/tests/python/acl.py

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