Attachment 18084 Details for Bug 15424 – Patch for master v5

[patch] Patch for master v5

CVE-2023-4154-dirsync-master-v5.patch (text/plain), 52.29 KB, created by Andrew Bartlett on 2023-09-08 03:25:44 UTC

(hide)

Description:

Filename:

MIME Type:

Creator: Andrew Bartlett

Created: 2023-09-08 03:25:44 UTC

Size: 52.29 KB

patch

obsolete

>From 0bbfb5f83794a60efae1805b2b88692e0502f113 Mon Sep 17 00:00:00 2001
>From: Andrew Bartlett <abartlet@samba.org>
>Date: Mon, 7 Aug 2023 11:55:55 +1200
>Subject: [PATCH 1/8] CVE-2023-4154 dsdb/tests: Do not run SimpleDirsyncTests
> twice
>
>To re-use setup code, the super-class must have no test_*() methods
>otherwise these will be run as well as the class-local tests.
>
>We rename tests that would otherwise have duplicate names
>
>BUG: https://bugzilla.samba.org/show_bug.cgi?id=15424
>
>Signed-off-by: Andrew Bartlett <abartlet@samba.org>
>---
> selftest/knownfail                   | 2 +-
> source4/dsdb/tests/python/dirsync.py | 7 ++-----
> 2 files changed, 3 insertions(+), 6 deletions(-)
>
>diff --git a/selftest/knownfail b/selftest/knownfail
>index 37c75d7ca33..4e34effbbd1 100644
>--- a/selftest/knownfail
>+++ b/selftest/knownfail
>@@ -151,7 +151,7 @@
> ^samba4.smb2.acls.*.inheritflags
> ^samba4.smb2.acls.*.owner
> ^samba4.smb2.acls.*.ACCESSBASED
>-^samba4.ldap.dirsync.python.ad_dc_ntvfs..__main__.ExtendedDirsyncTests.test_dirsync_deleted_items
>+^samba4.ldap.dirsync.python.ad_dc_ntvfs..__main__.SimpleDirsyncTests.test_dirsync_deleted_items_OBJECT_SECURITY
> #^samba4.ldap.dirsync.python.ad_dc_ntvfs..__main__.ExtendedDirsyncTests.*
> ^samba4.libsmbclient.opendir.(NT1|SMB3).opendir # This requires netbios browsing
> ^samba4.rpc.drsuapi.*.drsuapi.DsGetDomainControllerInfo$.*$$
>diff --git a/source4/dsdb/tests/python/dirsync.py b/source4/dsdb/tests/python/dirsync.py
>index b2f4eb27ae8..ef2316e37da 100755
>--- a/source4/dsdb/tests/python/dirsync.py
>+++ b/source4/dsdb/tests/python/dirsync.py
>@@ -458,7 +458,7 @@ class SimpleDirsyncTests(DirsyncBaseTests):
>         self.assertTrue(res[0].get("name") is not None)
>         delete_force(self.ldb_admin, ouname)
> 
>-    def test_dirsync_linkedattributes(self):
>+    def test_dirsync_linkedattributes_OBJECT_SECURITY(self):
>         """Check that dirsync returned deleted objects too"""
>         # Let's search for members
>         self.ldb_simple = self.get_ldb_connection(self.simple_user, self.user_pass)
>@@ -585,9 +585,6 @@ class SimpleDirsyncTests(DirsyncBaseTests):
>                                     expression="(&(objectClass=organizationalUnit)(!(isDeleted=*)))",
>                                     controls=controls)
> 
>-
>-class ExtendedDirsyncTests(SimpleDirsyncTests):
>-
>     def test_dirsync_linkedattributes_range(self):
>         self.ldb_simple = self.get_ldb_connection(self.simple_user, self.user_pass)
>         res = self.ldb_admin.search(self.base_dn,
>@@ -711,7 +708,7 @@ class ExtendedDirsyncTests(SimpleDirsyncTests):
>         self.assertIn(b"<GUID=", resEX0[0]["member"][0])
>         self.assertIn(b">;<SID=010500000000000515", resEX0[0]["member"][0])
> 
>-    def test_dirsync_deleted_items(self):
>+    def test_dirsync_deleted_items_OBJECT_SECURITY(self):
>         """Check that dirsync returned deleted objects too"""
>         # Let's create an OU
>         self.ldb_simple = self.get_ldb_connection(self.simple_user, self.user_pass)
>-- 
>2.25.1
>
>
>From 949e80edff8bf1f89998830a3e2683a16fb3db01 Mon Sep 17 00:00:00 2001
>From: Andrew Bartlett <abartlet@samba.org>
>Date: Mon, 7 Aug 2023 13:15:40 +1200
>Subject: [PATCH 2/8] CVE-2023-4154 dsdb/tests: Use self.addCleanup() and
> delete_force()
>
>Thie helps ensure this test is reliable even in spite of errors while
>running.
>
>BUG: https://bugzilla.samba.org/show_bug.cgi?id=15424
>
>Signed-off-by: Andrew Bartlett <abartlet@samba.org>
>---
> source4/dsdb/tests/python/confidential_attr.py | 6 ++----
> source4/dsdb/tests/python/dirsync.py           | 6 +-----
> 2 files changed, 3 insertions(+), 9 deletions(-)
>
>diff --git a/source4/dsdb/tests/python/confidential_attr.py b/source4/dsdb/tests/python/confidential_attr.py
>index 8ca56bd1023..3997848f8f9 100755
>--- a/source4/dsdb/tests/python/confidential_attr.py
>+++ b/source4/dsdb/tests/python/confidential_attr.py
>@@ -98,7 +98,9 @@ class ConfidentialAttrCommon(samba.tests.TestCase):
> 
>         userou = "OU=conf-attr-test"
>         self.ou = "{0},{1}".format(userou, self.base_dn)
>+        samba.tests.delete_force(self.ldb_admin, self.ou, controls=['tree_delete:1'])
>         self.ldb_admin.create_ou(self.ou)
>+        self.addCleanup(samba.tests.delete_force, self.ldb_admin, self.ou, controls=['tree_delete:1'])
> 
>         # use a common username prefix, so we can use sAMAccountName=CATC-* as
>         # a search filter to only return the users we're interested in
>@@ -139,10 +141,6 @@ class ConfidentialAttrCommon(samba.tests.TestCase):
>                          "{0} searchFlags already {1}".format(self.conf_attr,
>                                                               search_flags))
> 
>-    def tearDown(self):
>-        super(ConfidentialAttrCommon, self).tearDown()
>-        self.ldb_admin.delete(self.ou, ["tree_delete:1"])
>-
>     def add_attr(self, dn, attr, value):
>         m = Message()
>         m.dn = Dn(self.ldb_admin, dn)
>diff --git a/source4/dsdb/tests/python/dirsync.py b/source4/dsdb/tests/python/dirsync.py
>index ef2316e37da..0a72393dd0d 100755
>--- a/source4/dsdb/tests/python/dirsync.py
>+++ b/source4/dsdb/tests/python/dirsync.py
>@@ -137,10 +137,6 @@ class SimpleDirsyncTests(DirsyncBaseTests):
>         if self.ouname:
>             delete_force(self.ldb_admin, self.ouname)
>         self.sd_utils.modify_sd_on_dn(self.base_dn, self.desc_sddl)
>-        try:
>-            self.ldb_admin.deletegroup("testgroup")
>-        except Exception:
>-            pass
> 
>     # def test_dirsync_errors(self):
> 
>@@ -499,6 +495,7 @@ class SimpleDirsyncTests(DirsyncBaseTests):
>         self.assertEqual(len(res[0].get("member")), size)
> 
>         self.ldb_admin.newgroup("testgroup")
>+        self.addCleanup(self.ldb_admin.deletegroup, "testgroup")
>         self.ldb_admin.add_remove_group_members("testgroup", [self.simple_user],
>                                                 add_members_operation=True)
> 
>@@ -537,7 +534,6 @@ class SimpleDirsyncTests(DirsyncBaseTests):
>                                     attrs=["member"],
>                                     controls=[control1])
> 
>-        self.ldb_admin.deletegroup("testgroup")
>         self.assertEqual(len(res[0].get("member")), 0)
> 
>     def test_dirsync_deleted_items(self):
>-- 
>2.25.1
>
>
>From ea29462ff5bc13b0f94e26b00f43c4fbc494a60c Mon Sep 17 00:00:00 2001
>From: Andrew Bartlett <abartlet@samba.org>
>Date: Mon, 7 Aug 2023 14:44:28 +1200
>Subject: [PATCH 3/8] CVE-2023-4154 dsdb/tests: Force the test attribute to be
> not-confidential at the start
>
>Rather than fail, if the last run failed to reset things, just force
>the DC into the required state.
>
>BUG: https://bugzilla.samba.org/show_bug.cgi?id=15424
>
>Signed-off-by: Andrew Bartlett <abartlet@samba.org>
>---
> source4/dsdb/tests/python/confidential_attr.py | 10 ++++++----
> 1 file changed, 6 insertions(+), 4 deletions(-)
>
>diff --git a/source4/dsdb/tests/python/confidential_attr.py b/source4/dsdb/tests/python/confidential_attr.py
>index 3997848f8f9..ee7f554a008 100755
>--- a/source4/dsdb/tests/python/confidential_attr.py
>+++ b/source4/dsdb/tests/python/confidential_attr.py
>@@ -136,10 +136,12 @@ class ConfidentialAttrCommon(samba.tests.TestCase):
> 
>         # sanity-check the flag is not already set (this'll cause problems if
>         # previous test run didn't clean up properly)
>-        search_flags = self.get_attr_search_flags(self.attr_dn)
>-        self.assertEqual(0, int(search_flags) & SEARCH_FLAG_CONFIDENTIAL,
>-                         "{0} searchFlags already {1}".format(self.conf_attr,
>-                                                              search_flags))
>+        search_flags = int(self.get_attr_search_flags(self.attr_dn))
>+        if search_flags & SEARCH_FLAG_CONFIDENTIAL:
>+            self.set_attr_search_flags(self.attr_dn, str(search_flags &~ SEARCH_FLAG_CONFIDENTIAL))
>+        search_flags = int(self.get_attr_search_flags(self.attr_dn))
>+        self.assertEqual(0, search_flags & SEARCH_FLAG_CONFIDENTIAL,
>+                         f"{self.conf_attr} searchFlags did not reset to omit SEARCH_FLAG_CONFIDENTIAL ({search_flags})")
> 
>     def add_attr(self, dn, attr, value):
>         m = Message()
>-- 
>2.25.1
>
>
>From 14108b2229305ca1950f5266b019c0164f5047df Mon Sep 17 00:00:00 2001
>From: Andrew Bartlett <abartlet@samba.org>
>Date: Mon, 7 Aug 2023 11:56:56 +1200
>Subject: [PATCH 4/8] CVE-2023-4154 dsdb/tests: Check that secret attributes
> are not visible with DirSync ever.
>
>BUG: https://bugzilla.samba.org/show_bug.cgi?id=15424
>
>Signed-off-by: Andrew Bartlett <abartlet@samba.org>
>---
> selftest/knownfail.d/dirsync         |  1 +
> source4/dsdb/tests/python/dirsync.py | 12 ++++++++++++
> 2 files changed, 13 insertions(+)
> create mode 100644 selftest/knownfail.d/dirsync
>
>diff --git a/selftest/knownfail.d/dirsync b/selftest/knownfail.d/dirsync
>new file mode 100644
>index 00000000000..9367f92e109
>--- /dev/null
>+++ b/selftest/knownfail.d/dirsync
>@@ -0,0 +1 @@
>+^samba4.ldap.dirsync.python$.*$.__main__.SimpleDirsyncTests.test_dirsync_unicodePwd
>\ No newline at end of file
>diff --git a/source4/dsdb/tests/python/dirsync.py b/source4/dsdb/tests/python/dirsync.py
>index 0a72393dd0d..e9f70417c8d 100755
>--- a/source4/dsdb/tests/python/dirsync.py
>+++ b/source4/dsdb/tests/python/dirsync.py
>@@ -742,6 +742,18 @@ class SimpleDirsyncTests(DirsyncBaseTests):
>         self.assertEqual(guid2, guid)
>         self.assertEqual(str(res[0].dn), "")
> 
>+    def test_dirsync_unicodePwd(self):
>+        res = self.ldb_admin.search(self.base_dn,
>+                                    attrs=["unicodePwd", "supplementalCredentials", "samAccountName"],
>+                                    expression="(samAccountName=krbtgt)",
>+                                    controls=["dirsync:1:0:0"])
>+
>+        self.assertTrue(len(res) == 1)
>+        # This form ensures this is a case insensitive comparison
>+        self.assertTrue("samAccountName" in res[0])
>+        self.assertTrue(res[0].get("samAccountName"))
>+        self.assertTrue(res[0].get("unicodePwd") is None)
>+        self.assertTrue(res[0].get("supplementalCredentials") is None)
> 
> if not getattr(opts, "listtests", False):
>     lp = sambaopts.get_loadparm()
>-- 
>2.25.1
>
>
>From b645b94a9cf8d48ebb7497ad0c965716b41d9fb5 Mon Sep 17 00:00:00 2001
>From: Andrew Bartlett <abartlet@samba.org>
>Date: Tue, 8 Aug 2023 11:18:46 +1200
>Subject: [PATCH 5/8] CVE-2023-4154 dsdb/tests: Speed up DirSync test by only
> checking positive matches once
>
>When we (expect to) get back a result, do not waste time against a potentially
>slow server confirming we also get back results for all the other attribute
>combinations.
>
>BUG: https://bugzilla.samba.org/show_bug.cgi?id=15424
>
>Signed-off-by: Andrew Bartlett <abartlet@samba.org>
>---
> source4/dsdb/tests/python/confidential_attr.py | 8 +++++++-
> 1 file changed, 7 insertions(+), 1 deletion(-)
>
>diff --git a/source4/dsdb/tests/python/confidential_attr.py b/source4/dsdb/tests/python/confidential_attr.py
>index ee7f554a008..678a5a82948 100755
>--- a/source4/dsdb/tests/python/confidential_attr.py
>+++ b/source4/dsdb/tests/python/confidential_attr.py
>@@ -742,7 +742,13 @@ class ConfidentialAttrTestDirsync(ConfidentialAttrCommon):
>         #   want to weed out results from any previous test runs
>         search = "(&{0}{1})".format(expr, self.extra_filter)
> 
>-        for attr in self.attr_filters:
>+        # If we expect to return multiple results, only check the first
>+        if expected_num > 0:
>+            attr_filters = [self.attr_filters[0]]
>+        else:
>+            attr_filters = self.attr_filters
>+
>+        for attr in attr_filters:
>             res = samdb.search(base_dn, expression=search, scope=SCOPE_SUBTREE,
>                                attrs=attr, controls=self.dirsync)
>             self.assertEqual(len(res), expected_num,
>-- 
>2.25.1
>
>
>From c330fc1e35d8be3d01721daecbffda7f1414219c Mon Sep 17 00:00:00 2001
>From: Andrew Bartlett <abartlet@samba.org>
>Date: Tue, 8 Aug 2023 14:30:19 +1200
>Subject: [PATCH 6/8] CVE-2023-4154 dsdb/tests: Add test for
> SEARCH_FLAG_RODC_ATTRIBUTE behaviour
>
>SEARCH_FLAG_RODC_ATTRIBUTE should be like SEARCH_FLAG_CONFIDENTIAL,
>but for DirSync and DRS replication.  Accounts with
>GUID_DRS_GET_CHANGES rights should not be able to read this
>attribute.
>
>BUG: https://bugzilla.samba.org/show_bug.cgi?id=15424
>
>Signed-off-by: Andrew Bartlett <abartlet@samba.org>
>---
> .../dsdb/tests/python/confidential_attr.py    | 45 ++++++++++++++++---
> 1 file changed, 40 insertions(+), 5 deletions(-)
>
>diff --git a/source4/dsdb/tests/python/confidential_attr.py b/source4/dsdb/tests/python/confidential_attr.py
>index 678a5a82948..edbc9593204 100755
>--- a/source4/dsdb/tests/python/confidential_attr.py
>+++ b/source4/dsdb/tests/python/confidential_attr.py
>@@ -30,13 +30,15 @@ import time
> from samba.tests.subunitrun import SubunitOptions, TestProgram
> import samba.getopt as options
> from ldb import SCOPE_BASE, SCOPE_SUBTREE
>-from samba.dsdb import SEARCH_FLAG_CONFIDENTIAL, SEARCH_FLAG_PRESERVEONDELETE
>+from samba.dsdb import SEARCH_FLAG_CONFIDENTIAL, SEARCH_FLAG_RODC_ATTRIBUTE, SEARCH_FLAG_PRESERVEONDELETE
> from ldb import Message, MessageElement, Dn
> from ldb import FLAG_MOD_REPLACE, FLAG_MOD_ADD
> 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
>+from samba.dcerpc import security
>+
> import samba.tests
> import samba.dsdb
> 
>@@ -137,11 +139,11 @@ class ConfidentialAttrCommon(samba.tests.TestCase):
>         # sanity-check the flag is not already set (this'll cause problems if
>         # previous test run didn't clean up properly)
>         search_flags = int(self.get_attr_search_flags(self.attr_dn))
>-        if search_flags & SEARCH_FLAG_CONFIDENTIAL:
>-            self.set_attr_search_flags(self.attr_dn, str(search_flags &~ SEARCH_FLAG_CONFIDENTIAL))
>+        if search_flags & SEARCH_FLAG_CONFIDENTIAL|SEARCH_FLAG_RODC_ATTRIBUTE:
>+            self.set_attr_search_flags(self.attr_dn, str(search_flags &~ (SEARCH_FLAG_CONFIDENTIAL|SEARCH_FLAG_RODC_ATTRIBUTE)))
>         search_flags = int(self.get_attr_search_flags(self.attr_dn))
>-        self.assertEqual(0, search_flags & SEARCH_FLAG_CONFIDENTIAL,
>-                         f"{self.conf_attr} searchFlags did not reset to omit SEARCH_FLAG_CONFIDENTIAL ({search_flags})")
>+        self.assertEqual(0, search_flags & (SEARCH_FLAG_CONFIDENTIAL|SEARCH_FLAG_RODC_ATTRIBUTE),
>+                         f"{self.conf_attr} searchFlags did not reset to omit SEARCH_FLAG_CONFIDENTIAL and SEARCH_FLAG_RODC_ATTRIBUTE ({search_flags})")
> 
>     def add_attr(self, dn, attr, value):
>         m = Message()
>@@ -1098,5 +1100,38 @@ class ConfidentialAttrTestDirsync(ConfidentialAttrCommon):
>         user_matching, user_non_matching = time_searches(self.ldb_user)
>         assertRangesOverlap(user_matching, user_non_matching)
> 
>+# Check that using the dirsync controls doesn't reveal confidential
>+# "RODC filtered attribute" values to users with only
>+# GUID_DRS_GET_CHANGES.  The tests is so similar to the Confidential
>+# attribute test we base it on that.
>+class RodcFilteredAttrDirsync(ConfidentialAttrTestDirsync):
>+
>+    def setUp(self):
>+        super().setUp()
>+        self.dirsync = ["dirsync:1:0:1000"]
>+
>+        user_sid = self.sd_utils.get_object_sid(self.get_user_dn(self.user))
>+        mod = "(OA;;CR;%s;;%s)" % (security.GUID_DRS_GET_CHANGES,
>+                                   str(user_sid))
>+        self.sd_utils.dacl_add_ace(self.base_dn, mod)
>+
>+        self.ldb_user = self.get_ldb_connection(self.user, self.user_pass)
>+
>+        self.addCleanup(self.sd_utils.dacl_delete_aces, self.base_dn, mod)
>+
>+    def make_attr_confidential(self):
>+        """Marks the attribute under test as being confidential AND RODC
>+           filtered (which should mean it is not visible with only
>+           GUID_DRS_GET_CHANGES)
>+        """
>+
>+        # work out the original 'searchFlags' value before we overwrite it
>+        old_value = self.get_attr_search_flags(self.attr_dn)
>+
>+        self.set_attr_search_flags(self.attr_dn, str(SEARCH_FLAG_RODC_ATTRIBUTE|SEARCH_FLAG_CONFIDENTIAL))
>+
>+        # reset the value after the test completes
>+        self.addCleanup(self.set_attr_search_flags, self.attr_dn, old_value)
>+
> 
> TestProgram(module=__name__, opts=subunitopts)
>-- 
>2.25.1
>
>
>From c98e7da6e4c9915d165bf5d913838806ed90599f Mon Sep 17 00:00:00 2001
>From: Andrew Bartlett <abartlet@samba.org>
>Date: Tue, 22 Aug 2023 15:08:17 +1200
>Subject: [PATCH 7/8] CVE-2023-4154 dsdb/tests: Extend attribute read DirSync
> tests
>
>The aim here is to document the expected (even if not implemented)
>SEARCH_FLAG_RODC_ATTRIBUTE vs SEARCH_FLAG_CONFIDENTIAL, behaviour, so
>that any change once CVE-2023-4154 is fixed can be noted.
>
>BUG: https://bugzilla.samba.org/show_bug.cgi?id=15424
>
>Signed-off-by: Andrew Bartlett <abartlet@samba.org>
>---
> selftest/knownfail.d/dirsync         |  15 +-
> source4/dsdb/tests/python/dirsync.py | 456 +++++++++++++++++++++++----
> 2 files changed, 414 insertions(+), 57 deletions(-)
>
>diff --git a/selftest/knownfail.d/dirsync b/selftest/knownfail.d/dirsync
>index 9367f92e109..db098549a08 100644
>--- a/selftest/knownfail.d/dirsync
>+++ b/selftest/knownfail.d/dirsync
>@@ -1 +1,14 @@
>-^samba4.ldap.dirsync.python$.*$.__main__.SimpleDirsyncTests.test_dirsync_unicodePwd
>\ No newline at end of file
>+^samba4.ldap.dirsync.python$.*$.__main__.ConfidentialDirsyncTests.test_dirsync_OBJECT_SECURITY_insist_on_empty_element$.*$
>+^samba4.ldap.dirsync.python$.*$.__main__.ConfidentialDirsyncTests.test_dirsync_unicodePwd_OBJ_SEC_insist_on_empty_element$.*$
>+^samba4.ldap.dirsync.python$.*$.__main__.ConfidentialDirsyncTests.test_dirsync_unicodePwd_with_GET_CHANGES$.*$
>+^samba4.ldap.dirsync.python$.*$.__main__.ConfidentialDirsyncTests.test_dirsync_unicodePwd_with_GET_CHANGES_OBJ_SEC_insist_on_empty_element$.*$
>+^samba4.ldap.dirsync.python$.*$.__main__.ConfidentialDirsyncTests.test_dirsync_unicodePwd_with_GET_CHANGES_insist_on_empty_element$.*$
>+^samba4.ldap.dirsync.python$.*$.__main__.ConfidentialDirsyncTests.test_dirsync_with_GET_CHANGES_OBJECT_SECURITY_insist_on_empty_element$.*$
>+^samba4.ldap.dirsync.python$.*$.__main__.ConfidentialFilteredDirsyncTests.test_dirsync_OBJECT_SECURITY_insist_on_empty_element$.*$
>+^samba4.ldap.dirsync.python$.*$.__main__.ConfidentialFilteredDirsyncTests.test_dirsync_OBJECT_SECURITY_with_GET_CHANGES_insist_on_empty_element$.*$
>+^samba4.ldap.dirsync.python$.*$.__main__.ConfidentialFilteredDirsyncTests.test_dirsync_with_GET_CHANGES$.*$
>+^samba4.ldap.dirsync.python$.*$.__main__.ConfidentialFilteredDirsyncTests.test_dirsync_with_GET_CHANGES_attr$.*$
>+^samba4.ldap.dirsync.python$.*$.__main__.ConfidentialFilteredDirsyncTests.test_dirsync_with_GET_CHANGES_insist_on_empty_element$.*$
>+^samba4.ldap.dirsync.python$.*$.__main__.FilteredDirsyncTests.test_dirsync_with_GET_CHANGES$.*$
>+^samba4.ldap.dirsync.python$.*$.__main__.FilteredDirsyncTests.test_dirsync_with_GET_CHANGES_attr$.*$
>+^samba4.ldap.dirsync.python$.*$.__main__.FilteredDirsyncTests.test_dirsync_with_GET_CHANGES_insist_on_empty_element$.*$
>diff --git a/source4/dsdb/tests/python/dirsync.py b/source4/dsdb/tests/python/dirsync.py
>index e9f70417c8d..db3969215a6 100755
>--- a/source4/dsdb/tests/python/dirsync.py
>+++ b/source4/dsdb/tests/python/dirsync.py
>@@ -3,6 +3,7 @@
> # Unit tests for dirsync control
> # Copyright (C) Matthieu Patou <mat@matws.net> 2011
> # Copyright (C) Jelmer Vernooij <jelmer@samba.org> 2014
>+# Copyright (C) Catalyst.Net Ltd
> #
> # This program is free software; you can redistribute it and/or modify
> # it under the terms of the GNU General Public License as published by
>@@ -30,7 +31,8 @@ import base64
> import ldb
> from ldb import LdbError, SCOPE_BASE
> from ldb import Message, MessageElement, Dn
>-from ldb import FLAG_MOD_ADD, FLAG_MOD_DELETE
>+from ldb import FLAG_MOD_ADD, FLAG_MOD_DELETE, FLAG_MOD_REPLACE
>+from samba.dsdb import SEARCH_FLAG_CONFIDENTIAL, SEARCH_FLAG_RODC_ATTRIBUTE
> from samba.dcerpc import security, misc, drsblobs
> from samba.ndr import ndr_unpack, ndr_pack
> 
>@@ -60,7 +62,6 @@ if len(args) < 1:
> host = args.pop()
> if "://" not in host:
>     ldaphost = "ldap://%s" % host
>-    ldapshost = "ldaps://%s" % host
> else:
>     ldaphost = host
>     start = host.rindex("://")
>@@ -77,8 +78,8 @@ creds = credopts.get_credentials(lp)
> class DirsyncBaseTests(samba.tests.TestCase):
> 
>     def setUp(self):
>-        super(DirsyncBaseTests, self).setUp()
>-        self.ldb_admin = SamDB(ldapshost, credentials=creds, session_info=system_session(lp), lp=lp)
>+        super().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 = samba.generate_random_password(12, 16)
>@@ -87,63 +88,60 @@ class DirsyncBaseTests(samba.tests.TestCase):
>         # used for anonymous login
>         print("baseDN: %s" % self.base_dn)
> 
>-    def get_user_dn(self, name):
>-        return "CN=%s,CN=Users,%s" % (name, self.base_dn)
>+        userou = "OU=dirsync-test"
>+        self.ou = f"{userou},{self.base_dn}"
>+        samba.tests.delete_force(self.ldb_admin, self.ou, controls=['tree_delete:1'])
>+        self.ldb_admin.create_ou(self.ou)
>+        self.addCleanup(samba.tests.delete_force, self.ldb_admin, self.ou, controls=['tree_delete:1'])
> 
>-    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
>-
>-
>-# tests on ldap add operations
>-class SimpleDirsyncTests(DirsyncBaseTests):
>-
>-    def setUp(self):
>-        super(SimpleDirsyncTests, self).setUp()
>         # Regular user
>         self.dirsync_user = "test_dirsync_user"
>         self.simple_user = "test_simple_user"
>         self.admin_user = "test_admin_user"
>-        self.ouname = None
>+        self.dirsync_pass = self.user_pass
>+        self.simple_pass = self.user_pass
>+        self.admin_pass = self.user_pass
> 
>-        self.ldb_admin.newuser(self.dirsync_user, self.user_pass)
>-        self.ldb_admin.newuser(self.simple_user, self.user_pass)
>-        self.ldb_admin.newuser(self.admin_user, self.user_pass)
>+        self.ldb_admin.newuser(self.dirsync_user, self.dirsync_pass, userou=userou)
>+        self.ldb_admin.newuser(self.simple_user, self.simple_pass, userou=userou)
>+        self.ldb_admin.newuser(self.admin_user, self.admin_pass, userou=userou)
>         self.desc_sddl = self.sd_utils.get_sd_as_sddl(self.base_dn)
> 
>         user_sid = self.sd_utils.get_object_sid(self.get_user_dn(self.dirsync_user))
>         mod = "(OA;;CR;%s;;%s)" % (security.GUID_DRS_GET_CHANGES,
>                                    str(user_sid))
>         self.sd_utils.dacl_add_ace(self.base_dn, mod)
>+        self.addCleanup(self.sd_utils.dacl_delete_aces, self.base_dn, mod)
> 
>         # add admins to the Domain Admins group
>         self.ldb_admin.add_remove_group_members("Domain Admins", [self.admin_user],
>                                                 add_members_operation=True)
> 
>-    def tearDown(self):
>-        super(SimpleDirsyncTests, self).tearDown()
>-        delete_force(self.ldb_admin, self.get_user_dn(self.dirsync_user))
>-        delete_force(self.ldb_admin, self.get_user_dn(self.simple_user))
>-        delete_force(self.ldb_admin, self.get_user_dn(self.admin_user))
>-        if self.ouname:
>-            delete_force(self.ldb_admin, self.ouname)
>-        self.sd_utils.modify_sd_on_dn(self.base_dn, self.desc_sddl)
>+    def get_user_dn(self, name):
>+        return ldb.Dn(self.ldb_admin, "CN={0},{1}".format(name, self.ou))
>+
>+    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
>+
>+# tests on ldap add operations
>+class SimpleDirsyncTests(DirsyncBaseTests):
> 
>     # def test_dirsync_errors(self):
> 
>     def test_dirsync_supported(self):
>         """Test the basic of the dirsync is supported"""
>         self.ldb_dirsync = self.get_ldb_connection(self.dirsync_user, self.user_pass)
>-        self.ldb_simple = self.get_ldb_connection(self.simple_user, self.user_pass)
>+        self.ldb_simple = self.get_ldb_connection(self.simple_user, self.simple_pass)
>         res = self.ldb_admin.search(self.base_dn, expression="samaccountname=*", controls=["dirsync:1:0:1"])
>         res = self.ldb_dirsync.search(self.base_dn, expression="samaccountname=*", controls=["dirsync:1:0:1"])
>         try:
>@@ -169,8 +167,8 @@ class SimpleDirsyncTests(DirsyncBaseTests):
> 
>     def test_dirsync_errors(self):
>         """Test if dirsync returns the correct LDAP errors in case of pb"""
>-        self.ldb_simple = self.get_ldb_connection(self.simple_user, self.user_pass)
>-        self.ldb_dirsync = self.get_ldb_connection(self.dirsync_user, self.user_pass)
>+        self.ldb_simple = self.get_ldb_connection(self.simple_user, self.simple_pass)
>+        self.ldb_dirsync = self.get_ldb_connection(self.dirsync_user, self.dirsync_pass)
>         try:
>             self.ldb_simple.search(self.base_dn,
>                                    expression="samaccountname=*",
>@@ -292,11 +290,11 @@ class SimpleDirsyncTests(DirsyncBaseTests):
>                                     attrs=["parentGUID"],
>                                     controls=["dirsync:1:0:1"])
>         self.assertEqual(len(res.msgs), 0)
>-        ouname = "OU=testou,%s" % self.base_dn
>+        ouname = "OU=testou,%s" % self.ou
>         self.ouname = ouname
>         self.ldb_admin.create_ou(ouname)
>         delta = Message()
>-        delta.dn = Dn(self.ldb_admin, str(ouname))
>+        delta.dn = Dn(self.ldb_admin, ouname)
>         delta["cn"] = MessageElement("test ou",
>                                      FLAG_MOD_ADD,
>                                      "cn")
>@@ -457,7 +455,7 @@ class SimpleDirsyncTests(DirsyncBaseTests):
>     def test_dirsync_linkedattributes_OBJECT_SECURITY(self):
>         """Check that dirsync returned deleted objects too"""
>         # Let's search for members
>-        self.ldb_simple = self.get_ldb_connection(self.simple_user, self.user_pass)
>+        self.ldb_simple = self.get_ldb_connection(self.simple_user, self.simple_pass)
>         res = self.ldb_simple.search(self.base_dn,
>                                      expression="(name=Administrators)",
>                                      controls=["dirsync:1:1:1"])
>@@ -582,7 +580,7 @@ class SimpleDirsyncTests(DirsyncBaseTests):
>                                     controls=controls)
> 
>     def test_dirsync_linkedattributes_range(self):
>-        self.ldb_simple = self.get_ldb_connection(self.simple_user, self.user_pass)
>+        self.ldb_simple = self.get_ldb_connection(self.simple_user, self.simple_pass)
>         res = self.ldb_admin.search(self.base_dn,
>                                     attrs=["member;range=1-1"],
>                                     expression="(name=Administrators)",
>@@ -594,7 +592,7 @@ class SimpleDirsyncTests(DirsyncBaseTests):
>         self.assertTrue(len(res[0].get("member")) > 0)
> 
>     def test_dirsync_linkedattributes_range_user(self):
>-        self.ldb_simple = self.get_ldb_connection(self.simple_user, self.user_pass)
>+        self.ldb_simple = self.get_ldb_connection(self.simple_user, self.simple_pass)
>         try:
>             res = self.ldb_simple.search(self.base_dn,
>                                          attrs=["member;range=1-1"],
>@@ -608,7 +606,7 @@ class SimpleDirsyncTests(DirsyncBaseTests):
> 
>     def test_dirsync_linkedattributes(self):
>         flag_incr_linked = 2147483648
>-        self.ldb_simple = self.get_ldb_connection(self.simple_user, self.user_pass)
>+        self.ldb_simple = self.get_ldb_connection(self.simple_user, self.simple_pass)
>         res = self.ldb_admin.search(self.base_dn,
>                                     attrs=["member"],
>                                     expression="(name=Administrators)",
>@@ -676,7 +674,7 @@ class SimpleDirsyncTests(DirsyncBaseTests):
>     def test_dirsync_extended_dn(self):
>         """Check that dirsync works together with the extended_dn control"""
>         # Let's search for members
>-        self.ldb_simple = self.get_ldb_connection(self.simple_user, self.user_pass)
>+        self.ldb_simple = self.get_ldb_connection(self.simple_user, self.simple_pass)
>         res = self.ldb_simple.search(self.base_dn,
>                                      expression="(name=Administrators)",
>                                      controls=["dirsync:1:1:1"])
>@@ -707,7 +705,7 @@ class SimpleDirsyncTests(DirsyncBaseTests):
>     def test_dirsync_deleted_items_OBJECT_SECURITY(self):
>         """Check that dirsync returned deleted objects too"""
>         # Let's create an OU
>-        self.ldb_simple = self.get_ldb_connection(self.simple_user, self.user_pass)
>+        self.ldb_simple = self.get_ldb_connection(self.simple_user, self.simple_pass)
>         ouname = "OU=testou3,%s" % self.base_dn
>         self.ouname = ouname
>         self.ldb_admin.create_ou(ouname)
>@@ -742,18 +740,364 @@ class SimpleDirsyncTests(DirsyncBaseTests):
>         self.assertEqual(guid2, guid)
>         self.assertEqual(str(res[0].dn), "")
> 
>-    def test_dirsync_unicodePwd(self):
>+class SpecialDirsyncTests(DirsyncBaseTests):
>+
>+    def setUp(self):
>+        super().setUp()
>+
>+        self.schema_dn = self.ldb_admin.get_schema_basedn()
>+
>+        # the tests work by setting the 'Confidential' or 'RODC Filtered' bit in the searchFlags
>+        # for an existing schema attribute. This only works against Windows if
>+        # the systemFlags does not have FLAG_SCHEMA_BASE_OBJECT set for the
>+        # schema attribute being modified. There are only a few attributes that
>+        # meet this criteria (most of which only apply to 'user' objects)
>+        self.conf_attr = "homePostalAddress"
>+        attr_cn = "CN=Address-Home"
>+        # schemaIdGuid for homePostalAddress (used for ACE tests)
>+        self.attr_dn = f"{attr_cn},{self.schema_dn}"
>+
>+        userou = "OU=conf-attr-test"
>+        self.ou = "{0},{1}".format(userou, self.base_dn)
>+        samba.tests.delete_force(self.ldb_admin, self.ou, controls=['tree_delete:1'])
>+        self.ldb_admin.create_ou(self.ou)
>+        self.addCleanup(samba.tests.delete_force, self.ldb_admin, self.ou, controls=['tree_delete:1'])
>+
>+        # add a test object with this attribute set
>+        self.conf_value = "abcdef"
>+        self.conf_user = "conf-user"
>+        self.ldb_admin.newuser(self.conf_user, self.user_pass, userou=userou)
>+        self.conf_dn = self.get_user_dn(self.conf_user)
>+        self.add_attr(self.conf_dn, self.conf_attr, self.conf_value)
>+
>+        # sanity-check the flag is not already set (this'll cause problems if
>+        # previous test run didn't clean up properly)
>+
>+        search_flags = int(self.get_attr_search_flags(self.attr_dn))
>+        if search_flags & SEARCH_FLAG_CONFIDENTIAL|SEARCH_FLAG_RODC_ATTRIBUTE:
>+            self.set_attr_search_flags(self.attr_dn, str(search_flags &~ (SEARCH_FLAG_CONFIDENTIAL|SEARCH_FLAG_RODC_ATTRIBUTE)))
>+        search_flags = int(self.get_attr_search_flags(self.attr_dn))
>+        self.assertEqual(0, search_flags & (SEARCH_FLAG_CONFIDENTIAL|SEARCH_FLAG_RODC_ATTRIBUTE),
>+                         f"{self.conf_attr} searchFlags did not reset to omit SEARCH_FLAG_CONFIDENTIAL and SEARCH_FLAG_RODC_ATTRIBUTE ({search_flags})")
>+
>+        # work out the original 'searchFlags' value before we overwrite it
>+        old_value = self.get_attr_search_flags(self.attr_dn)
>+
>+        self.set_attr_search_flags(self.attr_dn, str(self.flag_under_test))
>+
>+        # reset the value after the test completes
>+        self.addCleanup(self.set_attr_search_flags, self.attr_dn, old_value)
>+
>+    def add_attr(self, dn, attr, value):
>+        m = Message()
>+        m.dn = dn
>+        m[attr] = MessageElement(value, FLAG_MOD_ADD, attr)
>+        self.ldb_admin.modify(m)
>+
>+    def set_attr_search_flags(self, attr_dn, flags):
>+        """Modifies the searchFlags for an object in the schema"""
>+        m = Message()
>+        m.dn = Dn(self.ldb_admin, attr_dn)
>+        m['searchFlags'] = MessageElement(flags, FLAG_MOD_REPLACE,
>+                                          'searchFlags')
>+        self.ldb_admin.modify(m)
>+
>+        # note we have to update the schema for this change to take effect (on
>+        # Windows, at least)
>+        self.ldb_admin.set_schema_update_now()
>+
>+    def get_attr_search_flags(self, attr_dn):
>+        res = self.ldb_admin.search(attr_dn, scope=SCOPE_BASE,
>+                                    attrs=['searchFlags'])
>+        return res[0]['searchFlags'][0]
>+
>+    def find_under_current_ou(self, res):
>+        for msg in res:
>+            if msg.dn == self.conf_dn:
>+                return msg
>+        self.fail(f"Failed to find object {self.conf_dn} in {len(res)} results")
>+
>+
>+class ConfidentialDirsyncTests(SpecialDirsyncTests):
>+
>+    def setUp(self):
>+        self.flag_under_test = SEARCH_FLAG_CONFIDENTIAL
>+        super().setUp()
>+
>+    def test_unicodePwd_normal(self):
>         res = self.ldb_admin.search(self.base_dn,
>                                     attrs=["unicodePwd", "supplementalCredentials", "samAccountName"],
>-                                    expression="(samAccountName=krbtgt)",
>-                                    controls=["dirsync:1:0:0"])
>+                                    expression=f"(samAccountName={self.conf_user})")
>+
>+        msg = res[0]
>+
>+        self.assertTrue("samAccountName" in msg)
>+        # This form ensures this is a case insensitive comparison
>+        self.assertTrue(msg.get("samAccountName"))
>+        self.assertTrue(msg.get("unicodePwd") is None)
>+        self.assertTrue(msg.get("supplementalCredentials") is None)
>+
>+    def _test_dirsync_unicodePwd(self, ldb_conn, control=None, insist_on_empty_element=False):
>+        res = ldb_conn.search(self.base_dn,
>+                         attrs=["unicodePwd", "supplementalCredentials", "samAccountName"],
>+                         expression=f"(samAccountName={self.conf_user})",
>+                         controls=[control])
>+
>+        msg = self.find_under_current_ou(res)
> 
>-        self.assertTrue(len(res) == 1)
>+        self.assertTrue("samAccountName" in msg)
>         # This form ensures this is a case insensitive comparison
>-        self.assertTrue("samAccountName" in res[0])
>-        self.assertTrue(res[0].get("samAccountName"))
>-        self.assertTrue(res[0].get("unicodePwd") is None)
>-        self.assertTrue(res[0].get("supplementalCredentials") is None)
>+        self.assertTrue(msg.get("samAccountName"))
>+        if insist_on_empty_element:
>+            self.assertTrue(msg.get("unicodePwd") is not None)
>+            self.assertEqual(len(msg.get("unicodePwd")), 0)
>+            self.assertTrue(msg.get("supplementalCredentials") is not None)
>+            self.assertEqual(len(msg.get("supplementalCredentials")), 0)
>+        else:
>+            self.assertTrue(msg.get("unicodePwd") is None
>+                            or len(msg.get("unicodePwd")) == 0)
>+            self.assertTrue(msg.get("supplementalCredentials") is None
>+                            or len(msg.get("supplementalCredentials")) == 0)
>+
>+    def test_dirsync_unicodePwd_OBJ_SEC(self):
>+        ldb_conn = self.get_ldb_connection(self.simple_user, self.simple_pass)
>+        self._test_dirsync_unicodePwd(ldb_conn, control="dirsync:1:1:0")
>+
>+    def test_dirsync_unicodePwd_OBJ_SEC_insist_on_empty_element(self):
>+        ldb_conn = self.get_ldb_connection(self.simple_user, self.simple_pass)
>+        self._test_dirsync_unicodePwd(ldb_conn, control="dirsync:1:1:0", insist_on_empty_element=True)
>+
>+    def test_dirsync_unicodePwd_with_GET_CHANGES_OBJ_SEC(self):
>+        ldb_conn = self.get_ldb_connection(self.dirsync_user, self.dirsync_pass)
>+        self._test_dirsync_unicodePwd(ldb_conn, control="dirsync:1:1:0")
>+
>+    def test_dirsync_unicodePwd_with_GET_CHANGES_OBJ_SEC_insist_on_empty_element(self):
>+        ldb_conn = self.get_ldb_connection(self.dirsync_user, self.dirsync_pass)
>+        self._test_dirsync_unicodePwd(ldb_conn, control="dirsync:1:1:0", insist_on_empty_element=True)
>+
>+    def test_dirsync_unicodePwd_with_GET_CHANGES(self):
>+        ldb_conn = self.get_ldb_connection(self.dirsync_user, self.dirsync_pass)
>+        self._test_dirsync_unicodePwd(ldb_conn, control="dirsync:1:0:0")
>+
>+    def test_dirsync_unicodePwd_with_GET_CHANGES_insist_on_empty_element(self):
>+        ldb_conn = self.get_ldb_connection(self.dirsync_user, self.dirsync_pass)
>+        self._test_dirsync_unicodePwd(ldb_conn, control="dirsync:1:0:0", insist_on_empty_element=True)
>+
>+    def test_normal(self):
>+        ldb_conn = self.get_ldb_connection(self.simple_user, self.simple_pass)
>+        res = ldb_conn.search(self.base_dn,
>+                         attrs=[self.conf_attr, "samAccountName"],
>+                         expression=f"(samAccountName={self.conf_user})")
>+
>+        msg = res[0]
>+        self.assertTrue("samAccountName" in msg)
>+        # This form ensures this is a case insensitive comparison
>+        self.assertTrue(msg.get("samAccountName"))
>+        self.assertTrue(msg.get(self.conf_attr) is None)
>+
>+    def _test_dirsync_OBJECT_SECURITY(self, ldb_conn, insist_on_empty_element=False):
>+        res = ldb_conn.search(self.base_dn,
>+                              attrs=[self.conf_attr, "samAccountName"],
>+                              expression=f"(samAccountName={self.conf_user})",
>+                              controls=["dirsync:1:1:0"])
>+
>+        msg = self.find_under_current_ou(res)
>+        self.assertTrue("samAccountName" in msg)
>+        # This form ensures this is a case insensitive comparison
>+        self.assertTrue(msg.get("samAccountName"))
>+        if insist_on_empty_element:
>+            self.assertTrue(msg.get(self.conf_attr) is not None)
>+            self.assertEqual(len(msg.get(self.conf_attr)), 0)
>+        else:
>+            self.assertTrue(msg.get(self.conf_attr) is None
>+                            or len(msg.get(self.conf_attr)) == 0)
>+
>+    def test_dirsync_OBJECT_SECURITY(self):
>+        ldb_conn = self.get_ldb_connection(self.simple_user, self.simple_pass)
>+        self._test_dirsync_OBJECT_SECURITY(ldb_conn)
>+
>+    def test_dirsync_OBJECT_SECURITY_insist_on_empty_element(self):
>+        ldb_conn = self.get_ldb_connection(self.simple_user, self.simple_pass)
>+        self._test_dirsync_OBJECT_SECURITY(ldb_conn, insist_on_empty_element=True)
>+
>+    def test_dirsync_with_GET_CHANGES(self):
>+        ldb_conn = self.get_ldb_connection(self.dirsync_user, self.dirsync_pass)
>+        res = ldb_conn.search(self.base_dn,
>+                         attrs=[self.conf_attr, "samAccountName"],
>+                         expression=f"(samAccountName={self.conf_user})",
>+                         controls=["dirsync:1:0:0"])
>+
>+        msg = self.find_under_current_ou(res)
>+        # This form ensures this is a case insensitive comparison
>+        self.assertTrue(msg.get("samAccountName"))
>+        self.assertTrue(msg.get(self.conf_attr))
>+        self.assertEqual(len(msg.get(self.conf_attr)), 1)
>+
>+    def test_dirsync_with_GET_CHANGES_OBJECT_SECURITY(self):
>+        ldb_conn = self.get_ldb_connection(self.dirsync_user, self.dirsync_pass)
>+        self._test_dirsync_OBJECT_SECURITY(ldb_conn)
>+
>+    def test_dirsync_with_GET_CHANGES_OBJECT_SECURITY_insist_on_empty_element(self):
>+        ldb_conn = self.get_ldb_connection(self.dirsync_user, self.dirsync_pass)
>+        self._test_dirsync_OBJECT_SECURITY(ldb_conn, insist_on_empty_element=True)
>+
>+class FilteredDirsyncTests(SpecialDirsyncTests):
>+
>+    def setUp(self):
>+        self.flag_under_test = SEARCH_FLAG_RODC_ATTRIBUTE
>+        super().setUp()
>+
>+    def test_attr(self):
>+        ldb_conn = self.get_ldb_connection(self.simple_user, self.simple_pass)
>+        res = ldb_conn.search(self.base_dn,
>+                         attrs=[self.conf_attr, "samAccountName"],
>+                         expression=f"(samAccountName={self.conf_user})")
>+
>+        msg = res[0]
>+        self.assertTrue("samAccountName" in msg)
>+        # This form ensures this is a case insensitive comparison
>+        self.assertTrue(msg.get("samAccountName"))
>+        self.assertTrue(msg.get(self.conf_attr))
>+        self.assertEqual(len(msg.get(self.conf_attr)), 1)
>+
>+    def _test_dirsync_OBJECT_SECURITY(self, ldb_conn):
>+        res = ldb_conn.search(self.base_dn,
>+                         attrs=[self.conf_attr, "samAccountName"],
>+                         expression=f"(samAccountName={self.conf_user})",
>+                         controls=["dirsync:1:1:0"])
>+
>+        msg = self.find_under_current_ou(res)
>+        self.assertTrue("samAccountName" in msg)
>+        # This form ensures this is a case insensitive comparison
>+        self.assertTrue(msg.get("samAccountName"))
>+        self.assertTrue(msg.get(self.conf_attr))
>+        self.assertEqual(len(msg.get(self.conf_attr)), 1)
>+
>+    def test_dirsync_OBJECT_SECURITY(self):
>+        ldb_conn = self.get_ldb_connection(self.simple_user, self.simple_pass)
>+        self._test_dirsync_OBJECT_SECURITY(ldb_conn)
>+
>+    def test_dirsync_OBJECT_SECURITY_with_GET_CHANGES(self):
>+        ldb_conn = self.get_ldb_connection(self.dirsync_user, self.dirsync_pass)
>+        self._test_dirsync_OBJECT_SECURITY(ldb_conn)
>+
>+    def _test_dirsync_with_GET_CHANGES(self, insist_on_empty_element=False):
>+        ldb_conn = self.get_ldb_connection(self.dirsync_user, self.dirsync_pass)
>+        res = ldb_conn.search(self.base_dn,
>+                         expression=f"(samAccountName={self.conf_user})",
>+                         controls=["dirsync:1:0:0"])
>+
>+        msg = self.find_under_current_ou(res)
>+        # This form ensures this is a case insensitive comparison
>+        self.assertTrue(msg.get("samAccountName"))
>+        if insist_on_empty_element:
>+            self.assertTrue(msg.get(self.conf_attr) is not None)
>+            self.assertEqual(len(msg.get(self.conf_attr)), 0)
>+        else:
>+            self.assertTrue(msg.get(self.conf_attr) is None
>+                            or len(msg.get(self.conf_attr)) == 0)
>+
>+    def test_dirsync_with_GET_CHANGES(self):
>+        self._test_dirsync_with_GET_CHANGES()
>+
>+    def test_dirsync_with_GET_CHANGES_insist_on_empty_element(self):
>+        self._test_dirsync_with_GET_CHANGES(insist_on_empty_element=True)
>+
>+    def test_dirsync_with_GET_CHANGES_attr(self):
>+        ldb_conn = self.get_ldb_connection(self.dirsync_user, self.dirsync_pass)
>+        try:
>+            res = ldb_conn.search(self.base_dn,
>+                                  attrs=[self.conf_attr, "samAccountName"],
>+                                  expression=f"(samAccountName={self.conf_user})",
>+                                  controls=["dirsync:1:0:0"])
>+            self.fail("ldb.search() should have failed with LDAP_INSUFFICIENT_ACCESS_RIGHTS")
>+        except ldb.LdbError as e:
>+            (errno, errstr) = e.args
>+            self.assertEqual(errno, ldb.ERR_INSUFFICIENT_ACCESS_RIGHTS)
>+
>+class ConfidentialFilteredDirsyncTests(SpecialDirsyncTests):
>+
>+    def setUp(self):
>+        self.flag_under_test = SEARCH_FLAG_RODC_ATTRIBUTE|SEARCH_FLAG_CONFIDENTIAL
>+        super().setUp()
>+
>+    def test_attr(self):
>+        ldb_conn = self.get_ldb_connection(self.simple_user, self.simple_pass)
>+        res = ldb_conn.search(self.base_dn,
>+                         attrs=["unicodePwd", "supplementalCredentials", "samAccountName"],
>+                         expression=f"(samAccountName={self.conf_user})")
>+
>+        msg = res[0]
>+        self.assertTrue(msg.get("samAccountName"))
>+        self.assertTrue(msg.get(self.conf_attr) is None)
>+
>+    def _test_dirsync_OBJECT_SECURITY(self, ldb_conn, insist_on_empty_element=False):
>+        res = ldb_conn.search(self.base_dn,
>+                              attrs=[self.conf_attr, "samAccountName"],
>+                              expression=f"(samAccountName={self.conf_user})",
>+                              controls=["dirsync:1:1:0"])
>+
>+        msg = self.find_under_current_ou(res)
>+        self.assertTrue("samAccountName" in msg)
>+        # This form ensures this is a case insensitive comparison
>+        self.assertTrue(msg.get("samAccountName"))
>+        if insist_on_empty_element:
>+            self.assertTrue(msg.get(self.conf_attr) is not None)
>+            self.assertEqual(len(msg.get(self.conf_attr)), 0)
>+        else:
>+            self.assertTrue(msg.get(self.conf_attr) is None
>+                            or len(msg.get(self.conf_attr)) == 0)
>+
>+    def test_dirsync_OBJECT_SECURITY(self):
>+        ldb_conn = self.get_ldb_connection(self.simple_user, self.simple_pass)
>+        self._test_dirsync_OBJECT_SECURITY(ldb_conn)
>+
>+    def test_dirsync_OBJECT_SECURITY_insist_on_empty_element(self):
>+        ldb_conn = self.get_ldb_connection(self.simple_user, self.simple_pass)
>+        self._test_dirsync_OBJECT_SECURITY(ldb_conn, insist_on_empty_element=True)
>+
>+    def test_dirsync_OBJECT_SECURITY_with_GET_CHANGES(self):
>+        ldb_conn = self.get_ldb_connection(self.dirsync_user, self.dirsync_pass)
>+        self._test_dirsync_OBJECT_SECURITY(ldb_conn)
>+
>+    def test_dirsync_OBJECT_SECURITY_with_GET_CHANGES_insist_on_empty_element(self):
>+        ldb_conn = self.get_ldb_connection(self.dirsync_user, self.dirsync_pass)
>+        self._test_dirsync_OBJECT_SECURITY(ldb_conn, insist_on_empty_element=True)
>+
>+    def _test_dirsync_with_GET_CHANGES(self, insist_on_empty_element=False):
>+        ldb_conn = self.get_ldb_connection(self.dirsync_user, self.dirsync_pass)
>+        res = ldb_conn.search(self.base_dn,
>+                         expression=f"(samAccountName={self.conf_user})",
>+                         controls=["dirsync:1:0:0"])
>+
>+        msg = self.find_under_current_ou(res)
>+        # This form ensures this is a case insensitive comparison
>+        self.assertTrue(msg.get("samAccountName"))
>+        if insist_on_empty_element:
>+            self.assertTrue(msg.get(self.conf_attr) is not None)
>+            self.assertEqual(len(msg.get(self.conf_attr)), 0)
>+        else:
>+            self.assertTrue(msg.get(self.conf_attr) is None
>+                            or len(msg.get(self.conf_attr)) == 0)
>+
>+    def test_dirsync_with_GET_CHANGES(self):
>+        self._test_dirsync_with_GET_CHANGES()
>+
>+    def test_dirsync_with_GET_CHANGES_insist_on_empty_element(self):
>+        self._test_dirsync_with_GET_CHANGES(insist_on_empty_element=True)
>+
>+    def test_dirsync_with_GET_CHANGES_attr(self):
>+        ldb_conn = self.get_ldb_connection(self.dirsync_user, self.dirsync_pass)
>+        try:
>+            res = ldb_conn.search(self.base_dn,
>+                                  attrs=[self.conf_attr, "samAccountName"],
>+                                  expression=f"(samAccountName={self.conf_user})",
>+                                  controls=["dirsync:1:0:0"])
>+            self.fail("ldb.search() should have failed with LDAP_INSUFFICIENT_ACCESS_RIGHTS")
>+        except ldb.LdbError as e:
>+            (errno, errstr) = e.args
>+            self.assertEqual(errno, ldb.ERR_INSUFFICIENT_ACCESS_RIGHTS)
>+
> 
> if not getattr(opts, "listtests", False):
>     lp = sambaopts.get_loadparm()
>-- 
>2.25.1
>
>
>From 78213380b8f68665f8496a95a0516e2107be659c Mon Sep 17 00:00:00 2001
>From: Andrew Bartlett <abartlet@samba.org>
>Date: Tue, 8 Aug 2023 17:58:27 +1200
>Subject: [PATCH 8/8] CVE-2023-4154: Unimplement the original DirSync behaviour
> without LDAP_DIRSYNC_OBJECT_SECURITY
>
>This makes LDAP_DIRSYNC_OBJECT_SECURITY the only behaviour provided by
>Samba.
>
>Having a second access control system withing the LDAP stack is unsafe
>and this layer is incomplete.
>
>The current system gives all accounts that have been given the
>GUID_DRS_GET_CHANGES extended right SYSTEM access.  Currently in Samba
>this equates to full access to passwords as well as "RODC Filtered
>attributes" (often used with confidential attributes).
>
>Rather than attempting to correctly filter for secrets (passwords) and
>these filtered attributes, as well as preventing search expressions for
>both, we leave this complexity to the acl_read module which has this
>facility already well tested.
>
>The implication is that callers will only see and filter by attribute
>in DirSync that they could without DirSync.
>
>BUG: https://bugzilla.samba.org/show_bug.cgi?id=15424
>
>Signed-off-by: Andrew Bartlett <abartlet@samba.org>
>---
> selftest/knownfail.d/dirsync             |  3 +--
> source4/dsdb/samdb/ldb_modules/dirsync.c | 22 +++++++++++-----------
> 2 files changed, 12 insertions(+), 13 deletions(-)
>
>diff --git a/selftest/knownfail.d/dirsync b/selftest/knownfail.d/dirsync
>index db098549a08..fcf4d469d6e 100644
>--- a/selftest/knownfail.d/dirsync
>+++ b/selftest/knownfail.d/dirsync
>@@ -1,12 +1,11 @@
> ^samba4.ldap.dirsync.python$.*$.__main__.ConfidentialDirsyncTests.test_dirsync_OBJECT_SECURITY_insist_on_empty_element$.*$
> ^samba4.ldap.dirsync.python$.*$.__main__.ConfidentialDirsyncTests.test_dirsync_unicodePwd_OBJ_SEC_insist_on_empty_element$.*$
>-^samba4.ldap.dirsync.python$.*$.__main__.ConfidentialDirsyncTests.test_dirsync_unicodePwd_with_GET_CHANGES$.*$
> ^samba4.ldap.dirsync.python$.*$.__main__.ConfidentialDirsyncTests.test_dirsync_unicodePwd_with_GET_CHANGES_OBJ_SEC_insist_on_empty_element$.*$
> ^samba4.ldap.dirsync.python$.*$.__main__.ConfidentialDirsyncTests.test_dirsync_unicodePwd_with_GET_CHANGES_insist_on_empty_element$.*$
> ^samba4.ldap.dirsync.python$.*$.__main__.ConfidentialDirsyncTests.test_dirsync_with_GET_CHANGES_OBJECT_SECURITY_insist_on_empty_element$.*$
>+^samba4.ldap.dirsync.python$.*$.__main__.ConfidentialDirsyncTests.test_dirsync_with_GET_CHANGES$.*$
> ^samba4.ldap.dirsync.python$.*$.__main__.ConfidentialFilteredDirsyncTests.test_dirsync_OBJECT_SECURITY_insist_on_empty_element$.*$
> ^samba4.ldap.dirsync.python$.*$.__main__.ConfidentialFilteredDirsyncTests.test_dirsync_OBJECT_SECURITY_with_GET_CHANGES_insist_on_empty_element$.*$
>-^samba4.ldap.dirsync.python$.*$.__main__.ConfidentialFilteredDirsyncTests.test_dirsync_with_GET_CHANGES$.*$
> ^samba4.ldap.dirsync.python$.*$.__main__.ConfidentialFilteredDirsyncTests.test_dirsync_with_GET_CHANGES_attr$.*$
> ^samba4.ldap.dirsync.python$.*$.__main__.ConfidentialFilteredDirsyncTests.test_dirsync_with_GET_CHANGES_insist_on_empty_element$.*$
> ^samba4.ldap.dirsync.python$.*$.__main__.FilteredDirsyncTests.test_dirsync_with_GET_CHANGES$.*$
>diff --git a/source4/dsdb/samdb/ldb_modules/dirsync.c b/source4/dsdb/samdb/ldb_modules/dirsync.c
>index bcbcd43b8ac..3d46ae70dee 100644
>--- a/source4/dsdb/samdb/ldb_modules/dirsync.c
>+++ b/source4/dsdb/samdb/ldb_modules/dirsync.c
>@@ -56,7 +56,6 @@ struct dirsync_context {
> 	bool linkIncrVal;
> 	bool localonly;
> 	bool partial;
>-	bool assystem;
> 	int functional_level;
> 	const struct GUID *our_invocation_id;
> 	const struct dsdb_schema *schema;
>@@ -872,10 +871,6 @@ static int dirsync_search_callback(struct ldb_request *req, struct ldb_reply *ar
> 			DSDB_SEARCH_SHOW_DELETED |
> 			DSDB_SEARCH_SHOW_EXTENDED_DN;
> 
>-		if (dsc->assystem) {
>-			flags = flags | DSDB_FLAG_AS_SYSTEM;
>-		}
>-
> 		ret = dsdb_module_search_tree(dsc->module, dsc, &res,
> 					dn, LDB_SCOPE_BASE,
> 					req->op.search.tree,
>@@ -1102,16 +1097,21 @@ static int dirsync_ldb_search(struct ldb_module *module, struct ldb_request *req
> 			return LDB_ERR_OPERATIONS_ERROR;
> 		}
> 		objectclass = dsdb_get_structural_oc_from_msg(schema, acl_res->msgs[0]);
>+
>+		/*
>+		 * While we never use the answer to this for access
>+		 * control (after CVE-2023-4154), we return a
>+		 * different error message depending on if the user
>+		 * was granted GUID_DRS_GET_CHANGES to provide a closer
>+		 * emulation and keep some tests passing.
>+		 *
>+		 * (Samba's ACL logic is not well suited to redacting
>+		 * only the secret and RODC filtered attributes).
>+		 */
> 		ret = acl_check_extended_right(dsc, module, req, objectclass,
> 					       sd, acl_user_token(module),
> 					       GUID_DRS_GET_CHANGES, SEC_ADS_CONTROL_ACCESS, sid);
> 
>-		if (ret == LDB_ERR_INSUFFICIENT_ACCESS_RIGHTS) {
>-			return ret;
>-		}
>-		dsc->assystem = true;
>-		ret = ldb_request_add_control(req, LDB_CONTROL_AS_SYSTEM_OID, false, NULL);
>-
> 		if (ret != LDB_SUCCESS) {
> 			return ret;
> 		}
>-- 
>2.25.1
>

Flags:

metze: review+
abartlet: ci-passed+

Actions: View

Attachments on bug 15424: 18025 | 18026 | 18027 | 18040 | 18075 | 18076 | 18084 | 18085 | 18086 | 18087 | 18088 | 18089 | 18090 | 18091 | 18092 | 18098