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