summaryrefslogtreecommitdiffstats
path: root/source4/dsdb/tests/python/acl.py
diff options
context:
space:
mode:
authorDaniel Baumann <daniel.baumann@progress-linux.org>2024-05-05 17:47:29 +0000
committerDaniel Baumann <daniel.baumann@progress-linux.org>2024-05-05 17:47:29 +0000
commit4f5791ebd03eaec1c7da0865a383175b05102712 (patch)
tree8ce7b00f7a76baa386372422adebbe64510812d4 /source4/dsdb/tests/python/acl.py
parentInitial commit. (diff)
downloadsamba-4f5791ebd03eaec1c7da0865a383175b05102712.tar.xz
samba-4f5791ebd03eaec1c7da0865a383175b05102712.zip
Adding upstream version 2:4.17.12+dfsg.upstream/2%4.17.12+dfsgupstream
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'source4/dsdb/tests/python/acl.py')
-rwxr-xr-xsource4/dsdb/tests/python/acl.py3395
1 files changed, 3395 insertions, 0 deletions
diff --git a/source4/dsdb/tests/python/acl.py b/source4/dsdb/tests/python/acl.py
new file mode 100755
index 0000000..eb2bafd
--- /dev/null
+++ b/source4/dsdb/tests/python/acl.py
@@ -0,0 +1,3395 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+# This is unit with tests for LDAP access checks
+
+import optparse
+import sys
+import base64
+import re
+sys.path.insert(0, "bin/python")
+import samba
+
+from samba.tests import DynamicTestCase
+from samba.tests.subunitrun import SubunitOptions, TestProgram
+from samba.common import get_string
+
+import samba.getopt as options
+from samba.join import DCJoinContext
+
+from ldb import (
+ SCOPE_BASE, SCOPE_ONELEVEL, SCOPE_SUBTREE, LdbError, ERR_NO_SUCH_OBJECT,
+ ERR_UNWILLING_TO_PERFORM, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+from ldb import ERR_CONSTRAINT_VIOLATION
+from ldb import ERR_OPERATIONS_ERROR
+from ldb import Message, MessageElement, Dn
+from ldb import FLAG_MOD_REPLACE, FLAG_MOD_ADD, FLAG_MOD_DELETE
+from samba.dcerpc import security, drsuapi, misc
+
+from samba.auth import system_session
+from samba import gensec, sd_utils
+from samba.samdb import SamDB
+from samba.credentials import Credentials, DONT_USE_KERBEROS
+import samba.tests
+from samba.tests import delete_force
+import samba.dsdb
+from samba.tests.password_test import PasswordCommon
+
+parser = optparse.OptionParser("acl.py [options] <host>")
+sambaopts = options.SambaOptions(parser)
+parser.add_option_group(sambaopts)
+parser.add_option_group(options.VersionOptions(parser))
+
+# use command line creds if available
+credopts = options.CredentialsOptions(parser)
+parser.add_option_group(credopts)
+subunitopts = SubunitOptions(parser)
+parser.add_option_group(subunitopts)
+
+opts, args = parser.parse_args()
+
+if len(args) < 1:
+ parser.print_usage()
+ sys.exit(1)
+
+host = args[0]
+if "://" not in host:
+ ldaphost = "ldap://%s" % host
+else:
+ ldaphost = host
+ start = host.rindex("://")
+ host = host.lstrip(start + 3)
+
+lp = sambaopts.get_loadparm()
+creds = credopts.get_credentials(lp)
+creds.set_gensec_features(creds.get_gensec_features() | gensec.FEATURE_SEAL)
+
+#
+# Tests start here
+#
+
+
+class AclTests(samba.tests.TestCase):
+
+ def setUp(self):
+ super(AclTests, self).setUp()
+ self.ldb_admin = SamDB(ldaphost, credentials=creds, session_info=system_session(lp), lp=lp)
+ self.base_dn = self.ldb_admin.domain_dn()
+ self.domain_sid = security.dom_sid(self.ldb_admin.get_domain_sid())
+ self.user_pass = "samba123@"
+ self.configuration_dn = self.ldb_admin.get_config_basedn().get_linearized()
+ self.sd_utils = sd_utils.SDUtils(self.ldb_admin)
+ self.addCleanup(self.delete_admin_connection)
+ # used for anonymous login
+ self.creds_tmp = Credentials()
+ self.creds_tmp.set_username("")
+ self.creds_tmp.set_password("")
+ self.creds_tmp.set_domain(creds.get_domain())
+ self.creds_tmp.set_realm(creds.get_realm())
+ self.creds_tmp.set_workstation(creds.get_workstation())
+ print("baseDN: %s" % self.base_dn)
+
+ def get_user_dn(self, name):
+ return "CN=%s,CN=Users,%s" % (name, self.base_dn)
+
+ def get_ldb_connection(self, target_username, target_password):
+ creds_tmp = Credentials()
+ creds_tmp.set_username(target_username)
+ creds_tmp.set_password(target_password)
+ creds_tmp.set_domain(creds.get_domain())
+ creds_tmp.set_realm(creds.get_realm())
+ creds_tmp.set_workstation(creds.get_workstation())
+ creds_tmp.set_gensec_features(creds_tmp.get_gensec_features()
+ | gensec.FEATURE_SEAL)
+ creds_tmp.set_kerberos_state(DONT_USE_KERBEROS) # kinit is too expensive to use in a tight loop
+ ldb_target = SamDB(url=ldaphost, credentials=creds_tmp, lp=lp)
+ return ldb_target
+
+ # Test if we have any additional groups for users than default ones
+ def assert_user_no_group_member(self, username):
+ res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" % self.get_user_dn(username))
+ try:
+ self.assertEqual(res[0]["memberOf"][0], "")
+ except KeyError:
+ pass
+ else:
+ self.fail()
+
+ def delete_admin_connection(self):
+ del self.sd_utils
+ del self.ldb_admin
+
+# tests on ldap add operations
+
+
+class AclAddTests(AclTests):
+
+ def setUp(self):
+ super(AclAddTests, self).setUp()
+ # Domain admin that will be creator of OU parent-child structure
+ self.usr_admin_owner = "acl_add_user1"
+ # Second domain admin that will not be creator of OU parent-child structure
+ self.usr_admin_not_owner = "acl_add_user2"
+ # Regular user
+ self.regular_user = "acl_add_user3"
+ self.test_user1 = "test_add_user1"
+ self.test_group1 = "test_add_group1"
+ self.ou1 = "OU=test_add_ou1"
+ self.ou2 = "OU=test_add_ou2,%s" % self.ou1
+ self.ldb_admin.newuser(self.usr_admin_owner, self.user_pass)
+ self.ldb_admin.newuser(self.usr_admin_not_owner, self.user_pass)
+ self.ldb_admin.newuser(self.regular_user, self.user_pass)
+
+ # add admins to the Domain Admins group
+ self.ldb_admin.add_remove_group_members("Domain Admins", [self.usr_admin_owner],
+ add_members_operation=True)
+ self.ldb_admin.add_remove_group_members("Domain Admins", [self.usr_admin_not_owner],
+ add_members_operation=True)
+
+ self.ldb_owner = self.get_ldb_connection(self.usr_admin_owner, self.user_pass)
+ self.ldb_notowner = self.get_ldb_connection(self.usr_admin_not_owner, self.user_pass)
+ self.ldb_user = self.get_ldb_connection(self.regular_user, self.user_pass)
+
+ def tearDown(self):
+ super(AclAddTests, self).tearDown()
+ delete_force(self.ldb_admin, "CN=%s,%s,%s" %
+ (self.test_user1, self.ou2, self.base_dn))
+ delete_force(self.ldb_admin, "CN=%s,%s,%s" %
+ (self.test_group1, self.ou2, self.base_dn))
+ delete_force(self.ldb_admin, "%s,%s" % (self.ou2, self.base_dn))
+ delete_force(self.ldb_admin, "%s,%s" % (self.ou1, self.base_dn))
+ delete_force(self.ldb_admin, self.get_user_dn(self.usr_admin_owner))
+ delete_force(self.ldb_admin, self.get_user_dn(self.usr_admin_not_owner))
+ delete_force(self.ldb_admin, self.get_user_dn(self.regular_user))
+ delete_force(self.ldb_admin, self.get_user_dn("test_add_anonymous"))
+
+ del self.ldb_notowner
+ del self.ldb_owner
+ del self.ldb_user
+
+ # Make sure top OU is deleted (and so everything under it)
+ def assert_top_ou_deleted(self):
+ res = self.ldb_admin.search(self.base_dn,
+ expression="(distinguishedName=%s,%s)" % (
+ "OU=test_add_ou1", self.base_dn))
+ self.assertEqual(len(res), 0)
+
+ def test_add_u1(self):
+ """Testing OU with the rights of Domain Admin not creator of the OU """
+ self.assert_top_ou_deleted()
+ # Change descriptor for top level OU
+ self.ldb_owner.create_ou("OU=test_add_ou1," + self.base_dn)
+ self.ldb_owner.create_ou("OU=test_add_ou2,OU=test_add_ou1," + self.base_dn)
+ user_sid = self.sd_utils.get_object_sid(self.get_user_dn(self.usr_admin_not_owner))
+ mod = "(D;CI;WPCC;;;%s)" % str(user_sid)
+ self.sd_utils.dacl_add_ace("OU=test_add_ou1," + self.base_dn, mod)
+ # Test user and group creation with another domain admin's credentials
+ self.ldb_notowner.newuser(self.test_user1, self.user_pass, userou=self.ou2)
+ self.ldb_notowner.newgroup("test_add_group1", groupou="OU=test_add_ou2,OU=test_add_ou1",
+ grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
+ # Make sure we HAVE created the two objects -- user and group
+ # !!! We should not be able to do that, but however because of ACE ordering our inherited Deny ACE
+ # !!! comes after explicit (A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA) that comes from somewhere
+ 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))
+ self.assertTrue(len(res) > 0)
+ 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))
+ self.assertTrue(len(res) > 0)
+
+ def test_add_u2(self):
+ """Testing OU with the regular user that has no rights granted over the OU """
+ self.assert_top_ou_deleted()
+ # Create a parent-child OU structure with domain admin credentials
+ self.ldb_owner.create_ou("OU=test_add_ou1," + self.base_dn)
+ self.ldb_owner.create_ou("OU=test_add_ou2,OU=test_add_ou1," + self.base_dn)
+ # Test user and group creation with regular user credentials
+ try:
+ self.ldb_user.newuser(self.test_user1, self.user_pass, userou=self.ou2)
+ self.ldb_user.newgroup("test_add_group1", groupou="OU=test_add_ou2,OU=test_add_ou1",
+ grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
+ except LdbError as e:
+ (num, _) = e.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+ else:
+ self.fail()
+ # Make sure we HAVEN'T created any of two objects -- user or group
+ 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))
+ self.assertEqual(len(res), 0)
+ 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))
+ self.assertEqual(len(res), 0)
+
+ def test_add_u3(self):
+ """Testing OU with the rights of regular user granted the right 'Create User child objects' """
+ self.assert_top_ou_deleted()
+ # Change descriptor for top level OU
+ self.ldb_owner.create_ou("OU=test_add_ou1," + self.base_dn)
+ user_sid = self.sd_utils.get_object_sid(self.get_user_dn(self.regular_user))
+ mod = "(OA;CI;CC;bf967aba-0de6-11d0-a285-00aa003049e2;;%s)" % str(user_sid)
+ self.sd_utils.dacl_add_ace("OU=test_add_ou1," + self.base_dn, mod)
+ self.ldb_owner.create_ou("OU=test_add_ou2,OU=test_add_ou1," + self.base_dn)
+ # Test user and group creation with granted user only to one of the objects
+ self.ldb_user.newuser(self.test_user1, self.user_pass, userou=self.ou2, setpassword=False)
+ try:
+ self.ldb_user.newgroup("test_add_group1", groupou="OU=test_add_ou2,OU=test_add_ou1",
+ grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
+ except LdbError as e1:
+ (num, _) = e1.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+ else:
+ self.fail()
+ # Make sure we HAVE created the one of two objects -- user
+ 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))
+ self.assertNotEqual(len(res), 0)
+ 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))
+ self.assertEqual(len(res), 0)
+
+ def test_add_u4(self):
+ """ 4 Testing OU with the rights of Domain Admin creator of the OU"""
+ self.assert_top_ou_deleted()
+ self.ldb_owner.create_ou("OU=test_add_ou1," + self.base_dn)
+ self.ldb_owner.create_ou("OU=test_add_ou2,OU=test_add_ou1," + self.base_dn)
+ self.ldb_owner.newuser(self.test_user1, self.user_pass, userou=self.ou2)
+ self.ldb_owner.newgroup("test_add_group1", groupou="OU=test_add_ou2,OU=test_add_ou1",
+ grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
+ # Make sure we have successfully created the two objects -- user and group
+ 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))
+ self.assertTrue(len(res) > 0)
+ 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))
+ self.assertTrue(len(res) > 0)
+
+ def test_add_anonymous(self):
+ """Test add operation with anonymous user"""
+ anonymous = SamDB(url=ldaphost, credentials=self.creds_tmp, lp=lp)
+ try:
+ anonymous.newuser("test_add_anonymous", self.user_pass)
+ except LdbError as e2:
+ (num, _) = e2.args
+ self.assertEqual(num, ERR_OPERATIONS_ERROR)
+ else:
+ self.fail()
+
+# tests on ldap modify operations
+
+
+class AclModifyTests(AclTests):
+
+ def setUp(self):
+ super(AclModifyTests, self).setUp()
+ self.user_with_wp = "acl_mod_user1"
+ self.user_with_sm = "acl_mod_user2"
+ self.user_with_group_sm = "acl_mod_user3"
+ self.ldb_admin.newuser(self.user_with_wp, self.user_pass)
+ self.ldb_admin.newuser(self.user_with_sm, self.user_pass)
+ self.ldb_admin.newuser(self.user_with_group_sm, self.user_pass)
+ self.ldb_user = self.get_ldb_connection(self.user_with_wp, self.user_pass)
+ self.ldb_user2 = self.get_ldb_connection(self.user_with_sm, self.user_pass)
+ self.ldb_user3 = self.get_ldb_connection(self.user_with_group_sm, self.user_pass)
+ self.user_sid = self.sd_utils.get_object_sid(self.get_user_dn(self.user_with_wp))
+ self.ldb_admin.newgroup("test_modify_group2", grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
+ self.ldb_admin.newgroup("test_modify_group3", grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
+ self.ldb_admin.newuser("test_modify_user2", self.user_pass)
+
+ def tearDown(self):
+ super(AclModifyTests, self).tearDown()
+ delete_force(self.ldb_admin, self.get_user_dn("test_modify_user1"))
+ delete_force(self.ldb_admin, "CN=test_modify_group1,CN=Users," + self.base_dn)
+ delete_force(self.ldb_admin, "CN=test_modify_group2,CN=Users," + self.base_dn)
+ delete_force(self.ldb_admin, "CN=test_modify_group3,CN=Users," + self.base_dn)
+ delete_force(self.ldb_admin, "CN=test_mod_hostname,OU=test_modify_ou1," + self.base_dn)
+ delete_force(self.ldb_admin, "OU=test_modify_ou1," + self.base_dn)
+ delete_force(self.ldb_admin, self.get_user_dn(self.user_with_wp))
+ delete_force(self.ldb_admin, self.get_user_dn(self.user_with_sm))
+ delete_force(self.ldb_admin, self.get_user_dn(self.user_with_group_sm))
+ delete_force(self.ldb_admin, self.get_user_dn("test_modify_user2"))
+ delete_force(self.ldb_admin, self.get_user_dn("test_anonymous"))
+
+ del self.ldb_user
+ del self.ldb_user2
+ del self.ldb_user3
+
+ def test_modify_u1(self):
+ """5 Modify one attribute if you have DS_WRITE_PROPERTY for it"""
+ mod = "(OA;;WP;bf967953-0de6-11d0-a285-00aa003049e2;;%s)" % str(self.user_sid)
+ # First test object -- User
+ print("Testing modify on User object")
+ self.ldb_admin.newuser("test_modify_user1", self.user_pass)
+ self.sd_utils.dacl_add_ace(self.get_user_dn("test_modify_user1"), mod)
+ ldif = """
+dn: """ + self.get_user_dn("test_modify_user1") + """
+changetype: modify
+replace: displayName
+displayName: test_changed"""
+ self.ldb_user.modify_ldif(ldif)
+ res = self.ldb_admin.search(self.base_dn,
+ expression="(distinguishedName=%s)" % self.get_user_dn("test_modify_user1"))
+ self.assertEqual(str(res[0]["displayName"][0]), "test_changed")
+ # Second test object -- Group
+ print("Testing modify on Group object")
+ self.ldb_admin.newgroup("test_modify_group1",
+ grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
+ self.sd_utils.dacl_add_ace("CN=test_modify_group1,CN=Users," + self.base_dn, mod)
+ ldif = """
+dn: CN=test_modify_group1,CN=Users,""" + self.base_dn + """
+changetype: modify
+replace: displayName
+displayName: test_changed"""
+ self.ldb_user.modify_ldif(ldif)
+ res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" % str("CN=test_modify_group1,CN=Users," + self.base_dn))
+ self.assertEqual(str(res[0]["displayName"][0]), "test_changed")
+ # Third test object -- Organizational Unit
+ print("Testing modify on OU object")
+ #delete_force(self.ldb_admin, "OU=test_modify_ou1," + self.base_dn)
+ self.ldb_admin.create_ou("OU=test_modify_ou1," + self.base_dn)
+ self.sd_utils.dacl_add_ace("OU=test_modify_ou1," + self.base_dn, mod)
+ ldif = """
+dn: OU=test_modify_ou1,""" + self.base_dn + """
+changetype: modify
+replace: displayName
+displayName: test_changed"""
+ self.ldb_user.modify_ldif(ldif)
+ res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" % str("OU=test_modify_ou1," + self.base_dn))
+ self.assertEqual(str(res[0]["displayName"][0]), "test_changed")
+
+ def test_modify_u2(self):
+ """6 Modify two attributes as you have DS_WRITE_PROPERTY granted only for one of them"""
+ mod = "(OA;;WP;bf967953-0de6-11d0-a285-00aa003049e2;;%s)" % str(self.user_sid)
+ # First test object -- User
+ print("Testing modify on User object")
+ #delete_force(self.ldb_admin, self.get_user_dn("test_modify_user1"))
+ self.ldb_admin.newuser("test_modify_user1", self.user_pass)
+ self.sd_utils.dacl_add_ace(self.get_user_dn("test_modify_user1"), mod)
+ # Modify on attribute you have rights for
+ ldif = """
+dn: """ + self.get_user_dn("test_modify_user1") + """
+changetype: modify
+replace: displayName
+displayName: test_changed"""
+ self.ldb_user.modify_ldif(ldif)
+ res = self.ldb_admin.search(self.base_dn,
+ expression="(distinguishedName=%s)" %
+ self.get_user_dn("test_modify_user1"))
+ self.assertEqual(str(res[0]["displayName"][0]), "test_changed")
+ # Modify on attribute you do not have rights for granted
+ ldif = """
+dn: """ + self.get_user_dn("test_modify_user1") + """
+changetype: modify
+replace: url
+url: www.samba.org"""
+ try:
+ self.ldb_user.modify_ldif(ldif)
+ except LdbError as e3:
+ (num, _) = e3.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+ else:
+ # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
+ self.fail()
+ # Second test object -- Group
+ print("Testing modify on Group object")
+ self.ldb_admin.newgroup("test_modify_group1",
+ grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
+ self.sd_utils.dacl_add_ace("CN=test_modify_group1,CN=Users," + self.base_dn, mod)
+ ldif = """
+dn: CN=test_modify_group1,CN=Users,""" + self.base_dn + """
+changetype: modify
+replace: displayName
+displayName: test_changed"""
+ self.ldb_user.modify_ldif(ldif)
+ res = self.ldb_admin.search(self.base_dn,
+ expression="(distinguishedName=%s)" %
+ str("CN=test_modify_group1,CN=Users," + self.base_dn))
+ self.assertEqual(str(res[0]["displayName"][0]), "test_changed")
+ # Modify on attribute you do not have rights for granted
+ ldif = """
+dn: CN=test_modify_group1,CN=Users,""" + self.base_dn + """
+changetype: modify
+replace: url
+url: www.samba.org"""
+ try:
+ self.ldb_user.modify_ldif(ldif)
+ except LdbError as e4:
+ (num, _) = e4.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+ else:
+ # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
+ self.fail()
+ # Modify on attribute you do not have rights for granted while also modifying something you do have rights for
+ ldif = """
+dn: CN=test_modify_group1,CN=Users,""" + self.base_dn + """
+changetype: modify
+replace: url
+url: www.samba.org
+replace: displayName
+displayName: test_changed"""
+ try:
+ self.ldb_user.modify_ldif(ldif)
+ except LdbError as e5:
+ (num, _) = e5.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+ else:
+ # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
+ self.fail()
+ # Second test object -- Organizational Unit
+ print("Testing modify on OU object")
+ self.ldb_admin.create_ou("OU=test_modify_ou1," + self.base_dn)
+ self.sd_utils.dacl_add_ace("OU=test_modify_ou1," + self.base_dn, mod)
+ ldif = """
+dn: OU=test_modify_ou1,""" + self.base_dn + """
+changetype: modify
+replace: displayName
+displayName: test_changed"""
+ self.ldb_user.modify_ldif(ldif)
+ res = self.ldb_admin.search(self.base_dn,
+ expression="(distinguishedName=%s)" % str("OU=test_modify_ou1,"
+ + self.base_dn))
+ self.assertEqual(str(res[0]["displayName"][0]), "test_changed")
+ # Modify on attribute you do not have rights for granted
+ ldif = """
+dn: OU=test_modify_ou1,""" + self.base_dn + """
+changetype: modify
+replace: url
+url: www.samba.org"""
+ try:
+ self.ldb_user.modify_ldif(ldif)
+ except LdbError as e6:
+ (num, _) = e6.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+ else:
+ # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
+ self.fail()
+
+ def test_modify_u3(self):
+ """7 Modify one attribute as you have no what so ever rights granted"""
+ # First test object -- User
+ print("Testing modify on User object")
+ self.ldb_admin.newuser("test_modify_user1", self.user_pass)
+ # Modify on attribute you do not have rights for granted
+ ldif = """
+dn: """ + self.get_user_dn("test_modify_user1") + """
+changetype: modify
+replace: url
+url: www.samba.org"""
+ try:
+ self.ldb_user.modify_ldif(ldif)
+ except LdbError as e7:
+ (num, _) = e7.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+ else:
+ # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
+ self.fail()
+
+ # Second test object -- Group
+ print("Testing modify on Group object")
+ self.ldb_admin.newgroup("test_modify_group1",
+ grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
+ # Modify on attribute you do not have rights for granted
+ ldif = """
+dn: CN=test_modify_group1,CN=Users,""" + self.base_dn + """
+changetype: modify
+replace: url
+url: www.samba.org"""
+ try:
+ self.ldb_user.modify_ldif(ldif)
+ except LdbError as e8:
+ (num, _) = e8.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+ else:
+ # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
+ self.fail()
+
+ # Second test object -- Organizational Unit
+ print("Testing modify on OU object")
+ #delete_force(self.ldb_admin, "OU=test_modify_ou1," + self.base_dn)
+ self.ldb_admin.create_ou("OU=test_modify_ou1," + self.base_dn)
+ # Modify on attribute you do not have rights for granted
+ ldif = """
+dn: OU=test_modify_ou1,""" + self.base_dn + """
+changetype: modify
+replace: url
+url: www.samba.org"""
+ try:
+ self.ldb_user.modify_ldif(ldif)
+ except LdbError as e9:
+ (num, _) = e9.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+ else:
+ # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
+ self.fail()
+
+ def test_modify_u4(self):
+ """11 Grant WP to PRINCIPAL_SELF and test modify"""
+ ldif = """
+dn: """ + self.get_user_dn(self.user_with_wp) + """
+changetype: modify
+add: adminDescription
+adminDescription: blah blah blah"""
+ try:
+ self.ldb_user.modify_ldif(ldif)
+ except LdbError as e10:
+ (num, _) = e10.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+ else:
+ # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
+ self.fail()
+
+ mod = "(OA;;WP;bf967919-0de6-11d0-a285-00aa003049e2;;PS)"
+ self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
+ # Modify on attribute you have rights for
+ self.ldb_user.modify_ldif(ldif)
+ res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)"
+ % self.get_user_dn(self.user_with_wp), attrs=["adminDescription"])
+ self.assertEqual(str(res[0]["adminDescription"][0]), "blah blah blah")
+
+ def test_modify_u5(self):
+ """12 test self membership"""
+ ldif = """
+dn: CN=test_modify_group2,CN=Users,""" + self.base_dn + """
+changetype: modify
+add: Member
+Member: """ + self.get_user_dn(self.user_with_sm)
+# the user has no rights granted, this should fail
+ try:
+ self.ldb_user2.modify_ldif(ldif)
+ except LdbError as e11:
+ (num, _) = e11.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+ else:
+ # This 'modify' operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
+ self.fail()
+
+# grant self-membership, should be able to add himself
+ user_sid = self.sd_utils.get_object_sid(self.get_user_dn(self.user_with_sm))
+ mod = "(OA;;SW;bf9679c0-0de6-11d0-a285-00aa003049e2;;%s)" % str(user_sid)
+ self.sd_utils.dacl_add_ace("CN=test_modify_group2,CN=Users," + self.base_dn, mod)
+ self.ldb_user2.modify_ldif(ldif)
+ res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)"
+ % ("CN=test_modify_group2,CN=Users," + self.base_dn), attrs=["Member"])
+ self.assertEqual(str(res[0]["Member"][0]), self.get_user_dn(self.user_with_sm))
+# but not other users
+ ldif = """
+dn: CN=test_modify_group2,CN=Users,""" + self.base_dn + """
+changetype: modify
+add: Member
+Member: CN=test_modify_user2,CN=Users,""" + self.base_dn
+ try:
+ self.ldb_user2.modify_ldif(ldif)
+ except LdbError as e12:
+ (num, _) = e12.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+ else:
+ self.fail()
+
+ def test_modify_u6(self):
+ """13 test self membership"""
+ ldif = """
+dn: CN=test_modify_group2,CN=Users,""" + self.base_dn + """
+changetype: modify
+add: Member
+Member: """ + self.get_user_dn(self.user_with_sm) + """
+Member: CN=test_modify_user2,CN=Users,""" + self.base_dn
+
+# grant self-membership, should be able to add himself but not others at the same time
+ user_sid = self.sd_utils.get_object_sid(self.get_user_dn(self.user_with_sm))
+ mod = "(OA;;SW;bf9679c0-0de6-11d0-a285-00aa003049e2;;%s)" % str(user_sid)
+ self.sd_utils.dacl_add_ace("CN=test_modify_group2,CN=Users," + self.base_dn, mod)
+ try:
+ self.ldb_user2.modify_ldif(ldif)
+ except LdbError as e13:
+ (num, _) = e13.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+ else:
+ self.fail()
+
+ def test_modify_u7(self):
+ """13 User with WP modifying Member"""
+# a second user is given write property permission
+ user_sid = self.sd_utils.get_object_sid(self.get_user_dn(self.user_with_wp))
+ mod = "(A;;WP;;;%s)" % str(user_sid)
+ self.sd_utils.dacl_add_ace("CN=test_modify_group2,CN=Users," + self.base_dn, mod)
+ ldif = """
+dn: CN=test_modify_group2,CN=Users,""" + self.base_dn + """
+changetype: modify
+add: Member
+Member: """ + self.get_user_dn(self.user_with_wp)
+ self.ldb_user.modify_ldif(ldif)
+ res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)"
+ % ("CN=test_modify_group2,CN=Users," + self.base_dn), attrs=["Member"])
+ self.assertEqual(str(res[0]["Member"][0]), self.get_user_dn(self.user_with_wp))
+ ldif = """
+dn: CN=test_modify_group2,CN=Users,""" + self.base_dn + """
+changetype: modify
+delete: Member"""
+ self.ldb_user.modify_ldif(ldif)
+ ldif = """
+dn: CN=test_modify_group2,CN=Users,""" + self.base_dn + """
+changetype: modify
+add: Member
+Member: CN=test_modify_user2,CN=Users,""" + self.base_dn
+ self.ldb_user.modify_ldif(ldif)
+ res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)"
+ % ("CN=test_modify_group2,CN=Users," + self.base_dn), attrs=["Member"])
+ self.assertEqual(str(res[0]["Member"][0]), "CN=test_modify_user2,CN=Users," + self.base_dn)
+
+ def test_modify_anonymous(self):
+ """Test add operation with anonymous user"""
+ anonymous = SamDB(url=ldaphost, credentials=self.creds_tmp, lp=lp)
+ self.ldb_admin.newuser("test_anonymous", "samba123@")
+ m = Message()
+ m.dn = Dn(anonymous, self.get_user_dn("test_anonymous"))
+
+ m["description"] = MessageElement("sambauser2",
+ FLAG_MOD_ADD,
+ "description")
+ try:
+ anonymous.modify(m)
+ except LdbError as e14:
+ (num, _) = e14.args
+ self.assertEqual(num, ERR_OPERATIONS_ERROR)
+ else:
+ self.fail()
+
+ def test_modify_dns_host_name(self):
+ '''Test modifying dNSHostName with validated write'''
+
+ ou_dn = f'OU=test_modify_ou1,{self.base_dn}'
+
+ account_name = 'test_mod_hostname'
+ dn = f'CN={account_name},{ou_dn}'
+
+ self.ldb_admin.create_ou(ou_dn)
+
+ # Grant Validated Write.
+ mod = (f'(OA;CI;SW;{security.GUID_DRS_DNS_HOST_NAME};;'
+ f'{self.user_sid})')
+ self.sd_utils.dacl_add_ace(ou_dn, mod)
+
+ # Create the account.
+ self.ldb_admin.add({
+ 'dn': dn,
+ 'objectClass': 'computer',
+ 'sAMAccountName': f'{account_name}$',
+ })
+
+ host_name = f'{account_name}.{self.ldb_user.domain_dns_name()}'
+
+ m = Message(Dn(self.ldb_user, dn))
+ m['dNSHostName'] = MessageElement(host_name,
+ FLAG_MOD_REPLACE,
+ 'dNSHostName')
+ try:
+ self.ldb_user.modify(m)
+ except LdbError:
+ self.fail()
+
+ def test_modify_dns_host_name_no_validated_write(self):
+ '''Test modifying dNSHostName without validated write'''
+
+ ou_dn = f'OU=test_modify_ou1,{self.base_dn}'
+
+ account_name = 'test_mod_hostname'
+ dn = f'CN={account_name},{ou_dn}'
+
+ self.ldb_admin.create_ou(ou_dn)
+
+ # Create the account.
+ self.ldb_admin.add({
+ 'dn': dn,
+ 'objectClass': 'computer',
+ 'sAMAccountName': f'{account_name}$',
+ })
+
+ host_name = f'{account_name}.{self.ldb_user.domain_dns_name()}'
+
+ m = Message(Dn(self.ldb_user, dn))
+ m['dNSHostName'] = MessageElement(host_name,
+ FLAG_MOD_REPLACE,
+ 'dNSHostName')
+ try:
+ self.ldb_user.modify(m)
+ except LdbError as err:
+ num, estr = err.args
+ self.assertEqual(ERR_INSUFFICIENT_ACCESS_RIGHTS, num)
+ else:
+ self.fail()
+
+ def test_modify_dns_host_name_invalid(self):
+ '''Test modifying dNSHostName to an invalid value'''
+
+ ou_dn = f'OU=test_modify_ou1,{self.base_dn}'
+
+ account_name = 'test_mod_hostname'
+ dn = f'CN={account_name},{ou_dn}'
+
+ self.ldb_admin.create_ou(ou_dn)
+
+ # Grant Validated Write.
+ mod = (f'(OA;CI;SW;{security.GUID_DRS_DNS_HOST_NAME};;'
+ f'{self.user_sid})')
+ self.sd_utils.dacl_add_ace(ou_dn, mod)
+
+ # Create the account.
+ self.ldb_admin.add({
+ 'dn': dn,
+ 'objectClass': 'computer',
+ 'sAMAccountName': f'{account_name}$',
+ })
+
+ host_name = 'invalid'
+
+ m = Message(Dn(self.ldb_user, dn))
+ m['dNSHostName'] = MessageElement(host_name,
+ FLAG_MOD_REPLACE,
+ 'dNSHostName')
+ try:
+ self.ldb_user.modify(m)
+ except LdbError as err:
+ num, estr = err.args
+ self.assertEqual(ERR_CONSTRAINT_VIOLATION, num)
+ else:
+ self.fail()
+
+ def test_modify_dns_host_name_invalid_wp(self):
+ '''Test modifying dNSHostName to an invalid value when we have WP'''
+
+ ou_dn = f'OU=test_modify_ou1,{self.base_dn}'
+
+ account_name = 'test_mod_hostname'
+ dn = f'CN={account_name},{ou_dn}'
+
+ self.ldb_admin.create_ou(ou_dn)
+
+ # Grant Write Property.
+ mod = (f'(OA;CI;WP;{security.GUID_DRS_DNS_HOST_NAME};;'
+ f'{self.user_sid})')
+ self.sd_utils.dacl_add_ace(ou_dn, mod)
+
+ # Create the account.
+ self.ldb_admin.add({
+ 'dn': dn,
+ 'objectClass': 'computer',
+ 'sAMAccountName': f'{account_name}$',
+ })
+
+ host_name = 'invalid'
+
+ m = Message(Dn(self.ldb_user, dn))
+ m['dNSHostName'] = MessageElement(host_name,
+ FLAG_MOD_REPLACE,
+ 'dNSHostName')
+ try:
+ self.ldb_user.modify(m)
+ except LdbError:
+ self.fail()
+
+ def test_modify_dns_host_name_invalid_non_computer(self):
+ '''Test modifying dNSHostName to an invalid value on a non-computer'''
+
+ ou_dn = f'OU=test_modify_ou1,{self.base_dn}'
+
+ account_name = 'test_mod_hostname'
+ dn = f'CN={account_name},{ou_dn}'
+
+ self.ldb_admin.create_ou(ou_dn)
+
+ # Grant Validated Write.
+ mod = (f'(OA;CI;SW;{security.GUID_DRS_DNS_HOST_NAME};;'
+ f'{self.user_sid})')
+ self.sd_utils.dacl_add_ace(ou_dn, mod)
+
+ # Create the account.
+ self.ldb_admin.add({
+ 'dn': dn,
+ 'objectClass': 'user',
+ 'sAMAccountName': f'{account_name}',
+ })
+
+ host_name = 'invalid'
+
+ m = Message(Dn(self.ldb_user, dn))
+ m['dNSHostName'] = MessageElement(host_name,
+ FLAG_MOD_REPLACE,
+ 'dNSHostName')
+ try:
+ self.ldb_user.modify(m)
+ except LdbError as err:
+ num, estr = err.args
+ self.assertEqual(ERR_INSUFFICIENT_ACCESS_RIGHTS, num)
+ else:
+ self.fail()
+
+ def test_modify_dns_host_name_no_value(self):
+ '''Test modifying dNSHostName with validated write with no value'''
+
+ ou_dn = f'OU=test_modify_ou1,{self.base_dn}'
+
+ account_name = 'test_mod_hostname'
+ dn = f'CN={account_name},{ou_dn}'
+
+ self.ldb_admin.create_ou(ou_dn)
+
+ # Grant Validated Write.
+ mod = (f'(OA;CI;SW;{security.GUID_DRS_DNS_HOST_NAME};;'
+ f'{self.user_sid})')
+ self.sd_utils.dacl_add_ace(ou_dn, mod)
+
+ # Create the account.
+ self.ldb_admin.add({
+ 'dn': dn,
+ 'objectClass': 'computer',
+ 'sAMAccountName': f'{account_name}$',
+ })
+
+ m = Message(Dn(self.ldb_user, dn))
+ m['dNSHostName'] = MessageElement([],
+ FLAG_MOD_REPLACE,
+ 'dNSHostName')
+ try:
+ self.ldb_user.modify(m)
+ except LdbError as err:
+ num, estr = err.args
+ self.assertEqual(ERR_OPERATIONS_ERROR, num)
+ else:
+ # Windows accepts this.
+ pass
+
+ def test_modify_dns_host_name_empty_string(self):
+ '''Test modifying dNSHostName with validated write of an empty string'''
+
+ ou_dn = f'OU=test_modify_ou1,{self.base_dn}'
+
+ account_name = 'test_mod_hostname'
+ dn = f'CN={account_name},{ou_dn}'
+
+ self.ldb_admin.create_ou(ou_dn)
+
+ # Grant Validated Write.
+ mod = (f'(OA;CI;SW;{security.GUID_DRS_DNS_HOST_NAME};;'
+ f'{self.user_sid})')
+ self.sd_utils.dacl_add_ace(ou_dn, mod)
+
+ # Create the account.
+ self.ldb_admin.add({
+ 'dn': dn,
+ 'objectClass': 'computer',
+ 'sAMAccountName': f'{account_name}$',
+ })
+
+ m = Message(Dn(self.ldb_user, dn))
+ m['dNSHostName'] = MessageElement('\0',
+ FLAG_MOD_REPLACE,
+ 'dNSHostName')
+ try:
+ self.ldb_user.modify(m)
+ except LdbError as err:
+ num, estr = err.args
+ self.assertEqual(ERR_CONSTRAINT_VIOLATION, num)
+ else:
+ self.fail()
+
+ def test_modify_dns_host_name_dollar(self):
+ '''Test modifying dNSHostName with validated write of a value including a dollar'''
+
+ ou_dn = f'OU=test_modify_ou1,{self.base_dn}'
+
+ account_name = 'test_mod_hostname'
+ dn = f'CN={account_name},{ou_dn}'
+
+ self.ldb_admin.create_ou(ou_dn)
+
+ # Grant Validated Write.
+ mod = (f'(OA;CI;SW;{security.GUID_DRS_DNS_HOST_NAME};;'
+ f'{self.user_sid})')
+ self.sd_utils.dacl_add_ace(ou_dn, mod)
+
+ # Create the account.
+ self.ldb_admin.add({
+ 'dn': dn,
+ 'objectClass': 'computer',
+ 'sAMAccountName': f'{account_name}$',
+ })
+
+ host_name = f'{account_name}$.{self.ldb_user.domain_dns_name()}'
+
+ m = Message(Dn(self.ldb_user, dn))
+ m['dNSHostName'] = MessageElement(host_name,
+ FLAG_MOD_REPLACE,
+ 'dNSHostName')
+ try:
+ self.ldb_user.modify(m)
+ except LdbError as err:
+ num, estr = err.args
+ self.assertEqual(ERR_CONSTRAINT_VIOLATION, num)
+ else:
+ self.fail()
+
+ def test_modify_dns_host_name_account_no_dollar(self):
+ '''Test modifying dNSHostName with validated write with no dollar in sAMAccountName'''
+
+ ou_dn = f'OU=test_modify_ou1,{self.base_dn}'
+
+ account_name = 'test_mod_hostname'
+ dn = f'CN={account_name},{ou_dn}'
+
+ self.ldb_admin.create_ou(ou_dn)
+
+ # Grant Validated Write.
+ mod = (f'(OA;CI;SW;{security.GUID_DRS_DNS_HOST_NAME};;'
+ f'{self.user_sid})')
+ self.sd_utils.dacl_add_ace(ou_dn, mod)
+
+ # Create the account.
+ self.ldb_admin.add({
+ 'dn': dn,
+ 'objectClass': 'computer',
+ 'sAMAccountName': f'{account_name}',
+ })
+
+ host_name = f'{account_name}.{self.ldb_user.domain_dns_name()}'
+
+ m = Message(Dn(self.ldb_user, dn))
+ m['dNSHostName'] = MessageElement(host_name,
+ FLAG_MOD_REPLACE,
+ 'dNSHostName')
+ try:
+ self.ldb_user.modify(m)
+ except LdbError:
+ self.fail()
+
+ def test_modify_dns_host_name_no_suffix(self):
+ '''Test modifying dNSHostName with validated write of a value missing the suffix'''
+
+ ou_dn = f'OU=test_modify_ou1,{self.base_dn}'
+
+ account_name = 'test_mod_hostname'
+ dn = f'CN={account_name},{ou_dn}'
+
+ self.ldb_admin.create_ou(ou_dn)
+
+ # Grant Validated Write.
+ mod = (f'(OA;CI;SW;{security.GUID_DRS_DNS_HOST_NAME};;'
+ f'{self.user_sid})')
+ self.sd_utils.dacl_add_ace(ou_dn, mod)
+
+ # Create the account.
+ self.ldb_admin.add({
+ 'dn': dn,
+ 'objectClass': 'computer',
+ 'sAMAccountName': f'{account_name}$',
+ })
+
+ host_name = f'{account_name}'
+
+ m = Message(Dn(self.ldb_user, dn))
+ m['dNSHostName'] = MessageElement(host_name,
+ FLAG_MOD_REPLACE,
+ 'dNSHostName')
+ try:
+ self.ldb_user.modify(m)
+ except LdbError as err:
+ num, estr = err.args
+ self.assertEqual(ERR_CONSTRAINT_VIOLATION, num)
+ else:
+ self.fail()
+
+ def test_modify_dns_host_name_wrong_prefix(self):
+ '''Test modifying dNSHostName with validated write of a value with the wrong prefix'''
+
+ ou_dn = f'OU=test_modify_ou1,{self.base_dn}'
+
+ account_name = 'test_mod_hostname'
+ dn = f'CN={account_name},{ou_dn}'
+
+ self.ldb_admin.create_ou(ou_dn)
+
+ # Grant Validated Write.
+ mod = (f'(OA;CI;SW;{security.GUID_DRS_DNS_HOST_NAME};;'
+ f'{self.user_sid})')
+ self.sd_utils.dacl_add_ace(ou_dn, mod)
+
+ # Create the account.
+ self.ldb_admin.add({
+ 'dn': dn,
+ 'objectClass': 'computer',
+ 'sAMAccountName': f'{account_name}$',
+ })
+
+ host_name = f'invalid.{self.ldb_user.domain_dns_name()}'
+
+ m = Message(Dn(self.ldb_user, dn))
+ m['dNSHostName'] = MessageElement(host_name,
+ FLAG_MOD_REPLACE,
+ 'dNSHostName')
+ try:
+ self.ldb_user.modify(m)
+ except LdbError as err:
+ num, estr = err.args
+ self.assertEqual(ERR_CONSTRAINT_VIOLATION, num)
+ else:
+ self.fail()
+
+ def test_modify_dns_host_name_wrong_suffix(self):
+ '''Test modifying dNSHostName with validated write of a value with the wrong suffix'''
+
+ ou_dn = f'OU=test_modify_ou1,{self.base_dn}'
+
+ account_name = 'test_mod_hostname'
+ dn = f'CN={account_name},{ou_dn}'
+
+ self.ldb_admin.create_ou(ou_dn)
+
+ # Grant Validated Write.
+ mod = (f'(OA;CI;SW;{security.GUID_DRS_DNS_HOST_NAME};;'
+ f'{self.user_sid})')
+ self.sd_utils.dacl_add_ace(ou_dn, mod)
+
+ # Create the account.
+ self.ldb_admin.add({
+ 'dn': dn,
+ 'objectClass': 'computer',
+ 'sAMAccountName': f'{account_name}$',
+ })
+
+ host_name = f'{account_name}.invalid.example.com'
+
+ m = Message(Dn(self.ldb_user, dn))
+ m['dNSHostName'] = MessageElement(host_name,
+ FLAG_MOD_REPLACE,
+ 'dNSHostName')
+ try:
+ self.ldb_user.modify(m)
+ except LdbError as err:
+ num, estr = err.args
+ self.assertEqual(ERR_CONSTRAINT_VIOLATION, num)
+ else:
+ self.fail()
+
+ def test_modify_dns_host_name_case(self):
+ '''Test modifying dNSHostName with validated write of a value with irregular case'''
+
+ ou_dn = f'OU=test_modify_ou1,{self.base_dn}'
+
+ account_name = 'test_mod_hostname'
+ dn = f'CN={account_name},{ou_dn}'
+
+ self.ldb_admin.create_ou(ou_dn)
+
+ # Grant Validated Write.
+ mod = (f'(OA;CI;SW;{security.GUID_DRS_DNS_HOST_NAME};;'
+ f'{self.user_sid})')
+ self.sd_utils.dacl_add_ace(ou_dn, mod)
+
+ # Create the account.
+ self.ldb_admin.add({
+ 'dn': dn,
+ 'objectClass': 'computer',
+ 'sAMAccountName': f'{account_name}$',
+ })
+
+ host_name = f'{account_name}.{self.ldb_user.domain_dns_name()}'
+ host_name = host_name.capitalize()
+
+ m = Message(Dn(self.ldb_user, dn))
+ m['dNSHostName'] = MessageElement(host_name,
+ FLAG_MOD_REPLACE,
+ 'dNSHostName')
+ try:
+ self.ldb_user.modify(m)
+ except LdbError:
+ self.fail()
+
+ def test_modify_dns_host_name_allowed_suffixes(self):
+ '''Test modifying dNSHostName with validated write and an allowed suffix'''
+
+ allowed_suffix = 'suffix.that.is.allowed'
+
+ # Add the allowed suffix.
+
+ res = self.ldb_admin.search(self.base_dn,
+ scope=SCOPE_BASE,
+ attrs=['msDS-AllowedDNSSuffixes'])
+ self.assertEqual(1, len(res))
+ old_allowed_suffixes = res[0].get('msDS-AllowedDNSSuffixes')
+
+ def modify_allowed_suffixes(suffixes):
+ if suffixes is None:
+ suffixes = []
+ flag = FLAG_MOD_DELETE
+ else:
+ flag = FLAG_MOD_REPLACE
+
+ m = Message(Dn(self.ldb_admin, self.base_dn))
+ m['msDS-AllowedDNSSuffixes'] = MessageElement(
+ suffixes,
+ flag,
+ 'msDS-AllowedDNSSuffixes')
+ self.ldb_admin.modify(m)
+
+ self.addCleanup(modify_allowed_suffixes, old_allowed_suffixes)
+
+ if old_allowed_suffixes is None:
+ allowed_suffixes = []
+ else:
+ allowed_suffixes = list(old_allowed_suffixes)
+
+ if (allowed_suffix not in allowed_suffixes and
+ allowed_suffix.encode('utf-8') not in allowed_suffixes):
+ allowed_suffixes.append(allowed_suffix)
+
+ modify_allowed_suffixes(allowed_suffixes)
+
+ # Create the account and run the test.
+
+ ou_dn = f'OU=test_modify_ou1,{self.base_dn}'
+
+ account_name = 'test_mod_hostname'
+ dn = f'CN={account_name},{ou_dn}'
+
+ self.ldb_admin.create_ou(ou_dn)
+
+ # Grant Validated Write.
+ mod = (f'(OA;CI;SW;{security.GUID_DRS_DNS_HOST_NAME};;'
+ f'{self.user_sid})')
+ self.sd_utils.dacl_add_ace(ou_dn, mod)
+
+ # Create the account.
+ self.ldb_admin.add({
+ 'dn': dn,
+ 'objectClass': 'computer',
+ 'sAMAccountName': f'{account_name}$',
+ })
+
+ host_name = f'{account_name}.{allowed_suffix}'
+
+ m = Message(Dn(self.ldb_user, dn))
+ m['dNSHostName'] = MessageElement(host_name,
+ FLAG_MOD_REPLACE,
+ 'dNSHostName')
+ try:
+ self.ldb_user.modify(m)
+ except LdbError:
+ self.fail()
+
+ def test_modify_dns_host_name_spn(self):
+ '''Test modifying dNSHostName and SPN with validated write'''
+
+ ou_dn = f'OU=test_modify_ou1,{self.base_dn}'
+
+ account_name = 'test_mod_hostname'
+ dn = f'CN={account_name},{ou_dn}'
+
+ self.ldb_admin.create_ou(ou_dn)
+
+ # Grant Validated Write.
+ mod = (f'(OA;CI;SW;{security.GUID_DRS_DNS_HOST_NAME};;'
+ f'{self.user_sid})')
+ self.sd_utils.dacl_add_ace(ou_dn, mod)
+ mod = (f'(OA;CI;SW;{security.GUID_DRS_VALIDATE_SPN};;'
+ f'{self.user_sid})')
+ self.sd_utils.dacl_add_ace(ou_dn, mod)
+
+ # Create the account.
+ self.ldb_admin.add({
+ 'dn': dn,
+ 'objectClass': 'computer',
+ 'sAMAccountName': f'{account_name}$',
+ })
+
+ host_name = f'{account_name}.{self.ldb_user.domain_dns_name()}'
+ spn = f'host/{host_name}'
+
+ m = Message(Dn(self.ldb_user, dn))
+ m['0'] = MessageElement(host_name,
+ FLAG_MOD_REPLACE,
+ 'dNSHostName')
+ m['1'] = MessageElement(spn,
+ FLAG_MOD_ADD,
+ 'servicePrincipalName')
+ try:
+ self.ldb_user.modify(m)
+ except LdbError:
+ self.fail()
+
+ def test_modify_spn_matching_dns_host_name_invalid(self):
+ '''Test modifying SPN with validated write, matching a valid dNSHostName '''
+
+ ou_dn = f'OU=test_modify_ou1,{self.base_dn}'
+
+ account_name = 'test_mod_hostname'
+ dn = f'CN={account_name},{ou_dn}'
+
+ self.ldb_admin.create_ou(ou_dn)
+
+ # Grant Write Property.
+ mod = (f'(OA;CI;WP;{security.GUID_DRS_DNS_HOST_NAME};;'
+ f'{self.user_sid})')
+ self.sd_utils.dacl_add_ace(ou_dn, mod)
+ # Grant Validated Write.
+ mod = (f'(OA;CI;SW;{security.GUID_DRS_VALIDATE_SPN};;'
+ f'{self.user_sid})')
+ self.sd_utils.dacl_add_ace(ou_dn, mod)
+
+ # Create the account.
+ self.ldb_admin.add({
+ 'dn': dn,
+ 'objectClass': 'computer',
+ 'sAMAccountName': f'{account_name}$',
+ })
+
+ invalid_host_name = 'invalid'
+
+ host_name = f'{account_name}.{self.ldb_user.domain_dns_name()}'
+ spn = f'host/{host_name}'
+
+ m = Message(Dn(self.ldb_user, dn))
+ m['0'] = MessageElement(invalid_host_name,
+ FLAG_MOD_REPLACE,
+ 'dNSHostName')
+ m['1'] = MessageElement(spn,
+ FLAG_MOD_ADD,
+ 'servicePrincipalName')
+ m['2'] = MessageElement(host_name,
+ FLAG_MOD_REPLACE,
+ 'dNSHostName')
+ try:
+ self.ldb_user.modify(m)
+ except LdbError:
+ self.fail()
+
+ def test_modify_spn_matching_dns_host_name_original(self):
+ '''Test modifying SPN with validated write, matching the original dNSHostName '''
+
+ ou_dn = f'OU=test_modify_ou1,{self.base_dn}'
+
+ account_name = 'test_mod_hostname'
+ dn = f'CN={account_name},{ou_dn}'
+
+ self.ldb_admin.create_ou(ou_dn)
+
+ # Grant Validated Write.
+ mod = (f'(OA;CI;SW;{security.GUID_DRS_DNS_HOST_NAME};;'
+ f'{self.user_sid})')
+ self.sd_utils.dacl_add_ace(ou_dn, mod)
+ mod = (f'(OA;CI;SW;{security.GUID_DRS_VALIDATE_SPN};;'
+ f'{self.user_sid})')
+ self.sd_utils.dacl_add_ace(ou_dn, mod)
+
+ original_host_name = 'invalid_host_name'
+ original_spn = 'host/{original_host_name}'
+
+ # Create the account.
+ self.ldb_admin.add({
+ 'dn': dn,
+ 'objectClass': 'computer',
+ 'sAMAccountName': f'{account_name}$',
+ 'dNSHostName': original_host_name,
+ })
+
+ host_name = f'{account_name}.{self.ldb_user.domain_dns_name()}'
+
+ m = Message(Dn(self.ldb_user, dn))
+ m['0'] = MessageElement(original_spn,
+ FLAG_MOD_ADD,
+ 'servicePrincipalName')
+ m['1'] = MessageElement(host_name,
+ FLAG_MOD_REPLACE,
+ 'dNSHostName')
+ try:
+ self.ldb_user.modify(m)
+ except LdbError as err:
+ num, estr = err.args
+ self.assertEqual(ERR_CONSTRAINT_VIOLATION, num)
+ else:
+ self.fail()
+
+ def test_modify_dns_host_name_spn_matching_account_name_original(self):
+ '''Test modifying dNSHostName and SPN with validated write, matching the original sAMAccountName'''
+
+ ou_dn = f'OU=test_modify_ou1,{self.base_dn}'
+
+ account_name = 'test_mod_hostname'
+ dn = f'CN={account_name},{ou_dn}'
+
+ self.ldb_admin.create_ou(ou_dn)
+
+ sam_account_name = '3e0abfd0-126a-11d0-a060-00aa006c33ed'
+
+ # Grant Write Property.
+ mod = (f'(OA;CI;WP;{sam_account_name};;'
+ f'{self.user_sid})')
+ self.sd_utils.dacl_add_ace(ou_dn, mod)
+ # Grant Validated Write.
+ mod = (f'(OA;CI;SW;{security.GUID_DRS_DNS_HOST_NAME};;'
+ f'{self.user_sid})')
+ self.sd_utils.dacl_add_ace(ou_dn, mod)
+ mod = (f'(OA;CI;SW;{security.GUID_DRS_VALIDATE_SPN};;'
+ f'{self.user_sid})')
+ self.sd_utils.dacl_add_ace(ou_dn, mod)
+
+ # Create the account.
+ self.ldb_admin.add({
+ 'dn': dn,
+ 'objectClass': 'computer',
+ 'sAMAccountName': f'{account_name}$',
+ })
+
+ new_account_name = 'test_mod_hostname2'
+ host_name = f'{account_name}.{self.ldb_user.domain_dns_name()}'
+ spn = f'host/{host_name}'
+
+ m = Message(Dn(self.ldb_user, dn))
+ m['0'] = MessageElement(host_name,
+ FLAG_MOD_REPLACE,
+ 'dNSHostName')
+ m['1'] = MessageElement(spn,
+ FLAG_MOD_ADD,
+ 'servicePrincipalName')
+ m['2'] = MessageElement(f'{new_account_name}$',
+ FLAG_MOD_REPLACE,
+ 'sAMAccountName')
+ try:
+ self.ldb_user.modify(m)
+ except LdbError as err:
+ num, estr = err.args
+ self.assertEqual(ERR_CONSTRAINT_VIOLATION, num)
+ else:
+ self.fail()
+
+ def test_modify_dns_host_name_spn_matching_account_name_new(self):
+ '''Test modifying dNSHostName and SPN with validated write, matching the new sAMAccountName'''
+
+ ou_dn = f'OU=test_modify_ou1,{self.base_dn}'
+
+ account_name = 'test_mod_hostname'
+ dn = f'CN={account_name},{ou_dn}'
+
+ self.ldb_admin.create_ou(ou_dn)
+
+ sam_account_name = '3e0abfd0-126a-11d0-a060-00aa006c33ed'
+
+ # Grant Write Property.
+ mod = (f'(OA;CI;WP;{sam_account_name};;'
+ f'{self.user_sid})')
+ self.sd_utils.dacl_add_ace(ou_dn, mod)
+ # Grant Validated Write.
+ mod = (f'(OA;CI;SW;{security.GUID_DRS_DNS_HOST_NAME};;'
+ f'{self.user_sid})')
+ self.sd_utils.dacl_add_ace(ou_dn, mod)
+ mod = (f'(OA;CI;SW;{security.GUID_DRS_VALIDATE_SPN};;'
+ f'{self.user_sid})')
+ self.sd_utils.dacl_add_ace(ou_dn, mod)
+
+ # Create the account.
+ self.ldb_admin.add({
+ 'dn': dn,
+ 'objectClass': 'computer',
+ 'sAMAccountName': f'{account_name}$',
+ })
+
+ new_account_name = 'test_mod_hostname2'
+ new_host_name = f'{new_account_name}.{self.ldb_user.domain_dns_name()}'
+ new_spn = f'host/{new_host_name}'
+
+ m = Message(Dn(self.ldb_user, dn))
+ m['0'] = MessageElement(new_spn,
+ FLAG_MOD_ADD,
+ 'servicePrincipalName')
+ m['1'] = MessageElement(new_host_name,
+ FLAG_MOD_REPLACE,
+ 'dNSHostName')
+ m['2'] = MessageElement(f'{new_account_name}$',
+ FLAG_MOD_REPLACE,
+ 'sAMAccountName')
+ try:
+ self.ldb_user.modify(m)
+ except LdbError:
+ self.fail()
+
+# enable these when we have search implemented
+
+
+class AclSearchTests(AclTests):
+
+ def setUp(self):
+ super(AclSearchTests, self).setUp()
+
+ # permit password changes during this test
+ PasswordCommon.allow_password_changes(self, self.ldb_admin)
+
+ self.u1 = "search_u1"
+ self.u2 = "search_u2"
+ self.u3 = "search_u3"
+ self.group1 = "group1"
+ self.ldb_admin.newuser(self.u1, self.user_pass)
+ self.ldb_admin.newuser(self.u2, self.user_pass)
+ self.ldb_admin.newuser(self.u3, self.user_pass)
+ self.ldb_admin.newgroup(self.group1, grouptype=samba.dsdb.GTYPE_SECURITY_GLOBAL_GROUP)
+ self.ldb_admin.add_remove_group_members(self.group1, [self.u2],
+ add_members_operation=True)
+ self.ldb_user = self.get_ldb_connection(self.u1, self.user_pass)
+ self.ldb_user2 = self.get_ldb_connection(self.u2, self.user_pass)
+ self.ldb_user3 = self.get_ldb_connection(self.u3, self.user_pass)
+ self.full_list = [Dn(self.ldb_admin, "OU=ou2,OU=ou1," + self.base_dn),
+ Dn(self.ldb_admin, "OU=ou1," + self.base_dn),
+ Dn(self.ldb_admin, "OU=ou3,OU=ou2,OU=ou1," + self.base_dn),
+ Dn(self.ldb_admin, "OU=ou4,OU=ou2,OU=ou1," + self.base_dn),
+ Dn(self.ldb_admin, "OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn),
+ Dn(self.ldb_admin, "OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn)]
+ self.user_sid = self.sd_utils.get_object_sid(self.get_user_dn(self.u1))
+ self.group_sid = self.sd_utils.get_object_sid(self.get_user_dn(self.group1))
+
+ def create_clean_ou(self, object_dn):
+ """ Base repeating setup for unittests to follow """
+ res = self.ldb_admin.search(base=self.base_dn, scope=SCOPE_SUBTREE,
+ expression="distinguishedName=%s" % object_dn)
+ # Make sure top testing OU has been deleted before starting the test
+ self.assertEqual(len(res), 0)
+ self.ldb_admin.create_ou(object_dn)
+ desc_sddl = self.sd_utils.get_sd_as_sddl(object_dn)
+ # Make sure there are inheritable ACEs initially
+ self.assertTrue("CI" in desc_sddl or "OI" in desc_sddl)
+ # Find and remove all inherit ACEs
+ res = re.findall(r"\(.*?\)", desc_sddl)
+ res = [x for x in res if ("CI" in x) or ("OI" in x)]
+ for x in res:
+ desc_sddl = desc_sddl.replace(x, "")
+ # Add flag 'protected' in both DACL and SACL so no inherit ACEs
+ # can propagate from above
+ # remove SACL, we are not interested
+ desc_sddl = desc_sddl.replace(":AI", ":AIP")
+ self.sd_utils.modify_sd_on_dn(object_dn, desc_sddl)
+ # Verify all inheritable ACEs are gone
+ desc_sddl = self.sd_utils.get_sd_as_sddl(object_dn)
+ self.assertFalse("CI" in desc_sddl)
+ self.assertFalse("OI" in desc_sddl)
+
+ def tearDown(self):
+ super(AclSearchTests, self).tearDown()
+ delete_force(self.ldb_admin, "OU=test_search_ou2,OU=test_search_ou1," + self.base_dn)
+ delete_force(self.ldb_admin, "OU=test_search_ou1," + self.base_dn)
+ delete_force(self.ldb_admin, "OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn)
+ delete_force(self.ldb_admin, "OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn)
+ delete_force(self.ldb_admin, "OU=ou4,OU=ou2,OU=ou1," + self.base_dn)
+ delete_force(self.ldb_admin, "OU=ou3,OU=ou2,OU=ou1," + self.base_dn)
+ delete_force(self.ldb_admin, "OU=ou2,OU=ou1," + self.base_dn)
+ delete_force(self.ldb_admin, "OU=ou1," + self.base_dn)
+ delete_force(self.ldb_admin, self.get_user_dn("search_u1"))
+ delete_force(self.ldb_admin, self.get_user_dn("search_u2"))
+ delete_force(self.ldb_admin, self.get_user_dn("search_u3"))
+ delete_force(self.ldb_admin, self.get_user_dn("group1"))
+
+ del self.ldb_user
+ del self.ldb_user2
+ del self.ldb_user3
+
+ def test_search_anonymous1(self):
+ """Verify access of rootDSE with the correct request"""
+ anonymous = SamDB(url=ldaphost, credentials=self.creds_tmp, lp=lp)
+ res = anonymous.search("", expression="(objectClass=*)", scope=SCOPE_BASE)
+ self.assertEqual(len(res), 1)
+ # verify some of the attributes
+ # don't care about values
+ self.assertTrue("ldapServiceName" in res[0])
+ self.assertTrue("namingContexts" in res[0])
+ self.assertTrue("isSynchronized" in res[0])
+ self.assertTrue("dsServiceName" in res[0])
+ self.assertTrue("supportedSASLMechanisms" in res[0])
+ self.assertTrue("isGlobalCatalogReady" in res[0])
+ self.assertTrue("domainControllerFunctionality" in res[0])
+ self.assertTrue("serverName" in res[0])
+
+ def test_search_anonymous2(self):
+ """Make sure we cannot access anything else"""
+ anonymous = SamDB(url=ldaphost, credentials=self.creds_tmp, lp=lp)
+ try:
+ res = anonymous.search("", expression="(objectClass=*)", scope=SCOPE_SUBTREE)
+ except LdbError as e15:
+ (num, _) = e15.args
+ self.assertEqual(num, ERR_OPERATIONS_ERROR)
+ else:
+ self.fail()
+ try:
+ res = anonymous.search(self.base_dn, expression="(objectClass=*)", scope=SCOPE_SUBTREE)
+ except LdbError as e16:
+ (num, _) = e16.args
+ self.assertEqual(num, ERR_OPERATIONS_ERROR)
+ else:
+ self.fail()
+ try:
+ res = anonymous.search(anonymous.get_config_basedn(), expression="(objectClass=*)",
+ scope=SCOPE_SUBTREE)
+ except LdbError as e17:
+ (num, _) = e17.args
+ self.assertEqual(num, ERR_OPERATIONS_ERROR)
+ else:
+ self.fail()
+
+ def test_search_anonymous3(self):
+ """Set dsHeuristics and repeat"""
+ self.ldb_admin.set_dsheuristics("0000002")
+ self.ldb_admin.create_ou("OU=test_search_ou1," + self.base_dn)
+ mod = "(A;CI;LC;;;AN)"
+ self.sd_utils.dacl_add_ace("OU=test_search_ou1," + self.base_dn, mod)
+ self.ldb_admin.create_ou("OU=test_search_ou2,OU=test_search_ou1," + self.base_dn)
+ anonymous = SamDB(url=ldaphost, credentials=self.creds_tmp, lp=lp)
+ res = anonymous.search("OU=test_search_ou2,OU=test_search_ou1," + self.base_dn,
+ expression="(objectClass=*)", scope=SCOPE_SUBTREE)
+ self.assertEqual(len(res), 1)
+ self.assertTrue("dn" in res[0])
+ self.assertTrue(res[0]["dn"] == Dn(self.ldb_admin,
+ "OU=test_search_ou2,OU=test_search_ou1," + self.base_dn))
+ res = anonymous.search(anonymous.get_config_basedn(), expression="(objectClass=*)",
+ scope=SCOPE_SUBTREE)
+ self.assertEqual(len(res), 1)
+ self.assertTrue("dn" in res[0])
+ self.assertTrue(res[0]["dn"] == Dn(self.ldb_admin, self.configuration_dn))
+
+ def test_search1(self):
+ """Make sure users can see us if given LC to user and group"""
+ self.create_clean_ou("OU=ou1," + self.base_dn)
+ mod = "(A;;LC;;;%s)(A;;LC;;;%s)" % (str(self.user_sid), str(self.group_sid))
+ self.sd_utils.dacl_add_ace("OU=ou1," + self.base_dn, mod)
+ tmp_desc = security.descriptor.from_sddl("D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)" + mod,
+ self.domain_sid)
+ self.ldb_admin.create_ou("OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
+ self.ldb_admin.create_ou("OU=ou3,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
+ self.ldb_admin.create_ou("OU=ou4,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
+ self.ldb_admin.create_ou("OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
+ self.ldb_admin.create_ou("OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
+
+ # regular users must see only ou1 and ou2
+ res = self.ldb_user3.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
+ scope=SCOPE_SUBTREE)
+ self.assertEqual(len(res), 2)
+ ok_list = [Dn(self.ldb_admin, "OU=ou2,OU=ou1," + self.base_dn),
+ Dn(self.ldb_admin, "OU=ou1," + self.base_dn)]
+
+ res_list = [x["dn"] for x in res if x["dn"] in ok_list]
+ self.assertEqual(sorted(res_list), sorted(ok_list))
+
+ # these users should see all ous
+ res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
+ scope=SCOPE_SUBTREE)
+ self.assertEqual(len(res), 6)
+ res_list = [x["dn"] for x in res if x["dn"] in self.full_list]
+ self.assertEqual(sorted(res_list), sorted(self.full_list))
+
+ res = self.ldb_user2.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
+ scope=SCOPE_SUBTREE)
+ self.assertEqual(len(res), 6)
+ res_list = [x["dn"] for x in res if x["dn"] in self.full_list]
+ self.assertEqual(sorted(res_list), sorted(self.full_list))
+
+ def test_search2(self):
+ """Make sure users can't see us if access is explicitly denied"""
+ self.create_clean_ou("OU=ou1," + self.base_dn)
+ self.ldb_admin.create_ou("OU=ou2,OU=ou1," + self.base_dn)
+ self.ldb_admin.create_ou("OU=ou3,OU=ou2,OU=ou1," + self.base_dn)
+ self.ldb_admin.create_ou("OU=ou4,OU=ou2,OU=ou1," + self.base_dn)
+ self.ldb_admin.create_ou("OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn)
+ self.ldb_admin.create_ou("OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn)
+ mod = "(D;;LC;;;%s)(D;;LC;;;%s)" % (str(self.user_sid), str(self.group_sid))
+ self.sd_utils.dacl_add_ace("OU=ou2,OU=ou1," + self.base_dn, mod)
+ res = self.ldb_user3.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
+ scope=SCOPE_SUBTREE)
+ # this user should see all ous
+ res_list = [x["dn"] for x in res if x["dn"] in self.full_list]
+ self.assertEqual(sorted(res_list), sorted(self.full_list))
+
+ # these users should see ou1, 2, 5 and 6 but not 3 and 4
+ res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
+ scope=SCOPE_SUBTREE)
+ ok_list = [Dn(self.ldb_admin, "OU=ou2,OU=ou1," + self.base_dn),
+ Dn(self.ldb_admin, "OU=ou1," + self.base_dn),
+ Dn(self.ldb_admin, "OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn),
+ Dn(self.ldb_admin, "OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn)]
+ res_list = [x["dn"] for x in res if x["dn"] in ok_list]
+ self.assertEqual(sorted(res_list), sorted(ok_list))
+
+ res = self.ldb_user2.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
+ scope=SCOPE_SUBTREE)
+ self.assertEqual(len(res), 4)
+ res_list = [x["dn"] for x in res if x["dn"] in ok_list]
+ self.assertEqual(sorted(res_list), sorted(ok_list))
+
+ def test_search3(self):
+ """Make sure users can't see ous if access is explicitly denied - 2"""
+ self.create_clean_ou("OU=ou1," + self.base_dn)
+ mod = "(A;CI;LC;;;%s)(A;CI;LC;;;%s)" % (str(self.user_sid), str(self.group_sid))
+ self.sd_utils.dacl_add_ace("OU=ou1," + self.base_dn, mod)
+ tmp_desc = security.descriptor.from_sddl("D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)" + mod,
+ self.domain_sid)
+ self.ldb_admin.create_ou("OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
+ self.ldb_admin.create_ou("OU=ou3,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
+ self.ldb_admin.create_ou("OU=ou4,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
+ self.ldb_admin.create_ou("OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
+ self.ldb_admin.create_ou("OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
+
+ print("Testing correct behavior on nonaccessible search base")
+ try:
+ self.ldb_user3.search("OU=ou3,OU=ou2,OU=ou1," + self.base_dn, expression="(objectClass=*)",
+ scope=SCOPE_BASE)
+ except LdbError as e18:
+ (num, _) = e18.args
+ self.assertEqual(num, ERR_NO_SUCH_OBJECT)
+ else:
+ self.fail()
+
+ mod = "(D;;LC;;;%s)(D;;LC;;;%s)" % (str(self.user_sid), str(self.group_sid))
+ self.sd_utils.dacl_add_ace("OU=ou2,OU=ou1," + self.base_dn, mod)
+
+ ok_list = [Dn(self.ldb_admin, "OU=ou2,OU=ou1," + self.base_dn),
+ Dn(self.ldb_admin, "OU=ou1," + self.base_dn)]
+
+ res = self.ldb_user3.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
+ scope=SCOPE_SUBTREE)
+ res_list = [x["dn"] for x in res if x["dn"] in ok_list]
+ self.assertEqual(sorted(res_list), sorted(ok_list))
+
+ ok_list = [Dn(self.ldb_admin, "OU=ou2,OU=ou1," + self.base_dn),
+ Dn(self.ldb_admin, "OU=ou1," + self.base_dn),
+ Dn(self.ldb_admin, "OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn),
+ Dn(self.ldb_admin, "OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn)]
+
+ # should not see ou3 and ou4, but should see ou5 and ou6
+ res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
+ scope=SCOPE_SUBTREE)
+ self.assertEqual(len(res), 4)
+ res_list = [x["dn"] for x in res if x["dn"] in ok_list]
+ self.assertEqual(sorted(res_list), sorted(ok_list))
+
+ res = self.ldb_user2.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
+ scope=SCOPE_SUBTREE)
+ self.assertEqual(len(res), 4)
+ res_list = [x["dn"] for x in res if x["dn"] in ok_list]
+ self.assertEqual(sorted(res_list), sorted(ok_list))
+
+ def test_search4(self):
+ """There is no difference in visibility if the user is also creator"""
+ self.create_clean_ou("OU=ou1," + self.base_dn)
+ mod = "(A;CI;CC;;;%s)" % (str(self.user_sid))
+ self.sd_utils.dacl_add_ace("OU=ou1," + self.base_dn, mod)
+ tmp_desc = security.descriptor.from_sddl("D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)" + mod,
+ self.domain_sid)
+ self.ldb_user.create_ou("OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
+ self.ldb_user.create_ou("OU=ou3,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
+ self.ldb_user.create_ou("OU=ou4,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
+ self.ldb_user.create_ou("OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
+ self.ldb_user.create_ou("OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
+
+ ok_list = [Dn(self.ldb_admin, "OU=ou2,OU=ou1," + self.base_dn),
+ Dn(self.ldb_admin, "OU=ou1," + self.base_dn)]
+ res = self.ldb_user3.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
+ scope=SCOPE_SUBTREE)
+ self.assertEqual(len(res), 2)
+ res_list = [x["dn"] for x in res if x["dn"] in ok_list]
+ self.assertEqual(sorted(res_list), sorted(ok_list))
+
+ res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(objectClass=*)",
+ scope=SCOPE_SUBTREE)
+ self.assertEqual(len(res), 2)
+ res_list = [x["dn"] for x in res if x["dn"] in ok_list]
+ self.assertEqual(sorted(res_list), sorted(ok_list))
+
+ def test_search5(self):
+ """Make sure users can see only attributes they are allowed to see"""
+ self.create_clean_ou("OU=ou1," + self.base_dn)
+ mod = "(A;CI;LC;;;%s)" % (str(self.user_sid))
+ self.sd_utils.dacl_add_ace("OU=ou1," + self.base_dn, mod)
+ tmp_desc = security.descriptor.from_sddl("D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)" + mod,
+ self.domain_sid)
+ self.ldb_admin.create_ou("OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
+ # assert user can only see dn
+ res = self.ldb_user.search("OU=ou2,OU=ou1," + self.base_dn, expression="(objectClass=*)",
+ scope=SCOPE_SUBTREE)
+ ok_list = ['dn']
+ self.assertEqual(len(res), 1)
+ res_list = list(res[0].keys())
+ self.assertEqual(res_list, ok_list)
+
+ res = self.ldb_user.search("OU=ou2,OU=ou1," + self.base_dn, expression="(objectClass=*)",
+ scope=SCOPE_BASE, attrs=["ou"])
+
+ self.assertEqual(len(res), 1)
+ res_list = list(res[0].keys())
+ self.assertEqual(res_list, ok_list)
+
+ # give read property on ou and assert user can only see dn and ou
+ mod = "(OA;;RP;bf9679f0-0de6-11d0-a285-00aa003049e2;;%s)" % (str(self.user_sid))
+ self.sd_utils.dacl_add_ace("OU=ou1," + self.base_dn, mod)
+ self.sd_utils.dacl_add_ace("OU=ou2,OU=ou1," + self.base_dn, mod)
+ res = self.ldb_user.search("OU=ou2,OU=ou1," + self.base_dn, expression="(objectClass=*)",
+ scope=SCOPE_SUBTREE)
+ ok_list = ['dn', 'ou']
+ self.assertEqual(len(res), 1)
+ res_list = list(res[0].keys())
+ self.assertEqual(sorted(res_list), sorted(ok_list))
+
+ # give read property on Public Information and assert user can see ou and other members
+ mod = "(OA;;RP;e48d0154-bcf8-11d1-8702-00c04fb96050;;%s)" % (str(self.user_sid))
+ self.sd_utils.dacl_add_ace("OU=ou1," + self.base_dn, mod)
+ self.sd_utils.dacl_add_ace("OU=ou2,OU=ou1," + self.base_dn, mod)
+ res = self.ldb_user.search("OU=ou2,OU=ou1," + self.base_dn, expression="(objectClass=*)",
+ scope=SCOPE_SUBTREE)
+
+ ok_list = ['dn', 'objectClass', 'ou', 'distinguishedName', 'name', 'objectGUID', 'objectCategory']
+ res_list = list(res[0].keys())
+ self.assertEqual(sorted(res_list), sorted(ok_list))
+
+ def test_search6(self):
+ """If an attribute that cannot be read is used in a filter, it is as if the attribute does not exist"""
+ self.create_clean_ou("OU=ou1," + self.base_dn)
+ mod = "(A;CI;LCCC;;;%s)" % (str(self.user_sid))
+ self.sd_utils.dacl_add_ace("OU=ou1," + self.base_dn, mod)
+ tmp_desc = security.descriptor.from_sddl("D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)" + mod,
+ self.domain_sid)
+ self.ldb_admin.create_ou("OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
+ self.ldb_user.create_ou("OU=ou3,OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
+
+ res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(ou=ou3)",
+ scope=SCOPE_SUBTREE)
+ # nothing should be returned as ou is not accessible
+ self.assertEqual(len(res), 0)
+
+ # give read property on ou and assert user can only see dn and ou
+ mod = "(OA;;RP;bf9679f0-0de6-11d0-a285-00aa003049e2;;%s)" % (str(self.user_sid))
+ self.sd_utils.dacl_add_ace("OU=ou3,OU=ou2,OU=ou1," + self.base_dn, mod)
+ res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(ou=ou3)",
+ scope=SCOPE_SUBTREE)
+ self.assertEqual(len(res), 1)
+ ok_list = ['dn', 'ou']
+ res_list = list(res[0].keys())
+ self.assertEqual(sorted(res_list), sorted(ok_list))
+
+ # give read property on Public Information and assert user can see ou and other members
+ mod = "(OA;;RP;e48d0154-bcf8-11d1-8702-00c04fb96050;;%s)" % (str(self.user_sid))
+ self.sd_utils.dacl_add_ace("OU=ou2,OU=ou1," + self.base_dn, mod)
+ res = self.ldb_user.search("OU=ou1," + self.base_dn, expression="(ou=ou2)",
+ scope=SCOPE_SUBTREE)
+ self.assertEqual(len(res), 1)
+ ok_list = ['dn', 'objectClass', 'ou', 'distinguishedName', 'name', 'objectGUID', 'objectCategory']
+ res_list = list(res[0].keys())
+ self.assertEqual(sorted(res_list), sorted(ok_list))
+
+ def assert_search_on_attr(self, dn, samdb, attr, expected_list):
+
+ expected_num = len(expected_list)
+ res = samdb.search(dn, expression="(%s=*)" % attr, scope=SCOPE_SUBTREE)
+ self.assertEqual(len(res), expected_num)
+
+ res_list = [ x["dn"] for x in res if x["dn"] in expected_list ]
+ self.assertEqual(sorted(res_list), sorted(expected_list))
+
+ def test_search7(self):
+ """Checks object search visibility when users don't have full rights"""
+ self.create_clean_ou("OU=ou1," + self.base_dn)
+ mod = "(A;;LC;;;%s)(A;;LC;;;%s)" % (str(self.user_sid),
+ str(self.group_sid))
+ self.sd_utils.dacl_add_ace("OU=ou1," + self.base_dn, mod)
+ tmp_desc = security.descriptor.from_sddl("D:(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)" + mod,
+ self.domain_sid)
+ self.ldb_admin.create_ou("OU=ou2,OU=ou1," + self.base_dn, sd=tmp_desc)
+ self.ldb_admin.create_ou("OU=ou3,OU=ou2,OU=ou1," + self.base_dn,
+ sd=tmp_desc)
+ self.ldb_admin.create_ou("OU=ou4,OU=ou2,OU=ou1," + self.base_dn,
+ sd=tmp_desc)
+ self.ldb_admin.create_ou("OU=ou5,OU=ou3,OU=ou2,OU=ou1," + self.base_dn,
+ sd=tmp_desc)
+ self.ldb_admin.create_ou("OU=ou6,OU=ou4,OU=ou2,OU=ou1," + self.base_dn,
+ sd=tmp_desc)
+
+ ou2_dn = Dn(self.ldb_admin, "OU=ou2,OU=ou1," + self.base_dn)
+ ou1_dn = Dn(self.ldb_admin, "OU=ou1," + self.base_dn)
+
+ # even though unprivileged users can't read these attributes for OU2,
+ # the object should still be visible in searches, because they have
+ # 'List Contents' rights still. This isn't really disclosive because
+ # ALL objects have these attributes
+ visible_attrs = ["objectClass", "distinguishedName", "name",
+ "objectGUID"]
+ two_objects = [ou2_dn, ou1_dn]
+
+ for attr in visible_attrs:
+ # a regular user should just see the 2 objects
+ self.assert_search_on_attr(str(ou1_dn), self.ldb_user3, attr,
+ expected_list=two_objects)
+
+ # whereas the following users have LC rights for all the objects,
+ # so they should see them all
+ self.assert_search_on_attr(str(ou1_dn), self.ldb_user, attr,
+ expected_list=self.full_list)
+ self.assert_search_on_attr(str(ou1_dn), self.ldb_user2, attr,
+ expected_list=self.full_list)
+
+ # however when searching on the following attributes, objects will not
+ # be visible unless the user has Read Property rights
+ hidden_attrs = ["objectCategory", "instanceType", "ou", "uSNChanged",
+ "uSNCreated", "whenCreated"]
+ one_object = [ou1_dn]
+
+ for attr in hidden_attrs:
+ self.assert_search_on_attr(str(ou1_dn), self.ldb_user3, attr,
+ expected_list=one_object)
+ self.assert_search_on_attr(str(ou1_dn), self.ldb_user, attr,
+ expected_list=one_object)
+ self.assert_search_on_attr(str(ou1_dn), self.ldb_user2, attr,
+ expected_list=one_object)
+
+ # admin has RP rights so can still see all the objects
+ self.assert_search_on_attr(str(ou1_dn), self.ldb_admin, attr,
+ expected_list=self.full_list)
+
+
+# tests on ldap delete operations
+
+
+class AclDeleteTests(AclTests):
+
+ def setUp(self):
+ super(AclDeleteTests, self).setUp()
+ self.regular_user = "acl_delete_user1"
+ # Create regular user
+ self.ldb_admin.newuser(self.regular_user, self.user_pass)
+ self.ldb_user = self.get_ldb_connection(self.regular_user, self.user_pass)
+
+ def tearDown(self):
+ super(AclDeleteTests, self).tearDown()
+ delete_force(self.ldb_admin, self.get_user_dn("test_delete_user1"))
+ delete_force(self.ldb_admin, self.get_user_dn(self.regular_user))
+ delete_force(self.ldb_admin, self.get_user_dn("test_anonymous"))
+
+ del self.ldb_user
+
+ def test_delete_u1(self):
+ """User is prohibited by default to delete another User object"""
+ # Create user that we try to delete
+ self.ldb_admin.newuser("test_delete_user1", self.user_pass)
+ # Here delete User object should ALWAYS through exception
+ try:
+ self.ldb_user.delete(self.get_user_dn("test_delete_user1"))
+ except LdbError as e19:
+ (num, _) = e19.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+ else:
+ self.fail()
+
+ def test_delete_u2(self):
+ """User's group has RIGHT_DELETE to another User object"""
+ user_dn = self.get_user_dn("test_delete_user1")
+ # Create user that we try to delete
+ self.ldb_admin.newuser("test_delete_user1", self.user_pass)
+ mod = "(A;;SD;;;AU)"
+ self.sd_utils.dacl_add_ace(user_dn, mod)
+ # Try to delete User object
+ self.ldb_user.delete(user_dn)
+ res = self.ldb_admin.search(self.base_dn,
+ expression="(distinguishedName=%s)" % user_dn)
+ self.assertEqual(len(res), 0)
+
+ def test_delete_u3(self):
+ """User identified by SID has RIGHT_DELETE to another User object"""
+ user_dn = self.get_user_dn("test_delete_user1")
+ # Create user that we try to delete
+ self.ldb_admin.newuser("test_delete_user1", self.user_pass)
+ mod = "(A;;SD;;;%s)" % self.sd_utils.get_object_sid(self.get_user_dn(self.regular_user))
+ self.sd_utils.dacl_add_ace(user_dn, mod)
+ # Try to delete User object
+ self.ldb_user.delete(user_dn)
+ res = self.ldb_admin.search(self.base_dn,
+ expression="(distinguishedName=%s)" % user_dn)
+ self.assertEqual(len(res), 0)
+
+ def test_delete_anonymous(self):
+ """Test add operation with anonymous user"""
+ anonymous = SamDB(url=ldaphost, credentials=self.creds_tmp, lp=lp)
+ self.ldb_admin.newuser("test_anonymous", "samba123@")
+
+ try:
+ anonymous.delete(self.get_user_dn("test_anonymous"))
+ except LdbError as e20:
+ (num, _) = e20.args
+ self.assertEqual(num, ERR_OPERATIONS_ERROR)
+ else:
+ self.fail()
+
+# tests on ldap rename operations
+
+
+class AclRenameTests(AclTests):
+
+ def setUp(self):
+ super(AclRenameTests, self).setUp()
+ self.regular_user = "acl_rename_user1"
+ self.ou1 = "OU=test_rename_ou1"
+ self.ou2 = "OU=test_rename_ou2"
+ self.ou3 = "OU=test_rename_ou3,%s" % self.ou2
+ self.testuser1 = "test_rename_user1"
+ self.testuser2 = "test_rename_user2"
+ self.testuser3 = "test_rename_user3"
+ self.testuser4 = "test_rename_user4"
+ self.testuser5 = "test_rename_user5"
+ # Create regular user
+ self.ldb_admin.newuser(self.regular_user, self.user_pass)
+ self.ldb_user = self.get_ldb_connection(self.regular_user, self.user_pass)
+
+ def tearDown(self):
+ super(AclRenameTests, self).tearDown()
+ # Rename OU3
+ delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser1, self.ou3, self.base_dn))
+ delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser2, self.ou3, self.base_dn))
+ delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser5, self.ou3, self.base_dn))
+ delete_force(self.ldb_admin, "%s,%s" % (self.ou3, self.base_dn))
+ # Rename OU2
+ delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser1, self.ou2, self.base_dn))
+ delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser2, self.ou2, self.base_dn))
+ delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser5, self.ou2, self.base_dn))
+ delete_force(self.ldb_admin, "%s,%s" % (self.ou2, self.base_dn))
+ # Rename OU1
+ delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser1, self.ou1, self.base_dn))
+ delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser2, self.ou1, self.base_dn))
+ delete_force(self.ldb_admin, "CN=%s,%s,%s" % (self.testuser5, self.ou1, self.base_dn))
+ delete_force(self.ldb_admin, "OU=test_rename_ou3,%s,%s" % (self.ou1, self.base_dn))
+ delete_force(self.ldb_admin, "%s,%s" % (self.ou1, self.base_dn))
+ delete_force(self.ldb_admin, self.get_user_dn(self.regular_user))
+
+ del self.ldb_user
+
+ def test_rename_u1(self):
+ """Regular user fails to rename 'User object' within single OU"""
+ # Create OU structure
+ self.ldb_admin.create_ou("OU=test_rename_ou1," + self.base_dn)
+ self.ldb_admin.newuser(self.testuser1, self.user_pass, userou=self.ou1)
+ try:
+ self.ldb_user.rename("CN=%s,%s,%s" % (self.testuser1, self.ou1, self.base_dn),
+ "CN=%s,%s,%s" % (self.testuser5, self.ou1, self.base_dn))
+ except LdbError as e21:
+ (num, _) = e21.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+ else:
+ self.fail()
+
+ def test_rename_u2(self):
+ """Grant WRITE_PROPERTY to AU so regular user can rename 'User object' within single OU"""
+ ou_dn = "OU=test_rename_ou1," + self.base_dn
+ user_dn = "CN=test_rename_user1," + ou_dn
+ rename_user_dn = "CN=test_rename_user5," + ou_dn
+ # Create OU structure
+ self.ldb_admin.create_ou(ou_dn)
+ self.ldb_admin.newuser(self.testuser1, self.user_pass, userou=self.ou1)
+ mod = "(A;;WP;;;AU)"
+ self.sd_utils.dacl_add_ace(user_dn, mod)
+ # Rename 'User object' having WP to AU
+ self.ldb_user.rename(user_dn, rename_user_dn)
+ res = self.ldb_admin.search(self.base_dn,
+ expression="(distinguishedName=%s)" % user_dn)
+ self.assertEqual(len(res), 0)
+ res = self.ldb_admin.search(self.base_dn,
+ expression="(distinguishedName=%s)" % rename_user_dn)
+ self.assertNotEqual(len(res), 0)
+
+ def test_rename_u3(self):
+ """Test rename with rights granted to 'User object' SID"""
+ ou_dn = "OU=test_rename_ou1," + self.base_dn
+ user_dn = "CN=test_rename_user1," + ou_dn
+ rename_user_dn = "CN=test_rename_user5," + ou_dn
+ # Create OU structure
+ self.ldb_admin.create_ou(ou_dn)
+ self.ldb_admin.newuser(self.testuser1, self.user_pass, userou=self.ou1)
+ sid = self.sd_utils.get_object_sid(self.get_user_dn(self.regular_user))
+ mod = "(A;;WP;;;%s)" % str(sid)
+ self.sd_utils.dacl_add_ace(user_dn, mod)
+ # Rename 'User object' having WP to AU
+ self.ldb_user.rename(user_dn, rename_user_dn)
+ res = self.ldb_admin.search(self.base_dn,
+ expression="(distinguishedName=%s)" % user_dn)
+ self.assertEqual(len(res), 0)
+ res = self.ldb_admin.search(self.base_dn,
+ expression="(distinguishedName=%s)" % rename_user_dn)
+ self.assertNotEqual(len(res), 0)
+
+ def test_rename_u4(self):
+ """Rename 'User object' cross OU with WP, SD and CC right granted on reg. user to AU"""
+ ou1_dn = "OU=test_rename_ou1," + self.base_dn
+ ou2_dn = "OU=test_rename_ou2," + self.base_dn
+ user_dn = "CN=test_rename_user2," + ou1_dn
+ rename_user_dn = "CN=test_rename_user5," + ou2_dn
+ # Create OU structure
+ self.ldb_admin.create_ou(ou1_dn)
+ self.ldb_admin.create_ou(ou2_dn)
+ self.ldb_admin.newuser(self.testuser2, self.user_pass, userou=self.ou1)
+ mod = "(A;;WPSD;;;AU)"
+ self.sd_utils.dacl_add_ace(user_dn, mod)
+ mod = "(A;;CC;;;AU)"
+ self.sd_utils.dacl_add_ace(ou2_dn, mod)
+ # Rename 'User object' having SD and CC to AU
+ self.ldb_user.rename(user_dn, rename_user_dn)
+ res = self.ldb_admin.search(self.base_dn,
+ expression="(distinguishedName=%s)" % user_dn)
+ self.assertEqual(len(res), 0)
+ res = self.ldb_admin.search(self.base_dn,
+ expression="(distinguishedName=%s)" % rename_user_dn)
+ self.assertNotEqual(len(res), 0)
+
+ def test_rename_u5(self):
+ """Test rename with rights granted to 'User object' SID"""
+ ou1_dn = "OU=test_rename_ou1," + self.base_dn
+ ou2_dn = "OU=test_rename_ou2," + self.base_dn
+ user_dn = "CN=test_rename_user2," + ou1_dn
+ rename_user_dn = "CN=test_rename_user5," + ou2_dn
+ # Create OU structure
+ self.ldb_admin.create_ou(ou1_dn)
+ self.ldb_admin.create_ou(ou2_dn)
+ self.ldb_admin.newuser(self.testuser2, self.user_pass, userou=self.ou1)
+ sid = self.sd_utils.get_object_sid(self.get_user_dn(self.regular_user))
+ mod = "(A;;WPSD;;;%s)" % str(sid)
+ self.sd_utils.dacl_add_ace(user_dn, mod)
+ mod = "(A;;CC;;;%s)" % str(sid)
+ self.sd_utils.dacl_add_ace(ou2_dn, mod)
+ # Rename 'User object' having SD and CC to AU
+ self.ldb_user.rename(user_dn, rename_user_dn)
+ res = self.ldb_admin.search(self.base_dn,
+ expression="(distinguishedName=%s)" % user_dn)
+ self.assertEqual(len(res), 0)
+ res = self.ldb_admin.search(self.base_dn,
+ expression="(distinguishedName=%s)" % rename_user_dn)
+ self.assertNotEqual(len(res), 0)
+
+ def test_rename_u6(self):
+ """Rename 'User object' cross OU with WP, DC and CC right granted on OU & user to AU"""
+ ou1_dn = "OU=test_rename_ou1," + self.base_dn
+ ou2_dn = "OU=test_rename_ou2," + self.base_dn
+ user_dn = "CN=test_rename_user2," + ou1_dn
+ rename_user_dn = "CN=test_rename_user2," + ou2_dn
+ # Create OU structure
+ self.ldb_admin.create_ou(ou1_dn)
+ self.ldb_admin.create_ou(ou2_dn)
+ #mod = "(A;CI;DCWP;;;AU)"
+ mod = "(A;;DC;;;AU)"
+ self.sd_utils.dacl_add_ace(ou1_dn, mod)
+ mod = "(A;;CC;;;AU)"
+ self.sd_utils.dacl_add_ace(ou2_dn, mod)
+ self.ldb_admin.newuser(self.testuser2, self.user_pass, userou=self.ou1)
+ mod = "(A;;WP;;;AU)"
+ self.sd_utils.dacl_add_ace(user_dn, mod)
+ # Rename 'User object' having SD and CC to AU
+ self.ldb_user.rename(user_dn, rename_user_dn)
+ res = self.ldb_admin.search(self.base_dn,
+ expression="(distinguishedName=%s)" % user_dn)
+ self.assertEqual(len(res), 0)
+ res = self.ldb_admin.search(self.base_dn,
+ expression="(distinguishedName=%s)" % rename_user_dn)
+ self.assertNotEqual(len(res), 0)
+
+ def test_rename_u7(self):
+ """Rename 'User object' cross OU (second level) with WP, DC and CC right granted on OU to AU"""
+ ou1_dn = "OU=test_rename_ou1," + self.base_dn
+ ou2_dn = "OU=test_rename_ou2," + self.base_dn
+ ou3_dn = "OU=test_rename_ou3," + ou2_dn
+ user_dn = "CN=test_rename_user2," + ou1_dn
+ rename_user_dn = "CN=test_rename_user5," + ou3_dn
+ # Create OU structure
+ self.ldb_admin.create_ou(ou1_dn)
+ self.ldb_admin.create_ou(ou2_dn)
+ self.ldb_admin.create_ou(ou3_dn)
+ mod = "(A;CI;WPDC;;;AU)"
+ self.sd_utils.dacl_add_ace(ou1_dn, mod)
+ mod = "(A;;CC;;;AU)"
+ self.sd_utils.dacl_add_ace(ou3_dn, mod)
+ self.ldb_admin.newuser(self.testuser2, self.user_pass, userou=self.ou1)
+ # Rename 'User object' having SD and CC to AU
+ self.ldb_user.rename(user_dn, rename_user_dn)
+ res = self.ldb_admin.search(self.base_dn,
+ expression="(distinguishedName=%s)" % user_dn)
+ self.assertEqual(len(res), 0)
+ res = self.ldb_admin.search(self.base_dn,
+ expression="(distinguishedName=%s)" % rename_user_dn)
+ self.assertNotEqual(len(res), 0)
+
+ def test_rename_u8(self):
+ """Test rename on an object with and without modify access on the RDN attribute"""
+ ou1_dn = "OU=test_rename_ou1," + self.base_dn
+ ou2_dn = "OU=test_rename_ou2," + ou1_dn
+ ou3_dn = "OU=test_rename_ou3," + ou1_dn
+ # Create OU structure
+ self.ldb_admin.create_ou(ou1_dn)
+ self.ldb_admin.create_ou(ou2_dn)
+ sid = self.sd_utils.get_object_sid(self.get_user_dn(self.regular_user))
+ mod = "(OA;;WP;bf967a0e-0de6-11d0-a285-00aa003049e2;;%s)" % str(sid)
+ self.sd_utils.dacl_add_ace(ou2_dn, mod)
+ mod = "(OD;;WP;bf9679f0-0de6-11d0-a285-00aa003049e2;;%s)" % str(sid)
+ self.sd_utils.dacl_add_ace(ou2_dn, mod)
+ try:
+ self.ldb_user.rename(ou2_dn, ou3_dn)
+ except LdbError as e22:
+ (num, _) = e22.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+ else:
+ # This rename operation should always throw ERR_INSUFFICIENT_ACCESS_RIGHTS
+ self.fail()
+ sid = self.sd_utils.get_object_sid(self.get_user_dn(self.regular_user))
+ mod = "(A;;WP;bf9679f0-0de6-11d0-a285-00aa003049e2;;%s)" % str(sid)
+ self.sd_utils.dacl_add_ace(ou2_dn, mod)
+ self.ldb_user.rename(ou2_dn, ou3_dn)
+ res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" % ou2_dn)
+ self.assertEqual(len(res), 0)
+ res = self.ldb_admin.search(self.base_dn, expression="(distinguishedName=%s)" % ou3_dn)
+ self.assertNotEqual(len(res), 0)
+
+ def test_rename_u9(self):
+ """Rename 'User object' cross OU, with explicit deny on sd and dc"""
+ ou1_dn = "OU=test_rename_ou1," + self.base_dn
+ ou2_dn = "OU=test_rename_ou2," + self.base_dn
+ user_dn = "CN=test_rename_user2," + ou1_dn
+ rename_user_dn = "CN=test_rename_user5," + ou2_dn
+ # Create OU structure
+ self.ldb_admin.create_ou(ou1_dn)
+ self.ldb_admin.create_ou(ou2_dn)
+ self.ldb_admin.newuser(self.testuser2, self.user_pass, userou=self.ou1)
+ mod = "(D;;SD;;;DA)"
+ self.sd_utils.dacl_add_ace(user_dn, mod)
+ mod = "(D;;DC;;;DA)"
+ self.sd_utils.dacl_add_ace(ou1_dn, mod)
+ # Rename 'User object' having SD and CC to AU
+ try:
+ self.ldb_admin.rename(user_dn, rename_user_dn)
+ except LdbError as e23:
+ (num, _) = e23.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+ else:
+ self.fail()
+ # add an allow ace so we can delete this ou
+ mod = "(A;;DC;;;DA)"
+ self.sd_utils.dacl_add_ace(ou1_dn, mod)
+
+
+# tests on Control Access Rights
+class AclCARTests(AclTests):
+
+ def setUp(self):
+ super(AclCARTests, self).setUp()
+
+ # Get the old "dSHeuristics" if it was set
+ dsheuristics = self.ldb_admin.get_dsheuristics()
+ # Reset the "dSHeuristics" as they were before
+ self.addCleanup(self.ldb_admin.set_dsheuristics, dsheuristics)
+ # Set the "dSHeuristics" to activate the correct "userPassword" behaviour
+ self.ldb_admin.set_dsheuristics("000000001")
+ # Get the old "minPwdAge"
+ minPwdAge = self.ldb_admin.get_minPwdAge()
+ # Reset the "minPwdAge" as it was before
+ self.addCleanup(self.ldb_admin.set_minPwdAge, minPwdAge)
+ # Set it temporarily to "0"
+ self.ldb_admin.set_minPwdAge("0")
+
+ self.user_with_wp = "acl_car_user1"
+ self.user_with_pc = "acl_car_user2"
+ self.ldb_admin.newuser(self.user_with_wp, self.user_pass)
+ self.ldb_admin.newuser(self.user_with_pc, self.user_pass)
+ self.ldb_user = self.get_ldb_connection(self.user_with_wp, self.user_pass)
+ self.ldb_user2 = self.get_ldb_connection(self.user_with_pc, self.user_pass)
+
+ def tearDown(self):
+ super(AclCARTests, self).tearDown()
+ delete_force(self.ldb_admin, self.get_user_dn(self.user_with_wp))
+ delete_force(self.ldb_admin, self.get_user_dn(self.user_with_pc))
+
+ del self.ldb_user
+ del self.ldb_user2
+
+ def test_change_password1(self):
+ """Try a password change operation without any CARs given"""
+ # users have change password by default - remove for negative testing
+ desc = self.sd_utils.read_sd_on_dn(self.get_user_dn(self.user_with_wp))
+ sddl = desc.as_sddl(self.domain_sid)
+ sddl = sddl.replace("(OA;;CR;ab721a53-1e2f-11d0-9819-00aa0040529b;;WD)", "")
+ sddl = sddl.replace("(OA;;CR;ab721a53-1e2f-11d0-9819-00aa0040529b;;PS)", "")
+ self.sd_utils.modify_sd_on_dn(self.get_user_dn(self.user_with_wp), sddl)
+ try:
+ self.ldb_user.modify_ldif("""
+dn: """ + self.get_user_dn(self.user_with_wp) + """
+changetype: modify
+delete: unicodePwd
+unicodePwd:: """ + base64.b64encode("\"samba123@\"".encode('utf-16-le')).decode('utf8') + """
+add: unicodePwd
+unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS2\"".encode('utf-16-le')).decode('utf8') + """
+""")
+ except LdbError as e24:
+ (num, _) = e24.args
+ self.assertEqual(num, ERR_CONSTRAINT_VIOLATION)
+ else:
+ # for some reason we get constraint violation instead of insufficient access error
+ self.fail()
+
+ def test_change_password2(self):
+ """Make sure WP has no influence"""
+ desc = self.sd_utils.read_sd_on_dn(self.get_user_dn(self.user_with_wp))
+ sddl = desc.as_sddl(self.domain_sid)
+ sddl = sddl.replace("(OA;;CR;ab721a53-1e2f-11d0-9819-00aa0040529b;;WD)", "")
+ sddl = sddl.replace("(OA;;CR;ab721a53-1e2f-11d0-9819-00aa0040529b;;PS)", "")
+ self.sd_utils.modify_sd_on_dn(self.get_user_dn(self.user_with_wp), sddl)
+ mod = "(A;;WP;;;PS)"
+ self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
+ desc = self.sd_utils.read_sd_on_dn(self.get_user_dn(self.user_with_wp))
+ sddl = desc.as_sddl(self.domain_sid)
+ try:
+ self.ldb_user.modify_ldif("""
+dn: """ + self.get_user_dn(self.user_with_wp) + """
+changetype: modify
+delete: unicodePwd
+unicodePwd:: """ + base64.b64encode("\"samba123@\"".encode('utf-16-le')).decode('utf8') + """
+add: unicodePwd
+unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS2\"".encode('utf-16-le')).decode('utf8') + """
+""")
+ except LdbError as e25:
+ (num, _) = e25.args
+ self.assertEqual(num, ERR_CONSTRAINT_VIOLATION)
+ else:
+ # for some reason we get constraint violation instead of insufficient access error
+ self.fail()
+
+ def test_change_password3(self):
+ """Make sure WP has no influence"""
+ mod = "(D;;WP;;;PS)"
+ self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
+ desc = self.sd_utils.read_sd_on_dn(self.get_user_dn(self.user_with_wp))
+ sddl = desc.as_sddl(self.domain_sid)
+ self.ldb_user.modify_ldif("""
+dn: """ + self.get_user_dn(self.user_with_wp) + """
+changetype: modify
+delete: unicodePwd
+unicodePwd:: """ + base64.b64encode("\"samba123@\"".encode('utf-16-le')).decode('utf8') + """
+add: unicodePwd
+unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS2\"".encode('utf-16-le')).decode('utf8') + """
+""")
+
+ def test_change_password5(self):
+ """Make sure rights have no influence on dBCSPwd"""
+ desc = self.sd_utils.read_sd_on_dn(self.get_user_dn(self.user_with_wp))
+ sddl = desc.as_sddl(self.domain_sid)
+ sddl = sddl.replace("(OA;;CR;ab721a53-1e2f-11d0-9819-00aa0040529b;;WD)", "")
+ sddl = sddl.replace("(OA;;CR;ab721a53-1e2f-11d0-9819-00aa0040529b;;PS)", "")
+ self.sd_utils.modify_sd_on_dn(self.get_user_dn(self.user_with_wp), sddl)
+ mod = "(D;;WP;;;PS)"
+ self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
+ try:
+ self.ldb_user.modify_ldif("""
+dn: """ + self.get_user_dn(self.user_with_wp) + """
+changetype: modify
+delete: dBCSPwd
+dBCSPwd: XXXXXXXXXXXXXXXX
+add: dBCSPwd
+dBCSPwd: YYYYYYYYYYYYYYYY
+""")
+ except LdbError as e26:
+ (num, _) = e26.args
+ self.assertEqual(num, ERR_UNWILLING_TO_PERFORM)
+ else:
+ self.fail()
+
+ def test_change_password6(self):
+ """Test uneven delete/adds"""
+ try:
+ self.ldb_user.modify_ldif("""
+dn: """ + self.get_user_dn(self.user_with_wp) + """
+changetype: modify
+delete: userPassword
+userPassword: thatsAcomplPASS1
+delete: userPassword
+userPassword: thatsAcomplPASS1
+add: userPassword
+userPassword: thatsAcomplPASS2
+""")
+ except LdbError as e27:
+ (num, _) = e27.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+ else:
+ self.fail()
+ mod = "(OA;;CR;00299570-246d-11d0-a768-00aa006e0529;;PS)"
+ self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
+ try:
+ self.ldb_user.modify_ldif("""
+dn: """ + self.get_user_dn(self.user_with_wp) + """
+changetype: modify
+delete: userPassword
+userPassword: thatsAcomplPASS1
+delete: userPassword
+userPassword: thatsAcomplPASS1
+add: userPassword
+userPassword: thatsAcomplPASS2
+""")
+ # This fails on Windows 2000 domain level with constraint violation
+ except LdbError as e28:
+ (num, _) = e28.args
+ self.assertTrue(num == ERR_CONSTRAINT_VIOLATION or
+ num == ERR_UNWILLING_TO_PERFORM)
+ else:
+ self.fail()
+
+ def test_change_password7(self):
+ """Try a password change operation without any CARs given"""
+ # users have change password by default - remove for negative testing
+ desc = self.sd_utils.read_sd_on_dn(self.get_user_dn(self.user_with_wp))
+ sddl = desc.as_sddl(self.domain_sid)
+ self.sd_utils.modify_sd_on_dn(self.get_user_dn(self.user_with_wp), sddl)
+ # first change our own password
+ self.ldb_user2.modify_ldif("""
+dn: """ + self.get_user_dn(self.user_with_pc) + """
+changetype: modify
+delete: unicodePwd
+unicodePwd:: """ + base64.b64encode("\"samba123@\"".encode('utf-16-le')).decode('utf8') + """
+add: unicodePwd
+unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS1\"".encode('utf-16-le')).decode('utf8') + """
+""")
+ # then someone else's
+ self.ldb_user2.modify_ldif("""
+dn: """ + self.get_user_dn(self.user_with_wp) + """
+changetype: modify
+delete: unicodePwd
+unicodePwd:: """ + base64.b64encode("\"samba123@\"".encode('utf-16-le')).decode('utf8') + """
+add: unicodePwd
+unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS2\"".encode('utf-16-le')).decode('utf8') + """
+""")
+
+ def test_reset_password1(self):
+ """Try a user password reset operation (unicodePwd) before and after granting CAR"""
+ try:
+ self.ldb_user.modify_ldif("""
+dn: """ + self.get_user_dn(self.user_with_wp) + """
+changetype: modify
+replace: unicodePwd
+unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS1\"".encode('utf-16-le')).decode('utf8') + """
+""")
+ except LdbError as e29:
+ (num, _) = e29.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+ else:
+ self.fail()
+ mod = "(OA;;CR;00299570-246d-11d0-a768-00aa006e0529;;PS)"
+ self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
+ self.ldb_user.modify_ldif("""
+dn: """ + self.get_user_dn(self.user_with_wp) + """
+changetype: modify
+replace: unicodePwd
+unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS1\"".encode('utf-16-le')).decode('utf8') + """
+""")
+
+ def test_reset_password2(self):
+ """Try a user password reset operation (userPassword) before and after granting CAR"""
+ try:
+ self.ldb_user.modify_ldif("""
+dn: """ + self.get_user_dn(self.user_with_wp) + """
+changetype: modify
+replace: userPassword
+userPassword: thatsAcomplPASS1
+""")
+ except LdbError as e30:
+ (num, _) = e30.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+ else:
+ self.fail()
+ mod = "(OA;;CR;00299570-246d-11d0-a768-00aa006e0529;;PS)"
+ self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
+ try:
+ self.ldb_user.modify_ldif("""
+dn: """ + self.get_user_dn(self.user_with_wp) + """
+changetype: modify
+replace: userPassword
+userPassword: thatsAcomplPASS1
+""")
+ # This fails on Windows 2000 domain level with constraint violation
+ except LdbError as e31:
+ (num, _) = e31.args
+ self.assertEqual(num, ERR_CONSTRAINT_VIOLATION)
+ else:
+ pass # Not self.fail() as we normally want success.
+
+ def test_reset_password3(self):
+ """Grant WP and see what happens (unicodePwd)"""
+ mod = "(A;;WP;;;PS)"
+ self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
+ try:
+ self.ldb_user.modify_ldif("""
+dn: """ + self.get_user_dn(self.user_with_wp) + """
+changetype: modify
+replace: unicodePwd
+unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS1\"".encode('utf-16-le')).decode('utf8') + """
+""")
+ except LdbError as e32:
+ (num, _) = e32.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+ else:
+ self.fail()
+
+ def test_reset_password4(self):
+ """Grant WP and see what happens (userPassword)"""
+ mod = "(A;;WP;;;PS)"
+ self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
+ try:
+ self.ldb_user.modify_ldif("""
+dn: """ + self.get_user_dn(self.user_with_wp) + """
+changetype: modify
+replace: userPassword
+userPassword: thatsAcomplPASS1
+""")
+ except LdbError as e33:
+ (num, _) = e33.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+ else:
+ self.fail()
+
+ def test_reset_password5(self):
+ """Explicitly deny WP but grant CAR (unicodePwd)"""
+ mod = "(D;;WP;;;PS)(OA;;CR;00299570-246d-11d0-a768-00aa006e0529;;PS)"
+ self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
+ self.ldb_user.modify_ldif("""
+dn: """ + self.get_user_dn(self.user_with_wp) + """
+changetype: modify
+replace: unicodePwd
+unicodePwd:: """ + base64.b64encode("\"thatsAcomplPASS1\"".encode('utf-16-le')).decode('utf8') + """
+""")
+
+ def test_reset_password6(self):
+ """Explicitly deny WP but grant CAR (userPassword)"""
+ mod = "(D;;WP;;;PS)(OA;;CR;00299570-246d-11d0-a768-00aa006e0529;;PS)"
+ self.sd_utils.dacl_add_ace(self.get_user_dn(self.user_with_wp), mod)
+ try:
+ self.ldb_user.modify_ldif("""
+dn: """ + self.get_user_dn(self.user_with_wp) + """
+changetype: modify
+replace: userPassword
+userPassword: thatsAcomplPASS1
+""")
+ # This fails on Windows 2000 domain level with constraint violation
+ except LdbError as e34:
+ (num, _) = e34.args
+ self.assertEqual(num, ERR_CONSTRAINT_VIOLATION)
+ else:
+ pass # Not self.fail() as we normally want success
+
+
+class AclExtendedTests(AclTests):
+
+ def setUp(self):
+ super(AclExtendedTests, self).setUp()
+ # regular user, will be the creator
+ self.u1 = "ext_u1"
+ # regular user
+ self.u2 = "ext_u2"
+ # admin user
+ self.u3 = "ext_u3"
+ self.ldb_admin.newuser(self.u1, self.user_pass)
+ self.ldb_admin.newuser(self.u2, self.user_pass)
+ self.ldb_admin.newuser(self.u3, self.user_pass)
+ self.ldb_admin.add_remove_group_members("Domain Admins", [self.u3],
+ add_members_operation=True)
+ self.ldb_user1 = self.get_ldb_connection(self.u1, self.user_pass)
+ self.ldb_user2 = self.get_ldb_connection(self.u2, self.user_pass)
+ self.ldb_user3 = self.get_ldb_connection(self.u3, self.user_pass)
+ self.user_sid1 = self.sd_utils.get_object_sid(self.get_user_dn(self.u1))
+ self.user_sid2 = self.sd_utils.get_object_sid(self.get_user_dn(self.u2))
+
+ def tearDown(self):
+ super(AclExtendedTests, self).tearDown()
+ delete_force(self.ldb_admin, self.get_user_dn(self.u1))
+ delete_force(self.ldb_admin, self.get_user_dn(self.u2))
+ delete_force(self.ldb_admin, self.get_user_dn(self.u3))
+ delete_force(self.ldb_admin, "CN=ext_group1,OU=ext_ou1," + self.base_dn)
+ delete_force(self.ldb_admin, "ou=ext_ou1," + self.base_dn)
+
+ del self.ldb_user1
+ del self.ldb_user2
+ del self.ldb_user3
+
+ def test_ntSecurityDescriptor(self):
+ # create empty ou
+ self.ldb_admin.create_ou("ou=ext_ou1," + self.base_dn)
+ # give u1 Create children access
+ mod = "(A;;CC;;;%s)" % str(self.user_sid1)
+ self.sd_utils.dacl_add_ace("OU=ext_ou1," + self.base_dn, mod)
+ mod = "(A;;LC;;;%s)" % str(self.user_sid2)
+ self.sd_utils.dacl_add_ace("OU=ext_ou1," + self.base_dn, mod)
+ # create a group under that, grant RP to u2
+ self.ldb_user1.newgroup("ext_group1", groupou="OU=ext_ou1",
+ grouptype=samba.dsdb.GTYPE_DISTRIBUTION_DOMAIN_LOCAL_GROUP)
+ mod = "(A;;RP;;;%s)" % str(self.user_sid2)
+ self.sd_utils.dacl_add_ace("CN=ext_group1,OU=ext_ou1," + self.base_dn, mod)
+ # u2 must not read the descriptor
+ res = self.ldb_user2.search("CN=ext_group1,OU=ext_ou1," + self.base_dn,
+ SCOPE_BASE, None, ["nTSecurityDescriptor"])
+ self.assertNotEqual(len(res), 0)
+ self.assertFalse("nTSecurityDescriptor" in res[0].keys())
+ # grant RC to u2 - still no access
+ mod = "(A;;RC;;;%s)" % str(self.user_sid2)
+ self.sd_utils.dacl_add_ace("CN=ext_group1,OU=ext_ou1," + self.base_dn, mod)
+ res = self.ldb_user2.search("CN=ext_group1,OU=ext_ou1," + self.base_dn,
+ SCOPE_BASE, None, ["nTSecurityDescriptor"])
+ self.assertNotEqual(len(res), 0)
+ self.assertFalse("nTSecurityDescriptor" in res[0].keys())
+ # u3 is member of administrators group, should be able to read sd
+ res = self.ldb_user3.search("CN=ext_group1,OU=ext_ou1," + self.base_dn,
+ SCOPE_BASE, None, ["nTSecurityDescriptor"])
+ self.assertEqual(len(res), 1)
+ self.assertTrue("nTSecurityDescriptor" in res[0].keys())
+
+
+class AclUndeleteTests(AclTests):
+
+ def setUp(self):
+ super(AclUndeleteTests, self).setUp()
+ self.regular_user = "undeleter1"
+ self.ou1 = "OU=undeleted_ou,"
+ self.testuser1 = "to_be_undeleted1"
+ self.testuser2 = "to_be_undeleted2"
+ self.testuser3 = "to_be_undeleted3"
+ self.testuser4 = "to_be_undeleted4"
+ self.testuser5 = "to_be_undeleted5"
+ self.testuser6 = "to_be_undeleted6"
+
+ self.new_dn_ou = "CN=" + self.testuser4 + "," + self.ou1 + self.base_dn
+
+ # Create regular user
+ self.testuser1_dn = self.get_user_dn(self.testuser1)
+ self.testuser2_dn = self.get_user_dn(self.testuser2)
+ self.testuser3_dn = self.get_user_dn(self.testuser3)
+ self.testuser4_dn = self.get_user_dn(self.testuser4)
+ self.testuser5_dn = self.get_user_dn(self.testuser5)
+ self.deleted_dn1 = self.create_delete_user(self.testuser1)
+ self.deleted_dn2 = self.create_delete_user(self.testuser2)
+ self.deleted_dn3 = self.create_delete_user(self.testuser3)
+ self.deleted_dn4 = self.create_delete_user(self.testuser4)
+ self.deleted_dn5 = self.create_delete_user(self.testuser5)
+
+ self.ldb_admin.create_ou(self.ou1 + self.base_dn)
+
+ self.ldb_admin.newuser(self.regular_user, self.user_pass)
+ self.ldb_admin.add_remove_group_members("Domain Admins", [self.regular_user],
+ add_members_operation=True)
+ self.ldb_user = self.get_ldb_connection(self.regular_user, self.user_pass)
+ self.sid = self.sd_utils.get_object_sid(self.get_user_dn(self.regular_user))
+
+ def tearDown(self):
+ super(AclUndeleteTests, self).tearDown()
+ delete_force(self.ldb_admin, self.get_user_dn(self.regular_user))
+ delete_force(self.ldb_admin, self.get_user_dn(self.testuser1))
+ delete_force(self.ldb_admin, self.get_user_dn(self.testuser2))
+ delete_force(self.ldb_admin, self.get_user_dn(self.testuser3))
+ delete_force(self.ldb_admin, self.get_user_dn(self.testuser4))
+ delete_force(self.ldb_admin, self.get_user_dn(self.testuser5))
+ delete_force(self.ldb_admin, self.new_dn_ou)
+ delete_force(self.ldb_admin, self.ou1 + self.base_dn)
+
+ del self.ldb_user
+
+ def GUID_string(self, guid):
+ return get_string(ldb.schema_format_value("objectGUID", guid))
+
+ def create_delete_user(self, new_user):
+ self.ldb_admin.newuser(new_user, self.user_pass)
+
+ res = self.ldb_admin.search(expression="(objectClass=*)",
+ base=self.get_user_dn(new_user),
+ scope=SCOPE_BASE,
+ controls=["show_deleted:1"])
+ guid = res[0]["objectGUID"][0]
+ self.ldb_admin.delete(self.get_user_dn(new_user))
+ res = self.ldb_admin.search(base="<GUID=%s>" % self.GUID_string(guid),
+ scope=SCOPE_BASE, controls=["show_deleted:1"])
+ self.assertEqual(len(res), 1)
+ return str(res[0].dn)
+
+ def undelete_deleted(self, olddn, newdn):
+ msg = Message()
+ msg.dn = Dn(self.ldb_user, olddn)
+ msg["isDeleted"] = MessageElement([], FLAG_MOD_DELETE, "isDeleted")
+ msg["distinguishedName"] = MessageElement([newdn], FLAG_MOD_REPLACE, "distinguishedName")
+ res = self.ldb_user.modify(msg, ["show_recycled:1"])
+
+ def undelete_deleted_with_mod(self, olddn, newdn):
+ msg = Message()
+ msg.dn = Dn(ldb, olddn)
+ msg["isDeleted"] = MessageElement([], FLAG_MOD_DELETE, "isDeleted")
+ msg["distinguishedName"] = MessageElement([newdn], FLAG_MOD_REPLACE, "distinguishedName")
+ msg["url"] = MessageElement(["www.samba.org"], FLAG_MOD_REPLACE, "url")
+ res = self.ldb_user.modify(msg, ["show_deleted:1"])
+
+ def test_undelete(self):
+ # it appears the user has to have LC on the old parent to be able to move the object
+ # otherwise we get no such object. Since only System can modify the SD on deleted object
+ # we cannot grant this permission via LDAP, and this leaves us with "negative" tests at the moment
+
+ # deny write property on rdn, should fail
+ mod = "(OD;;WP;bf967a0e-0de6-11d0-a285-00aa003049e2;;%s)" % str(self.sid)
+ self.sd_utils.dacl_add_ace(self.deleted_dn1, mod)
+ try:
+ self.undelete_deleted(self.deleted_dn1, self.testuser1_dn)
+ self.fail()
+ except LdbError as e35:
+ (num, _) = e35.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+
+ # seems that permissions on isDeleted and distinguishedName are irrelevant
+ mod = "(OD;;WP;bf96798f-0de6-11d0-a285-00aa003049e2;;%s)" % str(self.sid)
+ self.sd_utils.dacl_add_ace(self.deleted_dn2, mod)
+ mod = "(OD;;WP;bf9679e4-0de6-11d0-a285-00aa003049e2;;%s)" % str(self.sid)
+ self.sd_utils.dacl_add_ace(self.deleted_dn2, mod)
+ self.undelete_deleted(self.deleted_dn2, self.testuser2_dn)
+
+ # attempt undelete with simultaneous addition of url, WP to which is denied
+ mod = "(OD;;WP;9a9a0221-4a5b-11d1-a9c3-0000f80367c1;;%s)" % str(self.sid)
+ self.sd_utils.dacl_add_ace(self.deleted_dn3, mod)
+ try:
+ self.undelete_deleted_with_mod(self.deleted_dn3, self.testuser3_dn)
+ self.fail()
+ except LdbError as e36:
+ (num, _) = e36.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+
+ # undelete in an ou, in which we have no right to create children
+ mod = "(D;;CC;;;%s)" % str(self.sid)
+ self.sd_utils.dacl_add_ace(self.ou1 + self.base_dn, mod)
+ try:
+ self.undelete_deleted(self.deleted_dn4, self.new_dn_ou)
+ self.fail()
+ except LdbError as e37:
+ (num, _) = e37.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+
+ # delete is not required
+ mod = "(D;;SD;;;%s)" % str(self.sid)
+ self.sd_utils.dacl_add_ace(self.deleted_dn5, mod)
+ self.undelete_deleted(self.deleted_dn5, self.testuser5_dn)
+
+ # deny Reanimate-Tombstone, should fail
+ mod = "(OD;;CR;45ec5156-db7e-47bb-b53f-dbeb2d03c40f;;%s)" % str(self.sid)
+ self.sd_utils.dacl_add_ace(self.base_dn, mod)
+ try:
+ self.undelete_deleted(self.deleted_dn4, self.testuser4_dn)
+ self.fail()
+ except LdbError as e38:
+ (num, _) = e38.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+
+
+class AclSPNTests(AclTests):
+
+ def setUp(self):
+ super(AclSPNTests, self).setUp()
+ self.dcname = "TESTSRV8"
+ self.rodcname = "TESTRODC8"
+ self.computername = "testcomp8"
+ self.test_user = "spn_test_user8"
+ self.computerdn = "CN=%s,CN=computers,%s" % (self.computername, self.base_dn)
+ self.user_object = "user_with_spn"
+ self.user_object_dn = "CN=%s,CN=Users,%s" % (self.user_object, self.base_dn)
+ self.dc_dn = "CN=%s,OU=Domain Controllers,%s" % (self.dcname, self.base_dn)
+ self.site = "Default-First-Site-Name"
+ self.rodcctx = DCJoinContext(server=host, creds=creds, lp=lp,
+ site=self.site, netbios_name=self.rodcname,
+ targetdir=None, domain=None)
+ self.dcctx = DCJoinContext(server=host, creds=creds, lp=lp,
+ site=self.site, netbios_name=self.dcname,
+ targetdir=None, domain=None)
+ self.ldb_admin.newuser(self.test_user, self.user_pass)
+ self.ldb_user1 = self.get_ldb_connection(self.test_user, self.user_pass)
+ self.user_sid1 = self.sd_utils.get_object_sid(self.get_user_dn(self.test_user))
+ self.create_computer(self.computername, self.dcctx.dnsdomain)
+ self.create_rodc(self.rodcctx)
+ self.create_dc(self.dcctx)
+
+ def tearDown(self):
+ super(AclSPNTests, self).tearDown()
+ self.rodcctx.cleanup_old_join()
+ self.dcctx.cleanup_old_join()
+ delete_force(self.ldb_admin, "cn=%s,cn=computers,%s" % (self.computername, self.base_dn))
+ delete_force(self.ldb_admin, self.get_user_dn(self.test_user))
+ delete_force(self.ldb_admin, self.user_object_dn)
+
+ del self.ldb_user1
+
+ def replace_spn(self, _ldb, dn, spn):
+ print("Setting spn %s on %s" % (spn, dn))
+ res = self.ldb_admin.search(dn, expression="(objectClass=*)",
+ scope=SCOPE_BASE, attrs=["servicePrincipalName"])
+ if "servicePrincipalName" in res[0].keys():
+ flag = FLAG_MOD_REPLACE
+ else:
+ flag = FLAG_MOD_ADD
+
+ msg = Message()
+ msg.dn = Dn(self.ldb_admin, dn)
+ msg["servicePrincipalName"] = MessageElement(spn, flag,
+ "servicePrincipalName")
+ _ldb.modify(msg)
+
+ def create_computer(self, computername, domainname):
+ dn = "CN=%s,CN=computers,%s" % (computername, self.base_dn)
+ samaccountname = computername + "$"
+ dnshostname = "%s.%s" % (computername, domainname)
+ self.ldb_admin.add({
+ "dn": dn,
+ "objectclass": "computer",
+ "sAMAccountName": samaccountname,
+ "userAccountControl": str(samba.dsdb.UF_WORKSTATION_TRUST_ACCOUNT),
+ "dNSHostName": dnshostname})
+
+ # same as for join_RODC, but do not set any SPNs
+ def create_rodc(self, ctx):
+ ctx.nc_list = [ctx.base_dn, ctx.config_dn, ctx.schema_dn]
+ ctx.full_nc_list = [ctx.base_dn, ctx.config_dn, ctx.schema_dn]
+ ctx.krbtgt_dn = "CN=krbtgt_%s,CN=Users,%s" % (ctx.myname, ctx.base_dn)
+
+ ctx.never_reveal_sid = ["<SID=%s-%s>" % (ctx.domsid, security.DOMAIN_RID_RODC_DENY),
+ "<SID=%s>" % security.SID_BUILTIN_ADMINISTRATORS,
+ "<SID=%s>" % security.SID_BUILTIN_SERVER_OPERATORS,
+ "<SID=%s>" % security.SID_BUILTIN_BACKUP_OPERATORS,
+ "<SID=%s>" % security.SID_BUILTIN_ACCOUNT_OPERATORS]
+ ctx.reveal_sid = "<SID=%s-%s>" % (ctx.domsid, security.DOMAIN_RID_RODC_ALLOW)
+
+ mysid = ctx.get_mysid()
+ admin_dn = "<SID=%s>" % mysid
+ ctx.managedby = admin_dn
+
+ ctx.userAccountControl = (samba.dsdb.UF_WORKSTATION_TRUST_ACCOUNT |
+ samba.dsdb.UF_TRUSTED_TO_AUTHENTICATE_FOR_DELEGATION |
+ samba.dsdb.UF_PARTIAL_SECRETS_ACCOUNT)
+
+ ctx.connection_dn = "CN=RODC Connection (FRS),%s" % ctx.ntds_dn
+ ctx.secure_channel_type = misc.SEC_CHAN_RODC
+ ctx.RODC = True
+ ctx.replica_flags = (drsuapi.DRSUAPI_DRS_INIT_SYNC |
+ drsuapi.DRSUAPI_DRS_PER_SYNC |
+ drsuapi.DRSUAPI_DRS_GET_ANC |
+ drsuapi.DRSUAPI_DRS_NEVER_SYNCED |
+ drsuapi.DRSUAPI_DRS_SPECIAL_SECRET_PROCESSING)
+
+ ctx.join_add_objects()
+
+ def create_dc(self, ctx):
+ ctx.nc_list = [ctx.base_dn, ctx.config_dn, ctx.schema_dn]
+ ctx.full_nc_list = [ctx.base_dn, ctx.config_dn, ctx.schema_dn]
+ ctx.userAccountControl = samba.dsdb.UF_SERVER_TRUST_ACCOUNT | samba.dsdb.UF_TRUSTED_FOR_DELEGATION
+ ctx.secure_channel_type = misc.SEC_CHAN_BDC
+ ctx.replica_flags = (drsuapi.DRSUAPI_DRS_WRIT_REP |
+ drsuapi.DRSUAPI_DRS_INIT_SYNC |
+ drsuapi.DRSUAPI_DRS_PER_SYNC |
+ drsuapi.DRSUAPI_DRS_FULL_SYNC_IN_PROGRESS |
+ drsuapi.DRSUAPI_DRS_NEVER_SYNCED)
+
+ ctx.join_add_objects()
+
+ def dc_spn_test(self, ctx):
+ netbiosdomain = self.dcctx.get_domain_name()
+ try:
+ self.replace_spn(self.ldb_user1, ctx.acct_dn, "HOST/%s/%s" % (ctx.myname, netbiosdomain))
+ except LdbError as e39:
+ (num, _) = e39.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+ else:
+ self.fail()
+
+ mod = "(OA;;SW;f3a64788-5306-11d1-a9c5-0000f80367c1;;%s)" % str(self.user_sid1)
+ self.sd_utils.dacl_add_ace(ctx.acct_dn, mod)
+ self.replace_spn(self.ldb_user1, ctx.acct_dn, "HOST/%s/%s" % (ctx.myname, netbiosdomain))
+ self.replace_spn(self.ldb_user1, ctx.acct_dn, "HOST/%s" % (ctx.myname))
+ self.replace_spn(self.ldb_user1, ctx.acct_dn, "HOST/%s.%s/%s" %
+ (ctx.myname, ctx.dnsdomain, netbiosdomain))
+ self.replace_spn(self.ldb_user1, ctx.acct_dn, "HOST/%s/%s" % (ctx.myname, ctx.dnsdomain))
+ self.replace_spn(self.ldb_user1, ctx.acct_dn, "HOST/%s.%s/%s" %
+ (ctx.myname, ctx.dnsdomain, ctx.dnsdomain))
+ self.replace_spn(self.ldb_user1, ctx.acct_dn, "GC/%s.%s/%s" %
+ (ctx.myname, ctx.dnsdomain, ctx.dnsforest))
+ self.replace_spn(self.ldb_user1, ctx.acct_dn, "ldap/%s/%s" % (ctx.myname, netbiosdomain))
+ self.replace_spn(self.ldb_user1, ctx.acct_dn, "ldap/%s.%s/%s" %
+ (ctx.myname, ctx.dnsdomain, netbiosdomain))
+ self.replace_spn(self.ldb_user1, ctx.acct_dn, "ldap/%s" % (ctx.myname))
+ self.replace_spn(self.ldb_user1, ctx.acct_dn, "ldap/%s/%s" % (ctx.myname, ctx.dnsdomain))
+ self.replace_spn(self.ldb_user1, ctx.acct_dn, "ldap/%s.%s/%s" %
+ (ctx.myname, ctx.dnsdomain, ctx.dnsdomain))
+ self.replace_spn(self.ldb_user1, ctx.acct_dn, "DNS/%s/%s" % (ctx.myname, ctx.dnsdomain))
+ self.replace_spn(self.ldb_user1, ctx.acct_dn, "RestrictedKrbHost/%s/%s" %
+ (ctx.myname, ctx.dnsdomain))
+ self.replace_spn(self.ldb_user1, ctx.acct_dn, "RestrictedKrbHost/%s" %
+ (ctx.myname))
+ self.replace_spn(self.ldb_user1, ctx.acct_dn, "Dfsr-12F9A27C-BF97-4787-9364-D31B6C55EB04/%s/%s" %
+ (ctx.myname, ctx.dnsdomain))
+ self.replace_spn(self.ldb_user1, ctx.acct_dn, "NtFrs-88f5d2bd-b646-11d2-a6d3-00c04fc9b232/%s/%s" %
+ (ctx.myname, ctx.dnsdomain))
+ self.replace_spn(self.ldb_user1, ctx.acct_dn, "ldap/%s._msdcs.%s" %
+ (ctx.ntds_guid, ctx.dnsdomain))
+
+ # the following spns do not match the restrictions and should fail
+ try:
+ self.replace_spn(self.ldb_user1, ctx.acct_dn, "ldap/%s.%s/ForestDnsZones.%s" %
+ (ctx.myname, ctx.dnsdomain, ctx.dnsdomain))
+ except LdbError as e40:
+ (num, _) = e40.args
+ self.assertEqual(num, ERR_CONSTRAINT_VIOLATION)
+ else:
+ self.fail()
+ try:
+ self.replace_spn(self.ldb_user1, ctx.acct_dn, "ldap/%s.%s/DomainDnsZones.%s" %
+ (ctx.myname, ctx.dnsdomain, ctx.dnsdomain))
+ except LdbError as e41:
+ (num, _) = e41.args
+ self.assertEqual(num, ERR_CONSTRAINT_VIOLATION)
+ else:
+ self.fail()
+ try:
+ self.replace_spn(self.ldb_user1, ctx.acct_dn, "nosuchservice/%s/%s" % ("abcd", "abcd"))
+ except LdbError as e42:
+ (num, _) = e42.args
+ self.assertEqual(num, ERR_CONSTRAINT_VIOLATION)
+ else:
+ self.fail()
+ try:
+ self.replace_spn(self.ldb_user1, ctx.acct_dn, "GC/%s.%s/%s" %
+ (ctx.myname, ctx.dnsdomain, netbiosdomain))
+ except LdbError as e43:
+ (num, _) = e43.args
+ self.assertEqual(num, ERR_CONSTRAINT_VIOLATION)
+ else:
+ self.fail()
+ try:
+ self.replace_spn(self.ldb_user1, ctx.acct_dn, "E3514235-4B06-11D1-AB04-00C04FC2DCD2/%s/%s" %
+ (ctx.ntds_guid, ctx.dnsdomain))
+ except LdbError as e44:
+ (num, _) = e44.args
+ self.assertEqual(num, ERR_CONSTRAINT_VIOLATION)
+ else:
+ self.fail()
+
+ def test_computer_spn(self):
+ # with WP, any value can be set
+ netbiosdomain = self.dcctx.get_domain_name()
+ self.replace_spn(self.ldb_admin, self.computerdn, "HOST/%s/%s" %
+ (self.computername, netbiosdomain))
+ self.replace_spn(self.ldb_admin, self.computerdn, "HOST/%s" % (self.computername))
+ self.replace_spn(self.ldb_admin, self.computerdn, "HOST/%s.%s/%s" %
+ (self.computername, self.dcctx.dnsdomain, netbiosdomain))
+ self.replace_spn(self.ldb_admin, self.computerdn, "HOST/%s/%s" %
+ (self.computername, self.dcctx.dnsdomain))
+ self.replace_spn(self.ldb_admin, self.computerdn, "HOST/%s.%s/%s" %
+ (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsdomain))
+ self.replace_spn(self.ldb_admin, self.computerdn, "GC/%s.%s/%s" %
+ (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsforest))
+ self.replace_spn(self.ldb_admin, self.computerdn, "ldap/%s/%s" % (self.computername, netbiosdomain))
+ self.replace_spn(self.ldb_admin, self.computerdn, "ldap/%s.%s/ForestDnsZones.%s" %
+ (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsdomain))
+ self.replace_spn(self.ldb_admin, self.computerdn, "ldap/%s.%s/DomainDnsZones.%s" %
+ (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsdomain))
+ self.replace_spn(self.ldb_admin, self.computerdn, "ldap/%s.%s/%s" %
+ (self.computername, self.dcctx.dnsdomain, netbiosdomain))
+ self.replace_spn(self.ldb_admin, self.computerdn, "ldap/%s" % (self.computername))
+ self.replace_spn(self.ldb_admin, self.computerdn, "ldap/%s/%s" %
+ (self.computername, self.dcctx.dnsdomain))
+ self.replace_spn(self.ldb_admin, self.computerdn, "ldap/%s.%s/%s" %
+ (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsdomain))
+ self.replace_spn(self.ldb_admin, self.computerdn, "DNS/%s/%s" %
+ (self.computername, self.dcctx.dnsdomain))
+ self.replace_spn(self.ldb_admin, self.computerdn, "RestrictedKrbHost/%s/%s" %
+ (self.computername, self.dcctx.dnsdomain))
+ self.replace_spn(self.ldb_admin, self.computerdn, "RestrictedKrbHost/%s" %
+ (self.computername))
+ self.replace_spn(self.ldb_admin, self.computerdn, "Dfsr-12F9A27C-BF97-4787-9364-D31B6C55EB04/%s/%s" %
+ (self.computername, self.dcctx.dnsdomain))
+ self.replace_spn(self.ldb_admin, self.computerdn, "NtFrs-88f5d2bd-b646-11d2-a6d3-00c04fc9b232/%s/%s" %
+ (self.computername, self.dcctx.dnsdomain))
+ self.replace_spn(self.ldb_admin, self.computerdn, "nosuchservice/%s/%s" % ("abcd", "abcd"))
+
+ # user has neither WP nor Validated-SPN, access denied expected
+ try:
+ self.replace_spn(self.ldb_user1, self.computerdn, "HOST/%s/%s" % (self.computername, netbiosdomain))
+ except LdbError as e45:
+ (num, _) = e45.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+ else:
+ self.fail()
+
+ mod = "(OA;;SW;f3a64788-5306-11d1-a9c5-0000f80367c1;;%s)" % str(self.user_sid1)
+ self.sd_utils.dacl_add_ace(self.computerdn, mod)
+ # grant Validated-SPN and check which values are accepted
+ # see 3.1.1.5.3.1.1.4 servicePrincipalName for reference
+
+ # for regular computer objects we shouldalways get constraint violation
+
+ # This does not pass against Windows, although it should according to docs
+ self.replace_spn(self.ldb_user1, self.computerdn, "HOST/%s" % (self.computername))
+ self.replace_spn(self.ldb_user1, self.computerdn, "HOST/%s.%s" %
+ (self.computername, self.dcctx.dnsdomain))
+
+ try:
+ self.replace_spn(self.ldb_user1, self.computerdn, "HOST/%s/%s" % (self.computername, netbiosdomain))
+ except LdbError as e46:
+ (num, _) = e46.args
+ self.assertEqual(num, ERR_CONSTRAINT_VIOLATION)
+ else:
+ self.fail()
+ try:
+ self.replace_spn(self.ldb_user1, self.computerdn, "HOST/%s.%s/%s" %
+ (self.computername, self.dcctx.dnsdomain, netbiosdomain))
+ except LdbError as e47:
+ (num, _) = e47.args
+ self.assertEqual(num, ERR_CONSTRAINT_VIOLATION)
+ else:
+ self.fail()
+ try:
+ self.replace_spn(self.ldb_user1, self.computerdn, "HOST/%s/%s" %
+ (self.computername, self.dcctx.dnsdomain))
+ except LdbError as e48:
+ (num, _) = e48.args
+ self.assertEqual(num, ERR_CONSTRAINT_VIOLATION)
+ else:
+ self.fail()
+ try:
+ self.replace_spn(self.ldb_user1, self.computerdn, "HOST/%s.%s/%s" %
+ (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsdomain))
+ except LdbError as e49:
+ (num, _) = e49.args
+ self.assertEqual(num, ERR_CONSTRAINT_VIOLATION)
+ else:
+ self.fail()
+ try:
+ self.replace_spn(self.ldb_user1, self.computerdn, "GC/%s.%s/%s" %
+ (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsforest))
+ except LdbError as e50:
+ (num, _) = e50.args
+ self.assertEqual(num, ERR_CONSTRAINT_VIOLATION)
+ else:
+ self.fail()
+ try:
+ self.replace_spn(self.ldb_user1, self.computerdn, "ldap/%s/%s" % (self.computername, netbiosdomain))
+ except LdbError as e51:
+ (num, _) = e51.args
+ self.assertEqual(num, ERR_CONSTRAINT_VIOLATION)
+ else:
+ self.fail()
+ try:
+ self.replace_spn(self.ldb_user1, self.computerdn, "ldap/%s.%s/ForestDnsZones.%s" %
+ (self.computername, self.dcctx.dnsdomain, self.dcctx.dnsdomain))
+ except LdbError as e52:
+ (num, _) = e52.args
+ self.assertEqual(num, ERR_CONSTRAINT_VIOLATION)
+ else:
+ self.fail()
+
+ def test_spn_rwdc(self):
+ self.dc_spn_test(self.dcctx)
+
+ def test_spn_rodc(self):
+ self.dc_spn_test(self.rodcctx)
+
+ def test_user_spn(self):
+ #grant SW to a regular user and try to set the spn on a user object
+ #should get ERR_INSUFFICIENT_ACCESS_RIGHTS, since Validate-SPN only applies to computer
+ self.ldb_admin.newuser(self.user_object, self.user_pass)
+ mod = "(OA;;SW;f3a64788-5306-11d1-a9c5-0000f80367c1;;%s)" % str(self.user_sid1)
+ self.sd_utils.dacl_add_ace(self.user_object_dn, mod)
+ try:
+ self.replace_spn(self.ldb_user1, self.user_object_dn, "nosuchservice/%s/%s" % ("abcd", "abcd"))
+ except LdbError as e60:
+ (num, _) = e60.args
+ self.assertEqual(num, ERR_INSUFFICIENT_ACCESS_RIGHTS)
+ else:
+ self.fail()
+
+ def test_delete_add_spn(self):
+ # Grant Validated-SPN property.
+ mod = f'(OA;;SW;{security.GUID_DRS_VALIDATE_SPN};;{self.user_sid1})'
+ self.sd_utils.dacl_add_ace(self.computerdn, mod)
+
+ spn_base = f'HOST/{self.computername}'
+
+ allowed_spn = f'{spn_base}.{self.dcctx.dnsdomain}'
+ not_allowed_spn = f'{spn_base}/{self.dcctx.get_domain_name()}'
+
+ # Ensure we are able to add an allowed SPN.
+ msg = Message(Dn(self.ldb_user1, self.computerdn))
+ msg['servicePrincipalName'] = MessageElement(allowed_spn,
+ FLAG_MOD_ADD,
+ 'servicePrincipalName')
+ self.ldb_user1.modify(msg)
+
+ # Ensure we are not able to add a disallowed SPN.
+ msg = Message(Dn(self.ldb_user1, self.computerdn))
+ msg['servicePrincipalName'] = MessageElement(not_allowed_spn,
+ FLAG_MOD_ADD,
+ 'servicePrincipalName')
+ try:
+ self.ldb_user1.modify(msg)
+ except LdbError as e:
+ num, _ = e.args
+ self.assertEqual(num, ERR_CONSTRAINT_VIOLATION)
+ else:
+ self.fail(f'able to add disallowed SPN {not_allowed_spn}')
+
+ # Ensure that deleting an existing SPN followed by adding a disallowed
+ # SPN fails.
+ msg = Message(Dn(self.ldb_user1, self.computerdn))
+ msg['0'] = MessageElement([],
+ FLAG_MOD_DELETE,
+ 'servicePrincipalName')
+ msg['1'] = MessageElement(not_allowed_spn,
+ FLAG_MOD_ADD,
+ 'servicePrincipalName')
+ try:
+ self.ldb_user1.modify(msg)
+ except LdbError as e:
+ num, _ = e.args
+ self.assertEqual(num, ERR_CONSTRAINT_VIOLATION)
+ else:
+ self.fail(f'able to add disallowed SPN {not_allowed_spn}')
+
+ def test_delete_disallowed_spn(self):
+ # Grant Validated-SPN property.
+ mod = f'(OA;;SW;{security.GUID_DRS_VALIDATE_SPN};;{self.user_sid1})'
+ self.sd_utils.dacl_add_ace(self.computerdn, mod)
+
+ spn_base = f'HOST/{self.computername}'
+
+ not_allowed_spn = f'{spn_base}/{self.dcctx.get_domain_name()}'
+
+ # Add a disallowed SPN as admin.
+ msg = Message(Dn(self.ldb_admin, self.computerdn))
+ msg['servicePrincipalName'] = MessageElement(not_allowed_spn,
+ FLAG_MOD_ADD,
+ 'servicePrincipalName')
+ self.ldb_admin.modify(msg)
+
+ # Ensure we are able to delete a disallowed SPN.
+ msg = Message(Dn(self.ldb_user1, self.computerdn))
+ msg['servicePrincipalName'] = MessageElement(not_allowed_spn,
+ FLAG_MOD_DELETE,
+ 'servicePrincipalName')
+ try:
+ self.ldb_user1.modify(msg)
+ except LdbError:
+ self.fail(f'unable to delete disallowed SPN {not_allowed_spn}')
+
+
+# tests SEC_ADS_LIST vs. SEC_ADS_LIST_OBJECT
+@DynamicTestCase
+class AclVisibiltyTests(AclTests):
+
+ envs = {
+ "No": False,
+ "Do": True,
+ }
+ modes = {
+ "Allow": False,
+ "Deny": True,
+ }
+ perms = {
+ "nn": 0,
+ "Cn": security.SEC_ADS_LIST,
+ "nO": security.SEC_ADS_LIST_OBJECT,
+ "CO": security.SEC_ADS_LIST | security.SEC_ADS_LIST_OBJECT,
+ }
+
+ @classmethod
+ def setUpDynamicTestCases(cls):
+ for le in cls.envs.keys():
+ for lm in cls.modes.keys():
+ for l1 in cls.perms.keys():
+ for l2 in cls.perms.keys():
+ for l3 in cls.perms.keys():
+ tname = "%s_%s_%s_%s_%s" % (le, lm, l1, l2, l3)
+ ve = cls.envs[le]
+ vm = cls.modes[lm]
+ v1 = cls.perms[l1]
+ v2 = cls.perms[l2]
+ v3 = cls.perms[l3]
+ targs = (tname, ve, vm, v1, v2, v3)
+ cls.generate_dynamic_test("test_visibility",
+ tname, *targs)
+ return
+
+ def setUp(self):
+ super(AclVisibiltyTests, self).setUp()
+
+ strict_checking = samba.tests.env_get_var_value('STRICT_CHECKING', allow_missing=True)
+ if strict_checking is None:
+ strict_checking = '1'
+ self.strict_checking = bool(int(strict_checking))
+
+ # Get the old "dSHeuristics" if it was set
+ self.dsheuristics = self.ldb_admin.get_dsheuristics()
+ # Reset the "dSHeuristics" as they were before
+ self.addCleanup(self.ldb_admin.set_dsheuristics, self.dsheuristics)
+
+ # Domain Admins and SYSTEM get full access
+ self.sddl_dacl = "D:PAI(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;DA)(A;;RPWPCRCCDCLCLORCWOWDSDDTSW;;;SY)"
+ self.set_dacl_control = ["sd_flags:1:%d" % security.SECINFO_DACL]
+
+ self.level_idxs = [ 1, 2, 3, 4 ]
+ self.oul1 = "OU=acl_visibility_oul1"
+ self.oul1_dn_str = "%s,%s" % (self.oul1, self.base_dn)
+ self.oul2 = "OU=oul2,%s" % self.oul1
+ self.oul2_dn_str = "%s,%s" % (self.oul2, self.base_dn)
+ self.oul3 = "OU=oul3,%s" % self.oul2
+ self.oul3_dn_str = "%s,%s" % (self.oul3, self.base_dn)
+ self.user_name = "acl_visibility_user"
+ self.user_dn_str = "CN=%s,%s" % (self.user_name, self.oul3_dn_str)
+ delete_force(self.ldb_admin, self.user_dn_str)
+ delete_force(self.ldb_admin, self.oul3_dn_str)
+ delete_force(self.ldb_admin, self.oul2_dn_str)
+ delete_force(self.ldb_admin, self.oul1_dn_str)
+ self.ldb_admin.create_ou(self.oul1_dn_str)
+ self.sd_utils.modify_sd_on_dn(self.oul1_dn_str,
+ self.sddl_dacl,
+ controls=self.set_dacl_control)
+ self.ldb_admin.create_ou(self.oul2_dn_str)
+ self.sd_utils.modify_sd_on_dn(self.oul2_dn_str,
+ self.sddl_dacl,
+ controls=self.set_dacl_control)
+ self.ldb_admin.create_ou(self.oul3_dn_str)
+ self.sd_utils.modify_sd_on_dn(self.oul3_dn_str,
+ self.sddl_dacl,
+ controls=self.set_dacl_control)
+
+ self.ldb_admin.newuser(self.user_name, self.user_pass, userou=self.oul3)
+ self.user_sid = self.sd_utils.get_object_sid(self.user_dn_str)
+ self.ldb_user = self.get_ldb_connection(self.user_name, self.user_pass)
+
+ def tearDown(self):
+ super(AclVisibiltyTests, self).tearDown()
+ delete_force(self.ldb_admin, self.user_dn_str)
+ delete_force(self.ldb_admin, self.oul3_dn_str)
+ delete_force(self.ldb_admin, self.oul2_dn_str)
+ delete_force(self.ldb_admin, self.oul1_dn_str)
+
+ del self.ldb_user
+
+ def _test_visibility_with_args(self,
+ tname,
+ fDoListObject,
+ modeDeny,
+ l1_allow,
+ l2_allow,
+ l3_allow):
+ l1_deny = 0
+ l2_deny = 0
+ l3_deny = 0
+ if modeDeny:
+ l1_deny = ~l1_allow
+ l2_deny = ~l2_allow
+ l3_deny = ~l3_allow
+ print("Testing: fDoListObject=%s, modeDeny=%s, l1_allow=0x%02x, l2_allow=0x%02x, l3_allow=0x%02x)" % (
+ fDoListObject, modeDeny, l1_allow, l2_allow, l3_allow))
+ if fDoListObject:
+ self.ldb_admin.set_dsheuristics("001")
+ else:
+ self.ldb_admin.set_dsheuristics("000")
+
+ def _generate_dacl(allow, deny):
+ dacl = self.sddl_dacl
+ drights = ""
+ if deny & security.SEC_ADS_LIST:
+ drights += "LC"
+ if deny & security.SEC_ADS_LIST_OBJECT:
+ drights += "LO"
+ if len(drights) > 0:
+ dacl += "(D;;%s;;;%s)" % (drights, self.user_sid)
+ arights = ""
+ if allow & security.SEC_ADS_LIST:
+ arights += "LC"
+ if allow & security.SEC_ADS_LIST_OBJECT:
+ arights += "LO"
+ if len(arights) > 0:
+ dacl += "(A;;%s;;;%s)" % (arights, self.user_sid)
+ print("dacl: %s" % dacl)
+ return dacl
+
+ l1_dacl = _generate_dacl(l1_allow, l1_deny)
+ l2_dacl = _generate_dacl(l2_allow, l2_deny)
+ l3_dacl = _generate_dacl(l3_allow, l3_deny)
+ self.sd_utils.modify_sd_on_dn(self.oul1_dn_str,
+ l1_dacl,
+ controls=self.set_dacl_control)
+ self.sd_utils.modify_sd_on_dn(self.oul2_dn_str,
+ l2_dacl,
+ controls=self.set_dacl_control)
+ self.sd_utils.modify_sd_on_dn(self.oul3_dn_str,
+ l3_dacl,
+ controls=self.set_dacl_control)
+
+ def _generate_levels(_l1_allow,
+ _l1_deny,
+ _l2_allow,
+ _l2_deny,
+ _l3_allow,
+ _l3_deny):
+ _l0_allow = security.SEC_ADS_LIST | security.SEC_ADS_LIST_OBJECT | security.SEC_ADS_READ_PROP
+ _l0_deny = 0
+ _l4_allow = security.SEC_ADS_LIST | security.SEC_ADS_LIST_OBJECT | security.SEC_ADS_READ_PROP
+ _l4_deny = 0
+ _levels = [{
+ "dn": str(self.base_dn),
+ "allow": _l0_allow,
+ "deny": _l0_deny,
+ },{
+ "dn": str(self.oul1_dn_str),
+ "allow": _l1_allow,
+ "deny": _l1_deny,
+ },{
+ "dn": str(self.oul2_dn_str),
+ "allow": _l2_allow,
+ "deny": _l2_deny,
+ },{
+ "dn": str(self.oul3_dn_str),
+ "allow": _l3_allow,
+ "deny": _l3_deny,
+ },{
+ "dn": str(self.user_dn_str),
+ "allow": _l4_allow,
+ "deny": _l4_deny,
+ }]
+ return _levels
+
+ def _generate_admin_levels():
+ _l1_allow = security.SEC_ADS_LIST | security.SEC_ADS_READ_PROP
+ _l1_deny = 0
+ _l2_allow = security.SEC_ADS_LIST | security.SEC_ADS_READ_PROP
+ _l2_deny = 0
+ _l3_allow = security.SEC_ADS_LIST | security.SEC_ADS_READ_PROP
+ _l3_deny = 0
+ return _generate_levels(_l1_allow, _l1_deny,
+ _l2_allow, _l2_deny,
+ _l3_allow, _l3_deny)
+
+ def _generate_user_levels():
+ return _generate_levels(l1_allow, l1_deny,
+ l2_allow, l2_deny,
+ l3_allow, l3_deny)
+
+ admin_levels = _generate_admin_levels()
+ user_levels = _generate_user_levels()
+
+ def _msg_require_name(msg, idx, e):
+ self.assertIn("name", msg)
+ self.assertEqual(len(msg["name"]), 1)
+
+ def _msg_no_name(msg, idx, e):
+ self.assertNotIn("name", msg)
+
+ def _has_right(allow, deny, bit):
+ if allow & bit:
+ if not (deny & bit):
+ return True
+ return False
+
+ def _is_visible(p_allow, p_deny, o_allow, o_deny):
+ plc = _has_right(p_allow, p_deny, security.SEC_ADS_LIST)
+ if plc:
+ return True
+ if not fDoListObject:
+ return False
+ plo = _has_right(p_allow, p_deny, security.SEC_ADS_LIST_OBJECT)
+ if not plo:
+ return False
+ olo = _has_right(o_allow, o_deny, security.SEC_ADS_LIST_OBJECT)
+ if not olo:
+ return False
+ return True
+
+ def _generate_expected(scope, base_level, levels):
+ expected = {}
+
+ p = levels[base_level-1]
+ o = levels[base_level]
+ base_visible = _is_visible(p["allow"], p["deny"],
+ o["allow"], o["deny"])
+
+ if scope == SCOPE_BASE:
+ lmin = base_level
+ lmax = base_level
+ elif scope == SCOPE_ONELEVEL:
+ lmin = base_level+1
+ lmax = base_level+1
+ else:
+ lmin = base_level
+ lmax = len(levels)
+
+ next_idx = 0
+ for li in self.level_idxs:
+ if li < lmin:
+ continue
+ if li > lmax:
+ break
+ p = levels[li-1]
+ o = levels[li]
+ visible = _is_visible(p["allow"], p["deny"],
+ o["allow"], o["deny"])
+ if not visible:
+ continue
+ read = _has_right(o["allow"], o["deny"], security.SEC_ADS_READ_PROP)
+ if read:
+ check_msg_fn = _msg_require_name
+ else:
+ check_msg_fn = _msg_no_name
+ expected[o["dn"]] = {
+ "idx": next_idx,
+ "check_msg_fn": check_msg_fn,
+ }
+ next_idx += 1
+
+ if len(expected) == 0 and not base_visible:
+ # This means we're expecting NO_SUCH_OBJECT
+ return None
+ return expected
+
+ def _verify_result_array(results,
+ description,
+ expected):
+ print("%s Results: %d" % (description, len(results)))
+ for msg in results:
+ print("%s" % msg)
+ self.assertIsNotNone(expected)
+ print("%s Expected: %d" % (description, len(expected)))
+ for e in expected:
+ print("%s" % e)
+ self.assertEqual(len(results), len(expected))
+ idx = 0
+ found = {}
+ for msg in results:
+ dn_str = str(msg.dn)
+ self.assertIn(dn_str, expected)
+ self.assertNotIn(dn_str, found)
+ found[dn_str] = idx
+ e = expected[dn_str]
+ if self.strict_checking:
+ self.assertEqual(idx, int(e["idx"]))
+ if "check_msg_fn" in e:
+ check_msg_fn = e["check_msg_fn"]
+ check_msg_fn(msg, idx, e)
+ idx += 1
+
+ return
+
+ for li in self.level_idxs:
+ base_dn = admin_levels[li]["dn"]
+ for scope in [SCOPE_BASE, SCOPE_ONELEVEL, SCOPE_SUBTREE]:
+ print("\nTesting SCOPE[%d] %s" % (scope, base_dn))
+ admin_expected = _generate_expected(scope, li, admin_levels)
+ admin_res = self.ldb_admin.search(base_dn, scope=scope, attrs=["name"])
+ _verify_result_array(admin_res, "Admin", admin_expected)
+
+ user_expected = _generate_expected(scope, li, user_levels)
+ try:
+ user_res = self.ldb_user.search(base_dn, scope=scope, attrs=["name"])
+ except LdbError as e:
+ (num, _) = e.args
+ if user_expected is None:
+ self.assertEqual(num, ERR_NO_SUCH_OBJECT)
+ print("User: NO_SUCH_OBJECT")
+ continue
+ self.fail(e)
+ _verify_result_array(user_res, "User", user_expected)
+
+# Important unit running information
+
+ldb = SamDB(ldaphost, credentials=creds, session_info=system_session(lp), lp=lp)
+
+TestProgram(module=__name__, opts=subunitopts)