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 from samba.compat import get_string
  15
  16 import samba.getopt as options
  17 from samba.join import DCJoinContext
  18
  19 from ldb import (
  20     SCOPE_BASE, SCOPE_SUBTREE, LdbError, ERR_NO_SUCH_OBJECT,
  21     ERR_UNWILLING_TO_PERFORM, ERR_INSUFFICIENT_ACCESS_RIGHTS)
  22 from ldb import ERR_CONSTRAINT_VIOLATION
  23 from ldb import ERR_OPERATIONS_ERROR
  24 from ldb import Message, MessageElement, Dn
  25 from ldb import FLAG_MOD_REPLACE, FLAG_MOD_ADD, FLAG_MOD_DELETE
  26 from samba.dcerpc import security, drsuapi, misc
  27
  28 from samba.auth import system_session
  29 from samba import gensec, sd_utils
  30 from samba.samdb import SamDB
  31 from samba.credentials import Credentials, DONT_USE_KERBEROS
  32 import samba.tests
  33 from samba.tests import delete_force
  34 import samba.dsdb
  35 from samba.tests.password_test import PasswordCommon
  36
  37 parser = optparse.OptionParser("acl.py [options] <host>")
  38 sambaopts = options.SambaOptions(parser)
  39 parser.add_option_group(sambaopts)
  40 parser.add_option_group(options.VersionOptions(parser))
  41
  42 # use command line creds if available
  43 credopts = options.CredentialsOptions(parser)
  44 parser.add_option_group(credopts)
  45 subunitopts = SubunitOptions(parser)
  46 parser.add_option_group(subunitopts)
  47
  48 opts, args = parser.parse_args()
  49
  50 if len(args) < 1:
  51     parser.print_usage()
  52     sys.exit(1)
  53
  54 host = args[0]
  55 if "://" not in host:
  56     ldaphost = "ldap://%s" % host
  57 else:
  58     ldaphost = host
  59     start = host.rindex("://")
  60     host = host.lstrip(start + 3)
  61
  62 lp = sambaopts.get_loadparm()
  63 creds = credopts.get_credentials(lp)
  64 creds.set_gensec_features(creds.get_gensec_features() | gensec.FEATURE_SEAL)
  65
  66 #
  67 # Tests start here
  68 #
  69
  70
  71 class AclTests(samba.tests.TestCase):
  72
  73     def setUp(self):
  74         super(AclTests, self).setUp()
  75         self.ldb_admin = SamDB(ldaphost, credentials=creds, session_info=system_session(lp), lp=lp)
  76         self.base_dn = self.ldb_admin.domain_dn()
  77         self.domain_sid = security.dom_sid(self.ldb_admin.get_domain_sid())
  78         self.user_pass = "samba123@"
  79         self.configuration_dn = self.ldb_admin.get_config_basedn().get_linearized()
  80         self.sd_utils = sd_utils.SDUtils(self.ldb_admin)
  81         self.addCleanup(self.delete_admin_connection)
  82         # used for anonymous login
  83         self.creds_tmp = Credentials()
  84         self.creds_tmp.set_username("")
  85         self.creds_tmp.set_password("")
  86         self.creds_tmp.set_domain(creds.get_domain())
  87         self.creds_tmp.set_realm(creds.get_realm())
  88         self.creds_tmp.set_workstation(creds.get_workstation())
  89         print("baseDN: %s" % self.base_dn)
  90
  91     def get_user_dn(self, name):
  92         return "CN=%s,CN=Users,%s" % (name, self.base_dn)
  93
  94     def get_ldb_connection(self, target_username, target_password):
  95         creds_tmp = Credentials()
  96         creds_tmp.set_username(target_username)
  97         creds_tmp.set_password(target_password)
  98         creds_tmp.set_domain(creds.get_domain())
  99         creds_tmp.set_realm(creds.get_realm())
 100         creds_tmp.set_workstation(creds.get_workstation())
 101         creds_tmp.set_gensec_features(creds_tmp.get_gensec_features()
 102                                       | gensec.FEATURE_SEAL)
 103         creds_tmp.set_kerberos_state(DONT_USE_KERBEROS)  # kinit is too expensive to use in a tight loop
 104         ldb_target = SamDB(url=ldaphost, credentials=creds_tmp, lp=lp)
 105         return ldb_target
 106
 107     # Test if we have any additional groups for users than default ones
 108     def assert_user_no_group_member(self, username):
 109         res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" % self.get_user_dn(username))
 110         try:
 111             self.assertEqual(res[0]["memberOf"][0], "")
 112         except KeyError:
 113             pass
 114         else:
 115             self.fail()
 116
 117     def delete_admin_connection(self):
 118         del self.sd_utils
 119         del self.ldb_admin
 120
 121 # tests on ldap add operations
 122
 123
 124 class AclAddTests(AclTests):
 125
 126     def setUp(self):
 127         super(AclAddTests, self).setUp()
 128         # Domain admin that will be creator of OU parent-child structure
 129         self.usr_admin_owner = "acl_add_user1"
 130         # Second domain admin that will not be creator of OU parent-child structure
 131         self.usr_admin_not_owner = "acl_add_user2"
 132         # Regular user
 133         self.regular_user = "acl_add_user3"
 134         self.test_user1 = "test_add_user1"
 135         self.test_group1 = "test_add_group1"
 136         self.ou1 = "OU=test_add_ou1"
 137         self.ou2 = "OU=test_add_ou2,%s" % self.ou1
 138         self.ldb_admin.newuser(self.usr_admin_owner, self.user_pass)
 139         self.ldb_admin.newuser(self.usr_admin_not_owner, self.user_pass)
 140         self.ldb_admin.newuser(self.regular_user, self.user_pass)
 141
 142         # add admins to the Domain Admins group
 143         self.ldb_admin.add_remove_group_members("Domain Admins", [self.usr_admin_owner],
 144                                                 add_members_operation=True)
 145         self.ldb_admin.add_remove_group_members("Domain Admins", [self.usr_admin_not_owner],
 146                                                 add_members_operation=True)
 147
 148         self.ldb_owner = self.get_ldb_connection(self.usr_admin_owner, self.user_pass)
 149         self.ldb_notowner = self.get_ldb_connection(self.usr_admin_not_owner, self.user_pass)
 150         self.ldb_user = self.get_ldb_connection(self.regular_user, self.user_pass)
 151
 152     def tearDown(self):
 153         super(AclAddTests, self).tearDown()
 154         delete_force(self.ldb_admin, "CN=%s,%s,%s" %
 155                      (self.test_user1, self.ou2, self.base_dn))
 156         delete_force(self.ldb_admin, "CN=%s,%s,%s" %
 157                      (self.test_group1, self.ou2, self.base_dn))
 158         delete_force(self.ldb_admin, "%s,%s" % (self.ou2, self.base_dn))
 159         delete_force(self.ldb_admin, "%s,%s" % (self.ou1, self.base_dn))
 160         delete_force(self.ldb_admin, self.get_user_dn(self.usr_admin_owner))
 161         delete_force(self.ldb_admin, self.get_user_dn(self.usr_admin_not_owner))
 162         delete_force(self.ldb_admin, self.get_user_dn(self.regular_user))
 163         delete_force(self.ldb_admin, self.get_user_dn("test_add_anonymous"))
 164
 165         del self.ldb_notowner
 166         del self.ldb_owner
 167         del self.ldb_user
 168
 169     # Make sure top OU is deleted (and so everything under it)
 170     def assert_top_ou_deleted(self):
 171         res = self.ldb_admin.search(self.base_dn,
 172                                     expression="(distinguishedName=%s,%s)" % (
 173                                         "OU=test_add_ou1", self.base_dn))
 174         self.assertEqual(len(res), 0)
 175
 176     def test_add_u1(self):
 177         """Testing OU with the rights of Doman Admin not creator of the OU """
 178         self.assert_top_ou_deleted()
 179         # Change descriptor for top level OU
 180         self.ldb_owner.create_ou("OU=test_add_ou1," + self.base_dn)
 181         self.ldb_owner.create_ou("OU=test_add_ou2,OU=test_add_ou1," + self.base_dn)
 182         user_sid = self.sd_utils.get_object_sid(self.get_user_dn(self.usr_admin_not_owner))
 183         mod = "(D;CI;WPCC;;;%s)" % str(user_sid)
 184         self.sd_utils.dacl_add_ace("OU=test_add_ou1," + self.base_dn, mod)
 185         # Test user and group creation with another domain admin's credentials
 186         self.ldb_notowner.newuser(self.test_user1, self.user_pass, userou=self.ou2)
 187         self.ldb_notowner.newgroup("test_add_group1", groupou="OU=test_add_ou2,OU=test_add_ou1",
 188                                    grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
 189         # Make sure we HAVE created the two objects -- user and group
 190         # !!! We should not be able to do that, but however beacuse of ACE ordering our inherited Deny ACE
 191         # !!! comes after explicit (A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA) that comes from somewhere
 192         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))
 193         self.assertTrue(len(res) > 0)
 194         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))
 195         self.assertTrue(len(res) > 0)
 196
 197     def test_add_u2(self):
 198         """Testing OU with the regular user that has no rights granted over the OU """
 199         self.assert_top_ou_deleted()
 200         # Create a parent-child OU structure with domain admin credentials
 201         self.ldb_owner.create_ou("OU=test_add_ou1," + self.base_dn)
 202         self.ldb_owner.create_ou("OU=test_add_ou2,OU=test_add_ou1," + self.base_dn)
 203         # Test user and group creation with regular user credentials
 204         try:
 205             self.ldb_user.newuser(self.test_user1, self.user_pass, userou=self.ou2)
 206             self.ldb_user.newgroup("test_add_group1", groupou="OU=test_add_ou2,OU=test_add_ou1",
 207                                    grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
 208         except LdbError as e:
 209             (num, _) = e.args
 210             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
 211         else:
 212             self.fail()
 213         # Make sure we HAVEN'T created any of two objects -- user or group
 214         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))
 215         self.assertEqual(len(res), 0)
 216         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))
 217         self.assertEqual(len(res), 0)
 218
 219     def test_add_u3(self):
 220         """Testing OU with the rights of regular user granted the right 'Create User child objects' """
 221         self.assert_top_ou_deleted()
 222         # Change descriptor for top level OU
 223         self.ldb_owner.create_ou("OU=test_add_ou1," + self.base_dn)
 224         user_sid = self.sd_utils.get_object_sid(self.get_user_dn(self.regular_user))
 225         mod = "(OA;CI;CC;bf967aba-0de6-11d0-a285-00aa003049e2;;%s)" % str(user_sid)
 226         self.sd_utils.dacl_add_ace("OU=test_add_ou1," + self.base_dn, mod)
 227         self.ldb_owner.create_ou("OU=test_add_ou2,OU=test_add_ou1," + self.base_dn)
 228         # Test user and group creation with granted user only to one of the objects
 229         self.ldb_user.newuser(self.test_user1, self.user_pass, userou=self.ou2, setpassword=False)
 230         try:
 231             self.ldb_user.newgroup("test_add_group1", groupou="OU=test_add_ou2,OU=test_add_ou1",
 232                                    grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
 233         except LdbError as e1:
 234             (num, _) = e1.args
 235             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
 236         else:
 237             self.fail()
 238         # Make sure we HAVE created the one of two objects -- user
 239         res = self.ldb_admin.search(self.base_dn,
 240                                     expression="(distinguishedName=%s,%s)" %
 241                                     ("CN=test_add_user1,OU=test_add_ou2,OU=test_add_ou1",
 242                                      self.base_dn))
 243         self.assertNotEqual(len(res), 0)
 244         res = self.ldb_admin.search(self.base_dn,
 245                                     expression="(distinguishedName=%s,%s)" %
 246                                     ("CN=test_add_group1,OU=test_add_ou2,OU=test_add_ou1",
 247                                      self.base_dn))
 248         self.assertEqual(len(res), 0)
 249
 250     def test_add_u4(self):
 251         """ 4 Testing OU with the rights of Doman Admin creator of the OU"""
 252         self.assert_top_ou_deleted()
 253         self.ldb_owner.create_ou("OU=test_add_ou1," + self.base_dn)
 254         self.ldb_owner.create_ou("OU=test_add_ou2,OU=test_add_ou1," + self.base_dn)
 255         self.ldb_owner.newuser(self.test_user1, self.user_pass, userou=self.ou2)
 256         self.ldb_owner.newgroup("test_add_group1", groupou="OU=test_add_ou2,OU=test_add_ou1",
 257                                 grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
 258         # Make sure we have successfully created the two objects -- user and group
 259         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))
 260         self.assertTrue(len(res) > 0)
 261         res = self.ldb_admin.search(self.base_dn,
 262                                     expression="(distinguishedName=%s,%s)" % ("CN=test_add_group1,OU=test_add_ou2,OU=test_add_ou1", self.base_dn))
 263         self.assertTrue(len(res) > 0)
 264
 265     def test_add_anonymous(self):
 266         """Test add operation with anonymous user"""
 267         anonymous = SamDB(url=ldaphost, credentials=self.creds_tmp, lp=lp)
 268         try:
 269             anonymous.newuser("test_add_anonymous", self.user_pass)
 270         except LdbError as e2:
 271             (num, _) = e2.args
 272             self.assertEquals(num, ERR_OPERATIONS_ERROR)
 273         else:
 274             self.fail()
 275
 276 # tests on ldap modify operations
 277
 278
 279 class AclModifyTests(AclTests):
 280
 281     def setUp(self):
 282         super(AclModifyTests, self).setUp()
 283         self.user_with_wp = "acl_mod_user1"
 284         self.user_with_sm = "acl_mod_user2"
 285         self.user_with_group_sm = "acl_mod_user3"
 286         self.ldb_admin.newuser(self.user_with_wp, self.user_pass)
 287         self.ldb_admin.newuser(self.user_with_sm, self.user_pass)
 288         self.ldb_admin.newuser(self.user_with_group_sm, self.user_pass)
 289         self.ldb_user = self.get_ldb_connection(self.user_with_wp, self.user_pass)
 290         self.ldb_user2 = self.get_ldb_connection(self.user_with_sm, self.user_pass)
 291         self.ldb_user3 = self.get_ldb_connection(self.user_with_group_sm, self.user_pass)
 292         self.user_sid = self.sd_utils.get_object_sid(self.get_user_dn(self.user_with_wp))
 293         self.ldb_admin.newgroup("test_modify_group2", grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
 294         self.ldb_admin.newgroup("test_modify_group3", grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
 295         self.ldb_admin.newuser("test_modify_user2", self.user_pass)
 296
 297     def tearDown(self):
 298         super(AclModifyTests, self).tearDown()
 299         delete_force(self.ldb_admin, self.get_user_dn("test_modify_user1"))
 300         delete_force(self.ldb_admin, "CN=test_modify_group1,CN=Users," + self.base_dn)
 301         delete_force(self.ldb_admin, "CN=test_modify_group2,CN=Users," + self.base_dn)
 302         delete_force(self.ldb_admin, "CN=test_modify_group3,CN=Users," + self.base_dn)
 303         delete_force(self.ldb_admin, "OU=test_modify_ou1," + self.base_dn)
 304         delete_force(self.ldb_admin, self.get_user_dn(self.user_with_wp))
 305         delete_force(self.ldb_admin, self.get_user_dn(self.user_with_sm))
 306         delete_force(self.ldb_admin, self.get_user_dn(self.user_with_group_sm))
 307         delete_force(self.ldb_admin, self.get_user_dn("test_modify_user2"))
 308         delete_force(self.ldb_admin, self.get_user_dn("test_anonymous"))
 309
 310         del self.ldb_user
 311         del self.ldb_user2
 312         del self.ldb_user3
 313
 314     def test_modify_u1(self):
 315         """5 Modify one attribute if you have DS_WRITE_PROPERTY for it"""
 316         mod = "(OA;;WP;bf967953-0de6-11d0-a285-00aa003049e2;;%s)" % str(self.user_sid)
 317         # First test object -- User
 318         print("Testing modify on User object")
 319         self.ldb_admin.newuser("test_modify_user1", self.user_pass)
 320         self.sd_utils.dacl_add_ace(self.get_user_dn("test_modify_user1"), mod)
 321         ldif = """
 322 dn: """ + self.get_user_dn("test_modify_user1") + """
 323 changetype: modify
 324 replace: displayName
 325 displayName: test_changed"""
 326         self.ldb_user.modify_ldif(ldif)
 327         res = self.ldb_admin.search(self.base_dn,
 328                                     expression="(distinguishedName=%s)" % self.get_user_dn("test_modify_user1"))
 329         self.assertEqual(str(res[0]["displayName"][0]), "test_changed")
 330         # Second test object -- Group
 331         print("Testing modify on Group object")
 332         self.ldb_admin.newgroup("test_modify_group1",
 333                                 grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
 334         self.sd_utils.dacl_add_ace("CN=test_modify_group1,CN=Users," + self.base_dn, mod)
 335         ldif = """
 336 dn: CN=test_modify_group1,CN=Users,""" + self.base_dn + """
 337 changetype: modify
 338 replace: displayName
 339 displayName: test_changed"""
 340         self.ldb_user.modify_ldif(ldif)
 341         res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" % str("CN=test_modify_group1,CN=Users," + self.base_dn))
 342         self.assertEqual(str(res[0]["displayName"][0]), "test_changed")
 343         # Third test object -- Organizational Unit
 344         print("Testing modify on OU object")
 345         #delete_force(self.ldb_admin, "OU=test_modify_ou1," + self.base_dn)
 346         self.ldb_admin.create_ou("OU=test_modify_ou1," + self.base_dn)
 347         self.sd_utils.dacl_add_ace("OU=test_modify_ou1," + self.base_dn, mod)
 348         ldif = """
 349 dn: OU=test_modify_ou1,""" + self.base_dn + """
 350 changetype: modify
 351 replace: displayName
 352 displayName: test_changed"""
 353         self.ldb_user.modify_ldif(ldif)
 354         res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" % str("OU=test_modify_ou1," + self.base_dn))
 355         self.assertEqual(str(res[0]["displayName"][0]), "test_changed")
 356
 357     def test_modify_u2(self):
 358         """6 Modify two attributes as you have DS_WRITE_PROPERTY granted only for one of them"""
 359         mod = "(OA;;WP;bf967953-0de6-11d0-a285-00aa003049e2;;%s)" % str(self.user_sid)
 360         # First test object -- User
 361         print("Testing modify on User object")
 362         #delete_force(self.ldb_admin, self.get_user_dn("test_modify_user1"))
 363         self.ldb_admin.newuser("test_modify_user1", self.user_pass)
 364         self.sd_utils.dacl_add_ace(self.get_user_dn("test_modify_user1"), mod)
 365         # Modify on attribute you have rights for
 366         ldif = """
 367 dn: """ + self.get_user_dn("test_modify_user1") + """
 368 changetype: modify
 369 replace: displayName
 370 displayName: test_changed"""
 371         self.ldb_user.modify_ldif(ldif)
 372         res = self.ldb_admin.search(self.base_dn,
 373                                     expression="(distinguishedName=%s)" %
 374                                     self.get_user_dn("test_modify_user1"))
 375         self.assertEqual(str(res[0]["displayName"][0]), "test_changed")
 376         # Modify on attribute you do not have rights for granted
 377         ldif = """
 378 dn: """ + self.get_user_dn("test_modify_user1") + """
 379 changetype: modify
 380 replace: url
 381 url: www.samba.org"""
 382         try:
 383             self.ldb_user.modify_ldif(ldif)
 384         except LdbError as e3:
 385             (num, _) = e3.args
 386             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
 387         else:
 388             # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
 389             self.fail()
 390         # Second test object -- Group
 391         print("Testing modify on Group object")
 392         self.ldb_admin.newgroup("test_modify_group1",
 393                                 grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
 394         self.sd_utils.dacl_add_ace("CN=test_modify_group1,CN=Users," + self.base_dn, mod)
 395         ldif = """
 396 dn: CN=test_modify_group1,CN=Users,""" + self.base_dn + """
 397 changetype: modify
 398 replace: displayName
 399 displayName: test_changed"""
 400         self.ldb_user.modify_ldif(ldif)
 401         res = self.ldb_admin.search(self.base_dn,
 402                                     expression="(distinguishedName=%s)" %
 403                                     str("CN=test_modify_group1,CN=Users," + self.base_dn))
 404         self.assertEqual(str(res[0]["displayName"][0]), "test_changed")
 405         # Modify on attribute you do not have rights for granted
 406         ldif = """
 407 dn: CN=test_modify_group1,CN=Users,""" + self.base_dn + """
 408 changetype: modify
 409 replace: url
 410 url: www.samba.org"""
 411         try:
 412             self.ldb_user.modify_ldif(ldif)
 413         except LdbError as e4:
 414             (num, _) = e4.args
 415             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
 416         else:
 417             # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
 418             self.fail()
 419         # Modify on attribute you do not have rights for granted while also modifying something you do have rights for
 420         ldif = """
 421 dn: CN=test_modify_group1,CN=Users,""" + self.base_dn + """
 422 changetype: modify
 423 replace: url
 424 url: www.samba.org
 425 replace: displayName
 426 displayName: test_changed"""
 427         try:
 428             self.ldb_user.modify_ldif(ldif)
 429         except LdbError as e5:
 430             (num, _) = e5.args
 431             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
 432         else:
 433             # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
 434             self.fail()
 435         # Second test object -- Organizational Unit
 436         print("Testing modify on OU object")
 437         self.ldb_admin.create_ou("OU=test_modify_ou1," + self.base_dn)
 438         self.sd_utils.dacl_add_ace("OU=test_modify_ou1," + self.base_dn, mod)
 439         ldif = """
 440 dn: OU=test_modify_ou1,""" + self.base_dn + """
 441 changetype: modify
 442 replace: displayName
 443 displayName: test_changed"""
 444         self.ldb_user.modify_ldif(ldif)
 445         res = self.ldb_admin.search(self.base_dn,
 446                                     expression="(distinguishedName=%s)" % str("OU=test_modify_ou1,"
 447                                                                               + self.base_dn))
 448         self.assertEqual(str(res[0]["displayName"][0]), "test_changed")
 449         # Modify on attribute you do not have rights for granted
 450         ldif = """
 451 dn: OU=test_modify_ou1,""" + self.base_dn + """
 452 changetype: modify
 453 replace: url
 454 url: www.samba.org"""
 455         try:
 456             self.ldb_user.modify_ldif(ldif)
 457         except LdbError as e6:
 458             (num, _) = e6.args
 459             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
 460         else:
 461             # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
 462             self.fail()
 463
 464     def test_modify_u3(self):
 465         """7 Modify one attribute as you have no what so ever rights granted"""
 466         # First test object -- User
 467         print("Testing modify on User object")
 468         self.ldb_admin.newuser("test_modify_user1", self.user_pass)
 469         # Modify on attribute you do not have rights for granted
 470         ldif = """
 471 dn: """ + self.get_user_dn("test_modify_user1") + """
 472 changetype: modify
 473 replace: url
 474 url: www.samba.org"""
 475         try:
 476             self.ldb_user.modify_ldif(ldif)
 477         except LdbError as e7:
 478             (num, _) = e7.args
 479             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
 480         else:
 481             # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
 482             self.fail()
 483
 484         # Second test object -- Group
 485         print("Testing modify on Group object")
 486         self.ldb_admin.newgroup("test_modify_group1",
 487                                 grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
 488         # Modify on attribute you do not have rights for granted
 489         ldif = """
 490 dn: CN=test_modify_group1,CN=Users,""" + self.base_dn + """
 491 changetype: modify
 492 replace: url
 493 url: www.samba.org"""
 494         try:
 495             self.ldb_user.modify_ldif(ldif)
 496         except LdbError as e8:
 497             (num, _) = e8.args
 498             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
 499         else:
 500             # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
 501             self.fail()
 502
 503         # Second test object -- Organizational Unit
 504         print("Testing modify on OU object")
 505         #delete_force(self.ldb_admin, "OU=test_modify_ou1," + self.base_dn)
 506         self.ldb_admin.create_ou("OU=test_modify_ou1," + self.base_dn)
 507         # Modify on attribute you do not have rights for granted
 508         ldif = """
 509 dn: OU=test_modify_ou1,""" + self.base_dn + """
 510 changetype: modify
 511 replace: url
 512 url: www.samba.org"""
 513         try:
 514             self.ldb_user.modify_ldif(ldif)
 515         except LdbError as e9:
 516             (num, _) = e9.args
 517             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
 518         else:
 519             # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
 520             self.fail()
 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(str(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(str(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(str(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(str(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     def test_change_password7(self):
1575         """Try a password change operation without any CARs given"""
1576         # users have change password by default - remove for negative testing
1577         desc = self.sd_utils.read_sd_on_dn(self.get_user_dn(self.user_with_wp))
1578         sddl = desc.as_sddl(self.domain_sid)
1579         self.sd_utils.modify_sd_on_dn(self.get_user_dn(self.user_with_wp), sddl)
1580         # first change our own password
1581         self.ldb_user2.modify_ldif("""
1582 dn: """ + self.get_user_dn(self.user_with_pc) + """
1583 changetype: modify
1584 delete: unicodePwd
1585 unicodePwd:: """ + base64.b64encode("\"samba123@\"".encode('utf-16-le')).decode('utf8') + """
1586 add: unicodePwd
1587 unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS1\"".encode('utf-16-le')).decode('utf8') + """
1588 """)
1589         # then someone else's
1590         self.ldb_user2.modify_ldif("""
1591 dn: """ + self.get_user_dn(self.user_with_wp) + """
1592 changetype: modify
1593 delete: unicodePwd
1594 unicodePwd:: """ + base64.b64encode("\"samba123@\"".encode('utf-16-le')).decode('utf8') + """
1595 add: unicodePwd
1596 unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS2\"".encode('utf-16-le')).decode('utf8') + """
1597 """)
1598
1599     def test_reset_password1(self):
1600         """Try a user password reset operation (unicodePwd) before and after granting CAR"""
1601         try:
1602             self.ldb_user.modify_ldif("""
1603 dn: """ + self.get_user_dn(self.user_with_wp) + """
1604 changetype: modify
1605 replace: unicodePwd
1606 unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS1\"".encode('utf-16-le')).decode('utf8') + """
1607 """)
1608         except LdbError as e29:
1609             (num, _) = e29.args
1610             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1611         else:
1612             self.fail()
1613         mod = "(OA;;CR;00299570-246d-11d0-a768-00aa006e0529;;PS)"
1614         self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
1615         self.ldb_user.modify_ldif("""
1616 dn: """ + self.get_user_dn(self.user_with_wp) + """
1617 changetype: modify
1618 replace: unicodePwd
1619 unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS1\"".encode('utf-16-le')).decode('utf8') + """
1620 """)
1621
1622     def test_reset_password2(self):
1623         """Try a user password reset operation (userPassword) before and after granting CAR"""
1624         try:
1625             self.ldb_user.modify_ldif("""
1626 dn: """ + self.get_user_dn(self.user_with_wp) + """
1627 changetype: modify
1628 replace: userPassword
1629 userPassword: thatsAcomplPASS1
1630 """)
1631         except LdbError as e30:
1632             (num, _) = e30.args
1633             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1634         else:
1635             self.fail()
1636         mod = "(OA;;CR;00299570-246d-11d0-a768-00aa006e0529;;PS)"
1637         self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
1638         try:
1639             self.ldb_user.modify_ldif("""
1640 dn: """ + self.get_user_dn(self.user_with_wp) + """
1641 changetype: modify
1642 replace: userPassword
1643 userPassword: thatsAcomplPASS1
1644 """)
1645             # This fails on Windows 2000 domain level with constraint violation
1646         except LdbError as e31:
1647             (num, _) = e31.args
1648             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
1649
1650     def test_reset_password3(self):
1651         """Grant WP and see what happens (unicodePwd)"""
1652         mod = "(A;;WP;;;PS)"
1653         self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
1654         try:
1655             self.ldb_user.modify_ldif("""
1656 dn: """ + self.get_user_dn(self.user_with_wp) + """
1657 changetype: modify
1658 replace: unicodePwd
1659 unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS1\"".encode('utf-16-le')).decode('utf8') + """
1660 """)
1661         except LdbError as e32:
1662             (num, _) = e32.args
1663             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1664         else:
1665             self.fail()
1666
1667     def test_reset_password4(self):
1668         """Grant WP and see what happens (userPassword)"""
1669         mod = "(A;;WP;;;PS)"
1670         self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
1671         try:
1672             self.ldb_user.modify_ldif("""
1673 dn: """ + self.get_user_dn(self.user_with_wp) + """
1674 changetype: modify
1675 replace: userPassword
1676 userPassword: thatsAcomplPASS1
1677 """)
1678         except LdbError as e33:
1679             (num, _) = e33.args
1680             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1681         else:
1682             self.fail()
1683
1684     def test_reset_password5(self):
1685         """Explicitly deny WP but grant CAR (unicodePwd)"""
1686         mod = "(D;;WP;;;PS)(OA;;CR;00299570-246d-11d0-a768-00aa006e0529;;PS)"
1687         self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
1688         self.ldb_user.modify_ldif("""
1689 dn: """ + self.get_user_dn(self.user_with_wp) + """
1690 changetype: modify
1691 replace: unicodePwd
1692 unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS1\"".encode('utf-16-le')).decode('utf8') + """
1693 """)
1694
1695     def test_reset_password6(self):
1696         """Explicitly deny WP but grant CAR (userPassword)"""
1697         mod = "(D;;WP;;;PS)(OA;;CR;00299570-246d-11d0-a768-00aa006e0529;;PS)"
1698         self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
1699         try:
1700             self.ldb_user.modify_ldif("""
1701 dn: """ + self.get_user_dn(self.user_with_wp) + """
1702 changetype: modify
1703 replace: userPassword
1704 userPassword: thatsAcomplPASS1
1705 """)
1706             # This fails on Windows 2000 domain level with constraint violation
1707         except LdbError as e34:
1708             (num, _) = e34.args
1709             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
1710
1711
1712 class AclExtendedTests(AclTests):
1713
1714     def setUp(self):
1715         super(AclExtendedTests, self).setUp()
1716         # regular user, will be the creator
1717         self.u1 = "ext_u1"
1718         # regular user
1719         self.u2 = "ext_u2"
1720         # admin user
1721         self.u3 = "ext_u3"
1722         self.ldb_admin.newuser(self.u1, self.user_pass)
1723         self.ldb_admin.newuser(self.u2, self.user_pass)
1724         self.ldb_admin.newuser(self.u3, self.user_pass)
1725         self.ldb_admin.add_remove_group_members("Domain Admins", [self.u3],
1726                                                 add_members_operation=True)
1727         self.ldb_user1 = self.get_ldb_connection(self.u1, self.user_pass)
1728         self.ldb_user2 = self.get_ldb_connection(self.u2, self.user_pass)
1729         self.ldb_user3 = self.get_ldb_connection(self.u3, self.user_pass)
1730         self.user_sid1 = self.sd_utils.get_object_sid(self.get_user_dn(self.u1))
1731         self.user_sid2 = self.sd_utils.get_object_sid(self.get_user_dn(self.u2))
1732
1733     def tearDown(self):
1734         super(AclExtendedTests, self).tearDown()
1735         delete_force(self.ldb_admin, self.get_user_dn(self.u1))
1736         delete_force(self.ldb_admin, self.get_user_dn(self.u2))
1737         delete_force(self.ldb_admin, self.get_user_dn(self.u3))
1738         delete_force(self.ldb_admin, "CN=ext_group1,OU=ext_ou1," + self.base_dn)
1739         delete_force(self.ldb_admin, "ou=ext_ou1," + self.base_dn)
1740
1741         del self.ldb_user1
1742         del self.ldb_user2
1743         del self.ldb_user3
1744
1745     def test_ntSecurityDescriptor(self):
1746         # create empty ou
1747         self.ldb_admin.create_ou("ou=ext_ou1," + self.base_dn)
1748         # give u1 Create children access
1749         mod = "(A;;CC;;;%s)" % str(self.user_sid1)
1750         self.sd_utils.dacl_add_ace("OU=ext_ou1," + self.base_dn, mod)
1751         mod = "(A;;LC;;;%s)" % str(self.user_sid2)
1752         self.sd_utils.dacl_add_ace("OU=ext_ou1," + self.base_dn, mod)
1753         # create a group under that, grant RP to u2
1754         self.ldb_user1.newgroup("ext_group1", groupou="OU=ext_ou1",
1755                                 grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
1756         mod = "(A;;RP;;;%s)" % str(self.user_sid2)
1757         self.sd_utils.dacl_add_ace("CN=ext_group1,OU=ext_ou1," + self.base_dn, mod)
1758         # u2 must not read the descriptor
1759         res = self.ldb_user2.search("CN=ext_group1,OU=ext_ou1," + self.base_dn,
1760                                     SCOPE_BASE, None, ["nTSecurityDescriptor"])
1761         self.assertNotEqual(len(res), 0)
1762         self.assertFalse("nTSecurityDescriptor" in res[0].keys())
1763         # grant RC to u2 - still no access
1764         mod = "(A;;RC;;;%s)" % str(self.user_sid2)
1765         self.sd_utils.dacl_add_ace("CN=ext_group1,OU=ext_ou1," + self.base_dn, mod)
1766         res = self.ldb_user2.search("CN=ext_group1,OU=ext_ou1," + self.base_dn,
1767                                     SCOPE_BASE, None, ["nTSecurityDescriptor"])
1768         self.assertNotEqual(len(res), 0)
1769         self.assertFalse("nTSecurityDescriptor" in res[0].keys())
1770         # u3 is member of administrators group, should be able to read sd
1771         res = self.ldb_user3.search("CN=ext_group1,OU=ext_ou1," + self.base_dn,
1772                                     SCOPE_BASE, None, ["nTSecurityDescriptor"])
1773         self.assertEqual(len(res), 1)
1774         self.assertTrue("nTSecurityDescriptor" in res[0].keys())
1775
1776
1777 class AclUndeleteTests(AclTests):
1778
1779     def setUp(self):
1780         super(AclUndeleteTests, self).setUp()
1781         self.regular_user = "undeleter1"
1782         self.ou1 = "OU=undeleted_ou,"
1783         self.testuser1 = "to_be_undeleted1"
1784         self.testuser2 = "to_be_undeleted2"
1785         self.testuser3 = "to_be_undeleted3"
1786         self.testuser4 = "to_be_undeleted4"
1787         self.testuser5 = "to_be_undeleted5"
1788         self.testuser6 = "to_be_undeleted6"
1789
1790         self.new_dn_ou = "CN=" + self.testuser4 + "," + self.ou1 + self.base_dn
1791
1792         # Create regular user
1793         self.testuser1_dn = self.get_user_dn(self.testuser1)
1794         self.testuser2_dn = self.get_user_dn(self.testuser2)
1795         self.testuser3_dn = self.get_user_dn(self.testuser3)
1796         self.testuser4_dn = self.get_user_dn(self.testuser4)
1797         self.testuser5_dn = self.get_user_dn(self.testuser5)
1798         self.deleted_dn1 = self.create_delete_user(self.testuser1)
1799         self.deleted_dn2 = self.create_delete_user(self.testuser2)
1800         self.deleted_dn3 = self.create_delete_user(self.testuser3)
1801         self.deleted_dn4 = self.create_delete_user(self.testuser4)
1802         self.deleted_dn5 = self.create_delete_user(self.testuser5)
1803
1804         self.ldb_admin.create_ou(self.ou1 + self.base_dn)
1805
1806         self.ldb_admin.newuser(self.regular_user, self.user_pass)
1807         self.ldb_admin.add_remove_group_members("Domain Admins", [self.regular_user],
1808                                                 add_members_operation=True)
1809         self.ldb_user = self.get_ldb_connection(self.regular_user, self.user_pass)
1810         self.sid = self.sd_utils.get_object_sid(self.get_user_dn(self.regular_user))
1811
1812     def tearDown(self):
1813         super(AclUndeleteTests, self).tearDown()
1814         delete_force(self.ldb_admin, self.get_user_dn(self.regular_user))
1815         delete_force(self.ldb_admin, self.get_user_dn(self.testuser1))
1816         delete_force(self.ldb_admin, self.get_user_dn(self.testuser2))
1817         delete_force(self.ldb_admin, self.get_user_dn(self.testuser3))
1818         delete_force(self.ldb_admin, self.get_user_dn(self.testuser4))
1819         delete_force(self.ldb_admin, self.get_user_dn(self.testuser5))
1820         delete_force(self.ldb_admin, self.new_dn_ou)
1821         delete_force(self.ldb_admin, self.ou1 + self.base_dn)
1822
1823         del self.ldb_user
1824
1825     def GUID_string(self, guid):
1826         return get_string(ldb.schema_format_value("objectGUID", guid))
1827
1828     def create_delete_user(self, new_user):
1829         self.ldb_admin.newuser(new_user, self.user_pass)
1830
1831         res = self.ldb_admin.search(expression="(objectClass=*)",
1832                                     base=self.get_user_dn(new_user),
1833                                     scope=SCOPE_BASE,
1834                                     controls=["show_deleted:1"])
1835         guid = res[0]["objectGUID"][0]
1836         self.ldb_admin.delete(self.get_user_dn(new_user))
1837         res = self.ldb_admin.search(base="<GUID=%s>" % self.GUID_string(guid),
1838                                     scope=SCOPE_BASE, controls=["show_deleted:1"])
1839         self.assertEquals(len(res), 1)
1840         return str(res[0].dn)
1841
1842     def undelete_deleted(self, olddn, newdn):
1843         msg = Message()
1844         msg.dn = Dn(self.ldb_user, olddn)
1845         msg["isDeleted"] = MessageElement([], FLAG_MOD_DELETE, "isDeleted")
1846         msg["distinguishedName"] = MessageElement([newdn], FLAG_MOD_REPLACE, "distinguishedName")
1847         res = self.ldb_user.modify(msg, ["show_recycled:1"])
1848
1849     def undelete_deleted_with_mod(self, olddn, newdn):
1850         msg = Message()
1851         msg.dn = Dn(ldb, olddn)
1852         msg["isDeleted"] = MessageElement([], FLAG_MOD_DELETE, "isDeleted")
1853         msg["distinguishedName"] = MessageElement([newdn], FLAG_MOD_REPLACE, "distinguishedName")
1854         msg["url"] = MessageElement(["www.samba.org"], FLAG_MOD_REPLACE, "url")
1855         res = self.ldb_user.modify(msg, ["show_deleted:1"])
1856
1857     def test_undelete(self):
1858         # it appears the user has to have LC on the old parent to be able to move the object
1859         # otherwise we get no such object. Since only System can modify the SD on deleted object
1860         # we cannot grant this permission via LDAP, and this leaves us with "negative" tests at the moment
1861
1862         # deny write property on rdn, should fail
1863         mod = "(OD;;WP;bf967a0e-0de6-11d0-a285-00aa003049e2;;%s)" % str(self.sid)
1864         self.sd_utils.dacl_add_ace(self.deleted_dn1, mod)
1865         try:
1866             self.undelete_deleted(self.deleted_dn1, self.testuser1_dn)
1867             self.fail()
1868         except LdbError as e35:
1869             (num, _) = e35.args
1870             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1871
1872         # seems that permissions on isDeleted and distinguishedName are irrelevant
1873         mod = "(OD;;WP;bf96798f-0de6-11d0-a285-00aa003049e2;;%s)" % str(self.sid)
1874         self.sd_utils.dacl_add_ace(self.deleted_dn2, mod)
1875         mod = "(OD;;WP;bf9679e4-0de6-11d0-a285-00aa003049e2;;%s)" % str(self.sid)
1876         self.sd_utils.dacl_add_ace(self.deleted_dn2, mod)
1877         self.undelete_deleted(self.deleted_dn2, self.testuser2_dn)
1878
1879         # attempt undelete with simultanious addition of url, WP to which is denied
1880         mod = "(OD;;WP;9a9a0221-4a5b-11d1-a9c3-0000f80367c1;;%s)" % str(self.sid)
1881         self.sd_utils.dacl_add_ace(self.deleted_dn3, mod)
1882         try:
1883             self.undelete_deleted_with_mod(self.deleted_dn3, self.testuser3_dn)
1884             self.fail()
1885         except LdbError as e36:
1886             (num, _) = e36.args
1887             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1888
1889         # undelete in an ou, in which we have no right to create children
1890         mod = "(D;;CC;;;%s)" % str(self.sid)
1891         self.sd_utils.dacl_add_ace(self.ou1 + self.base_dn, mod)
1892         try:
1893             self.undelete_deleted(self.deleted_dn4, self.new_dn_ou)
1894             self.fail()
1895         except LdbError as e37:
1896             (num, _) = e37.args
1897             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1898
1899         # delete is not required
1900         mod = "(D;;SD;;;%s)" % str(self.sid)
1901         self.sd_utils.dacl_add_ace(self.deleted_dn5, mod)
1902         self.undelete_deleted(self.deleted_dn5, self.testuser5_dn)
1903
1904         # deny Reanimate-Tombstone, should fail
1905         mod = "(OD;;CR;45ec5156-db7e-47bb-b53f-dbeb2d03c40f;;%s)" % str(self.sid)
1906         self.sd_utils.dacl_add_ace(self.base_dn, mod)
1907         try:
1908             self.undelete_deleted(self.deleted_dn4, self.testuser4_dn)
1909             self.fail()
1910         except LdbError as e38:
1911             (num, _) = e38.args
1912             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
1913
1914
1915 class AclSPNTests(AclTests):
1916
1917     def setUp(self):
1918         super(AclSPNTests, self).setUp()
1919         self.dcname = "TESTSRV8"
1920         self.rodcname = "TESTRODC8"
1921         self.computername = "testcomp8"
1922         self.test_user = "spn_test_user8"
1923         self.computerdn = "CN=%s,CN=computers,%s" % (self.computername, self.base_dn)
1924         self.dc_dn = "CN=%s,OU=Domain Controllers,%s" % (self.dcname, self.base_dn)
1925         self.site = "Default-First-Site-Name"
1926         self.rodcctx = DCJoinContext(server=host, creds=creds, lp=lp,
1927                                      site=self.site, netbios_name=self.rodcname,
1928                                      targetdir=None, domain=None)
1929         self.dcctx = DCJoinContext(server=host, creds=creds, lp=lp,
1930                                    site=self.site, netbios_name=self.dcname,
1931                                    targetdir=None, domain=None)
1932         self.ldb_admin.newuser(self.test_user, self.user_pass)
1933         self.ldb_user1 = self.get_ldb_connection(self.test_user, self.user_pass)
1934         self.user_sid1 = self.sd_utils.get_object_sid(self.get_user_dn(self.test_user))
1935         self.create_computer(self.computername, self.dcctx.dnsdomain)
1936         self.create_rodc(self.rodcctx)
1937         self.create_dc(self.dcctx)
1938
1939     def tearDown(self):
1940         super(AclSPNTests, self).tearDown()
1941         self.rodcctx.cleanup_old_join()
1942         self.dcctx.cleanup_old_join()
1943         delete_force(self.ldb_admin, "cn=%s,cn=computers,%s" % (self.computername, self.base_dn))
1944         delete_force(self.ldb_admin, self.get_user_dn(self.test_user))
1945
1946         del self.ldb_user1
1947
1948     def replace_spn(self, _ldb, dn, spn):
1949         print("Setting spn %s on %s" % (spn, dn))
1950         res = self.ldb_admin.search(dn, expression="(objectClass=*)",
1951                                     scope=SCOPE_BASE, attrs=["servicePrincipalName"])
1952         if "servicePrincipalName" in res[0].keys():
1953             flag = FLAG_MOD_REPLACE
1954         else:
1955             flag = FLAG_MOD_ADD
1956
1957         msg = Message()
1958         msg.dn = Dn(self.ldb_admin, dn)
1959         msg["servicePrincipalName"] = MessageElement(spn, flag,
1960                                                      "servicePrincipalName")
1961         _ldb.modify(msg)
1962
1963     def create_computer(self, computername, domainname):
1964         dn = "CN=%s,CN=computers,%s" % (computername, self.base_dn)
1965         samaccountname = computername + "$"
1966         dnshostname = "%s.%s" % (computername, domainname)
1967         self.ldb_admin.add({
1968             "dn": dn,
1969             "objectclass": "computer",
1970             "sAMAccountName": samaccountname,
1971             "userAccountControl": str(samba.dsdb.UF_WORKSTATION_TRUST_ACCOUNT),
1972             "dNSHostName": dnshostname})
1973
1974     # same as for join_RODC, but do not set any SPNs
1975     def create_rodc(self, ctx):
1976         ctx.nc_list = [ctx.base_dn, ctx.config_dn, ctx.schema_dn]
1977         ctx.full_nc_list = [ctx.base_dn, ctx.config_dn, ctx.schema_dn]
1978         ctx.krbtgt_dn = "CN=krbtgt_%s,CN=Users,%s" % (ctx.myname, ctx.base_dn)
1979
1980         ctx.never_reveal_sid = ["<SID=%s-%s>" % (ctx.domsid, security.DOMAIN_RID_RODC_DENY),
1981                                 "<SID=%s>" % security.SID_BUILTIN_ADMINISTRATORS,
1982                                 "<SID=%s>" % security.SID_BUILTIN_SERVER_OPERATORS,
1983                                 "<SID=%s>" % security.SID_BUILTIN_BACKUP_OPERATORS,
1984                                 "<SID=%s>" % security.SID_BUILTIN_ACCOUNT_OPERATORS]
1985         ctx.reveal_sid = "<SID=%s-%s>" % (ctx.domsid, security.DOMAIN_RID_RODC_ALLOW)
1986
1987         mysid = ctx.get_mysid()
1988         admin_dn = "<SID=%s>" % mysid
1989         ctx.managedby = admin_dn
1990
1991         ctx.userAccountControl = (samba.dsdb.UF_WORKSTATION_TRUST_ACCOUNT |
1992                                   samba.dsdb.UF_TRUSTED_TO_AUTHENTICATE_FOR_DELEGATION |
1993                                   samba.dsdb.UF_PARTIAL_SECRETS_ACCOUNT)
1994
1995         ctx.connection_dn = "CN=RODC Connection (FRS),%s" % ctx.ntds_dn
1996         ctx.secure_channel_type = misc.SEC_CHAN_RODC
1997         ctx.RODC = True
1998         ctx.replica_flags = (drsuapi.DRSUAPI_DRS_INIT_SYNC |
1999                              drsuapi.DRSUAPI_DRS_PER_SYNC |
2000                              drsuapi.DRSUAPI_DRS_GET_ANC |
2001                              drsuapi.DRSUAPI_DRS_NEVER_SYNCED |
2002                              drsuapi.DRSUAPI_DRS_SPECIAL_SECRET_PROCESSING)
2003
2004         ctx.join_add_objects()
2005
2006     def create_dc(self, ctx):
2007         ctx.nc_list = [ctx.base_dn, ctx.config_dn, ctx.schema_dn]
2008         ctx.full_nc_list = [ctx.base_dn, ctx.config_dn, ctx.schema_dn]
2009         ctx.userAccountControl = samba.dsdb.UF_SERVER_TRUST_ACCOUNT | samba.dsdb.UF_TRUSTED_FOR_DELEGATION
2010         ctx.secure_channel_type = misc.SEC_CHAN_BDC
2011         ctx.replica_flags = (drsuapi.DRSUAPI_DRS_WRIT_REP |
2012                              drsuapi.DRSUAPI_DRS_INIT_SYNC |
2013                              drsuapi.DRSUAPI_DRS_PER_SYNC |
2014                              drsuapi.DRSUAPI_DRS_FULL_SYNC_IN_PROGRESS |
2015                              drsuapi.DRSUAPI_DRS_NEVER_SYNCED)
2016
2017         ctx.join_add_objects()
2018
2019     def dc_spn_test(self, ctx):
2020         netbiosdomain = self.dcctx.get_domain_name()
2021         try:
2022             self.replace_spn(self.ldb_user1, ctx.acct_dn, "HOST/%s/%s" % (ctx.myname, netbiosdomain))
2023         except LdbError as e39:
2024             (num, _) = e39.args
2025             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
2026
2027         mod = "(OA;;SW;f3a64788-5306-11d1-a9c5-0000f80367c1;;%s)" % str(self.user_sid1)
2028         self.sd_utils.dacl_add_ace(ctx.acct_dn, mod)
2029         self.replace_spn(self.ldb_user1, ctx.acct_dn, "HOST/%s/%s" % (ctx.myname, netbiosdomain))
2030         self.replace_spn(self.ldb_user1, ctx.acct_dn, "HOST/%s" % (ctx.myname))
2031         self.replace_spn(self.ldb_user1, ctx.acct_dn, "HOST/%s.%s/%s" %
2032                          (ctx.myname, ctx.dnsdomain, netbiosdomain))
2033         self.replace_spn(self.ldb_user1, ctx.acct_dn, "HOST/%s/%s" % (ctx.myname, ctx.dnsdomain))
2034         self.replace_spn(self.ldb_user1, ctx.acct_dn, "HOST/%s.%s/%s" %
2035                          (ctx.myname, ctx.dnsdomain, ctx.dnsdomain))
2036         self.replace_spn(self.ldb_user1, ctx.acct_dn, "GC/%s.%s/%s" %
2037                          (ctx.myname, ctx.dnsdomain, ctx.dnsforest))
2038         self.replace_spn(self.ldb_user1, ctx.acct_dn, "ldap/%s/%s" % (ctx.myname, netbiosdomain))
2039         self.replace_spn(self.ldb_user1, ctx.acct_dn, "ldap/%s.%s/%s" %
2040                          (ctx.myname, ctx.dnsdomain, netbiosdomain))
2041         self.replace_spn(self.ldb_user1, ctx.acct_dn, "ldap/%s" % (ctx.myname))
2042         self.replace_spn(self.ldb_user1, ctx.acct_dn, "ldap/%s/%s" % (ctx.myname, ctx.dnsdomain))
2043         self.replace_spn(self.ldb_user1, ctx.acct_dn, "ldap/%s.%s/%s" %
2044                          (ctx.myname, ctx.dnsdomain, ctx.dnsdomain))
2045         self.replace_spn(self.ldb_user1, ctx.acct_dn, "DNS/%s/%s" % (ctx.myname, ctx.dnsdomain))
2046         self.replace_spn(self.ldb_user1, ctx.acct_dn, "RestrictedKrbHost/%s/%s" %
2047                          (ctx.myname, ctx.dnsdomain))
2048         self.replace_spn(self.ldb_user1, ctx.acct_dn, "RestrictedKrbHost/%s" %
2049                          (ctx.myname))
2050         self.replace_spn(self.ldb_user1, ctx.acct_dn, "Dfsr-12F9A27C-BF97-4787-9364-D31B6C55EB04/%s/%s" %
2051                          (ctx.myname, ctx.dnsdomain))
2052         self.replace_spn(self.ldb_user1, ctx.acct_dn, "NtFrs-88f5d2bd-b646-11d2-a6d3-00c04fc9b232/%s/%s" %
2053                          (ctx.myname, ctx.dnsdomain))
2054         self.replace_spn(self.ldb_user1, ctx.acct_dn, "ldap/%s._msdcs.%s" %
2055                          (ctx.ntds_guid, ctx.dnsdomain))
2056
2057         # the following spns do not match the restrictions and should fail
2058         try:
2059             self.replace_spn(self.ldb_user1, ctx.acct_dn, "ldap/%s.%s/ForestDnsZones.%s" %
2060                              (ctx.myname, ctx.dnsdomain, ctx.dnsdomain))
2061         except LdbError as e40:
2062             (num, _) = e40.args
2063             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
2064         try:
2065             self.replace_spn(self.ldb_user1, ctx.acct_dn, "ldap/%s.%s/DomainDnsZones.%s" %
2066                              (ctx.myname, ctx.dnsdomain, ctx.dnsdomain))
2067         except LdbError as e41:
2068             (num, _) = e41.args
2069             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
2070         try:
2071             self.replace_spn(self.ldb_user1, ctx.acct_dn, "nosuchservice/%s/%s" % ("abcd", "abcd"))
2072         except LdbError as e42:
2073             (num, _) = e42.args
2074             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
2075         try:
2076             self.replace_spn(self.ldb_user1, ctx.acct_dn, "GC/%s.%s/%s" %
2077                              (ctx.myname, ctx.dnsdomain, netbiosdomain))
2078         except LdbError as e43:
2079             (num, _) = e43.args
2080             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
2081         try:
2082             self.replace_spn(self.ldb_user1, ctx.acct_dn, "E3514235-4B06-11D1-AB04-00C04FC2DCD2/%s/%s" %
2083                              (ctx.ntds_guid, ctx.dnsdomain))
2084         except LdbError as e44:
2085             (num, _) = e44.args
2086             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
2087
2088     def test_computer_spn(self):
2089         # with WP, any value can be set
2090         netbiosdomain = self.dcctx.get_domain_name()
2091         self.replace_spn(self.ldb_admin, self.computerdn, "HOST/%s/%s" %
2092                          (self.computername, netbiosdomain))
2093         self.replace_spn(self.ldb_admin, self.computerdn, "HOST/%s" % (self.computername))
2094         self.replace_spn(self.ldb_admin, self.computerdn, "HOST/%s.%s/%s" %
2095                          (self.computername, self.dcctx.dnsdomain, netbiosdomain))
2096         self.replace_spn(self.ldb_admin, self.computerdn, "HOST/%s/%s" %
2097                          (self.computername, self.dcctx.dnsdomain))
2098         self.replace_spn(self.ldb_admin, self.computerdn, "HOST/%s.%s/%s" %
2099                          (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsdomain))
2100         self.replace_spn(self.ldb_admin, self.computerdn, "GC/%s.%s/%s" %
2101                          (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsforest))
2102         self.replace_spn(self.ldb_admin, self.computerdn, "ldap/%s/%s" % (self.computername, netbiosdomain))
2103         self.replace_spn(self.ldb_admin, self.computerdn, "ldap/%s.%s/ForestDnsZones.%s" %
2104                          (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsdomain))
2105         self.replace_spn(self.ldb_admin, self.computerdn, "ldap/%s.%s/DomainDnsZones.%s" %
2106                          (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsdomain))
2107         self.replace_spn(self.ldb_admin, self.computerdn, "ldap/%s.%s/%s" %
2108                          (self.computername, self.dcctx.dnsdomain, netbiosdomain))
2109         self.replace_spn(self.ldb_admin, self.computerdn, "ldap/%s" % (self.computername))
2110         self.replace_spn(self.ldb_admin, self.computerdn, "ldap/%s/%s" %
2111                          (self.computername, self.dcctx.dnsdomain))
2112         self.replace_spn(self.ldb_admin, self.computerdn, "ldap/%s.%s/%s" %
2113                          (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsdomain))
2114         self.replace_spn(self.ldb_admin, self.computerdn, "DNS/%s/%s" %
2115                          (self.computername, self.dcctx.dnsdomain))
2116         self.replace_spn(self.ldb_admin, self.computerdn, "RestrictedKrbHost/%s/%s" %
2117                          (self.computername, self.dcctx.dnsdomain))
2118         self.replace_spn(self.ldb_admin, self.computerdn, "RestrictedKrbHost/%s" %
2119                          (self.computername))
2120         self.replace_spn(self.ldb_admin, self.computerdn, "Dfsr-12F9A27C-BF97-4787-9364-D31B6C55EB04/%s/%s" %
2121                          (self.computername, self.dcctx.dnsdomain))
2122         self.replace_spn(self.ldb_admin, self.computerdn, "NtFrs-88f5d2bd-b646-11d2-a6d3-00c04fc9b232/%s/%s" %
2123                          (self.computername, self.dcctx.dnsdomain))
2124         self.replace_spn(self.ldb_admin, self.computerdn, "nosuchservice/%s/%s" % ("abcd", "abcd"))
2125
2126         # user has neither WP nor Validated-SPN, access denied expected
2127         try:
2128             self.replace_spn(self.ldb_user1, self.computerdn, "HOST/%s/%s" % (self.computername, netbiosdomain))
2129         except LdbError as e45:
2130             (num, _) = e45.args
2131             self.assertEquals(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
2132
2133         mod = "(OA;;SW;f3a64788-5306-11d1-a9c5-0000f80367c1;;%s)" % str(self.user_sid1)
2134         self.sd_utils.dacl_add_ace(self.computerdn, mod)
2135         # grant Validated-SPN and check which values are accepted
2136         # see 3.1.1.5.3.1.1.4 servicePrincipalName for reference
2137
2138         # for regular computer objects we shouldalways get constraint violation
2139
2140         # This does not pass against Windows, although it should according to docs
2141         self.replace_spn(self.ldb_user1, self.computerdn, "HOST/%s" % (self.computername))
2142         self.replace_spn(self.ldb_user1, self.computerdn, "HOST/%s.%s" %
2143                          (self.computername, self.dcctx.dnsdomain))
2144
2145         try:
2146             self.replace_spn(self.ldb_user1, self.computerdn, "HOST/%s/%s" % (self.computername, netbiosdomain))
2147         except LdbError as e46:
2148             (num, _) = e46.args
2149             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
2150         try:
2151             self.replace_spn(self.ldb_user1, self.computerdn, "HOST/%s.%s/%s" %
2152                              (self.computername, self.dcctx.dnsdomain, netbiosdomain))
2153         except LdbError as e47:
2154             (num, _) = e47.args
2155             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
2156         try:
2157             self.replace_spn(self.ldb_user1, self.computerdn, "HOST/%s/%s" %
2158                              (self.computername, self.dcctx.dnsdomain))
2159         except LdbError as e48:
2160             (num, _) = e48.args
2161             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
2162         try:
2163             self.replace_spn(self.ldb_user1, self.computerdn, "HOST/%s.%s/%s" %
2164                              (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsdomain))
2165         except LdbError as e49:
2166             (num, _) = e49.args
2167             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
2168         try:
2169             self.replace_spn(self.ldb_user1, self.computerdn, "GC/%s.%s/%s" %
2170                              (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsforest))
2171         except LdbError as e50:
2172             (num, _) = e50.args
2173             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
2174         try:
2175             self.replace_spn(self.ldb_user1, self.computerdn, "ldap/%s/%s" % (self.computername, netbiosdomain))
2176         except LdbError as e51:
2177             (num, _) = e51.args
2178             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
2179         try:
2180             self.replace_spn(self.ldb_user1, self.computerdn, "ldap/%s.%s/ForestDnsZones.%s" %
2181                              (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsdomain))
2182         except LdbError as e52:
2183             (num, _) = e52.args
2184             self.assertEquals(num, ERR_CONSTRAINT_VIOLATION)
2185
2186     def test_spn_rwdc(self):
2187         self.dc_spn_test(self.dcctx)
2188
2189     def test_spn_rodc(self):
2190         self.dc_spn_test(self.rodcctx)
2191
2192
2193 # Important unit running information
2194
2195 ldb = SamDB(ldaphost, credentials=creds, session_info=system_session(lp), lp=lp)
2196
2197 TestProgram(module=__name__, opts=subunitopts)