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 from __future__ import print_function
   6 import optparse
   7 import sys
   8 import base64
   9 import re
  10 sys.path.insert(0, "bin/python")
  11 import samba
  12
  13 from samba.tests.subunitrun import SubunitOptions, TestProgram
  14
  15 import samba.getopt as options
  16 from samba.join import DCJoinContext
  17
  18 from ldb import (
  19     SCOPE_BASE, SCOPE_SUBTREE, LdbError, ERR_NO_SUCH_OBJECT,
  20     ERR_UNWILLING_TO_PERFORM, ERR_INSUFFICIENT_ACCESS_RIGHTS)
  21 from ldb import ERR_CONSTRAINT_VIOLATION
  22 from ldb import ERR_OPERATIONS_ERROR
  23 from ldb import Message, MessageElement, Dn
  24 from ldb import FLAG_MOD_REPLACE, FLAG_MOD_ADD, FLAG_MOD_DELETE
  25 from samba.dcerpc import security, drsuapi, misc
  26
  27 from samba.auth import system_session
  28 from samba import gensec, sd_utils
  29 from samba.samdb import SamDB
  30 from samba.credentials import Credentials, DONT_USE_KERBEROS
  31 import samba.tests
  32 from samba.tests import delete_force
  33 import samba.dsdb
  34 from samba.tests.password_test import PasswordCommon
  35
  36 parser = optparse.OptionParser("acl.py [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 subunitopts = SubunitOptions(parser)
  45 parser.add_option_group(subunitopts)
  46
  47 opts, args = parser.parse_args()
  48
  49 if len(args) < 1:
  50     parser.print_usage()
  51     sys.exit(1)
  52
  53 host = args[0]
  54 if "://" not in host:
  55     ldaphost = "ldap://%s" % host
  56 else:
  57     ldaphost = host
  58     start = host.rindex("://")
  59     host = host.lstrip(start + 3)
  60
  61 lp = sambaopts.get_loadparm()
  62 creds = credopts.get_credentials(lp)
  63 creds.set_gensec_features(creds.get_gensec_features() | gensec.FEATURE_SEAL)
  64
  65 #
  66 # Tests start here
  67 #
  68
  69
  70 class AclTests(samba.tests.TestCase):
  71
  72     def setUp(self):
  73         super(AclTests, self).setUp()
  74         self.ldb_admin = SamDB(ldaphost, credentials=creds, session_info=system_session(lp), lp=lp)
  75         self.base_dn = self.ldb_admin.domain_dn()
  76         self.domain_sid = security.dom_sid(self.ldb_admin.get_domain_sid())
  77         self.user_pass = "samba123@"
  78         self.configuration_dn = self.ldb_admin.get_config_basedn().get_linearized()
  79         self.sd_utils = sd_utils.SDUtils(self.ldb_admin)
  80         self.addCleanup(self.delete_admin_connection)
  81         # used for anonymous login
  82         self.creds_tmp = Credentials()
  83         self.creds_tmp.set_username("")
  84         self.creds_tmp.set_password("")
  85         self.creds_tmp.set_domain(creds.get_domain())
  86         self.creds_tmp.set_realm(creds.get_realm())
  87         self.creds_tmp.set_workstation(creds.get_workstation())
  88         print("baseDN: %s" % self.base_dn)
  89
  90     def get_user_dn(self, name):
  91         return "CN=%s,CN=Users,%s" % (name, self.base_dn)
  92
  93     def get_ldb_connection(self, target_username, target_password):
  94         creds_tmp = Credentials()
  95         creds_tmp.set_username(target_username)
  96         creds_tmp.set_password(target_password)
  97         creds_tmp.set_domain(creds.get_domain())
  98         creds_tmp.set_realm(creds.get_realm())
  99         creds_tmp.set_workstation(creds.get_workstation())
 100         creds_tmp.set_gensec_features(creds_tmp.get_gensec_features()
 101                                       | gensec.FEATURE_SEAL)
 102         creds_tmp.set_kerberos_state(DONT_USE_KERBEROS)  # kinit is too expensive to use in a tight loop
 103         ldb_target = SamDB(url=ldaphost, credentials=creds_tmp, lp=lp)
 104         return ldb_target
 105
 106     # Test if we have any additional groups for users than default ones
 107     def assert_user_no_group_member(self, username):
 108         res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" % self.get_user_dn(username))
 109         try:
 110             self.assertEqual(res[0]["memberOf"][0], "")
 111         except KeyError:
 112             pass
 113         else:
 114             self.fail()
 115
 116     def delete_admin_connection(self):
 117         del self.sd_utils
 118         del self.ldb_admin
 119
 120 # tests on ldap add operations
 121
 122
 123 class AclAddTests(AclTests):
 124
 125     def setUp(self):
 126         super(AclAddTests, self).setUp()
 127         # Domain admin that will be creator of OU parent-child structure
 128         self.usr_admin_owner = "acl_add_user1"
 129         # Second domain admin that will not be creator of OU parent-child structure
 130         self.usr_admin_not_owner = "acl_add_user2"
 131         # Regular user
 132         self.regular_user = "acl_add_user3"
 133         self.test_user1 = "test_add_user1"
 134         self.test_group1 = "test_add_group1"
 135         self.ou1 = "OU=test_add_ou1"
 136         self.ou2 = "OU=test_add_ou2,%s" % self.ou1
 137         self.ldb_admin.newuser(self.usr_admin_owner, self.user_pass)
 138         self.ldb_admin.newuser(self.usr_admin_not_owner, self.user_pass)
 139         self.ldb_admin.newuser(self.regular_user, self.user_pass)
 140
 141         # add admins to the Domain Admins group
 142         self.ldb_admin.add_remove_group_members("Domain Admins", [self.usr_admin_owner],
 143                                                 add_members_operation=True)
 144         self.ldb_admin.add_remove_group_members("Domain Admins", [self.usr_admin_not_owner],
 145                                                 add_members_operation=True)
 146
 147         self.ldb_owner = self.get_ldb_connection(self.usr_admin_owner, self.user_pass)
 148         self.ldb_notowner = self.get_ldb_connection(self.usr_admin_not_owner, self.user_pass)
 149         self.ldb_user = self.get_ldb_connection(self.regular_user, self.user_pass)
 150
 151     def tearDown(self):
 152         super(AclAddTests, self).tearDown()
 153         delete_force(self.ldb_admin, "CN=%s,%s,%s" %
 154                      (self.test_user1, self.ou2, self.base_dn))
 155         delete_force(self.ldb_admin, "CN=%s,%s,%s" %
 156                      (self.test_group1, self.ou2, self.base_dn))
 157         delete_force(self.ldb_admin, "%s,%s" % (self.ou2, self.base_dn))
 158         delete_force(self.ldb_admin, "%s,%s" % (self.ou1, self.base_dn))
 159         delete_force(self.ldb_admin, self.get_user_dn(self.usr_admin_owner))
 160         delete_force(self.ldb_admin, self.get_user_dn(self.usr_admin_not_owner))
 161         delete_force(self.ldb_admin, self.get_user_dn(self.regular_user))
 162         delete_force(self.ldb_admin, self.get_user_dn("test_add_anonymous"))
 163
 164         del self.ldb_notowner
 165         del self.ldb_owner
 166         del self.ldb_user
 167
 168     # Make sure top OU is deleted (and so everything under it)
 169     def assert_top_ou_deleted(self):
 170         res = self.ldb_admin.search(self.base_dn,
 171                                     expression="(distinguishedName=%s,%s)" % (
 172                                         "OU=test_add_ou1", self.base_dn))
 173         self.assertEqual(len(res), 0)
 174
 175     def test_add_u1(self):
 176         """Testing OU with the rights of Doman Admin not creator of the OU """
 177         self.assert_top_ou_deleted()
 178         # Change descriptor for top level OU
 179         self.ldb_owner.create_ou("OU=test_add_ou1," + self.base_dn)
 180         self.ldb_owner.create_ou("OU=test_add_ou2,OU=test_add_ou1," + self.base_dn)
 181         user_sid = self.sd_utils.get_object_sid(self.get_user_dn(self.usr_admin_not_owner))
 182         mod = "(D;CI;WPCC;;;%s)" % str(user_sid)
 183         self.sd_utils.dacl_add_ace("OU=test_add_ou1," + self.base_dn, mod)
 184         # Test user and group creation with another domain admin's credentials
 185         self.ldb_notowner.newuser(self.test_user1, self.user_pass, userou=self.ou2)
 186         self.ldb_notowner.newgroup("test_add_group1", groupou="OU=test_add_ou2,OU=test_add_ou1",
 187                                    grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
 188         # Make sure we HAVE created the two objects -- user and group
 189         # !!! We should not be able to do that, but however beacuse of ACE ordering our inherited Deny ACE
 190         # !!! comes after explicit (A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA) that comes from somewhere
 191         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))
 192         self.assertTrue(len(res) > 0)
 193         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))
 194         self.assertTrue(len(res) > 0)
 195
 196     def test_add_u2(self):
 197         """Testing OU with the regular user that has no rights granted over the OU """
 198         self.assert_top_ou_deleted()
 199         # Create a parent-child OU structure with domain admin credentials
 200         self.ldb_owner.create_ou("OU=test_add_ou1," + self.base_dn)
 201         self.ldb_owner.create_ou("OU=test_add_ou2,OU=test_add_ou1," + self.base_dn)
 202         # Test user and group creation with regular user credentials
 203         try:
 204             self.ldb_user.newuser(self.test_user1, self.user_pass, userou=self.ou2)
 205             self.ldb_user.newgroup("test_add_group1", groupou="OU=test_add_ou2,OU=test_add_ou1",
 206                                    grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
 207         except LdbError as e:
 208             (num, _) = e.args
 209             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
 210         else:
 211             self.fail()
 212         # Make sure we HAVEN'T created any of two objects -- user or group
 213         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))
 214         self.assertEqual(len(res), 0)
 215         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))
 216         self.assertEqual(len(res), 0)
 217
 218     def test_add_u3(self):
 219         """Testing OU with the rights of regular user granted the right 'Create User child objects' """
 220         self.assert_top_ou_deleted()
 221         # Change descriptor for top level OU
 222         self.ldb_owner.create_ou("OU=test_add_ou1," + self.base_dn)
 223         user_sid = self.sd_utils.get_object_sid(self.get_user_dn(self.regular_user))
 224         mod = "(OA;CI;CC;bf967aba-0de6-11d0-a285-00aa003049e2;;%s)" % str(user_sid)
 225         self.sd_utils.dacl_add_ace("OU=test_add_ou1," + self.base_dn, mod)
 226         self.ldb_owner.create_ou("OU=test_add_ou2,OU=test_add_ou1," + self.base_dn)
 227         # Test user and group creation with granted user only to one of the objects
 228         self.ldb_user.newuser(self.test_user1, self.user_pass, userou=self.ou2, setpassword=False)
 229         try:
 230             self.ldb_user.newgroup("test_add_group1", groupou="OU=test_add_ou2,OU=test_add_ou1",
 231                                    grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
 232         except LdbError as e1:
 233             (num, _) = e1.args
 234             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
 235         else:
 236             self.fail()
 237         # Make sure we HAVE created the one of two objects -- user
 238         res = self.ldb_admin.search(self.base_dn,
 239                                     expression="(distinguishedName=%s,%s)" %
 240                                     ("CN=test_add_user1,OU=test_add_ou2,OU=test_add_ou1",
 241                                      self.base_dn))
 242         self.assertNotEqual(len(res), 0)
 243         res = self.ldb_admin.search(self.base_dn,
 244                                     expression="(distinguishedName=%s,%s)" %
 245                                     ("CN=test_add_group1,OU=test_add_ou2,OU=test_add_ou1",
 246                                      self.base_dn))
 247         self.assertEqual(len(res), 0)
 248
 249     def test_add_u4(self):
 250         """ 4 Testing OU with the rights of Doman Admin creator of the OU"""
 251         self.assert_top_ou_deleted()
 252         self.ldb_owner.create_ou("OU=test_add_ou1," + self.base_dn)
 253         self.ldb_owner.create_ou("OU=test_add_ou2,OU=test_add_ou1," + self.base_dn)
 254         self.ldb_owner.newuser(self.test_user1, self.user_pass, userou=self.ou2)
 255         self.ldb_owner.newgroup("test_add_group1", groupou="OU=test_add_ou2,OU=test_add_ou1",
 256                                 grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
 257         # Make sure we have successfully created the two objects -- user and group
 258         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))
 259         self.assertTrue(len(res) > 0)
 260         res = self.ldb_admin.search(self.base_dn,
 261                                     expression="(distinguishedName=%s,%s)" % ("CN=test_add_group1,OU=test_add_ou2,OU=test_add_ou1", self.base_dn))
 262         self.assertTrue(len(res) > 0)
 263
 264     def test_add_anonymous(self):
 265         """Test add operation with anonymous user"""
 266         anonymous = SamDB(url=ldaphost, credentials=self.creds_tmp, lp=lp)
 267         try:
 268             anonymous.newuser("test_add_anonymous", self.user_pass)
 269         except LdbError as e2:
 270             (num, _) = e2.args
 271             self.assertEquals(num, ERR_OPERATIONS_ERROR)
 272         else:
 273             self.fail()
 274
 275 # tests on ldap modify operations
 276
 277
 278 class AclModifyTests(AclTests):
 279
 280     def setUp(self):
 281         super(AclModifyTests, self).setUp()
 282         self.user_with_wp = "acl_mod_user1"
 283         self.user_with_sm = "acl_mod_user2"
 284         self.user_with_group_sm = "acl_mod_user3"
 285         self.ldb_admin.newuser(self.user_with_wp, self.user_pass)
 286         self.ldb_admin.newuser(self.user_with_sm, self.user_pass)
 287         self.ldb_admin.newuser(self.user_with_group_sm, self.user_pass)
 288         self.ldb_user = self.get_ldb_connection(self.user_with_wp, self.user_pass)
 289         self.ldb_user2 = self.get_ldb_connection(self.user_with_sm, self.user_pass)
 290         self.ldb_user3 = self.get_ldb_connection(self.user_with_group_sm, self.user_pass)
 291         self.user_sid = self.sd_utils.get_object_sid(self.get_user_dn(self.user_with_wp))
 292         self.ldb_admin.newgroup("test_modify_group2", grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
 293         self.ldb_admin.newgroup("test_modify_group3", grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
 294         self.ldb_admin.newuser("test_modify_user2", self.user_pass)
 295
 296     def tearDown(self):
 297         super(AclModifyTests, self).tearDown()
 298         delete_force(self.ldb_admin, self.get_user_dn("test_modify_user1"))
 299         delete_force(self.ldb_admin, "CN=test_modify_group1,CN=Users," + self.base_dn)
 300         delete_force(self.ldb_admin, "CN=test_modify_group2,CN=Users," + self.base_dn)
 301         delete_force(self.ldb_admin, "CN=test_modify_group3,CN=Users," + self.base_dn)
 302         delete_force(self.ldb_admin, "OU=test_modify_ou1," + self.base_dn)
 303         delete_force(self.ldb_admin, self.get_user_dn(self.user_with_wp))
 304         delete_force(self.ldb_admin, self.get_user_dn(self.user_with_sm))
 305         delete_force(self.ldb_admin, self.get_user_dn(self.user_with_group_sm))
 306         delete_force(self.ldb_admin, self.get_user_dn("test_modify_user2"))
 307         delete_force(self.ldb_admin, self.get_user_dn("test_anonymous"))
 308
 309         del self.ldb_user
 310         del self.ldb_user2
 311         del self.ldb_user3
 312
 313     def test_modify_u1(self):
 314         """5 Modify one attribute if you have DS_WRITE_PROPERTY for it"""
 315         mod = "(OA;;WP;bf967953-0de6-11d0-a285-00aa003049e2;;%s)" % str(self.user_sid)
 316         # First test object -- User
 317         print("Testing modify on User object")
 318         self.ldb_admin.newuser("test_modify_user1", self.user_pass)
 319         self.sd_utils.dacl_add_ace(self.get_user_dn("test_modify_user1"), mod)
 320         ldif = """
 321 dn: """ + self.get_user_dn("test_modify_user1") + """
 322 changetype: modify
 323 replace: displayName
 324 displayName: test_changed"""
 325         self.ldb_user.modify_ldif(ldif)
 326         res = self.ldb_admin.search(self.base_dn,
 327                                     expression="(distinguishedName=%s)" % self.get_user_dn("test_modify_user1"))
 328         self.assertEqual(res[0]["displayName"][0], "test_changed")
 329         # Second test object -- Group
 330         print("Testing modify on Group object")
 331         self.ldb_admin.newgroup("test_modify_group1",
 332                                 grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
 333         self.sd_utils.dacl_add_ace("CN=test_modify_group1,CN=Users," + self.base_dn, mod)
 334         ldif = """
 335 dn: CN=test_modify_group1,CN=Users,""" + self.base_dn + """
 336 changetype: modify
 337 replace: displayName
 338 displayName: test_changed"""
 339         self.ldb_user.modify_ldif(ldif)
 340         res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" % str("CN=test_modify_group1,CN=Users," + self.base_dn))
 341         self.assertEqual(res[0]["displayName"][0], "test_changed")
 342         # Third test object -- Organizational Unit
 343         print("Testing modify on OU object")
 344         #delete_force(self.ldb_admin, "OU=test_modify_ou1," + self.base_dn)
 345         self.ldb_admin.create_ou("OU=test_modify_ou1," + self.base_dn)
 346         self.sd_utils.dacl_add_ace("OU=test_modify_ou1," + self.base_dn, mod)
 347         ldif = """
 348 dn: OU=test_modify_ou1,""" + self.base_dn + """
 349 changetype: modify
 350 replace: displayName
 351 displayName: test_changed"""
 352         self.ldb_user.modify_ldif(ldif)
 353         res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" % str("OU=test_modify_ou1," + self.base_dn))
 354         self.assertEqual(res[0]["displayName"][0], "test_changed")
 355
 356     def test_modify_u2(self):
 357         """6 Modify two attributes as you have DS_WRITE_PROPERTY granted only for one of them"""
 358         mod = "(OA;;WP;bf967953-0de6-11d0-a285-00aa003049e2;;%s)" % str(self.user_sid)
 359         # First test object -- User
 360         print("Testing modify on User object")
 361         #delete_force(self.ldb_admin, self.get_user_dn("test_modify_user1"))
 362         self.ldb_admin.newuser("test_modify_user1", self.user_pass)
 363         self.sd_utils.dacl_add_ace(self.get_user_dn("test_modify_user1"), mod)
 364         # Modify on attribute you have rights for
 365         ldif = """
 366 dn: """ + self.get_user_dn("test_modify_user1") + """
 367 changetype: modify
 368 replace: displayName
 369 displayName: test_changed"""
 370         self.ldb_user.modify_ldif(ldif)
 371         res = self.ldb_admin.search(self.base_dn,
 372                                     expression="(distinguishedName=%s)" %
 373                                     self.get_user_dn("test_modify_user1"))
 374         self.assertEqual(res[0]["displayName"][0], "test_changed")
 375         # Modify on attribute you do not have rights for granted
 376         ldif = """
 377 dn: """ + self.get_user_dn("test_modify_user1") + """
 378 changetype: modify
 379 replace: url
 380 url: www.samba.org"""
 381         try:
 382             self.ldb_user.modify_ldif(ldif)
 383         except LdbError as e3:
 384             (num, _) = e3.args
 385             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
 386         else:
 387             # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
 388             self.fail()
 389         # Second test object -- Group
 390         print("Testing modify on Group object")
 391         self.ldb_admin.newgroup("test_modify_group1",
 392                                 grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
 393         self.sd_utils.dacl_add_ace("CN=test_modify_group1,CN=Users," + self.base_dn, mod)
 394         ldif = """
 395 dn: CN=test_modify_group1,CN=Users,""" + self.base_dn + """
 396 changetype: modify
 397 replace: displayName
 398 displayName: test_changed"""
 399         self.ldb_user.modify_ldif(ldif)
 400         res = self.ldb_admin.search(self.base_dn,
 401                                     expression="(distinguishedName=%s)" %
 402                                     str("CN=test_modify_group1,CN=Users," + self.base_dn))
 403         self.assertEqual(res[0]["displayName"][0], "test_changed")
 404         # Modify on attribute you do not have rights for granted
 405         ldif = """
 406 dn: CN=test_modify_group1,CN=Users,""" + self.base_dn + """
 407 changetype: modify
 408 replace: url
 409 url: www.samba.org"""
 410         try:
 411             self.ldb_user.modify_ldif(ldif)
 412         except LdbError as e4:
 413             (num, _) = e4.args
 414             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
 415         else:
 416             # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
 417             self.fail()
 418         # Modify on attribute you do not have rights for granted while also modifying something you do have rights for
 419         ldif = """
 420 dn: CN=test_modify_group1,CN=Users,""" + self.base_dn + """
 421 changetype: modify
 422 replace: url
 423 url: www.samba.org
 424 replace: displayName
 425 displayName: test_changed"""
 426         try:
 427             self.ldb_user.modify_ldif(ldif)
 428         except LdbError as e5:
 429             (num, _) = e5.args
 430             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
 431         else:
 432             # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
 433             self.fail()
 434         # Second test object -- Organizational Unit
 435         print("Testing modify on OU object")
 436         self.ldb_admin.create_ou("OU=test_modify_ou1," + self.base_dn)
 437         self.sd_utils.dacl_add_ace("OU=test_modify_ou1," + self.base_dn, mod)
 438         ldif = """
 439 dn: OU=test_modify_ou1,""" + self.base_dn + """
 440 changetype: modify
 441 replace: displayName
 442 displayName: test_changed"""
 443         self.ldb_user.modify_ldif(ldif)
 444         res = self.ldb_admin.search(self.base_dn,
 445                                     expression="(distinguishedName=%s)" % str("OU=test_modify_ou1,"
 446                                                                               + self.base_dn))
 447         self.assertEqual(res[0]["displayName"][0], "test_changed")
 448         # Modify on attribute you do not have rights for granted
 449         ldif = """
 450 dn: OU=test_modify_ou1,""" + self.base_dn + """
 451 changetype: modify
 452 replace: url
 453 url: www.samba.org"""
 454         try:
 455             self.ldb_user.modify_ldif(ldif)
 456         except LdbError as e6:
 457             (num, _) = e6.args
 458             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
 459         else:
 460             # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
 461             self.fail()
 462
 463     def test_modify_u3(self):
 464         """7 Modify one attribute as you have no what so ever rights granted"""
 465         # First test object -- User
 466         print("Testing modify on User object")
 467         self.ldb_admin.newuser("test_modify_user1", self.user_pass)
 468         # Modify on attribute you do not have rights for granted
 469         ldif = """
 470 dn: """ + self.get_user_dn("test_modify_user1") + """
 471 changetype: modify
 472 replace: url
 473 url: www.samba.org"""
 474         try:
 475             self.ldb_user.modify_ldif(ldif)
 476         except LdbError as e7:
 477             (num, _) = e7.args
 478             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
 479         else:
 480             # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
 481             self.fail()
 482
 483         # Second test object -- Group
 484         print("Testing modify on Group object")
 485         self.ldb_admin.newgroup("test_modify_group1",
 486                                 grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
 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 as e8:
 496             (num, _) = e8.args
 497             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
 498         else:
 499             # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
 500             self.fail()
 501
 502         # Second test object -- Organizational Unit
 503         print("Testing modify on OU object")
 504         #delete_force(self.ldb_admin, "OU=test_modify_ou1," + self.base_dn)
 505         self.ldb_admin.create_ou("OU=test_modify_ou1," + self.base_dn)
 506         # Modify on attribute you do not have rights for granted
 507         ldif = """
 508 dn: OU=test_modify_ou1,""" + self.base_dn + """
 509 changetype: modify
 510 replace: url
 511 url: www.samba.org"""
 512         try:
 513             self.ldb_user.modify_ldif(ldif)
 514         except LdbError as e9:
 515             (num, _) = e9.args
 516             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
 517         else:
 518             # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
 519             self.fail()
 520
 521     def test_modify_u4(self):
 522         """11 Grant WP to PRINCIPAL_SELF and test modify"""
 523         ldif = """
 524 dn: """ + self.get_user_dn(self.user_with_wp) + """
 525 changetype: modify
 526 add: adminDescription
 527 adminDescription: blah blah blah"""
 528         try:
 529             self.ldb_user.modify_ldif(ldif)
 530         except LdbError as e10:
 531             (num, _) = e10.args
 532             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
 533         else:
 534             # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
 535             self.fail()
 536
 537         mod = "(OA;;WP;bf967919-0de6-11d0-a285-00aa003049e2;;PS)"
 538         self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
 539         # Modify on attribute you have rights for
 540         self.ldb_user.modify_ldif(ldif)
 541         res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)"
 542                                     % self.get_user_dn(self.user_with_wp), attrs=["adminDescription"])
 543         self.assertEqual(res[0]["adminDescription"][0], "blah blah blah")
 544
 545     def test_modify_u5(self):
 546         """12 test self membership"""
 547         ldif = """
 548 dn: CN=test_modify_group2,CN=Users,""" + self.base_dn + """
 549 changetype: modify
 550 add: Member
 551 Member: """ + self.get_user_dn(self.user_with_sm)
 552 # the user has no rights granted, this should fail
 553         try:
 554             self.ldb_user2.modify_ldif(ldif)
 555         except LdbError as e11:
 556             (num, _) = e11.args
 557             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
 558         else:
 559             # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
 560             self.fail()
 561
 562 # grant self-membership, should be able to add himself
 563         user_sid = self.sd_utils.get_object_sid(self.get_user_dn(self.user_with_sm))
 564         mod = "(OA;;SW;bf9679c0-0de6-11d0-a285-00aa003049e2;;%s)" % str(user_sid)
 565         self.sd_utils.dacl_add_ace("CN=test_modify_group2,CN=Users," + self.base_dn, mod)
 566         self.ldb_user2.modify_ldif(ldif)
 567         res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)"
 568                                     % ("CN=test_modify_group2,CN=Users," + self.base_dn), attrs=["Member"])
 569         self.assertEqual(res[0]["Member"][0], self.get_user_dn(self.user_with_sm))
 570 # but not other users
 571         ldif = """
 572 dn: CN=test_modify_group2,CN=Users,""" + self.base_dn + """
 573 changetype: modify
 574 add: Member
 575 Member: CN=test_modify_user2,CN=Users,""" + self.base_dn
 576         try:
 577             self.ldb_user2.modify_ldif(ldif)
 578         except LdbError as e12:
 579             (num, _) = e12.args
 580             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
 581         else:
 582             self.fail()
 583
 584     def test_modify_u6(self):
 585         """13 test self membership"""
 586         ldif = """
 587 dn: CN=test_modify_group2,CN=Users,""" + self.base_dn + """
 588 changetype: modify
 589 add: Member
 590 Member: """ + self.get_user_dn(self.user_with_sm) + """
 591 Member: CN=test_modify_user2,CN=Users,""" + self.base_dn
 592
 593 # grant self-membership, should be able to add himself  but not others at the same time
 594         user_sid = self.sd_utils.get_object_sid(self.get_user_dn(self.user_with_sm))
 595         mod = "(OA;;SW;bf9679c0-0de6-11d0-a285-00aa003049e2;;%s)" % str(user_sid)
 596         self.sd_utils.dacl_add_ace("CN=test_modify_group2,CN=Users," + self.base_dn, mod)
 597         try:
 598             self.ldb_user2.modify_ldif(ldif)
 599         except LdbError as e13:
 600             (num, _) = e13.args
 601             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
 602         else:
 603             self.fail()
 604
 605     def test_modify_u7(self):
 606         """13 User with WP modifying Member"""
 607 # a second user is given write property permission
 608         user_sid = self.sd_utils.get_object_sid(self.get_user_dn(self.user_with_wp))
 609         mod = "(A;;WP;;;%s)" % str(user_sid)
 610         self.sd_utils.dacl_add_ace("CN=test_modify_group2,CN=Users," + self.base_dn, mod)
 611         ldif = """
 612 dn: CN=test_modify_group2,CN=Users,""" + self.base_dn + """
 613 changetype: modify
 614 add: Member
 615 Member: """ + self.get_user_dn(self.user_with_wp)
 616         self.ldb_user.modify_ldif(ldif)
 617         res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)"
 618                                     % ("CN=test_modify_group2,CN=Users," + self.base_dn), attrs=["Member"])
 619         self.assertEqual(res[0]["Member"][0], self.get_user_dn(self.user_with_wp))
 620         ldif = """
 621 dn: CN=test_modify_group2,CN=Users,""" + self.base_dn + """
 622 changetype: modify
 623 delete: Member"""
 624         self.ldb_user.modify_ldif(ldif)
 625         ldif = """
 626 dn: CN=test_modify_group2,CN=Users,""" + self.base_dn + """
 627 changetype: modify
 628 add: Member
 629 Member: CN=test_modify_user2,CN=Users,""" + self.base_dn
 630         self.ldb_user.modify_ldif(ldif)
 631         res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)"
 632                                     % ("CN=test_modify_group2,CN=Users," + self.base_dn), attrs=["Member"])
 633         self.assertEqual(res[0]["Member"][0], "CN=test_modify_user2,CN=Users," + self.base_dn)
 634
 635     def test_modify_anonymous(self):
 636         """Test add operation with anonymous user"""
 637         anonymous = SamDB(url=ldaphost, credentials=self.creds_tmp, lp=lp)
 638         self.ldb_admin.newuser("test_anonymous", "samba123@")
 639         m = Message()
 640         m.dn = Dn(anonymous, self.get_user_dn("test_anonymous"))
 641
 642         m["description"] = MessageElement("sambauser2",
 643                                           FLAG_MOD_ADD,
 644                                           "description")
 645         try:
 646             anonymous.modify(m)
 647         except LdbError as e14:
 648             (num, _) = e14.args
 649             self.assertEquals(num, ERR_OPERATIONS_ERROR)
 650         else:
 651             self.fail()
 652
 653 # enable these when we have search implemented
 654
 655
 656 class AclSearchTests(AclTests):
 657
 658     def setUp(self):
 659         super(AclSearchTests, self).setUp()
 660
 661         # permit password changes during this test
 662         PasswordCommon.allow_password_changes(self, self.ldb_admin)
 663
 664         self.u1 = "search_u1"
 665         self.u2 = "search_u2"
 666         self.u3 = "search_u3"
 667         self.group1 = "group1"
 668         self.ldb_admin.newuser(self.u1, self.user_pass)
 669         self.ldb_admin.newuser(self.u2, self.user_pass)
 670         self.ldb_admin.newuser(self.u3, self.user_pass)
 671         self.ldb_admin.newgroup(self.group1, grouptype=samba.dsdb.GTYPE_SECURITY_GLOBAL_GROUP)
 672         self.ldb_admin.add_remove_group_members(self.group1, [self.u2],
 673                                                 add_members_operation=True)
 674         self.ldb_user = self.get_ldb_connection(self.u1, self.user_pass)
 675         self.ldb_user2 = self.get_ldb_connection(self.u2, self.user_pass)
 676         self.ldb_user3 = self.get_ldb_connection(self.u3, self.user_pass)
 677         self.full_list = [Dn(self.ldb_admin, "OU=ou2,OU=ou1," + self.base_dn),
 678                           Dn(self.ldb_admin, "OU=ou1," + self.base_dn),
 679                           Dn(self.ldb_admin, "OU=ou3,OU=ou2,OU=ou1," + self.base_dn),
 680                           Dn(self.ldb_admin, "OU=ou4,OU=ou2,OU=ou1," + self.base_dn),
 681                           Dn(self.ldb_admin, "OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn),
 682                           Dn(self.ldb_admin, "OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn)]
 683         self.user_sid = self.sd_utils.get_object_sid(self.get_user_dn(self.u1))
 684         self.group_sid = self.sd_utils.get_object_sid(self.get_user_dn(self.group1))
 685
 686     def create_clean_ou(self, object_dn):
 687         """ Base repeating setup for unittests to follow """
 688         res = self.ldb_admin.search(base=self.base_dn, scope=SCOPE_SUBTREE,
 689                                     expression="distinguishedName=%s" % object_dn)
 690         # Make sure top testing OU has been deleted before starting the test
 691         self.assertEqual(len(res), 0)
 692         self.ldb_admin.create_ou(object_dn)
 693         desc_sddl = self.sd_utils.get_sd_as_sddl(object_dn)
 694         # Make sure there are inheritable ACEs initially
 695         self.assertTrue("CI" in desc_sddl or "OI" in desc_sddl)
 696         # Find and remove all inherit ACEs
 697         res = re.findall("\(.*?\)", desc_sddl)
 698         res = [x for x in res if ("CI" in x) or ("OI" in x)]
 699         for x in res:
 700             desc_sddl = desc_sddl.replace(x, "")
 701         # Add flag 'protected' in both DACL and SACL so no inherit ACEs
 702         # can propagate from above
 703         # remove SACL, we are not interested
 704         desc_sddl = desc_sddl.replace(":AI", ":AIP")
 705         self.sd_utils.modify_sd_on_dn(object_dn, desc_sddl)
 706         # Verify all inheritable ACEs are gone
 707         desc_sddl = self.sd_utils.get_sd_as_sddl(object_dn)
 708         self.assertFalse("CI" in desc_sddl)
 709         self.assertFalse("OI" in desc_sddl)
 710
 711     def tearDown(self):
 712         super(AclSearchTests, self).tearDown()
 713         delete_force(self.ldb_admin, "OU=test_search_ou2,OU=test_search_ou1," + self.base_dn)
 714         delete_force(self.ldb_admin, "OU=test_search_ou1," + self.base_dn)
 715         delete_force(self.ldb_admin, "OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn)
 716         delete_force(self.ldb_admin, "OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn)
 717         delete_force(self.ldb_admin, "OU=ou4,OU=ou2,OU=ou1," + self.base_dn)
 718         delete_force(self.ldb_admin, "OU=ou3,OU=ou2,OU=ou1," + self.base_dn)
 719         delete_force(self.ldb_admin, "OU=ou2,OU=ou1," + self.base_dn)
 720         delete_force(self.ldb_admin, "OU=ou1," + self.base_dn)
 721         delete_force(self.ldb_admin, self.get_user_dn("search_u1"))
 722         delete_force(self.ldb_admin, self.get_user_dn("search_u2"))
 723         delete_force(self.ldb_admin, self.get_user_dn("search_u3"))
 724         delete_force(self.ldb_admin, self.get_user_dn("group1"))
 725
 726         del self.ldb_user
 727         del self.ldb_user2
 728         del self.ldb_user3
 729
 730     def test_search_anonymous1(self):
 731         """Verify access of rootDSE with the correct request"""
 732         anonymous = SamDB(url=ldaphost, credentials=self.creds_tmp, lp=lp)
 733         res = anonymous.search("", expression="(objectClass=*)", scope=SCOPE_BASE)
 734         self.assertEquals(len(res), 1)
 735         # verify some of the attributes
 736         # don't care about values
 737         self.assertTrue("ldapServiceName" in res[0])
 738         self.assertTrue("namingContexts" in res[0])
 739         self.assertTrue("isSynchronized" in res[0])
 740         self.assertTrue("dsServiceName" in res[0])
 741         self.assertTrue("supportedSASLMechanisms" in res[0])
 742         self.assertTrue("isGlobalCatalogReady" in res[0])
 743         self.assertTrue("domainControllerFunctionality" in res[0])
 744         self.assertTrue("serverName" in res[0])
 745
 746     def test_search_anonymous2(self):
 747         """Make sure we cannot access anything else"""
 748         anonymous = SamDB(url=ldaphost, credentials=self.creds_tmp, lp=lp)
 749         try:
 750             res = anonymous.search("", expression="(objectClass=*)", scope=SCOPE_SUBTREE)
 751         except LdbError as e15:
 752             (num, _) = e15.args
 753             self.assertEquals(num, ERR_OPERATIONS_ERROR)
 754         else:
 755             self.fail()
 756         try:
 757             res = anonymous.search(self.base_dn, expression="(objectClass=*)", scope=SCOPE_SUBTREE)
 758         except LdbError as e16:
 759             (num, _) = e16.args
 760             self.assertEquals(num, ERR_OPERATIONS_ERROR)
 761         else:
 762             self.fail()
 763         try:
 764             res = anonymous.search(anonymous.get_config_basedn(), expression="(objectClass=*)",
 765                                    scope=SCOPE_SUBTREE)
 766         except LdbError as e17:
 767             (num, _) = e17.args
 768             self.assertEquals(num, ERR_OPERATIONS_ERROR)
 769         else:
 770             self.fail()
 771
 772     def test_search_anonymous3(self):
 773         """Set dsHeuristics and repeat"""
 774         self.ldb_admin.set_dsheuristics("0000002")
 775         self.ldb_admin.create_ou("OU=test_search_ou1," + self.base_dn)
 776         mod = "(A;CI;LC;;;AN)"
 777         self.sd_utils.dacl_add_ace("OU=test_search_ou1," + self.base_dn, mod)
 778         self.ldb_admin.create_ou("OU=test_search_ou2,OU=test_search_ou1," + self.base_dn)
 779         anonymous = SamDB(url=ldaphost, credentials=self.creds_tmp, lp=lp)
 780         res = anonymous.search("OU=test_search_ou2,OU=test_search_ou1," + self.base_dn,
 781                                expression="(objectClass=*)", scope=SCOPE_SUBTREE)
 782         self.assertEquals(len(res), 1)
 783         self.assertTrue("dn" in res[0])
 784         self.assertTrue(res[0]["dn"] == Dn(self.ldb_admin,
 785                                            "OU=test_search_ou2,OU=test_search_ou1," + self.base_dn))
 786         res = anonymous.search(anonymous.get_config_basedn(), expression="(objectClass=*)",
 787                                scope=SCOPE_SUBTREE)
 788         self.assertEquals(len(res), 1)
 789         self.assertTrue("dn" in res[0])
 790         self.assertTrue(res[0]["dn"] == Dn(self.ldb_admin, self.configuration_dn))
 791
 792     def test_search1(self):
 793         """Make sure users can see us if given LC to user and group"""
 794         self.create_clean_ou("OU=ou1," + self.base_dn)
 795         mod = "(A;;LC;;;%s)(A;;LC;;;%s)" % (str(self.user_sid), str(self.group_sid))
 796         self.sd_utils.dacl_add_ace("OU=ou1," + self.base_dn, mod)
 797         tmp_desc = security.descriptor.from_sddl("D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)" + mod,
 798                                                  self.domain_sid)
 799         self.ldb_admin.create_ou("OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
 800         self.ldb_admin.create_ou("OU=ou3,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
 801         self.ldb_admin.create_ou("OU=ou4,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
 802         self.ldb_admin.create_ou("OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
 803         self.ldb_admin.create_ou("OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
 804
 805         # regular users must see only ou1 and ou2
 806         res = self.ldb_user3.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
 807                                     scope=SCOPE_SUBTREE)
 808         self.assertEquals(len(res), 2)
 809         ok_list = [Dn(self.ldb_admin, "OU=ou2,OU=ou1," + self.base_dn),
 810                    Dn(self.ldb_admin, "OU=ou1," + self.base_dn)]
 811
 812         res_list = [x["dn"] for x in res if x["dn"] in ok_list]
 813         self.assertEquals(sorted(res_list), sorted(ok_list))
 814
 815         # these users should see all ous
 816         res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
 817                                    scope=SCOPE_SUBTREE)
 818         self.assertEquals(len(res), 6)
 819         res_list = [x["dn"] for x in res if x["dn"] in self.full_list]
 820         self.assertEquals(sorted(res_list), sorted(self.full_list))
 821
 822         res = self.ldb_user2.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
 823                                     scope=SCOPE_SUBTREE)
 824         self.assertEquals(len(res), 6)
 825         res_list = [x["dn"] for x in res if x["dn"] in self.full_list]
 826         self.assertEquals(sorted(res_list), sorted(self.full_list))
 827
 828     def test_search2(self):
 829         """Make sure users can't see us if access is explicitly denied"""
 830         self.create_clean_ou("OU=ou1," + self.base_dn)
 831         self.ldb_admin.create_ou("OU=ou2,OU=ou1," + self.base_dn)
 832         self.ldb_admin.create_ou("OU=ou3,OU=ou2,OU=ou1," + self.base_dn)
 833         self.ldb_admin.create_ou("OU=ou4,OU=ou2,OU=ou1," + self.base_dn)
 834         self.ldb_admin.create_ou("OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn)
 835         self.ldb_admin.create_ou("OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn)
 836         mod = "(D;;LC;;;%s)(D;;LC;;;%s)" % (str(self.user_sid), str(self.group_sid))
 837         self.sd_utils.dacl_add_ace("OU=ou2,OU=ou1," + self.base_dn, mod)
 838         res = self.ldb_user3.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
 839                                     scope=SCOPE_SUBTREE)
 840         # this user should see all ous
 841         res_list = [x["dn"] for x in res if x["dn"] in self.full_list]
 842         self.assertEquals(sorted(res_list), sorted(self.full_list))
 843
 844         # these users should see ou1, 2, 5 and 6 but not 3 and 4
 845         res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
 846                                    scope=SCOPE_SUBTREE)
 847         ok_list = [Dn(self.ldb_admin, "OU=ou2,OU=ou1," + self.base_dn),
 848                    Dn(self.ldb_admin, "OU=ou1," + self.base_dn),
 849                    Dn(self.ldb_admin, "OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn),
 850                    Dn(self.ldb_admin, "OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn)]
 851         res_list = [x["dn"] for x in res if x["dn"] in ok_list]
 852         self.assertEquals(sorted(res_list), sorted(ok_list))
 853
 854         res = self.ldb_user2.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
 855                                     scope=SCOPE_SUBTREE)
 856         self.assertEquals(len(res), 4)
 857         res_list = [x["dn"] for x in res if x["dn"] in ok_list]
 858         self.assertEquals(sorted(res_list), sorted(ok_list))
 859
 860     def test_search3(self):
 861         """Make sure users can't see ous if access is explicitly denied - 2"""
 862         self.create_clean_ou("OU=ou1," + self.base_dn)
 863         mod = "(A;CI;LC;;;%s)(A;CI;LC;;;%s)" % (str(self.user_sid), str(self.group_sid))
 864         self.sd_utils.dacl_add_ace("OU=ou1," + self.base_dn, mod)
 865         tmp_desc = security.descriptor.from_sddl("D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)" + mod,
 866                                                  self.domain_sid)
 867         self.ldb_admin.create_ou("OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
 868         self.ldb_admin.create_ou("OU=ou3,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
 869         self.ldb_admin.create_ou("OU=ou4,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
 870         self.ldb_admin.create_ou("OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
 871         self.ldb_admin.create_ou("OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
 872
 873         print("Testing correct behavior on nonaccessible search base")
 874         try:
 875             self.ldb_user3.search("OU=ou3,OU=ou2,OU=ou1," + self.base_dn, expression="(objectClass=*)",
 876                                   scope=SCOPE_BASE)
 877         except LdbError as e18:
 878             (num, _) = e18.args
 879             self.assertEquals(num, ERR_NO_SUCH_OBJECT)
 880         else:
 881             self.fail()
 882
 883         mod = "(D;;LC;;;%s)(D;;LC;;;%s)" % (str(self.user_sid), str(self.group_sid))
 884         self.sd_utils.dacl_add_ace("OU=ou2,OU=ou1," + self.base_dn, mod)
 885
 886         ok_list = [Dn(self.ldb_admin, "OU=ou2,OU=ou1," + self.base_dn),
 887                    Dn(self.ldb_admin, "OU=ou1," + self.base_dn)]
 888
 889         res = self.ldb_user3.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
 890                                     scope=SCOPE_SUBTREE)
 891         res_list = [x["dn"] for x in res if x["dn"] in ok_list]
 892         self.assertEquals(sorted(res_list), sorted(ok_list))
 893
 894         ok_list = [Dn(self.ldb_admin, "OU=ou2,OU=ou1," + self.base_dn),
 895                    Dn(self.ldb_admin, "OU=ou1," + self.base_dn),
 896                    Dn(self.ldb_admin, "OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn),
 897                    Dn(self.ldb_admin, "OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn)]
 898
 899         # should not see ou3 and ou4, but should see ou5 and ou6
 900         res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
 901                                    scope=SCOPE_SUBTREE)
 902         self.assertEquals(len(res), 4)
 903         res_list = [x["dn"] for x in res if x["dn"] in ok_list]
 904         self.assertEquals(sorted(res_list), sorted(ok_list))
 905
 906         res = self.ldb_user2.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
 907                                     scope=SCOPE_SUBTREE)
 908         self.assertEquals(len(res), 4)
 909         res_list = [x["dn"] for x in res if x["dn"] in ok_list]
 910         self.assertEquals(sorted(res_list), sorted(ok_list))
 911
 912     def test_search4(self):
 913         """There is no difference in visibility if the user is also creator"""
 914         self.create_clean_ou("OU=ou1," + self.base_dn)
 915         mod = "(A;CI;CC;;;%s)" % (str(self.user_sid))
 916         self.sd_utils.dacl_add_ace("OU=ou1," + self.base_dn, mod)
 917         tmp_desc = security.descriptor.from_sddl("D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)" + mod,
 918                                                  self.domain_sid)
 919         self.ldb_user.create_ou("OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
 920         self.ldb_user.create_ou("OU=ou3,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
 921         self.ldb_user.create_ou("OU=ou4,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
 922         self.ldb_user.create_ou("OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
 923         self.ldb_user.create_ou("OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
 924
 925         ok_list = [Dn(self.ldb_admin, "OU=ou2,OU=ou1," + self.base_dn),
 926                    Dn(self.ldb_admin, "OU=ou1," + self.base_dn)]
 927         res = self.ldb_user3.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
 928                                     scope=SCOPE_SUBTREE)
 929         self.assertEquals(len(res), 2)
 930         res_list = [x["dn"] for x in res if x["dn"] in ok_list]
 931         self.assertEquals(sorted(res_list), sorted(ok_list))
 932
 933         res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
 934                                    scope=SCOPE_SUBTREE)
 935         self.assertEquals(len(res), 2)
 936         res_list = [x["dn"] for x in res if x["dn"] in ok_list]
 937         self.assertEquals(sorted(res_list), sorted(ok_list))
 938
 939     def test_search5(self):
 940         """Make sure users can see only attributes they are allowed to see"""
 941         self.create_clean_ou("OU=ou1," + self.base_dn)
 942         mod = "(A;CI;LC;;;%s)" % (str(self.user_sid))
 943         self.sd_utils.dacl_add_ace("OU=ou1," + self.base_dn, mod)
 944         tmp_desc = security.descriptor.from_sddl("D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)" + mod,
 945                                                  self.domain_sid)
 946         self.ldb_admin.create_ou("OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
 947         # assert user can only see dn
 948         res = self.ldb_user.search("OU=ou2,OU=ou1," + self.base_dn, expression="(objectClass=*)",
 949                                    scope=SCOPE_SUBTREE)
 950         ok_list = ['dn']
 951         self.assertEquals(len(res), 1)
 952         res_list = list(res[0].keys())
 953         self.assertEquals(res_list, ok_list)
 954
 955         res = self.ldb_user.search("OU=ou2,OU=ou1," + self.base_dn, expression="(objectClass=*)",
 956                                    scope=SCOPE_BASE, attrs=["ou"])
 957
 958         self.assertEquals(len(res), 1)
 959         res_list = list(res[0].keys())
 960         self.assertEquals(res_list, ok_list)
 961
 962         # give read property on ou and assert user can only see dn and ou
 963         mod = "(OA;;RP;bf9679f0-0de6-11d0-a285-00aa003049e2;;%s)" % (str(self.user_sid))
 964         self.sd_utils.dacl_add_ace("OU=ou1," + self.base_dn, mod)
 965         self.sd_utils.dacl_add_ace("OU=ou2,OU=ou1," + self.base_dn, mod)
 966         res = self.ldb_user.search("OU=ou2,OU=ou1," + self.base_dn, expression="(objectClass=*)",
 967                                    scope=SCOPE_SUBTREE)
 968         ok_list = ['dn', 'ou']
 969         self.assertEquals(len(res), 1)
 970         res_list = list(res[0].keys())
 971         self.assertEquals(sorted(res_list), sorted(ok_list))
 972
 973         # give read property on Public Information and assert user can see ou and other members
 974         mod = "(OA;;RP;e48d0154-bcf8-11d1-8702-00c04fb96050;;%s)" % (str(self.user_sid))
 975         self.sd_utils.dacl_add_ace("OU=ou1," + self.base_dn, mod)
 976         self.sd_utils.dacl_add_ace("OU=ou2,OU=ou1," + self.base_dn, mod)
 977         res = self.ldb_user.search("OU=ou2,OU=ou1," + self.base_dn, expression="(objectClass=*)",
 978                                    scope=SCOPE_SUBTREE)
 979
 980         ok_list = ['dn', 'objectClass', 'ou', 'distinguishedName', 'name', 'objectGUID', 'objectCategory']
 981         res_list = list(res[0].keys())
 982         self.assertEquals(sorted(res_list), sorted(ok_list))
 983
 984     def test_search6(self):
 985         """If an attribute that cannot be read is used in a filter, it is as if the attribute does not exist"""
 986         self.create_clean_ou("OU=ou1," + self.base_dn)
 987         mod = "(A;CI;LCCC;;;%s)" % (str(self.user_sid))
 988         self.sd_utils.dacl_add_ace("OU=ou1," + self.base_dn, mod)
 989         tmp_desc = security.descriptor.from_sddl("D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)" + mod,
 990                                                  self.domain_sid)
 991         self.ldb_admin.create_ou("OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
 992         self.ldb_user.create_ou("OU=ou3,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
 993
 994         res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(ou=ou3)",
 995                                    scope=SCOPE_SUBTREE)
 996         # nothing should be returned as ou is not accessible
 997         self.assertEquals(len(res), 0)
 998
 999         # give read property on ou and assert user can only see dn and ou
1000         mod = "(OA;;RP;bf9679f0-0de6-11d0-a285-00aa003049e2;;%s)" % (str(self.user_sid))
1001         self.sd_utils.dacl_add_ace("OU=ou3,OU=ou2,OU=ou1," + self.base_dn, mod)
1002         res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(ou=ou3)",
1003                                    scope=SCOPE_SUBTREE)
1004         self.assertEquals(len(res), 1)
1005         ok_list = ['dn', 'ou']
1006         res_list = list(res[0].keys())
1007         self.assertEquals(sorted(res_list), sorted(ok_list))
1008
1009         # give read property on Public Information and assert user can see ou and other members
1010         mod = "(OA;;RP;e48d0154-bcf8-11d1-8702-00c04fb96050;;%s)" % (str(self.user_sid))
1011         self.sd_utils.dacl_add_ace("OU=ou2,OU=ou1," + self.base_dn, mod)
1012         res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(ou=ou2)",
1013                                    scope=SCOPE_SUBTREE)
1014         self.assertEquals(len(res), 1)
1015         ok_list = ['dn', 'objectClass', 'ou', 'distinguishedName', 'name', 'objectGUID', 'objectCategory']
1016         res_list = list(res[0].keys())
1017         self.assertEquals(sorted(res_list), sorted(ok_list))
1018
1019     def assert_search_on_attr(self, dn, samdb, attr, expected_list):
1020
1021         expected_num = len(expected_list)
1022         res = samdb.search(dn, expression="(%s=*)" % attr, scope=SCOPE_SUBTREE)
1023         self.assertEquals(len(res), expected_num)
1024
1025         res_list = [ x["dn"] for x in res if x["dn"] in expected_list ]
1026         self.assertEquals(sorted(res_list), sorted(expected_list))
1027
1028     def test_search7(self):
1029         """Checks object search visibility when users don't have full rights"""
1030         self.create_clean_ou("OU=ou1," + self.base_dn)
1031         mod = "(A;;LC;;;%s)(A;;LC;;;%s)" % (str(self.user_sid),
1032                                             str(self.group_sid))
1033         self.sd_utils.dacl_add_ace("OU=ou1," + self.base_dn, mod)
1034         tmp_desc = security.descriptor.from_sddl("D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)" + mod,
1035                                                  self.domain_sid)
1036         self.ldb_admin.create_ou("OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
1037         self.ldb_admin.create_ou("OU=ou3,OU=ou2,OU=ou1," + self.base_dn,
1038                                  sd=tmp_desc)
1039         self.ldb_admin.create_ou("OU=ou4,OU=ou2,OU=ou1," + self.base_dn,
1040                                  sd=tmp_desc)
1041         self.ldb_admin.create_ou("OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn,
1042                                  sd=tmp_desc)
1043         self.ldb_admin.create_ou("OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn,
1044                                  sd=tmp_desc)
1045
1046         ou2_dn = Dn(self.ldb_admin,  "OU=ou2,OU=ou1," + self.base_dn)
1047         ou1_dn = Dn(self.ldb_admin,  "OU=ou1," + self.base_dn)
1048
1049         # even though unprivileged users can't read these attributes for OU2,
1050         # the object should still be visible in searches, because they have
1051         # 'List Contents' rights still. This isn't really disclosive because
1052         # ALL objects have these attributes
1053         visible_attrs = ["objectClass", "distinguishedName", "name",
1054                          "objectGUID"]
1055         two_objects = [ou2_dn, ou1_dn]
1056
1057         for attr in visible_attrs:
1058             # a regular user should just see the 2 objects
1059             self.assert_search_on_attr(str(ou1_dn), self.ldb_user3, attr,
1060                                        expected_list=two_objects)
1061
1062             # whereas the following users have LC rights for all the objects,
1063             # so they should see them all
1064             self.assert_search_on_attr(str(ou1_dn), self.ldb_user, attr,
1065                                        expected_list=self.full_list)
1066             self.assert_search_on_attr(str(ou1_dn), self.ldb_user2, attr,
1067                                        expected_list=self.full_list)
1068
1069         # however when searching on the following attributes, objects will not
1070         # be visible unless the user has Read Property rights
1071         hidden_attrs = ["objectCategory", "instanceType", "ou", "uSNChanged",
1072                         "uSNCreated", "whenCreated"]
1073         one_object = [ou1_dn]
1074
1075         for attr in hidden_attrs:
1076             self.assert_search_on_attr(str(ou1_dn), self.ldb_user3, attr,
1077                                        expected_list=one_object)
1078             self.assert_search_on_attr(str(ou1_dn), self.ldb_user, attr,
1079                                        expected_list=one_object)
1080             self.assert_search_on_attr(str(ou1_dn), self.ldb_user2, attr,
1081                                        expected_list=one_object)
1082
1083             # admin has RP rights so can still see all the objects
1084             self.assert_search_on_attr(str(ou1_dn), self.ldb_admin, attr,
1085                                        expected_list=self.full_list)
1086
1087
1088 # tests on ldap delete operations
1089
1090
1091 class AclDeleteTests(AclTests):
1092
1093     def setUp(self):
1094         super(AclDeleteTests, self).setUp()
1095         self.regular_user = "acl_delete_user1"
1096         # Create regular user
1097         self.ldb_admin.newuser(self.regular_user, self.user_pass)
1098         self.ldb_user = self.get_ldb_connection(self.regular_user, self.user_pass)
1099
1100     def tearDown(self):
1101         super(AclDeleteTests, self).tearDown()
1102         delete_force(self.ldb_admin, self.get_user_dn("test_delete_user1"))
1103         delete_force(self.ldb_admin, self.get_user_dn(self.regular_user))
1104         delete_force(self.ldb_admin, self.get_user_dn("test_anonymous"))
1105
1106         del self.ldb_user
1107
1108     def test_delete_u1(self):
1109         """User is prohibited by default to delete another User object"""
1110         # Create user that we try to delete
1111         self.ldb_admin.newuser("test_delete_user1", self.user_pass)
1112         # Here delete User object should ALWAYS through exception
1113         try:
1114             self.ldb_user.delete(self.get_user_dn("test_delete_user1"))
1115         except LdbError as e19:
1116             (num, _) = e19.args
1117             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1118         else:
1119             self.fail()
1120
1121     def test_delete_u2(self):
1122         """User's group has RIGHT_DELETE to another User object"""
1123         user_dn = self.get_user_dn("test_delete_user1")
1124         # Create user that we try to delete
1125         self.ldb_admin.newuser("test_delete_user1", self.user_pass)
1126         mod = "(A;;SD;;;AU)"
1127         self.sd_utils.dacl_add_ace(user_dn, mod)
1128         # Try to delete User object
1129         self.ldb_user.delete(user_dn)
1130         res = self.ldb_admin.search(self.base_dn,
1131                                     expression="(distinguishedName=%s)" % user_dn)
1132         self.assertEqual(len(res), 0)
1133
1134     def test_delete_u3(self):
1135         """User indentified by SID has RIGHT_DELETE to another User object"""
1136         user_dn = self.get_user_dn("test_delete_user1")
1137         # Create user that we try to delete
1138         self.ldb_admin.newuser("test_delete_user1", self.user_pass)
1139         mod = "(A;;SD;;;%s)" % self.sd_utils.get_object_sid(self.get_user_dn(self.regular_user))
1140         self.sd_utils.dacl_add_ace(user_dn, mod)
1141         # Try to delete User object
1142         self.ldb_user.delete(user_dn)
1143         res = self.ldb_admin.search(self.base_dn,
1144                                     expression="(distinguishedName=%s)" % user_dn)
1145         self.assertEqual(len(res), 0)
1146
1147     def test_delete_anonymous(self):
1148         """Test add operation with anonymous user"""
1149         anonymous = SamDB(url=ldaphost, credentials=self.creds_tmp, lp=lp)
1150         self.ldb_admin.newuser("test_anonymous", "samba123@")
1151
1152         try:
1153             anonymous.delete(self.get_user_dn("test_anonymous"))
1154         except LdbError as e20:
1155             (num, _) = e20.args
1156             self.assertEquals(num, ERR_OPERATIONS_ERROR)
1157         else:
1158             self.fail()
1159
1160 # tests on ldap rename operations
1161
1162
1163 class AclRenameTests(AclTests):
1164
1165     def setUp(self):
1166         super(AclRenameTests, self).setUp()
1167         self.regular_user = "acl_rename_user1"
1168         self.ou1 = "OU=test_rename_ou1"
1169         self.ou2 = "OU=test_rename_ou2"
1170         self.ou3 = "OU=test_rename_ou3,%s" % self.ou2
1171         self.testuser1 = "test_rename_user1"
1172         self.testuser2 = "test_rename_user2"
1173         self.testuser3 = "test_rename_user3"
1174         self.testuser4 = "test_rename_user4"
1175         self.testuser5 = "test_rename_user5"
1176         # Create regular user
1177         self.ldb_admin.newuser(self.regular_user, self.user_pass)
1178         self.ldb_user = self.get_ldb_connection(self.regular_user, self.user_pass)
1179
1180     def tearDown(self):
1181         super(AclRenameTests, self).tearDown()
1182         # Rename OU3
1183         delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser1, self.ou3, self.base_dn))
1184         delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser2, self.ou3, self.base_dn))
1185         delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser5, self.ou3, self.base_dn))
1186         delete_force(self.ldb_admin, "%s,%s" % (self.ou3, self.base_dn))
1187         # Rename OU2
1188         delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser1, self.ou2, self.base_dn))
1189         delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser2, self.ou2, self.base_dn))
1190         delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser5, self.ou2, self.base_dn))
1191         delete_force(self.ldb_admin, "%s,%s" % (self.ou2, self.base_dn))
1192         # Rename OU1
1193         delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser1, self.ou1, self.base_dn))
1194         delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser2, self.ou1, self.base_dn))
1195         delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser5, self.ou1, self.base_dn))
1196         delete_force(self.ldb_admin, "OU=test_rename_ou3,%s,%s" % (self.ou1, self.base_dn))
1197         delete_force(self.ldb_admin, "%s,%s" % (self.ou1, self.base_dn))
1198         delete_force(self.ldb_admin, self.get_user_dn(self.regular_user))
1199
1200         del self.ldb_user
1201
1202     def test_rename_u1(self):
1203         """Regular user fails to rename 'User object' within single OU"""
1204         # Create OU structure
1205         self.ldb_admin.create_ou("OU=test_rename_ou1," + self.base_dn)
1206         self.ldb_admin.newuser(self.testuser1, self.user_pass, userou=self.ou1)
1207         try:
1208             self.ldb_user.rename("CN=%s,%s,%s" % (self.testuser1, self.ou1, self.base_dn),
1209                                  "CN=%s,%s,%s" % (self.testuser5, self.ou1, self.base_dn))
1210         except LdbError as e21:
1211             (num, _) = e21.args
1212             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1213         else:
1214             self.fail()
1215
1216     def test_rename_u2(self):
1217         """Grant WRITE_PROPERTY to AU so regular user can rename 'User object' within single OU"""
1218         ou_dn = "OU=test_rename_ou1," + self.base_dn
1219         user_dn = "CN=test_rename_user1," + ou_dn
1220         rename_user_dn = "CN=test_rename_user5," + ou_dn
1221         # Create OU structure
1222         self.ldb_admin.create_ou(ou_dn)
1223         self.ldb_admin.newuser(self.testuser1, self.user_pass, userou=self.ou1)
1224         mod = "(A;;WP;;;AU)"
1225         self.sd_utils.dacl_add_ace(user_dn, mod)
1226         # Rename 'User object' having WP to AU
1227         self.ldb_user.rename(user_dn, rename_user_dn)
1228         res = self.ldb_admin.search(self.base_dn,
1229                                     expression="(distinguishedName=%s)" % user_dn)
1230         self.assertEqual(len(res), 0)
1231         res = self.ldb_admin.search(self.base_dn,
1232                                     expression="(distinguishedName=%s)" % rename_user_dn)
1233         self.assertNotEqual(len(res), 0)
1234
1235     def test_rename_u3(self):
1236         """Test rename with rights granted to 'User object' SID"""
1237         ou_dn = "OU=test_rename_ou1," + self.base_dn
1238         user_dn = "CN=test_rename_user1," + ou_dn
1239         rename_user_dn = "CN=test_rename_user5," + ou_dn
1240         # Create OU structure
1241         self.ldb_admin.create_ou(ou_dn)
1242         self.ldb_admin.newuser(self.testuser1, self.user_pass, userou=self.ou1)
1243         sid = self.sd_utils.get_object_sid(self.get_user_dn(self.regular_user))
1244         mod = "(A;;WP;;;%s)" % str(sid)
1245         self.sd_utils.dacl_add_ace(user_dn, mod)
1246         # Rename 'User object' having WP to AU
1247         self.ldb_user.rename(user_dn, rename_user_dn)
1248         res = self.ldb_admin.search(self.base_dn,
1249                                     expression="(distinguishedName=%s)" % user_dn)
1250         self.assertEqual(len(res), 0)
1251         res = self.ldb_admin.search(self.base_dn,
1252                                     expression="(distinguishedName=%s)" % rename_user_dn)
1253         self.assertNotEqual(len(res), 0)
1254
1255     def test_rename_u4(self):
1256         """Rename 'User object' cross OU with WP, SD and CC right granted on reg. user to AU"""
1257         ou1_dn = "OU=test_rename_ou1," + self.base_dn
1258         ou2_dn = "OU=test_rename_ou2," + self.base_dn
1259         user_dn = "CN=test_rename_user2," + ou1_dn
1260         rename_user_dn = "CN=test_rename_user5," + ou2_dn
1261         # Create OU structure
1262         self.ldb_admin.create_ou(ou1_dn)
1263         self.ldb_admin.create_ou(ou2_dn)
1264         self.ldb_admin.newuser(self.testuser2, self.user_pass, userou=self.ou1)
1265         mod = "(A;;WPSD;;;AU)"
1266         self.sd_utils.dacl_add_ace(user_dn, mod)
1267         mod = "(A;;CC;;;AU)"
1268         self.sd_utils.dacl_add_ace(ou2_dn, mod)
1269         # Rename 'User object' having SD and CC to AU
1270         self.ldb_user.rename(user_dn, rename_user_dn)
1271         res = self.ldb_admin.search(self.base_dn,
1272                                     expression="(distinguishedName=%s)" % user_dn)
1273         self.assertEqual(len(res), 0)
1274         res = self.ldb_admin.search(self.base_dn,
1275                                     expression="(distinguishedName=%s)" % rename_user_dn)
1276         self.assertNotEqual(len(res), 0)
1277
1278     def test_rename_u5(self):
1279         """Test rename with rights granted to 'User object' SID"""
1280         ou1_dn = "OU=test_rename_ou1," + self.base_dn
1281         ou2_dn = "OU=test_rename_ou2," + self.base_dn
1282         user_dn = "CN=test_rename_user2," + ou1_dn
1283         rename_user_dn = "CN=test_rename_user5," + ou2_dn
1284         # Create OU structure
1285         self.ldb_admin.create_ou(ou1_dn)
1286         self.ldb_admin.create_ou(ou2_dn)
1287         self.ldb_admin.newuser(self.testuser2, self.user_pass, userou=self.ou1)
1288         sid = self.sd_utils.get_object_sid(self.get_user_dn(self.regular_user))
1289         mod = "(A;;WPSD;;;%s)" % str(sid)
1290         self.sd_utils.dacl_add_ace(user_dn, mod)
1291         mod = "(A;;CC;;;%s)" % str(sid)
1292         self.sd_utils.dacl_add_ace(ou2_dn, mod)
1293         # Rename 'User object' having SD and CC to AU
1294         self.ldb_user.rename(user_dn, rename_user_dn)
1295         res = self.ldb_admin.search(self.base_dn,
1296                                     expression="(distinguishedName=%s)" % user_dn)
1297         self.assertEqual(len(res), 0)
1298         res = self.ldb_admin.search(self.base_dn,
1299                                     expression="(distinguishedName=%s)" % rename_user_dn)
1300         self.assertNotEqual(len(res), 0)
1301
1302     def test_rename_u6(self):
1303         """Rename 'User object' cross OU with WP, DC and CC right granted on OU & user to AU"""
1304         ou1_dn = "OU=test_rename_ou1," + self.base_dn
1305         ou2_dn = "OU=test_rename_ou2," + self.base_dn
1306         user_dn = "CN=test_rename_user2," + ou1_dn
1307         rename_user_dn = "CN=test_rename_user2," + ou2_dn
1308         # Create OU structure
1309         self.ldb_admin.create_ou(ou1_dn)
1310         self.ldb_admin.create_ou(ou2_dn)
1311         #mod = "(A;CI;DCWP;;;AU)"
1312         mod = "(A;;DC;;;AU)"
1313         self.sd_utils.dacl_add_ace(ou1_dn, mod)
1314         mod = "(A;;CC;;;AU)"
1315         self.sd_utils.dacl_add_ace(ou2_dn, mod)
1316         self.ldb_admin.newuser(self.testuser2, self.user_pass, userou=self.ou1)
1317         mod = "(A;;WP;;;AU)"
1318         self.sd_utils.dacl_add_ace(user_dn, mod)
1319         # Rename 'User object' having SD and CC to AU
1320         self.ldb_user.rename(user_dn, rename_user_dn)
1321         res = self.ldb_admin.search(self.base_dn,
1322                                     expression="(distinguishedName=%s)" % user_dn)
1323         self.assertEqual(len(res), 0)
1324         res = self.ldb_admin.search(self.base_dn,
1325                                     expression="(distinguishedName=%s)" % rename_user_dn)
1326         self.assertNotEqual(len(res), 0)
1327
1328     def test_rename_u7(self):
1329         """Rename 'User object' cross OU (second level) with WP, DC and CC right granted on OU to AU"""
1330         ou1_dn = "OU=test_rename_ou1," + self.base_dn
1331         ou2_dn = "OU=test_rename_ou2," + self.base_dn
1332         ou3_dn = "OU=test_rename_ou3," + ou2_dn
1333         user_dn = "CN=test_rename_user2," + ou1_dn
1334         rename_user_dn = "CN=test_rename_user5," + ou3_dn
1335         # Create OU structure
1336         self.ldb_admin.create_ou(ou1_dn)
1337         self.ldb_admin.create_ou(ou2_dn)
1338         self.ldb_admin.create_ou(ou3_dn)
1339         mod = "(A;CI;WPDC;;;AU)"
1340         self.sd_utils.dacl_add_ace(ou1_dn, mod)
1341         mod = "(A;;CC;;;AU)"
1342         self.sd_utils.dacl_add_ace(ou3_dn, mod)
1343         self.ldb_admin.newuser(self.testuser2, self.user_pass, userou=self.ou1)
1344         # Rename 'User object' having SD and CC to AU
1345         self.ldb_user.rename(user_dn, rename_user_dn)
1346         res = self.ldb_admin.search(self.base_dn,
1347                                     expression="(distinguishedName=%s)" % user_dn)
1348         self.assertEqual(len(res), 0)
1349         res = self.ldb_admin.search(self.base_dn,
1350                                     expression="(distinguishedName=%s)" % rename_user_dn)
1351         self.assertNotEqual(len(res), 0)
1352
1353     def test_rename_u8(self):
1354         """Test rename on an object with and without modify access on the RDN attribute"""
1355         ou1_dn = "OU=test_rename_ou1," + self.base_dn
1356         ou2_dn = "OU=test_rename_ou2," + ou1_dn
1357         ou3_dn = "OU=test_rename_ou3," + ou1_dn
1358         # Create OU structure
1359         self.ldb_admin.create_ou(ou1_dn)
1360         self.ldb_admin.create_ou(ou2_dn)
1361         sid = self.sd_utils.get_object_sid(self.get_user_dn(self.regular_user))
1362         mod = "(OA;;WP;bf967a0e-0de6-11d0-a285-00aa003049e2;;%s)" % str(sid)
1363         self.sd_utils.dacl_add_ace(ou2_dn, mod)
1364         mod = "(OD;;WP;bf9679f0-0de6-11d0-a285-00aa003049e2;;%s)" % str(sid)
1365         self.sd_utils.dacl_add_ace(ou2_dn, mod)
1366         try:
1367             self.ldb_user.rename(ou2_dn, ou3_dn)
1368         except LdbError as e22:
1369             (num, _) = e22.args
1370             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1371         else:
1372             # This rename operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
1373             self.fail()
1374         sid = self.sd_utils.get_object_sid(self.get_user_dn(self.regular_user))
1375         mod = "(A;;WP;bf9679f0-0de6-11d0-a285-00aa003049e2;;%s)" % str(sid)
1376         self.sd_utils.dacl_add_ace(ou2_dn, mod)
1377         self.ldb_user.rename(ou2_dn, ou3_dn)
1378         res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" % ou2_dn)
1379         self.assertEqual(len(res), 0)
1380         res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" % ou3_dn)
1381         self.assertNotEqual(len(res), 0)
1382
1383     def test_rename_u9(self):
1384         """Rename 'User object' cross OU, with explicit deny on sd and dc"""
1385         ou1_dn = "OU=test_rename_ou1," + self.base_dn
1386         ou2_dn = "OU=test_rename_ou2," + self.base_dn
1387         user_dn = "CN=test_rename_user2," + ou1_dn
1388         rename_user_dn = "CN=test_rename_user5," + ou2_dn
1389         # Create OU structure
1390         self.ldb_admin.create_ou(ou1_dn)
1391         self.ldb_admin.create_ou(ou2_dn)
1392         self.ldb_admin.newuser(self.testuser2, self.user_pass, userou=self.ou1)
1393         mod = "(D;;SD;;;DA)"
1394         self.sd_utils.dacl_add_ace(user_dn, mod)
1395         mod = "(D;;DC;;;DA)"
1396         self.sd_utils.dacl_add_ace(ou1_dn, mod)
1397         # Rename 'User object' having SD and CC to AU
1398         try:
1399             self.ldb_admin.rename(user_dn, rename_user_dn)
1400         except LdbError as e23:
1401             (num, _) = e23.args
1402             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1403         else:
1404             self.fail()
1405         # add an allow ace so we can delete this ou
1406         mod = "(A;;DC;;;DA)"
1407         self.sd_utils.dacl_add_ace(ou1_dn, mod)
1408
1409
1410 # tests on Control Access Rights
1411 class AclCARTests(AclTests):
1412
1413     def setUp(self):
1414         super(AclCARTests, self).setUp()
1415
1416         # Get the old "dSHeuristics" if it was set
1417         dsheuristics = self.ldb_admin.get_dsheuristics()
1418         # Reset the "dSHeuristics" as they were before
1419         self.addCleanup(self.ldb_admin.set_dsheuristics, dsheuristics)
1420         # Set the "dSHeuristics" to activate the correct "userPassword" behaviour
1421         self.ldb_admin.set_dsheuristics("000000001")
1422         # Get the old "minPwdAge"
1423         minPwdAge = self.ldb_admin.get_minPwdAge()
1424         # Reset the "minPwdAge" as it was before
1425         self.addCleanup(self.ldb_admin.set_minPwdAge, minPwdAge)
1426         # Set it temporarely to "0"
1427         self.ldb_admin.set_minPwdAge("0")
1428
1429         self.user_with_wp = "acl_car_user1"
1430         self.user_with_pc = "acl_car_user2"
1431         self.ldb_admin.newuser(self.user_with_wp, self.user_pass)
1432         self.ldb_admin.newuser(self.user_with_pc, self.user_pass)
1433         self.ldb_user = self.get_ldb_connection(self.user_with_wp, self.user_pass)
1434         self.ldb_user2 = self.get_ldb_connection(self.user_with_pc, self.user_pass)
1435
1436     def tearDown(self):
1437         super(AclCARTests, self).tearDown()
1438         delete_force(self.ldb_admin, self.get_user_dn(self.user_with_wp))
1439         delete_force(self.ldb_admin, self.get_user_dn(self.user_with_pc))
1440
1441         del self.ldb_user
1442         del self.ldb_user2
1443
1444     def test_change_password1(self):
1445         """Try a password change operation without any CARs given"""
1446         # users have change password by default - remove for negative testing
1447         desc = self.sd_utils.read_sd_on_dn(self.get_user_dn(self.user_with_wp))
1448         sddl = desc.as_sddl(self.domain_sid)
1449         sddl = sddl.replace("(OA;;CR;ab721a53-1e2f-11d0-9819-00aa0040529b;;WD)", "")
1450         sddl = sddl.replace("(OA;;CR;ab721a53-1e2f-11d0-9819-00aa0040529b;;PS)", "")
1451         self.sd_utils.modify_sd_on_dn(self.get_user_dn(self.user_with_wp), sddl)
1452         try:
1453             self.ldb_user.modify_ldif("""
1454 dn: """ + self.get_user_dn(self.user_with_wp) + """
1455 changetype: modify
1456 delete: unicodePwd
1457 unicodePwd:: """ + base64.b64encode("\"samba123@\"".encode('utf-16-le')).decode('utf8') + """
1458 add: unicodePwd
1459 unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS2\"".encode('utf-16-le')).decode('utf8') + """
1460 """)
1461         except LdbError as e24:
1462             (num, _) = e24.args
1463             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
1464         else:
1465             # for some reason we get constraint violation instead of insufficient access error
1466             self.fail()
1467
1468     def test_change_password2(self):
1469         """Make sure WP has no influence"""
1470         desc = self.sd_utils.read_sd_on_dn(self.get_user_dn(self.user_with_wp))
1471         sddl = desc.as_sddl(self.domain_sid)
1472         sddl = sddl.replace("(OA;;CR;ab721a53-1e2f-11d0-9819-00aa0040529b;;WD)", "")
1473         sddl = sddl.replace("(OA;;CR;ab721a53-1e2f-11d0-9819-00aa0040529b;;PS)", "")
1474         self.sd_utils.modify_sd_on_dn(self.get_user_dn(self.user_with_wp), sddl)
1475         mod = "(A;;WP;;;PS)"
1476         self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
1477         desc = self.sd_utils.read_sd_on_dn(self.get_user_dn(self.user_with_wp))
1478         sddl = desc.as_sddl(self.domain_sid)
1479         try:
1480             self.ldb_user.modify_ldif("""
1481 dn: """ + self.get_user_dn(self.user_with_wp) + """
1482 changetype: modify
1483 delete: unicodePwd
1484 unicodePwd:: """ + base64.b64encode("\"samba123@\"".encode('utf-16-le')).decode('utf8') + """
1485 add: unicodePwd
1486 unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS2\"".encode('utf-16-le')).decode('utf8') + """
1487 """)
1488         except LdbError as e25:
1489             (num, _) = e25.args
1490             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
1491         else:
1492             # for some reason we get constraint violation instead of insufficient access error
1493             self.fail()
1494
1495     def test_change_password3(self):
1496         """Make sure WP has no influence"""
1497         mod = "(D;;WP;;;PS)"
1498         self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
1499         desc = self.sd_utils.read_sd_on_dn(self.get_user_dn(self.user_with_wp))
1500         sddl = desc.as_sddl(self.domain_sid)
1501         self.ldb_user.modify_ldif("""
1502 dn: """ + self.get_user_dn(self.user_with_wp) + """
1503 changetype: modify
1504 delete: unicodePwd
1505 unicodePwd:: """ + base64.b64encode("\"samba123@\"".encode('utf-16-le')).decode('utf8') + """
1506 add: unicodePwd
1507 unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS2\"".encode('utf-16-le')).decode('utf8') + """
1508 """)
1509
1510     def test_change_password5(self):
1511         """Make sure rights have no influence on dBCSPwd"""
1512         desc = self.sd_utils.read_sd_on_dn(self.get_user_dn(self.user_with_wp))
1513         sddl = desc.as_sddl(self.domain_sid)
1514         sddl = sddl.replace("(OA;;CR;ab721a53-1e2f-11d0-9819-00aa0040529b;;WD)", "")
1515         sddl = sddl.replace("(OA;;CR;ab721a53-1e2f-11d0-9819-00aa0040529b;;PS)", "")
1516         self.sd_utils.modify_sd_on_dn(self.get_user_dn(self.user_with_wp), sddl)
1517         mod = "(D;;WP;;;PS)"
1518         self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
1519         try:
1520             self.ldb_user.modify_ldif("""
1521 dn: """ + self.get_user_dn(self.user_with_wp) + """
1522 changetype: modify
1523 delete: dBCSPwd
1524 dBCSPwd: XXXXXXXXXXXXXXXX
1525 add: dBCSPwd
1526 dBCSPwd: YYYYYYYYYYYYYYYY
1527 """)
1528         except LdbError as e26:
1529             (num, _) = e26.args
1530             self.assertEquals(num, ERR_UNWILLING_TO_PERFORM)
1531         else:
1532             self.fail()
1533
1534     def test_change_password6(self):
1535         """Test uneven delete/adds"""
1536         try:
1537             self.ldb_user.modify_ldif("""
1538 dn: """ + self.get_user_dn(self.user_with_wp) + """
1539 changetype: modify
1540 delete: userPassword
1541 userPassword: thatsAcomplPASS1
1542 delete: userPassword
1543 userPassword: thatsAcomplPASS1
1544 add: userPassword
1545 userPassword: thatsAcomplPASS2
1546 """)
1547         except LdbError as e27:
1548             (num, _) = e27.args
1549             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1550         else:
1551             self.fail()
1552         mod = "(OA;;CR;00299570-246d-11d0-a768-00aa006e0529;;PS)"
1553         self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
1554         try:
1555             self.ldb_user.modify_ldif("""
1556 dn: """ + self.get_user_dn(self.user_with_wp) + """
1557 changetype: modify
1558 delete: userPassword
1559 userPassword: thatsAcomplPASS1
1560 delete: userPassword
1561 userPassword: thatsAcomplPASS1
1562 add: userPassword
1563 userPassword: thatsAcomplPASS2
1564 """)
1565             # This fails on Windows 2000 domain level with constraint violation
1566         except LdbError as e28:
1567             (num, _) = e28.args
1568             self.assertTrue(num == ERR_CONSTRAINT_VIOLATION or
1569                             num == ERR_UNWILLING_TO_PERFORM)
1570         else:
1571             self.fail()
1572
1573     def test_change_password7(self):
1574         """Try a password change operation without any CARs given"""
1575         # users have change password by default - remove for negative testing
1576         desc = self.sd_utils.read_sd_on_dn(self.get_user_dn(self.user_with_wp))
1577         sddl = desc.as_sddl(self.domain_sid)
1578         self.sd_utils.modify_sd_on_dn(self.get_user_dn(self.user_with_wp), sddl)
1579         # first change our own password
1580         self.ldb_user2.modify_ldif("""
1581 dn: """ + self.get_user_dn(self.user_with_pc) + """
1582 changetype: modify
1583 delete: unicodePwd
1584 unicodePwd:: """ + base64.b64encode("\"samba123@\"".encode('utf-16-le')).decode('utf8') + """
1585 add: unicodePwd
1586 unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS1\"".encode('utf-16-le')).decode('utf8') + """
1587 """)
1588         # then someone else's
1589         self.ldb_user2.modify_ldif("""
1590 dn: """ + self.get_user_dn(self.user_with_wp) + """
1591 changetype: modify
1592 delete: unicodePwd
1593 unicodePwd:: """ + base64.b64encode("\"samba123@\"".encode('utf-16-le')).decode('utf8') + """
1594 add: unicodePwd
1595 unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS2\"".encode('utf-16-le')).decode('utf8') + """
1596 """)
1597
1598     def test_reset_password1(self):
1599         """Try a user password reset operation (unicodePwd) before and after granting CAR"""
1600         try:
1601             self.ldb_user.modify_ldif("""
1602 dn: """ + self.get_user_dn(self.user_with_wp) + """
1603 changetype: modify
1604 replace: unicodePwd
1605 unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS1\"".encode('utf-16-le')).decode('utf8') + """
1606 """)
1607         except LdbError as e29:
1608             (num, _) = e29.args
1609             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1610         else:
1611             self.fail()
1612         mod = "(OA;;CR;00299570-246d-11d0-a768-00aa006e0529;;PS)"
1613         self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
1614         self.ldb_user.modify_ldif("""
1615 dn: """ + self.get_user_dn(self.user_with_wp) + """
1616 changetype: modify
1617 replace: unicodePwd
1618 unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS1\"".encode('utf-16-le')).decode('utf8') + """
1619 """)
1620
1621     def test_reset_password2(self):
1622         """Try a user password reset operation (userPassword) before and after granting CAR"""
1623         try:
1624             self.ldb_user.modify_ldif("""
1625 dn: """ + self.get_user_dn(self.user_with_wp) + """
1626 changetype: modify
1627 replace: userPassword
1628 userPassword: thatsAcomplPASS1
1629 """)
1630         except LdbError as e30:
1631             (num, _) = e30.args
1632             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1633         else:
1634             self.fail()
1635         mod = "(OA;;CR;00299570-246d-11d0-a768-00aa006e0529;;PS)"
1636         self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
1637         try:
1638             self.ldb_user.modify_ldif("""
1639 dn: """ + self.get_user_dn(self.user_with_wp) + """
1640 changetype: modify
1641 replace: userPassword
1642 userPassword: thatsAcomplPASS1
1643 """)
1644             # This fails on Windows 2000 domain level with constraint violation
1645         except LdbError as e31:
1646             (num, _) = e31.args
1647             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
1648
1649     def test_reset_password3(self):
1650         """Grant WP and see what happens (unicodePwd)"""
1651         mod = "(A;;WP;;;PS)"
1652         self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
1653         try:
1654             self.ldb_user.modify_ldif("""
1655 dn: """ + self.get_user_dn(self.user_with_wp) + """
1656 changetype: modify
1657 replace: unicodePwd
1658 unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS1\"".encode('utf-16-le')).decode('utf8') + """
1659 """)
1660         except LdbError as e32:
1661             (num, _) = e32.args
1662             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1663         else:
1664             self.fail()
1665
1666     def test_reset_password4(self):
1667         """Grant WP and see what happens (userPassword)"""
1668         mod = "(A;;WP;;;PS)"
1669         self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
1670         try:
1671             self.ldb_user.modify_ldif("""
1672 dn: """ + self.get_user_dn(self.user_with_wp) + """
1673 changetype: modify
1674 replace: userPassword
1675 userPassword: thatsAcomplPASS1
1676 """)
1677         except LdbError as e33:
1678             (num, _) = e33.args
1679             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1680         else:
1681             self.fail()
1682
1683     def test_reset_password5(self):
1684         """Explicitly deny WP but grant CAR (unicodePwd)"""
1685         mod = "(D;;WP;;;PS)(OA;;CR;00299570-246d-11d0-a768-00aa006e0529;;PS)"
1686         self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
1687         self.ldb_user.modify_ldif("""
1688 dn: """ + self.get_user_dn(self.user_with_wp) + """
1689 changetype: modify
1690 replace: unicodePwd
1691 unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS1\"".encode('utf-16-le')).decode('utf8') + """
1692 """)
1693
1694     def test_reset_password6(self):
1695         """Explicitly deny WP but grant CAR (userPassword)"""
1696         mod = "(D;;WP;;;PS)(OA;;CR;00299570-246d-11d0-a768-00aa006e0529;;PS)"
1697         self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
1698         try:
1699             self.ldb_user.modify_ldif("""
1700 dn: """ + self.get_user_dn(self.user_with_wp) + """
1701 changetype: modify
1702 replace: userPassword
1703 userPassword: thatsAcomplPASS1
1704 """)
1705             # This fails on Windows 2000 domain level with constraint violation
1706         except LdbError as e34:
1707             (num, _) = e34.args
1708             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
1709
1710
1711 class AclExtendedTests(AclTests):
1712
1713     def setUp(self):
1714         super(AclExtendedTests, self).setUp()
1715         # regular user, will be the creator
1716         self.u1 = "ext_u1"
1717         # regular user
1718         self.u2 = "ext_u2"
1719         # admin user
1720         self.u3 = "ext_u3"
1721         self.ldb_admin.newuser(self.u1, self.user_pass)
1722         self.ldb_admin.newuser(self.u2, self.user_pass)
1723         self.ldb_admin.newuser(self.u3, self.user_pass)
1724         self.ldb_admin.add_remove_group_members("Domain Admins", [self.u3],
1725                                                 add_members_operation=True)
1726         self.ldb_user1 = self.get_ldb_connection(self.u1, self.user_pass)
1727         self.ldb_user2 = self.get_ldb_connection(self.u2, self.user_pass)
1728         self.ldb_user3 = self.get_ldb_connection(self.u3, self.user_pass)
1729         self.user_sid1 = self.sd_utils.get_object_sid(self.get_user_dn(self.u1))
1730         self.user_sid2 = self.sd_utils.get_object_sid(self.get_user_dn(self.u2))
1731
1732     def tearDown(self):
1733         super(AclExtendedTests, self).tearDown()
1734         delete_force(self.ldb_admin, self.get_user_dn(self.u1))
1735         delete_force(self.ldb_admin, self.get_user_dn(self.u2))
1736         delete_force(self.ldb_admin, self.get_user_dn(self.u3))
1737         delete_force(self.ldb_admin, "CN=ext_group1,OU=ext_ou1," + self.base_dn)
1738         delete_force(self.ldb_admin, "ou=ext_ou1," + self.base_dn)
1739
1740         del self.ldb_user1
1741         del self.ldb_user2
1742         del self.ldb_user3
1743
1744     def test_ntSecurityDescriptor(self):
1745         # create empty ou
1746         self.ldb_admin.create_ou("ou=ext_ou1," + self.base_dn)
1747         # give u1 Create children access
1748         mod = "(A;;CC;;;%s)" % str(self.user_sid1)
1749         self.sd_utils.dacl_add_ace("OU=ext_ou1," + self.base_dn, mod)
1750         mod = "(A;;LC;;;%s)" % str(self.user_sid2)
1751         self.sd_utils.dacl_add_ace("OU=ext_ou1," + self.base_dn, mod)
1752         # create a group under that, grant RP to u2
1753         self.ldb_user1.newgroup("ext_group1", groupou="OU=ext_ou1",
1754                                 grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
1755         mod = "(A;;RP;;;%s)" % str(self.user_sid2)
1756         self.sd_utils.dacl_add_ace("CN=ext_group1,OU=ext_ou1," + self.base_dn, mod)
1757         # u2 must not read the descriptor
1758         res = self.ldb_user2.search("CN=ext_group1,OU=ext_ou1," + self.base_dn,
1759                                     SCOPE_BASE, None, ["nTSecurityDescriptor"])
1760         self.assertNotEqual(len(res), 0)
1761         self.assertFalse("nTSecurityDescriptor" in res[0].keys())
1762         # grant RC to u2 - still no access
1763         mod = "(A;;RC;;;%s)" % str(self.user_sid2)
1764         self.sd_utils.dacl_add_ace("CN=ext_group1,OU=ext_ou1," + self.base_dn, mod)
1765         res = self.ldb_user2.search("CN=ext_group1,OU=ext_ou1," + self.base_dn,
1766                                     SCOPE_BASE, None, ["nTSecurityDescriptor"])
1767         self.assertNotEqual(len(res), 0)
1768         self.assertFalse("nTSecurityDescriptor" in res[0].keys())
1769         # u3 is member of administrators group, should be able to read sd
1770         res = self.ldb_user3.search("CN=ext_group1,OU=ext_ou1," + self.base_dn,
1771                                     SCOPE_BASE, None, ["nTSecurityDescriptor"])
1772         self.assertEqual(len(res), 1)
1773         self.assertTrue("nTSecurityDescriptor" in res[0].keys())
1774
1775
1776 class AclUndeleteTests(AclTests):
1777
1778     def setUp(self):
1779         super(AclUndeleteTests, self).setUp()
1780         self.regular_user = "undeleter1"
1781         self.ou1 = "OU=undeleted_ou,"
1782         self.testuser1 = "to_be_undeleted1"
1783         self.testuser2 = "to_be_undeleted2"
1784         self.testuser3 = "to_be_undeleted3"
1785         self.testuser4 = "to_be_undeleted4"
1786         self.testuser5 = "to_be_undeleted5"
1787         self.testuser6 = "to_be_undeleted6"
1788
1789         self.new_dn_ou = "CN=" + self.testuser4 + "," + self.ou1 + self.base_dn
1790
1791         # Create regular user
1792         self.testuser1_dn = self.get_user_dn(self.testuser1)
1793         self.testuser2_dn = self.get_user_dn(self.testuser2)
1794         self.testuser3_dn = self.get_user_dn(self.testuser3)
1795         self.testuser4_dn = self.get_user_dn(self.testuser4)
1796         self.testuser5_dn = self.get_user_dn(self.testuser5)
1797         self.deleted_dn1 = self.create_delete_user(self.testuser1)
1798         self.deleted_dn2 = self.create_delete_user(self.testuser2)
1799         self.deleted_dn3 = self.create_delete_user(self.testuser3)
1800         self.deleted_dn4 = self.create_delete_user(self.testuser4)
1801         self.deleted_dn5 = self.create_delete_user(self.testuser5)
1802
1803         self.ldb_admin.create_ou(self.ou1 + self.base_dn)
1804
1805         self.ldb_admin.newuser(self.regular_user, self.user_pass)
1806         self.ldb_admin.add_remove_group_members("Domain Admins", [self.regular_user],
1807                                                 add_members_operation=True)
1808         self.ldb_user = self.get_ldb_connection(self.regular_user, self.user_pass)
1809         self.sid = self.sd_utils.get_object_sid(self.get_user_dn(self.regular_user))
1810
1811     def tearDown(self):
1812         super(AclUndeleteTests, self).tearDown()
1813         delete_force(self.ldb_admin, self.get_user_dn(self.regular_user))
1814         delete_force(self.ldb_admin, self.get_user_dn(self.testuser1))
1815         delete_force(self.ldb_admin, self.get_user_dn(self.testuser2))
1816         delete_force(self.ldb_admin, self.get_user_dn(self.testuser3))
1817         delete_force(self.ldb_admin, self.get_user_dn(self.testuser4))
1818         delete_force(self.ldb_admin, self.get_user_dn(self.testuser5))
1819         delete_force(self.ldb_admin, self.new_dn_ou)
1820         delete_force(self.ldb_admin, self.ou1 + self.base_dn)
1821
1822         del self.ldb_user
1823
1824     def GUID_string(self, guid):
1825         return ldb.schema_format_value("objectGUID", guid)
1826
1827     def create_delete_user(self, new_user):
1828         self.ldb_admin.newuser(new_user, self.user_pass)
1829
1830         res = self.ldb_admin.search(expression="(objectClass=*)",
1831                                     base=self.get_user_dn(new_user),
1832                                     scope=SCOPE_BASE,
1833                                     controls=["show_deleted:1"])
1834         guid = res[0]["objectGUID"][0]
1835         self.ldb_admin.delete(self.get_user_dn(new_user))
1836         res = self.ldb_admin.search(base="<GUID=%s>" % self.GUID_string(guid),
1837                                     scope=SCOPE_BASE, controls=["show_deleted:1"])
1838         self.assertEquals(len(res), 1)
1839         return str(res[0].dn)
1840
1841     def undelete_deleted(self, olddn, newdn):
1842         msg = Message()
1843         msg.dn = Dn(self.ldb_user, olddn)
1844         msg["isDeleted"] = MessageElement([], FLAG_MOD_DELETE, "isDeleted")
1845         msg["distinguishedName"] = MessageElement([newdn], FLAG_MOD_REPLACE, "distinguishedName")
1846         res = self.ldb_user.modify(msg, ["show_recycled:1"])
1847
1848     def undelete_deleted_with_mod(self, olddn, newdn):
1849         msg = Message()
1850         msg.dn = Dn(ldb, olddn)
1851         msg["isDeleted"] = MessageElement([], FLAG_MOD_DELETE, "isDeleted")
1852         msg["distinguishedName"] = MessageElement([newdn], FLAG_MOD_REPLACE, "distinguishedName")
1853         msg["url"] = MessageElement(["www.samba.org"], FLAG_MOD_REPLACE, "url")
1854         res = self.ldb_user.modify(msg, ["show_deleted:1"])
1855
1856     def test_undelete(self):
1857         # it appears the user has to have LC on the old parent to be able to move the object
1858         # otherwise we get no such object. Since only System can modify the SD on deleted object
1859         # we cannot grant this permission via LDAP, and this leaves us with "negative" tests at the moment
1860
1861         # deny write property on rdn, should fail
1862         mod = "(OD;;WP;bf967a0e-0de6-11d0-a285-00aa003049e2;;%s)" % str(self.sid)
1863         self.sd_utils.dacl_add_ace(self.deleted_dn1, mod)
1864         try:
1865             self.undelete_deleted(self.deleted_dn1, self.testuser1_dn)
1866             self.fail()
1867         except LdbError as e35:
1868             (num, _) = e35.args
1869             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1870
1871         # seems that permissions on isDeleted and distinguishedName are irrelevant
1872         mod = "(OD;;WP;bf96798f-0de6-11d0-a285-00aa003049e2;;%s)" % str(self.sid)
1873         self.sd_utils.dacl_add_ace(self.deleted_dn2, mod)
1874         mod = "(OD;;WP;bf9679e4-0de6-11d0-a285-00aa003049e2;;%s)" % str(self.sid)
1875         self.sd_utils.dacl_add_ace(self.deleted_dn2, mod)
1876         self.undelete_deleted(self.deleted_dn2, self.testuser2_dn)
1877
1878         # attempt undelete with simultanious addition of url, WP to which is denied
1879         mod = "(OD;;WP;9a9a0221-4a5b-11d1-a9c3-0000f80367c1;;%s)" % str(self.sid)
1880         self.sd_utils.dacl_add_ace(self.deleted_dn3, mod)
1881         try:
1882             self.undelete_deleted_with_mod(self.deleted_dn3, self.testuser3_dn)
1883             self.fail()
1884         except LdbError as e36:
1885             (num, _) = e36.args
1886             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1887
1888         # undelete in an ou, in which we have no right to create children
1889         mod = "(D;;CC;;;%s)" % str(self.sid)
1890         self.sd_utils.dacl_add_ace(self.ou1 + self.base_dn, mod)
1891         try:
1892             self.undelete_deleted(self.deleted_dn4, self.new_dn_ou)
1893             self.fail()
1894         except LdbError as e37:
1895             (num, _) = e37.args
1896             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1897
1898         # delete is not required
1899         mod = "(D;;SD;;;%s)" % str(self.sid)
1900         self.sd_utils.dacl_add_ace(self.deleted_dn5, mod)
1901         self.undelete_deleted(self.deleted_dn5, self.testuser5_dn)
1902
1903         # deny Reanimate-Tombstone, should fail
1904         mod = "(OD;;CR;45ec5156-db7e-47bb-b53f-dbeb2d03c40f;;%s)" % str(self.sid)
1905         self.sd_utils.dacl_add_ace(self.base_dn, mod)
1906         try:
1907             self.undelete_deleted(self.deleted_dn4, self.testuser4_dn)
1908             self.fail()
1909         except LdbError as e38:
1910             (num, _) = e38.args
1911             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1912
1913
1914 class AclSPNTests(AclTests):
1915
1916     def setUp(self):
1917         super(AclSPNTests, self).setUp()
1918         self.dcname = "TESTSRV8"
1919         self.rodcname = "TESTRODC8"
1920         self.computername = "testcomp8"
1921         self.test_user = "spn_test_user8"
1922         self.computerdn = "CN=%s,CN=computers,%s" % (self.computername, self.base_dn)
1923         self.dc_dn = "CN=%s,OU=Domain Controllers,%s" % (self.dcname, self.base_dn)
1924         self.site = "Default-First-Site-Name"
1925         self.rodcctx = DCJoinContext(server=host, creds=creds, lp=lp,
1926                                      site=self.site, netbios_name=self.rodcname,
1927                                      targetdir=None, domain=None)
1928         self.dcctx = DCJoinContext(server=host, creds=creds, lp=lp,
1929                                    site=self.site, netbios_name=self.dcname,
1930                                    targetdir=None, domain=None)
1931         self.ldb_admin.newuser(self.test_user, self.user_pass)
1932         self.ldb_user1 = self.get_ldb_connection(self.test_user, self.user_pass)
1933         self.user_sid1 = self.sd_utils.get_object_sid(self.get_user_dn(self.test_user))
1934         self.create_computer(self.computername, self.dcctx.dnsdomain)
1935         self.create_rodc(self.rodcctx)
1936         self.create_dc(self.dcctx)
1937
1938     def tearDown(self):
1939         super(AclSPNTests, self).tearDown()
1940         self.rodcctx.cleanup_old_join()
1941         self.dcctx.cleanup_old_join()
1942         delete_force(self.ldb_admin, "cn=%s,cn=computers,%s" % (self.computername, self.base_dn))
1943         delete_force(self.ldb_admin, self.get_user_dn(self.test_user))
1944
1945         del self.ldb_user1
1946
1947     def replace_spn(self, _ldb, dn, spn):
1948         print("Setting spn %s on %s" % (spn, dn))
1949         res = self.ldb_admin.search(dn, expression="(objectClass=*)",
1950                                     scope=SCOPE_BASE, attrs=["servicePrincipalName"])
1951         if "servicePrincipalName" in res[0].keys():
1952             flag = FLAG_MOD_REPLACE
1953         else:
1954             flag = FLAG_MOD_ADD
1955
1956         msg = Message()
1957         msg.dn = Dn(self.ldb_admin, dn)
1958         msg["servicePrincipalName"] = MessageElement(spn, flag,
1959                                                      "servicePrincipalName")
1960         _ldb.modify(msg)
1961
1962     def create_computer(self, computername, domainname):
1963         dn = "CN=%s,CN=computers,%s" % (computername, self.base_dn)
1964         samaccountname = computername + "$"
1965         dnshostname = "%s.%s" % (computername, domainname)
1966         self.ldb_admin.add({
1967             "dn": dn,
1968             "objectclass": "computer",
1969             "sAMAccountName": samaccountname,
1970             "userAccountControl": str(samba.dsdb.UF_WORKSTATION_TRUST_ACCOUNT),
1971             "dNSHostName": dnshostname})
1972
1973     # same as for join_RODC, but do not set any SPNs
1974     def create_rodc(self, ctx):
1975         ctx.nc_list = [ctx.base_dn, ctx.config_dn, ctx.schema_dn]
1976         ctx.full_nc_list = [ctx.base_dn, ctx.config_dn, ctx.schema_dn]
1977         ctx.krbtgt_dn = "CN=krbtgt_%s,CN=Users,%s" % (ctx.myname, ctx.base_dn)
1978
1979         ctx.never_reveal_sid = ["<SID=%s-%s>" % (ctx.domsid, security.DOMAIN_RID_RODC_DENY),
1980                                 "<SID=%s>" % security.SID_BUILTIN_ADMINISTRATORS,
1981                                 "<SID=%s>" % security.SID_BUILTIN_SERVER_OPERATORS,
1982                                 "<SID=%s>" % security.SID_BUILTIN_BACKUP_OPERATORS,
1983                                 "<SID=%s>" % security.SID_BUILTIN_ACCOUNT_OPERATORS]
1984         ctx.reveal_sid = "<SID=%s-%s>" % (ctx.domsid, security.DOMAIN_RID_RODC_ALLOW)
1985
1986         mysid = ctx.get_mysid()
1987         admin_dn = "<SID=%s>" % mysid
1988         ctx.managedby = admin_dn
1989
1990         ctx.userAccountControl = (samba.dsdb.UF_WORKSTATION_TRUST_ACCOUNT |
1991                                   samba.dsdb.UF_TRUSTED_TO_AUTHENTICATE_FOR_DELEGATION |
1992                                   samba.dsdb.UF_PARTIAL_SECRETS_ACCOUNT)
1993
1994         ctx.connection_dn = "CN=RODC Connection (FRS),%s" % ctx.ntds_dn
1995         ctx.secure_channel_type = misc.SEC_CHAN_RODC
1996         ctx.RODC = True
1997         ctx.replica_flags = (drsuapi.DRSUAPI_DRS_INIT_SYNC |
1998                              drsuapi.DRSUAPI_DRS_PER_SYNC |
1999                              drsuapi.DRSUAPI_DRS_GET_ANC |
2000                              drsuapi.DRSUAPI_DRS_NEVER_SYNCED |
2001                              drsuapi.DRSUAPI_DRS_SPECIAL_SECRET_PROCESSING)
2002
2003         ctx.join_add_objects()
2004
2005     def create_dc(self, ctx):
2006         ctx.nc_list = [ctx.base_dn, ctx.config_dn, ctx.schema_dn]
2007         ctx.full_nc_list = [ctx.base_dn, ctx.config_dn, ctx.schema_dn]
2008         ctx.userAccountControl = samba.dsdb.UF_SERVER_TRUST_ACCOUNT | samba.dsdb.UF_TRUSTED_FOR_DELEGATION
2009         ctx.secure_channel_type = misc.SEC_CHAN_BDC
2010         ctx.replica_flags = (drsuapi.DRSUAPI_DRS_WRIT_REP |
2011                              drsuapi.DRSUAPI_DRS_INIT_SYNC |
2012                              drsuapi.DRSUAPI_DRS_PER_SYNC |
2013                              drsuapi.DRSUAPI_DRS_FULL_SYNC_IN_PROGRESS |
2014                              drsuapi.DRSUAPI_DRS_NEVER_SYNCED)
2015
2016         ctx.join_add_objects()
2017
2018     def dc_spn_test(self, ctx):
2019         netbiosdomain = self.dcctx.get_domain_name()
2020         try:
2021             self.replace_spn(self.ldb_user1, ctx.acct_dn, "HOST/%s/%s" % (ctx.myname, netbiosdomain))
2022         except LdbError as e39:
2023             (num, _) = e39.args
2024             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
2025
2026         mod = "(OA;;SW;f3a64788-5306-11d1-a9c5-0000f80367c1;;%s)" % str(self.user_sid1)
2027         self.sd_utils.dacl_add_ace(ctx.acct_dn, mod)
2028         self.replace_spn(self.ldb_user1, ctx.acct_dn, "HOST/%s/%s" % (ctx.myname, netbiosdomain))
2029         self.replace_spn(self.ldb_user1, ctx.acct_dn, "HOST/%s" % (ctx.myname))
2030         self.replace_spn(self.ldb_user1, ctx.acct_dn, "HOST/%s.%s/%s" %
2031                          (ctx.myname, ctx.dnsdomain, netbiosdomain))
2032         self.replace_spn(self.ldb_user1, ctx.acct_dn, "HOST/%s/%s" % (ctx.myname, ctx.dnsdomain))
2033         self.replace_spn(self.ldb_user1, ctx.acct_dn, "HOST/%s.%s/%s" %
2034                          (ctx.myname, ctx.dnsdomain, ctx.dnsdomain))
2035         self.replace_spn(self.ldb_user1, ctx.acct_dn, "GC/%s.%s/%s" %
2036                          (ctx.myname, ctx.dnsdomain, ctx.dnsforest))
2037         self.replace_spn(self.ldb_user1, ctx.acct_dn, "ldap/%s/%s" % (ctx.myname, netbiosdomain))
2038         self.replace_spn(self.ldb_user1, ctx.acct_dn, "ldap/%s.%s/%s" %
2039                          (ctx.myname, ctx.dnsdomain, netbiosdomain))
2040         self.replace_spn(self.ldb_user1, ctx.acct_dn, "ldap/%s" % (ctx.myname))
2041         self.replace_spn(self.ldb_user1, ctx.acct_dn, "ldap/%s/%s" % (ctx.myname, ctx.dnsdomain))
2042         self.replace_spn(self.ldb_user1, ctx.acct_dn, "ldap/%s.%s/%s" %
2043                          (ctx.myname, ctx.dnsdomain, ctx.dnsdomain))
2044         self.replace_spn(self.ldb_user1, ctx.acct_dn, "DNS/%s/%s" % (ctx.myname, ctx.dnsdomain))
2045         self.replace_spn(self.ldb_user1, ctx.acct_dn, "RestrictedKrbHost/%s/%s" %
2046                          (ctx.myname, ctx.dnsdomain))
2047         self.replace_spn(self.ldb_user1, ctx.acct_dn, "RestrictedKrbHost/%s" %
2048                          (ctx.myname))
2049         self.replace_spn(self.ldb_user1, ctx.acct_dn, "Dfsr-12F9A27C-BF97-4787-9364-D31B6C55EB04/%s/%s" %
2050                          (ctx.myname, ctx.dnsdomain))
2051         self.replace_spn(self.ldb_user1, ctx.acct_dn, "NtFrs-88f5d2bd-b646-11d2-a6d3-00c04fc9b232/%s/%s" %
2052                          (ctx.myname, ctx.dnsdomain))
2053         self.replace_spn(self.ldb_user1, ctx.acct_dn, "ldap/%s._msdcs.%s" %
2054                          (ctx.ntds_guid, ctx.dnsdomain))
2055
2056         # the following spns do not match the restrictions and should fail
2057         try:
2058             self.replace_spn(self.ldb_user1, ctx.acct_dn, "ldap/%s.%s/ForestDnsZones.%s" %
2059                              (ctx.myname, ctx.dnsdomain, ctx.dnsdomain))
2060         except LdbError as e40:
2061             (num, _) = e40.args
2062             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
2063         try:
2064             self.replace_spn(self.ldb_user1, ctx.acct_dn, "ldap/%s.%s/DomainDnsZones.%s" %
2065                              (ctx.myname, ctx.dnsdomain, ctx.dnsdomain))
2066         except LdbError as e41:
2067             (num, _) = e41.args
2068             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
2069         try:
2070             self.replace_spn(self.ldb_user1, ctx.acct_dn, "nosuchservice/%s/%s" % ("abcd", "abcd"))
2071         except LdbError as e42:
2072             (num, _) = e42.args
2073             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
2074         try:
2075             self.replace_spn(self.ldb_user1, ctx.acct_dn, "GC/%s.%s/%s" %
2076                              (ctx.myname, ctx.dnsdomain, netbiosdomain))
2077         except LdbError as e43:
2078             (num, _) = e43.args
2079             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
2080         try:
2081             self.replace_spn(self.ldb_user1, ctx.acct_dn, "E3514235-4B06-11D1-AB04-00C04FC2DCD2/%s/%s" %
2082                              (ctx.ntds_guid, ctx.dnsdomain))
2083         except LdbError as e44:
2084             (num, _) = e44.args
2085             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
2086
2087     def test_computer_spn(self):
2088         # with WP, any value can be set
2089         netbiosdomain = self.dcctx.get_domain_name()
2090         self.replace_spn(self.ldb_admin, self.computerdn, "HOST/%s/%s" %
2091                          (self.computername, netbiosdomain))
2092         self.replace_spn(self.ldb_admin, self.computerdn, "HOST/%s" % (self.computername))
2093         self.replace_spn(self.ldb_admin, self.computerdn, "HOST/%s.%s/%s" %
2094                          (self.computername, self.dcctx.dnsdomain, netbiosdomain))
2095         self.replace_spn(self.ldb_admin, self.computerdn, "HOST/%s/%s" %
2096                          (self.computername, self.dcctx.dnsdomain))
2097         self.replace_spn(self.ldb_admin, self.computerdn, "HOST/%s.%s/%s" %
2098                          (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsdomain))
2099         self.replace_spn(self.ldb_admin, self.computerdn, "GC/%s.%s/%s" %
2100                          (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsforest))
2101         self.replace_spn(self.ldb_admin, self.computerdn, "ldap/%s/%s" % (self.computername, netbiosdomain))
2102         self.replace_spn(self.ldb_admin, self.computerdn, "ldap/%s.%s/ForestDnsZones.%s" %
2103                          (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsdomain))
2104         self.replace_spn(self.ldb_admin, self.computerdn, "ldap/%s.%s/DomainDnsZones.%s" %
2105                          (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsdomain))
2106         self.replace_spn(self.ldb_admin, self.computerdn, "ldap/%s.%s/%s" %
2107                          (self.computername, self.dcctx.dnsdomain, netbiosdomain))
2108         self.replace_spn(self.ldb_admin, self.computerdn, "ldap/%s" % (self.computername))
2109         self.replace_spn(self.ldb_admin, self.computerdn, "ldap/%s/%s" %
2110                          (self.computername, self.dcctx.dnsdomain))
2111         self.replace_spn(self.ldb_admin, self.computerdn, "ldap/%s.%s/%s" %
2112                          (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsdomain))
2113         self.replace_spn(self.ldb_admin, self.computerdn, "DNS/%s/%s" %
2114                          (self.computername, self.dcctx.dnsdomain))
2115         self.replace_spn(self.ldb_admin, self.computerdn, "RestrictedKrbHost/%s/%s" %
2116                          (self.computername, self.dcctx.dnsdomain))
2117         self.replace_spn(self.ldb_admin, self.computerdn, "RestrictedKrbHost/%s" %
2118                          (self.computername))
2119         self.replace_spn(self.ldb_admin, self.computerdn, "Dfsr-12F9A27C-BF97-4787-9364-D31B6C55EB04/%s/%s" %
2120                          (self.computername, self.dcctx.dnsdomain))
2121         self.replace_spn(self.ldb_admin, self.computerdn, "NtFrs-88f5d2bd-b646-11d2-a6d3-00c04fc9b232/%s/%s" %
2122                          (self.computername, self.dcctx.dnsdomain))
2123         self.replace_spn(self.ldb_admin, self.computerdn, "nosuchservice/%s/%s" % ("abcd", "abcd"))
2124
2125         # user has neither WP nor Validated-SPN, access denied expected
2126         try:
2127             self.replace_spn(self.ldb_user1, self.computerdn, "HOST/%s/%s" % (self.computername, netbiosdomain))
2128         except LdbError as e45:
2129             (num, _) = e45.args
2130             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
2131
2132         mod = "(OA;;SW;f3a64788-5306-11d1-a9c5-0000f80367c1;;%s)" % str(self.user_sid1)
2133         self.sd_utils.dacl_add_ace(self.computerdn, mod)
2134         # grant Validated-SPN and check which values are accepted
2135         # see 3.1.1.5.3.1.1.4 servicePrincipalName for reference
2136
2137         # for regular computer objects we shouldalways get constraint violation
2138
2139         # This does not pass against Windows, although it should according to docs
2140         self.replace_spn(self.ldb_user1, self.computerdn, "HOST/%s" % (self.computername))
2141         self.replace_spn(self.ldb_user1, self.computerdn, "HOST/%s.%s" %
2142                          (self.computername, self.dcctx.dnsdomain))
2143
2144         try:
2145             self.replace_spn(self.ldb_user1, self.computerdn, "HOST/%s/%s" % (self.computername, netbiosdomain))
2146         except LdbError as e46:
2147             (num, _) = e46.args
2148             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
2149         try:
2150             self.replace_spn(self.ldb_user1, self.computerdn, "HOST/%s.%s/%s" %
2151                              (self.computername, self.dcctx.dnsdomain, netbiosdomain))
2152         except LdbError as e47:
2153             (num, _) = e47.args
2154             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
2155         try:
2156             self.replace_spn(self.ldb_user1, self.computerdn, "HOST/%s/%s" %
2157                              (self.computername, self.dcctx.dnsdomain))
2158         except LdbError as e48:
2159             (num, _) = e48.args
2160             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
2161         try:
2162             self.replace_spn(self.ldb_user1, self.computerdn, "HOST/%s.%s/%s" %
2163                              (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsdomain))
2164         except LdbError as e49:
2165             (num, _) = e49.args
2166             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
2167         try:
2168             self.replace_spn(self.ldb_user1, self.computerdn, "GC/%s.%s/%s" %
2169                              (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsforest))
2170         except LdbError as e50:
2171             (num, _) = e50.args
2172             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
2173         try:
2174             self.replace_spn(self.ldb_user1, self.computerdn, "ldap/%s/%s" % (self.computername, netbiosdomain))
2175         except LdbError as e51:
2176             (num, _) = e51.args
2177             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
2178         try:
2179             self.replace_spn(self.ldb_user1, self.computerdn, "ldap/%s.%s/ForestDnsZones.%s" %
2180                              (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsdomain))
2181         except LdbError as e52:
2182             (num, _) = e52.args
2183             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
2184
2185     def test_spn_rwdc(self):
2186         self.dc_spn_test(self.dcctx)
2187
2188     def test_spn_rodc(self):
2189         self.dc_spn_test(self.rodcctx)
2190
2191
2192 # Important unit running information
2193
2194 ldb = SamDB(ldaphost, credentials=creds, session_info=system_session(lp), lp=lp)
2195
2196 TestProgram(module=__name__, opts=subunitopts)