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