From 229af7ceba6722eb8a54c42704d8e7a04f9857da Mon Sep 17 00:00:00 2001
From: Tim Beale <timbeale@catalyst.net.nz>
Date: Thu, 19 Jul 2018 16:03:36 +1200
Subject: [PATCH 01/10] security: Move object-specific access checks into
 separate function

Object-specific access checks refer to a specific section of the
MS-ADTS, and the code closely matches the spec. We need to extend this
logic to properly handle the Control-Access Right (CR), so it makes
sense to split the logic out into its own function.

This patch just moves the code, and should not alter the logic (apart
from ading in the boolean grant_access return variable.

BUG: https://bugzilla.samba.org/show_bug.cgi?id=13434

Signed-off-by: Tim Beale <timbeale@catalyst.net.nz>
---
 libcli/security/access_check.c | 86 +++++++++++++++++++++++++++++-------------
 1 file changed, 59 insertions(+), 27 deletions(-)

diff --git a/libcli/security/access_check.c b/libcli/security/access_check.c
index b4c850b..b4e6244 100644
--- a/libcli/security/access_check.c
+++ b/libcli/security/access_check.c
@@ -375,6 +375,57 @@ static const struct GUID *get_ace_object_type(struct security_ace *ace)
 }
 
 /**
+ * Evaluates access rights specified in a object-specific ACE for an AD object.
+ * This logic corresponds to MS-ADTS 5.1.3.3.3 Checking Object-Specific Access.
+ * @param[in] ace - the ACE being processed
+ * @param[in/out] tree - remaining_access gets updated for the tree
+ * @param[out] grant_access - set to true if the ACE grants sufficient access
+ *                            rights to the object/attribute
+ * @returns NT_STATUS_OK, unless access was denied
+ */
+static NTSTATUS check_object_specific_access(struct security_ace *ace,
+					     struct object_tree *tree,
+					     bool *grant_access)
+{
+	struct object_tree *node = NULL;
+	const struct GUID *type = NULL;
+
+	*grant_access = false;
+
+	/*
+	 * check only in case we have provided a tree,
+	 * the ACE has an object type and that type
+	 * is in the tree
+	 */
+	type = get_ace_object_type(ace);
+
+	if (!tree) {
+		return NT_STATUS_OK;
+	}
+
+	if (!type) {
+		node = tree;
+	} else {
+		if (!(node = get_object_tree_by_GUID(tree, type))) {
+			return NT_STATUS_OK;
+		}
+	}
+
+	if (ace->type == SEC_ACE_TYPE_ACCESS_ALLOWED_OBJECT) {
+		object_tree_modify_access(node, ace->access_mask);
+		if (node->remaining_access == 0) {
+			*grant_access = true;
+			return NT_STATUS_OK;
+		}
+	} else {
+		if (node->remaining_access & ace->access_mask){
+			return NT_STATUS_ACCESS_DENIED;
+		}
+	}
+	return NT_STATUS_OK;
+}
+
+/**
  * @brief Perform directoryservice (DS) related access checks for a given user
  *
  * Perform DS access checks for the user represented by its security_token, on
@@ -405,8 +456,6 @@ NTSTATUS sec_access_check_ds(const struct security_descriptor *sd,
 {
 	uint32_t i;
 	uint32_t bits_remaining;
-	struct object_tree *node;
-	const struct GUID *type;
 	struct dom_sid self_sid;
 
 	dom_sid_parse(SID_NT_SELF, &self_sid);
@@ -456,6 +505,8 @@ NTSTATUS sec_access_check_ds(const struct security_descriptor *sd,
 	for (i=0; bits_remaining && i < sd->dacl->num_aces; i++) {
 		struct dom_sid *trustee;
 		struct security_ace *ace = &sd->dacl->aces[i];
+		NTSTATUS status;
+		bool grant_access = false;
 
 		if (ace->flags & SEC_ACE_FLAG_INHERIT_ONLY) {
 			continue;
@@ -486,34 +537,15 @@ NTSTATUS sec_access_check_ds(const struct security_descriptor *sd,
 			break;
 		case SEC_ACE_TYPE_ACCESS_DENIED_OBJECT:
 		case SEC_ACE_TYPE_ACCESS_ALLOWED_OBJECT:
-			/*
-			 * check only in case we have provided a tree,
-			 * the ACE has an object type and that type
-			 * is in the tree
-			 */
-			type = get_ace_object_type(ace);
-
-			if (!tree) {
-				continue;
-			}
+			status = check_object_specific_access(ace, tree,
+							      &grant_access);
 
-			if (!type) {
-				node = tree;
-			} else {
-				if (!(node = get_object_tree_by_GUID(tree, type))) {
-					continue;
-				}
+			if (!NT_STATUS_IS_OK(status)) {
+				return status;
 			}
 
-			if (ace->type == SEC_ACE_TYPE_ACCESS_ALLOWED_OBJECT) {
-				object_tree_modify_access(node, ace->access_mask);
-				if (node->remaining_access == 0) {
-					return NT_STATUS_OK;
-				}
-			} else {
-				if (node->remaining_access & ace->access_mask){
-					return NT_STATUS_ACCESS_DENIED;
-				}
+			if (grant_access) {
+				return NT_STATUS_OK;
 			}
 			break;
 		default:	/* Other ACE types not handled/supported */
-- 
2.7.4


From 4d37e03eb23673b54c6d7a444050b7e4c9e24b18 Mon Sep 17 00:00:00 2001
From: Tim Beale <timbeale@catalyst.net.nz>
Date: Fri, 20 Jul 2018 13:13:50 +1200
Subject: [PATCH 02/10] security: Add more comments to the object-specific
 access checks

Reading the spec and then reading the code makes sense, but we could
comment the code more so it makes sense on its own.

BUG: https://bugzilla.samba.org/show_bug.cgi?id=13434

Signed-off-by: Tim Beale <timbeale@catalyst.net.nz>
---
 libcli/security/access_check.c | 30 ++++++++++++++++++++++--------
 1 file changed, 22 insertions(+), 8 deletions(-)

diff --git a/libcli/security/access_check.c b/libcli/security/access_check.c
index b4e6244..93eb85d 100644
--- a/libcli/security/access_check.c
+++ b/libcli/security/access_check.c
@@ -392,32 +392,46 @@ static NTSTATUS check_object_specific_access(struct security_ace *ace,
 
 	*grant_access = false;
 
-	/*
-	 * check only in case we have provided a tree,
-	 * the ACE has an object type and that type
-	 * is in the tree
-	 */
-	type = get_ace_object_type(ace);
-
+	/* if no tree was supplied, we can't do object-specific access checks */
 	if (!tree) {
 		return NT_STATUS_OK;
 	}
 
+	/* Get the ObjectType GUID this ACE applies to */
+	type = get_ace_object_type(ace);
+
+	/*
+	 * If the ACE doesn't have a type, then apply it to the whole tree, i.e.
+	 * treat 'OA' ACEs as 'A' and 'OD' as 'D'
+	 */
 	if (!type) {
 		node = tree;
 	} else {
-		if (!(node = get_object_tree_by_GUID(tree, type))) {
+
+		/* skip it if the ACE's ObjectType GUID is not in the tree */
+		node = get_object_tree_by_GUID(tree, type);
+		if (!node) {
 			return NT_STATUS_OK;
 		}
 	}
 
 	if (ace->type == SEC_ACE_TYPE_ACCESS_ALLOWED_OBJECT) {
+
+		/* apply the access rights to this node, and any children */
 		object_tree_modify_access(node, ace->access_mask);
+
+		/*
+		 * Currently all nodes in the tree request the same access mask,
+		 * so we can use any node to check if processing this ACE now
+		 * means the requested access has been granted
+		 */
 		if (node->remaining_access == 0) {
 			*grant_access = true;
 			return NT_STATUS_OK;
 		}
 	} else {
+
+		/* this ACE denies access to the requested object/attribute */
 		if (node->remaining_access & ace->access_mask){
 			return NT_STATUS_ACCESS_DENIED;
 		}
-- 
2.7.4


From c5a7d270e3a6b155b68009b5e44a05002937f148 Mon Sep 17 00:00:00 2001
From: Tim Beale <timbeale@catalyst.net.nz>
Date: Mon, 9 Jul 2018 15:57:59 +1200
Subject: [PATCH 03/10] tests: Add tests for guessing confidential attributes

Adds tests that assert that a confidential attribute cannot be guessed
by an unprivileged user through wildcard DB searches.

The tests basically consist of a set of DB searches/assertions that
get run for:
- basic searches against a confidential attribute
- confidential attributes that get overridden by giving access to the
  user via an ACE (run against a variety of ACEs)
- protecting a non-confidential attribute via an ACL that denies read-
  access (run against a variety of ACEs)
- querying confidential attributes via the dirsync controls

These tests all pass when run against a Windows Dc and all fail against
a Samba DC.

BUG: https://bugzilla.samba.org/show_bug.cgi?id=13434

Signed-off-by: Tim Beale <timbeale@catalyst.net.nz>
---
 selftest/knownfail.d/confidential_attr         |  15 +
 source4/dsdb/tests/python/confidential_attr.py | 568 +++++++++++++++++++++++++
 source4/selftest/tests.py                      |   3 +
 3 files changed, 586 insertions(+)
 create mode 100644 selftest/knownfail.d/confidential_attr
 create mode 100755 source4/dsdb/tests/python/confidential_attr.py

diff --git a/selftest/knownfail.d/confidential_attr b/selftest/knownfail.d/confidential_attr
new file mode 100644
index 0000000..253ea15
--- /dev/null
+++ b/selftest/knownfail.d/confidential_attr
@@ -0,0 +1,15 @@
+samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_basic_search\(ad_dc_ntvfs\)
+samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_search_with_acl_override\(ad_dc_ntvfs\)
+samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_search_with_attr_acl_override\(ad_dc_ntvfs\)
+samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_search_with_blanket_oa_acl\(ad_dc_ntvfs\)
+samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_search_with_blanket_oa_deny_acl\(ad_dc_ntvfs\)
+samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_search_with_deny_acl\(ad_dc_ntvfs\)
+samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_search_with_deny_attr_acl\(ad_dc_ntvfs\)
+samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_search_with_deny_propset_acl\(ad_dc_ntvfs\)
+samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_search_with_dirsync\(ad_dc_ntvfs\)
+samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_search_with_neutral_acl\(ad_dc_ntvfs\)
+samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_search_with_neutral_acl\(ad_dc_ntvfs\)
+samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_search_with_neutral_attr_acl\(ad_dc_ntvfs\)
+samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_search_with_neutral_cr_acl\(ad_dc_ntvfs\)
+samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_search_with_propset_acl_override\(ad_dc_ntvfs\)
+
diff --git a/source4/dsdb/tests/python/confidential_attr.py b/source4/dsdb/tests/python/confidential_attr.py
new file mode 100755
index 0000000..02a4b33
--- /dev/null
+++ b/source4/dsdb/tests/python/confidential_attr.py
@@ -0,0 +1,568 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+#
+# Tests that confidential attributes (or attributes protected by a ACL that
+# denies read access) cannot be guessed through wildcard DB searches.
+#
+# 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
+# the Free Software Foundation; either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+import optparse
+import sys
+sys.path.insert(0, "bin/python")
+
+import samba
+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
+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
+import samba.tests
+from samba.tests import delete_force
+import samba.dsdb
+
+parser = optparse.OptionParser("confidential_attr.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 ConfidentialAttrTest(samba.tests.TestCase):
+
+    def setUp(self):
+        super(ConfidentialAttrTest, self).setUp()
+
+        self.ldb_admin = SamDB(ldaphost, credentials=creds,
+                               session_info=system_session(lp), lp=lp)
+        self.user_pass = "samba123@"
+        self.base_dn = self.ldb_admin.domain_dn()
+        self.schema_dn = self.ldb_admin.get_schema_basedn()
+        self.sd_utils = sd_utils.SDUtils(self.ldb_admin)
+
+        # the tests work by setting the 'Confidential' 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"
+        self.conf_attr_cn = "CN=Address-Home"
+        # schemaIdGuid for homePostalAddress:
+        self.conf_attr_guid = "16775781-47f3-11d1-a9c3-0000f80367c1"
+        self.conf_attr_sec_guid = "77b5b886-944a-11d1-aebd-0000f80367c1"
+
+        # add a test object with this attribute set
+        self.conf_value = "abcdef"
+        self.conf_user = "confidential-user"
+        self.ldb_admin.newuser(self.conf_user, self.user_pass)
+        self.test_dn = self.get_user_dn(self.conf_user)
+        self.add_attr(self.test_dn, self.conf_attr, self.conf_value)
+
+        # add a sneaky user that will try to steal our secrets
+        self.user = "confattr_u1"
+        self.ldb_admin.newuser(self.user, self.user_pass)
+        self.ldb_user = self.get_ldb_connection(self.user, self.user_pass)
+
+        self.dirsync = ["dirsync:1:1:1000"]
+
+    def tearDown(self):
+        super(ConfidentialAttrTest, self).tearDown()
+        delete_force(self.ldb_admin, self.get_user_dn(self.user))
+        delete_force(self.ldb_admin, self.test_dn)
+
+    def add_attr(self, dn, attr, value):
+        m = Message()
+        m.dn = Dn(self.ldb_admin, 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 make_attr_confidential(self):
+        """Marks the attribute under test as being confidential"""
+        attr_dn = "{},{}".format(self.conf_attr_cn, self.schema_dn)
+
+        # work out the original 'searchFlags' value before we overwrite it
+        res = self.ldb_admin.search(attr_dn, scope=SCOPE_BASE,
+                                    attrs=['searchFlags'])
+        old_value = res[0]['searchFlags'][0]
+
+        self.set_attr_search_flags(attr_dn, str(SEARCH_FLAG_CONFIDENTIAL))
+
+        # reset the value after the test completes
+        self.addCleanup(self.set_attr_search_flags, attr_dn, old_value)
+
+    def get_user_dn(self, name):
+        return "CN={},CN=Users,{}".format(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())
+        features = creds_tmp.get_gensec_features() | gensec.FEATURE_SEAL
+        creds_tmp.set_gensec_features(features)
+        creds_tmp.set_kerberos_state(DONT_USE_KERBEROS)
+        ldb_target = SamDB(url=ldaphost, credentials=creds_tmp, lp=lp)
+        return ldb_target
+
+    def assert_search_result(self, expected_num, base_dn, expr,
+                             samdb, controls, attrs):
+        # try varying the scope of the search, just in case that has an impact
+        for scope in [SCOPE_SUBTREE, SCOPE_BASE]:
+            res = samdb.search(base_dn, expression=expr, scope=scope,
+                               attrs=attrs, controls=controls)
+            self.assertTrue(len(res) == expected_num,
+                            "%u results, not %u for search %s, attrs %s" %
+                            (len(res), expected_num, expr, str(attrs)))
+
+    def assert_conf_attr_searches(self, expected_num, base_dn=None,
+                                  samdb=None, controls=None, total_objects=1,
+                                  inverse_search=None):
+        """Check searches against the attribute under test work as expected"""
+
+        if base_dn is None:
+            base_dn = self.test_dn
+
+        if samdb is None:
+            samdb = self.ldb_user
+
+        # try asking for different attributes back: None/all, the confidential
+        # attribute itself, and a random unrelated attribute
+        attr_filters = [None, [self.conf_attr], ['name']]
+
+        first_char = self.conf_value[:1]
+        last_char = self.conf_value[-1:]
+        test_attr = self.conf_attr
+
+        searches = [
+            # search for the attribute using a sub-string wildcard
+            # (which could reveal the attribute's actual value)
+            "({}={}*)".format(test_attr, first_char),
+            "({}=*{})".format(test_attr, last_char),
+
+            # check a full wildcard against the confidential attribute
+            # (which could reveal the attribute's presence/absence)
+            "({}=*)".format(test_attr),
+
+            # sanity-check equality against an exact match on value
+            "({}={})".format(test_attr, self.conf_value),
+
+            # check wildcard in an AND search...
+            "(&(objectclass=*)({}=*))".format(test_attr),
+            "(&({}={}*)(objectclass=*))".format(test_attr, first_char),
+
+            # ...an OR search (against another term that will never match)
+            "(|(objectclass=banana)({}=*))".format(test_attr),
+            "(|({}={}*)(objectclass=banana))".format(test_attr, first_char),
+
+            # '~=' searches don't work against Samba
+            # sanity-check an approx search against an exact match on value
+            # "({}~={})".format(test_attr, self.conf_value),
+
+            # check <=, and >= expressions that would normally find a match
+            "({}>=0)".format(test_attr),
+            "({}<=ZZZZZZZZZZZ)".format(test_attr)]
+
+        for search in searches:
+            for attrs in attr_filters:
+                self.assert_search_result(expected_num, base_dn, search, samdb,
+                                          controls, attrs)
+
+        # Check '!' search conditions. This gets a bit more complicated, as
+        # it may return the inverse number of matching objects
+        self.assert_inverse_searches(expected_num, base_dn, samdb, controls,
+                                     total_objects, inverse_search)
+
+    def assert_inverse_searches(self, expected_num, base_dn, samdb, controls,
+                                total_objects, and_expression):
+        """Asserts users cannot guess attr values by using 'NOT' searches"""
+
+        first_char = self.conf_value[:1]
+        last_char = self.conf_value[-1:]
+
+        # The following are double negative searches (i.e. NOT non-matching-
+        # condition) which will therefore match ALL objects, including the test
+        # object(s). The NOT search filter makes things a bit more complicated.
+        # A windows DC will always act as if the hidden attribute doesn't exist
+        # AT ALL (for an unprivileged user). So, even for a user that lacks
+        # access rights, these queries should return ALL objects.
+        # However, on Samba (for implementation simplicity) it never returns
+        # a matching result for an unprivileged user. The point here is to
+        # check that the search is non-disclosive, i.e. an unprivileged user
+        # either gets consistently told about all objects or no objects, and
+        # therefore cannot guess an attribute's value.
+        matching_searches = [
+            "(!({}={}*))".format(self.conf_attr, chr(ord(first_char) + 1)),
+            "(!({}=*{}))".format(self.conf_attr, chr(ord(last_char) + 1))]
+
+        # work out if this user has sufficient access rights
+        insufficient_rights = True if expected_num == 0 else False
+        expected_inverse_num = total_objects - expected_num
+        first_search = True
+        attr_filters = [None, [self.conf_attr], ['name']]
+
+        # these first searches match ALL objects, so regardless of access
+        # rights, a Windows DC will always return all objects
+        expected_num = total_objects
+
+        for search in matching_searches:
+            # workaround for dirsync - we add an extra search clause to filter
+            # out the thousands of non-matching objects we're not interested in
+            if and_expression:
+                search = "(&{}{})".format(search, and_expression)
+
+            # on the first search, check if we're an unprivileged user talking
+            # to a Samba DC, in which case we're told about NO objects. This is
+            # acceptable, as long as the DC returns the same thing every time
+            if first_search and insufficient_rights:
+                res = samdb.search(base_dn, expression=search,
+                                   scope=SCOPE_SUBTREE, controls=controls)
+                if len(res) == 0:
+                    expected_num = 0
+                    expected_inverse_num = 0
+                first_search = False
+
+            for attrs in attr_filters:
+                self.assert_search_result(expected_num, base_dn, search,
+                                          samdb, controls, attrs)
+
+        # the following searches will not match against the test object(s). So
+        # a user with sufficient rights will see the inverse number of objects.
+        # An unprivileged user should still see all objects (Windows) or no
+        # objects (Samba)
+        non_matching_searches = [
+            "(!({}=*))".format(self.conf_attr),
+            "(!({}={}*))".format(self.conf_attr, first_char),
+            "(!({}=*{}))".format(self.conf_attr, last_char)]
+
+        for search in non_matching_searches:
+            # workaround for dirsync - we add an extra search clause to filter
+            # out the thousands of non-matching objects we're not interested in
+            if and_expression:
+                search = "(&{}{})".format(search, and_expression)
+            for attrs in attr_filters:
+                self.assert_search_result(expected_inverse_num, base_dn,
+                                          search, samdb, controls, attrs)
+
+    def assert_attr_returned(self, expect_attr, samdb, attrs):
+        # does a query that should always return a successful result, and
+        # checks whether the confidential attribute is present
+        res = samdb.search(self.test_dn, expression="(objectClass=*)",
+                           scope=SCOPE_SUBTREE, attrs=attrs)
+        self.assertTrue(len(res) == 1)
+        if expect_attr:
+            self.assertTrue(self.conf_attr in res[0])
+        else:
+            self.assertTrue(self.conf_attr not in res[0])
+
+    def assert_attr_visible(self, expect_attr, samdb=None):
+        if samdb is None:
+            samdb = self.ldb_user
+
+        # sanity-check confidential attribute is/isn't returned as expected
+        # based on the filter attributes we ask for
+        self.assert_attr_returned(expect_attr, samdb, attrs=None)
+        self.assert_attr_returned(expect_attr, samdb, attrs=[self.conf_attr])
+
+        # filtering on a different attribute should never return the conf_attr
+        self.assert_attr_returned(expect_attr=False, samdb=samdb,
+                                  attrs=['name'])
+
+    def assert_attr_visible_to_admin(self):
+        # sanity-check the admin user can always see the confidential attribute
+        self.assert_conf_attr_searches(expected_num=1, samdb=self.ldb_admin)
+        self.assert_attr_visible(expect_attr=True, samdb=self.ldb_admin)
+
+    def test_basic_search(self):
+        """Checks basic confidential attribute search (without any ACEs)"""
+
+        # check we can see a non-confidential attribute in a basic searches
+        self.assert_conf_attr_searches(expected_num=1)
+        self.assert_attr_visible(expect_attr=True)
+
+        # now make the attribute confidential. Repeat the tests and check that
+        # an ordinary user can't see the attribute, or indirectly match on the
+        # attribute via the search expression
+        self.make_attr_confidential()
+
+        self.assert_conf_attr_searches(expected_num=0)
+        self.assert_attr_visible(expect_attr=False)
+
+        # sanity-check we haven't hidden the attribute from the admin as well
+        self.assert_attr_visible_to_admin()
+
+    def _test_search_with_allow_acl(self, allow_ace):
+        """Checks a ACE with 'CR' rights can override a confidential attr"""
+        # make the test attribute confidential and check user can't see it
+        self.make_attr_confidential()
+        self.assert_conf_attr_searches(expected_num=0)
+        self.assert_attr_visible(expect_attr=False)
+
+        # apply the allow ACE to the object under test
+        self.sd_utils.dacl_add_ace(self.test_dn, allow_ace)
+
+        # the user should now be able to see the attribute
+        self.assert_conf_attr_searches(expected_num=1)
+        self.assert_attr_visible(expect_attr=True)
+
+        # sanity-check the admin can still see the attribute
+        self.assert_attr_visible_to_admin()
+
+    def test_search_with_attr_acl_override(self):
+        """Make the confidential attr visible via an OA attr ACE"""
+
+        # set the SEC_ADS_CONTROL_ACCESS bit ('CR') for the user for the
+        # attribute under test, so the user can see it once more
+        user_dn = self.get_user_dn(self.user)
+        user_sid = self.sd_utils.get_object_sid(user_dn)
+        ace = "(OA;;CR;{};;{})".format(self.conf_attr_guid, str(user_sid))
+
+        self._test_search_with_allow_acl(ace)
+
+    def test_search_with_propset_acl_override(self):
+        """Make the confidential attr visible via a Property-set ACE"""
+
+        # set the SEC_ADS_CONTROL_ACCESS bit ('CR') for the user for the
+        # property-set containing the attribute under test (i.e. the
+        # attributeSecurityGuid), so the user can see it once more
+        user_dn = self.get_user_dn(self.user)
+        user_sid = self.sd_utils.get_object_sid(user_dn)
+        ace = "(OA;;CR;{};;{})".format(self.conf_attr_sec_guid, str(user_sid))
+
+        self._test_search_with_allow_acl(ace)
+
+    def test_search_with_acl_override(self):
+        """Make the confidential attr visible via a general 'allow' ACE"""
+
+        # set the allow SEC_ADS_CONTROL_ACCESS bit ('CR') for the user
+        user_dn = self.get_user_dn(self.user)
+        user_sid = self.sd_utils.get_object_sid(user_dn)
+        ace = "(A;;CR;;;{})".format(str(user_sid))
+
+        self._test_search_with_allow_acl(ace)
+
+    def test_search_with_blanket_oa_acl(self):
+        """Make the confidential attr visible via a non-specific OA ACE"""
+
+        # this just checks that an Object Access (OA) ACE without a GUID
+        # specified will work the same as an 'Access' (A) ACE
+        user_dn = self.get_user_dn(self.user)
+        user_sid = self.sd_utils.get_object_sid(user_dn)
+        ace = "(OA;;CR;;;{})".format(str(user_sid))
+
+        self._test_search_with_allow_acl(ace)
+
+    def _test_search_with_neutral_acl(self, neutral_ace):
+        """Checks that a user does NOT gain access via an unrelated ACE"""
+        # make the test attribute confidential and check user can't see it
+        self.make_attr_confidential()
+        self.assert_conf_attr_searches(expected_num=0)
+        self.assert_attr_visible(expect_attr=False)
+
+        # apply the ACE to the object under test
+        self.sd_utils.dacl_add_ace(self.test_dn, neutral_ace)
+
+        # this should make no difference to the user's ability to see the attr
+        self.assert_conf_attr_searches(expected_num=0)
+        self.assert_attr_visible(expect_attr=False)
+
+        # sanity-check the admin can still see the attribute
+        self.assert_attr_visible_to_admin()
+
+    def test_search_with_neutral_acl(self):
+        """Give the user all rights *except* CR for any attributes"""
+
+        # give the user all rights *except* CR and check it makes no difference
+        user_dn = self.get_user_dn(self.user)
+        user_sid = self.sd_utils.get_object_sid(user_dn)
+        ace = "(A;;RPWPCCDCLCLORCWOWDSDDTSW;;;{})".format(str(user_sid))
+        self._test_search_with_neutral_acl(ace)
+
+    def test_search_with_neutral_attr_acl(self):
+        """Give the user all rights *except* CR for the attribute under test"""
+
+        # giving user object-access rights *except* CR should make no difference
+        user_dn = self.get_user_dn(self.user)
+        user_sid = self.sd_utils.get_object_sid(user_dn)
+        rights = "RPWPCCDCLCLORCWOWDSDDTSW"
+        ace = "(OA;;{};{};;{})".format(rights, self.conf_attr_guid,
+                                       str(user_sid))
+        self._test_search_with_neutral_acl(ace)
+
+    def test_search_with_neutral_cr_acl(self):
+        """Give the user CR rights for *another* unrelated attribute"""
+
+        # giving user object-access CR rights to an unrelated attribute
+        user_dn = self.get_user_dn(self.user)
+        user_sid = self.sd_utils.get_object_sid(user_dn)
+        # use the GUID for sAMAccountName here (for no particular reason)
+        unrelated_attr = "3e0abfd0-126a-11d0-a060-00aa006c33ed"
+        ace = "(OA;;CR;{};;{})".format(unrelated_attr, str(user_sid))
+        self._test_search_with_neutral_acl(ace)
+
+    def _test_search_with_deny_acl(self, ace):
+        # check the user can see the attribute initially
+        self.assert_conf_attr_searches(expected_num=1)
+        self.assert_attr_visible(expect_attr=True)
+
+        # add the ACE that denies access to the attr under test
+        self.sd_utils.dacl_add_ace(self.test_dn, ace)
+
+        # the user shouldn't be able to see the attribute anymore
+        self.assert_conf_attr_searches(expected_num=0)
+        self.assert_attr_visible(expect_attr=False)
+
+        # sanity-check we haven't hidden the attribute from the admin as well
+        self.assert_attr_visible_to_admin()
+
+    def test_search_with_deny_attr_acl(self):
+        """Checks a deny ACE works the same way as a confidential attribute"""
+
+        # add an ACE that denies the user Read Property (RP) access to the attr
+        # (which is similar to making the attribute confidential)
+        user_dn = self.get_user_dn(self.user)
+        user_sid = self.sd_utils.get_object_sid(user_dn)
+        ace = "(OD;;RP;{};;{})".format(self.conf_attr_guid, str(user_sid))
+
+        # check the user cannot see the attribute anymore
+        self._test_search_with_deny_acl(ace)
+
+    def test_search_with_deny_acl(self):
+        """Checks a blanket deny ACE denies access to an object's attributes"""
+
+        # add an blanket deny ACE for Read Property (RP) rights
+        user_dn = self.get_user_dn(self.user)
+        user_sid = self.sd_utils.get_object_sid(user_dn)
+        ace = "(D;;RP;;;{})".format(str(user_sid))
+
+        # check the user cannot see the attribute anymore
+        self._test_search_with_deny_acl(ace)
+
+    def test_search_with_deny_propset_acl(self):
+        """Checks a deny ACE on the attribute's Property-Set"""
+
+        # add an blanket deny ACE for Read Property (RP) rights
+        user_dn = self.get_user_dn(self.user)
+        user_sid = self.sd_utils.get_object_sid(user_dn)
+        ace = "(OD;;RP;{};;{})".format(self.conf_attr_sec_guid, str(user_sid))
+
+        # check the user cannot see the attribute anymore
+        self._test_search_with_deny_acl(ace)
+
+    def test_search_with_blanket_oa_deny_acl(self):
+        """Checks a non-specific 'OD' ACE works the same as a 'D' ACE"""
+
+        # this just checks that adding a 'Object Deny' (OD) ACE without
+        # specifying a GUID will work the same way as a 'Deny' (D) ACE
+        user_dn = self.get_user_dn(self.user)
+        user_sid = self.sd_utils.get_object_sid(user_dn)
+        ace = "(OD;;RP;;;{})".format(str(user_sid))
+
+        # check the user cannot see the attribute anymore
+        self._test_search_with_deny_acl(ace)
+
+    def assert_dirsync_attr_returned(self, samdb, expect_attr):
+        """Similar to assert_attr_returned() but adapted for dirsync"""
+
+        # When using dirsync, the base DN we search on needs to be a naming
+        # context. Also, the attrs we request just seem to be ignored
+        expr = "(&(samaccountname={})(!(isDeleted=*)))".format(self.conf_user)
+        res = samdb.search(self.base_dn, expression=expr,
+                           scope=SCOPE_SUBTREE, controls=self.dirsync)
+        self.assertTrue(len(res) == 1)
+
+        if expect_attr:
+            self.assertTrue(self.conf_attr in res[0])
+            self.assertTrue(len(res[0][self.conf_attr]) == 1)
+        else:
+            # Windows returns the attribute if it's present (it returns all
+            # attributes) but suppresses its actual value
+            self.assertTrue(self.conf_attr not in res[0] or
+                            len(res[0][self.conf_attr]) == 0)
+
+    def test_search_with_dirsync(self):
+        """Checks dirsync controls don't reveal confidential attributes"""
+
+        # try searching using the dirsync controls. Note we need to search on
+        # the base DN when using the dirsync controls. This means, we need an
+        # extra filter for the inverse ('!') search, so we don't get thousands
+        # of objects returned
+        and_expr = "(samaccountname={})(!(isDeleted=*))".format(self.conf_user)
+        self.assert_conf_attr_searches(expected_num=1, base_dn=self.base_dn,
+                                       controls=self.dirsync,
+                                       inverse_search=and_expr)
+        self.assert_dirsync_attr_returned(self.ldb_user, expect_attr=True)
+
+        # make the test attribute confidential and check user can't see it,
+        # even if they use the dirsync controls
+        self.make_attr_confidential()
+
+        self.assert_conf_attr_searches(expected_num=0, base_dn=self.base_dn,
+                                       controls=self.dirsync,
+                                       inverse_search=and_expr)
+        self.assert_dirsync_attr_returned(self.ldb_user, expect_attr=False)
+
+        # as a final sanity-check, make sure the admin can still see the attr
+        self.assert_conf_attr_searches(expected_num=1, base_dn=self.base_dn,
+                                       samdb=self.ldb_admin,
+                                       controls=self.dirsync,
+                                       inverse_search=and_expr)
+        self.assert_dirsync_attr_returned(self.ldb_admin, expect_attr=True)
+
+TestProgram(module=__name__, opts=subunitopts)
diff --git a/source4/selftest/tests.py b/source4/selftest/tests.py
index 4b77b7e..82ccf5e 100755
--- a/source4/selftest/tests.py
+++ b/source4/selftest/tests.py
@@ -857,6 +857,9 @@ for env in ["ad_dc_ntvfs", "fl2000dc", "fl2003dc", "fl2008r2dc"]:
         # therefore skip it in that configuration
         plantestsuite_loadlist("samba4.ldap.passwords.python(%s)" % env, env, [python, os.path.join(samba4srcdir, "dsdb/tests/python/passwords.py"), "$SERVER", '-U"$USERNAME%$PASSWORD"', "-W$DOMAIN", '$LOADLIST', '$LISTOPT'])
 
+env = "ad_dc_ntvfs"
+plantestsuite_loadlist("samba4.ldap.confidential_attr.python(%s)" % env, env, [python, os.path.join(samba4srcdir, "dsdb/tests/python/confidential_attr.py"), '$SERVER', '-U"$USERNAME%$PASSWORD"', '--workgroup=$DOMAIN', '$LOADLIST', '$LISTOPT'])
+
 for env in ["ad_dc_ntvfs"]:
     # This test takes a lot of time, so we run it against a minimum of
     # environments, please only add new ones if there's really a
-- 
2.7.4


From 23dd9c5efcfb63c592aa84d208136a71d6dc0406 Mon Sep 17 00:00:00 2001
From: Tim Beale <timbeale@catalyst.net.nz>
Date: Wed, 25 Jul 2018 10:08:34 +1200
Subject: [PATCH 04/10] tests: Add test case for object visibility with limited
 rights

Currently Samba is a bit disclosive with LDB_OP_PRESENT (i.e.
attribute=*) searches compared to Windows.

All the acl.py tests are based on objectClass=* searches, where Windows
will happily tell a user about objects they have List Contents rights,
but not Read Property rights for. However, if you change the attribute
being searched for, suddenly the objects are no longer visible on
Windows (whereas they are on Samba).

This is a problem, because Samba can tell you about which objects have
confidential attributes, which in itself could be disclosive.

This patch adds a acl.py test-case that highlights this behaviour. The
test passes against Windows but fails against Samba.

Signed-off-by: Tim Beale <timbeale@catalyst.net.nz>
---
 selftest/knownfail.d/acl         |  1 +
 source4/dsdb/tests/python/acl.py | 68 ++++++++++++++++++++++++++++++++++++++++
 2 files changed, 69 insertions(+)
 create mode 100644 selftest/knownfail.d/acl

diff --git a/selftest/knownfail.d/acl b/selftest/knownfail.d/acl
new file mode 100644
index 0000000..6772ea1
--- /dev/null
+++ b/selftest/knownfail.d/acl
@@ -0,0 +1 @@
+^samba4.ldap.acl.python.*test_search7
diff --git a/source4/dsdb/tests/python/acl.py b/source4/dsdb/tests/python/acl.py
index c698bf1..1df1c7b 100755
--- a/source4/dsdb/tests/python/acl.py
+++ b/source4/dsdb/tests/python/acl.py
@@ -1010,6 +1010,74 @@ class AclSearchTests(AclTests):
         res_list = list(res[0].keys())
         self.assertEquals(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.assertEquals(len(res), expected_num)
+
+        res_list = [ x["dn"] for x in res if x["dn"] in expected_list ]
+        self.assertEquals(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):
 
-- 
2.7.4


From 2ac0af5b3ae5bdcd18cc2e3b959545319a53db79 Mon Sep 17 00:00:00 2001
From: Tim Beale <timbeale@catalyst.net.nz>
Date: Fri, 20 Jul 2018 13:01:00 +1200
Subject: [PATCH 05/10] security: Fix checking of object-specific
 CONTROL_ACCESS rights

An 'Object Access Allowed' ACE that assigned 'Control Access' (CR)
rights to a specific attribute would not actually grant access.

What was happening was the remaining_access mask for the object_tree
nodes would be Read Property (RP) + Control Access (CR). The ACE mapped
to the schemaIDGUID for a given attribute, which would end up being a
child node in the tree. So the CR bit was cleared for a child node, but
not the rest of the tree. We would then check the user had the RP access
right, which it did. However, the RP right was cleared for another node
in the tree, which still had the CR bit set in its remaining_access
bitmap, so Samba would not grant access.

Generally, the remaining_access only ever has one bit set, which means
this isn't a problem normally. However, in the Control Access case there
are 2 separate bits being checked, i.e. RP + CR.

One option to fix this problem would be to clear the remaining_access
for the tree instead of just the node. However, the Windows spec is
actually pretty clear on this: if the ACE has a CR right present, then
you can stop any further access checks.

BUG: https://bugzilla.samba.org/show_bug.cgi?id=13434

Signed-off-by: Tim Beale <timbeale@catalyst.net.nz>
---
 libcli/security/access_check.c | 10 ++++++++++
 1 file changed, 10 insertions(+)

diff --git a/libcli/security/access_check.c b/libcli/security/access_check.c
index 93eb85d..03a7dca 100644
--- a/libcli/security/access_check.c
+++ b/libcli/security/access_check.c
@@ -429,6 +429,16 @@ static NTSTATUS check_object_specific_access(struct security_ace *ace,
 			*grant_access = true;
 			return NT_STATUS_OK;
 		}
+
+		/*
+		 * As per 5.1.3.3.4 Checking Control Access Right-Based Access,
+		 * if the CONTROL_ACCESS right is present, then we can grant
+		 * access and stop any further access checks
+		 */
+		if (ace->access_mask & SEC_ADS_CONTROL_ACCESS) {
+			*grant_access = true;
+			return NT_STATUS_OK;
+		}
 	} else {
 
 		/* this ACE denies access to the requested object/attribute */
-- 
2.7.4


From 2b7f72c2a24afaec88a7edc5c57f289eb545ee4f Mon Sep 17 00:00:00 2001
From: Tim Beale <timbeale@catalyst.net.nz>
Date: Fri, 20 Jul 2018 13:52:24 +1200
Subject: [PATCH 06/10] acl_read: Split access_mask logic out into helper
 function

So we can re-use the same logic laster for checking the search-ops.

BUG: https://bugzilla.samba.org/show_bug.cgi?id=13434

Signed-off-by: Tim Beale <timbeale@catalyst.net.nz>
---
 source4/dsdb/samdb/ldb_modules/acl_read.c | 54 ++++++++++++++++++++-----------
 1 file changed, 35 insertions(+), 19 deletions(-)

diff --git a/source4/dsdb/samdb/ldb_modules/acl_read.c b/source4/dsdb/samdb/ldb_modules/acl_read.c
index 3c9cf7c..f42b131 100644
--- a/source4/dsdb/samdb/ldb_modules/acl_read.c
+++ b/source4/dsdb/samdb/ldb_modules/acl_read.c
@@ -227,6 +227,40 @@ static int aclread_get_sd_from_ldb_message(struct aclread_context *ac,
 	return LDB_SUCCESS;
 }
 
+/*
+ * Returns the access mask required to read a given attribute
+ */
+static uint32_t get_attr_access_mask(const struct dsdb_attribute *attr,
+				     uint32_t sd_flags)
+{
+
+	uint32_t access_mask = 0;
+	bool is_sd;
+
+	/* nTSecurityDescriptor is a special case */
+	is_sd = (ldb_attr_cmp("nTSecurityDescriptor",
+			      attr->lDAPDisplayName) == 0);
+
+	if (is_sd) {
+		if (sd_flags & (SECINFO_OWNER|SECINFO_GROUP)) {
+			access_mask |= SEC_STD_READ_CONTROL;
+		}
+		if (sd_flags & SECINFO_DACL) {
+			access_mask |= SEC_STD_READ_CONTROL;
+		}
+		if (sd_flags & SECINFO_SACL) {
+			access_mask |= SEC_FLAG_SYSTEM_SECURITY;
+		}
+	} else {
+		access_mask = SEC_ADS_READ_PROP;
+	}
+
+	if (attr->searchFlags & SEARCH_FLAG_CONFIDENTIAL) {
+		access_mask |= SEC_ADS_CONTROL_ACCESS;
+	}
+
+	return access_mask;
+}
 
 static int aclread_callback(struct ldb_request *req, struct ldb_reply *ares)
 {
@@ -342,26 +376,8 @@ static int aclread_callback(struct ldb_request *req, struct ldb_reply *ares)
 				aclread_mark_inaccesslible(&msg->elements[i]);
 				continue;
 			}
-			/* nTSecurityDescriptor is a special case */
-			if (is_sd) {
-				access_mask = 0;
-
-				if (ac->sd_flags & (SECINFO_OWNER|SECINFO_GROUP)) {
-					access_mask |= SEC_STD_READ_CONTROL;
-				}
-				if (ac->sd_flags & SECINFO_DACL) {
-					access_mask |= SEC_STD_READ_CONTROL;
-				}
-				if (ac->sd_flags & SECINFO_SACL) {
-					access_mask |= SEC_FLAG_SYSTEM_SECURITY;
-				}
-			} else {
-				access_mask = SEC_ADS_READ_PROP;
-			}
 
-			if (attr->searchFlags & SEARCH_FLAG_CONFIDENTIAL) {
-				access_mask |= SEC_ADS_CONTROL_ACCESS;
-			}
+			access_mask = get_attr_access_mask(attr, ac->sd_flags);
 
 			if (access_mask == 0) {
 				aclread_mark_inaccesslible(&msg->elements[i]);
-- 
2.7.4


From 9f57cca1da697509df2231be95ddcc9320cda49b Mon Sep 17 00:00:00 2001
From: Tim Beale <timbeale@catalyst.net.nz>
Date: Thu, 26 Jul 2018 12:20:49 +1200
Subject: [PATCH 07/10] acl_read: Small refactor to aclread_callback()

Flip the dirsync check (to avoid a double negative), and use a helper
boolean variable.

BUG: https://bugzilla.samba.org/show_bug.cgi?id=13434

Signed-off-by: Tim Beale <timbeale@catalyst.net.nz>
---
 source4/dsdb/samdb/ldb_modules/acl_read.c | 31 +++++++++++++++++--------------
 1 file changed, 17 insertions(+), 14 deletions(-)

diff --git a/source4/dsdb/samdb/ldb_modules/acl_read.c b/source4/dsdb/samdb/ldb_modules/acl_read.c
index f42b131..17d6492 100644
--- a/source4/dsdb/samdb/ldb_modules/acl_read.c
+++ b/source4/dsdb/samdb/ldb_modules/acl_read.c
@@ -398,18 +398,12 @@ static int aclread_callback(struct ldb_request *req, struct ldb_reply *ares)
 			 * in anycase.
 			 */
 			if (ret == LDB_ERR_INSUFFICIENT_ACCESS_RIGHTS) {
-				if (!ac->indirsync) {
-					/*
-					 * do not return this entry if attribute is
-					 * part of the search filter
-					 */
-					if (dsdb_attr_in_parse_tree(ac->req->op.search.tree,
-								msg->elements[i].name)) {
-						talloc_free(tmp_ctx);
-						return LDB_SUCCESS;
-					}
-					aclread_mark_inaccesslible(&msg->elements[i]);
-				} else {
+				bool in_search_filter;
+
+				in_search_filter = dsdb_attr_in_parse_tree(ac->req->op.search.tree,
+								msg->elements[i].name);
+
+				if (ac->indirsync) {
 					/*
 					 * We are doing dirysnc answers
 					 * and the object shouldn't be returned (normally)
@@ -418,13 +412,22 @@ static int aclread_callback(struct ldb_request *req, struct ldb_reply *ares)
 					 * (remove the object if it is not deleted, or return
 					 * just the objectGUID if it's deleted).
 					 */
-					if (dsdb_attr_in_parse_tree(ac->req->op.search.tree,
-								msg->elements[i].name)) {
+					if (in_search_filter) {
 						ldb_msg_remove_attr(msg, "replPropertyMetaData");
 						break;
 					} else {
 						aclread_mark_inaccesslible(&msg->elements[i]);
 					}
+				} else {
+					/*
+					 * do not return this entry if attribute is
+					 * part of the search filter
+					 */
+					if (in_search_filter) {
+						talloc_free(tmp_ctx);
+						return LDB_SUCCESS;
+					}
+					aclread_mark_inaccesslible(&msg->elements[i]);
 				}
 			} else if (ret != LDB_SUCCESS) {
 				ldb_debug_set(ldb, LDB_DEBUG_FATAL,
-- 
2.7.4


From 0c23a378903afc8e8dede7a5008429a5de85c979 Mon Sep 17 00:00:00 2001
From: Tim Beale <timbeale@catalyst.net.nz>
Date: Mon, 30 Jul 2018 16:00:15 +1200
Subject: [PATCH 08/10] acl_read: Flip the logic in the dirsync check

This better reflects the special case we're making for dirsync, and gets
rid of a 'if-else' clause.

BUG: https://bugzilla.samba.org/show_bug.cgi?id=13434

Signed-off-by: Tim Beale <timbeale@catalyst.net.nz>
---
 source4/dsdb/samdb/ldb_modules/acl_read.c | 17 +++++++----------
 1 file changed, 7 insertions(+), 10 deletions(-)

diff --git a/source4/dsdb/samdb/ldb_modules/acl_read.c b/source4/dsdb/samdb/ldb_modules/acl_read.c
index 17d6492..9607ed0 100644
--- a/source4/dsdb/samdb/ldb_modules/acl_read.c
+++ b/source4/dsdb/samdb/ldb_modules/acl_read.c
@@ -400,10 +400,12 @@ static int aclread_callback(struct ldb_request *req, struct ldb_reply *ares)
 			if (ret == LDB_ERR_INSUFFICIENT_ACCESS_RIGHTS) {
 				bool in_search_filter;
 
+				/* check if attr is part of the search filter */
 				in_search_filter = dsdb_attr_in_parse_tree(ac->req->op.search.tree,
 								msg->elements[i].name);
 
-				if (ac->indirsync) {
+				if (in_search_filter) {
+
 					/*
 					 * We are doing dirysnc answers
 					 * and the object shouldn't be returned (normally)
@@ -412,21 +414,16 @@ static int aclread_callback(struct ldb_request *req, struct ldb_reply *ares)
 					 * (remove the object if it is not deleted, or return
 					 * just the objectGUID if it's deleted).
 					 */
-					if (in_search_filter) {
+					if (ac->indirsync) {
 						ldb_msg_remove_attr(msg, "replPropertyMetaData");
 						break;
 					} else {
-						aclread_mark_inaccesslible(&msg->elements[i]);
-					}
-				} else {
-					/*
-					 * do not return this entry if attribute is
-					 * part of the search filter
-					 */
-					if (in_search_filter) {
+
+						/* do not return this entry */
 						talloc_free(tmp_ctx);
 						return LDB_SUCCESS;
 					}
+				} else {
 					aclread_mark_inaccesslible(&msg->elements[i]);
 				}
 			} else if (ret != LDB_SUCCESS) {
-- 
2.7.4


From 6dd94f0ae49484b4939f11c26b928868fc7f560e Mon Sep 17 00:00:00 2001
From: Tim Beale <timbeale@catalyst.net.nz>
Date: Fri, 20 Jul 2018 15:42:36 +1200
Subject: [PATCH 09/10] acl_read: Fix unauthorized attribute access via
 searches

A user that doesn't have access to view an attribute can still guess the
attribute's value via repeated LDAP searches. This affects confidential
attributes, as well as ACLs applied to an object/attribute to deny
access.

Currently the code will hide objects if the attribute filter contains an
attribute they are not authorized to see. However, the code still
returns objects as results if confidential attribute is in the search
expression itself, but not in the attribute filter.

To fix this problem we have to check the access rights on the attributes
in the search-tree, as well as the attributes returned in the message.

Points of note:
- I've preserved the existing dirsync logic (the dirsync module code
  suppresses the result as long as the replPropertyMetaData attribute is
  removed). However, there doesn't appear to be any test that highlights
  that this functionality is required for dirsync.
- To avoid this fix breaking the acl.py tests, we need to still permit
  searches like 'objectClass=*', even though we don't have Read Property
  access rights for the objectClass attribute. The logic that Windows
  uses does not appear to be clearly documented, so I've made a best
  guess that seems to mirror Windows behaviour.

BUG: https://bugzilla.samba.org/show_bug.cgi?id=13434

Signed-off-by: Tim Beale <timbeale@catalyst.net.nz>
---
 selftest/knownfail.d/acl                  |   1 -
 selftest/knownfail.d/confidential_attr    |  15 --
 source4/dsdb/samdb/ldb_modules/acl_read.c | 246 ++++++++++++++++++++++++++++++
 3 files changed, 246 insertions(+), 16 deletions(-)
 delete mode 100644 selftest/knownfail.d/acl
 delete mode 100644 selftest/knownfail.d/confidential_attr

diff --git a/selftest/knownfail.d/acl b/selftest/knownfail.d/acl
deleted file mode 100644
index 6772ea1..0000000
--- a/selftest/knownfail.d/acl
+++ /dev/null
@@ -1 +0,0 @@
-^samba4.ldap.acl.python.*test_search7
diff --git a/selftest/knownfail.d/confidential_attr b/selftest/knownfail.d/confidential_attr
deleted file mode 100644
index 253ea15..0000000
--- a/selftest/knownfail.d/confidential_attr
+++ /dev/null
@@ -1,15 +0,0 @@
-samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_basic_search\(ad_dc_ntvfs\)
-samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_search_with_acl_override\(ad_dc_ntvfs\)
-samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_search_with_attr_acl_override\(ad_dc_ntvfs\)
-samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_search_with_blanket_oa_acl\(ad_dc_ntvfs\)
-samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_search_with_blanket_oa_deny_acl\(ad_dc_ntvfs\)
-samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_search_with_deny_acl\(ad_dc_ntvfs\)
-samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_search_with_deny_attr_acl\(ad_dc_ntvfs\)
-samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_search_with_deny_propset_acl\(ad_dc_ntvfs\)
-samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_search_with_dirsync\(ad_dc_ntvfs\)
-samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_search_with_neutral_acl\(ad_dc_ntvfs\)
-samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_search_with_neutral_acl\(ad_dc_ntvfs\)
-samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_search_with_neutral_attr_acl\(ad_dc_ntvfs\)
-samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_search_with_neutral_cr_acl\(ad_dc_ntvfs\)
-samba4.ldap.confidential_attr.python\(ad_dc_ntvfs\).__main__.ConfidentialAttrTest.test_search_with_propset_acl_override\(ad_dc_ntvfs\)
-
diff --git a/source4/dsdb/samdb/ldb_modules/acl_read.c b/source4/dsdb/samdb/ldb_modules/acl_read.c
index 9607ed0..37f2b97 100644
--- a/source4/dsdb/samdb/ldb_modules/acl_read.c
+++ b/source4/dsdb/samdb/ldb_modules/acl_read.c
@@ -38,6 +38,8 @@
 #include "param/param.h"
 #include "dsdb/samdb/ldb_modules/util.h"
 
+/* The attributeSecurityGuid for the Public Information Property-Set */
+#define PUBLIC_INFO_PROPERTY_SET "e48d0154-bcf8-11d1-8702-00c04fb96050"
 
 struct aclread_context {
 	struct ldb_module *module;
@@ -262,6 +264,217 @@ static uint32_t get_attr_access_mask(const struct dsdb_attribute *attr,
 	return access_mask;
 }
 
+/* helper struct for traversing the attributes in the search-tree */
+struct parse_tree_aclread_ctx {
+	struct aclread_context *ac;
+	TALLOC_CTX *mem_ctx;
+	struct dom_sid *sid;
+	struct ldb_dn *dn;
+	struct security_descriptor *sd;
+	const struct dsdb_class *objectclass;
+	bool suppress_result;
+};
+
+/*
+ * Checks that the user has sufficient access rights to view an attribute
+ */
+static int check_attr_access_rights(TALLOC_CTX *mem_ctx, const char *attr_name,
+				    struct aclread_context *ac,
+				    struct security_descriptor *sd,
+				    const struct dsdb_class *objectclass,
+				    struct dom_sid *sid, struct ldb_dn *dn,
+				    bool *is_public_info)
+{
+	int ret;
+	const struct dsdb_attribute *attr = NULL;
+	uint32_t access_mask;
+	struct ldb_context *ldb = ldb_module_get_ctx(ac->module);
+
+	*is_public_info = false;
+
+	attr = dsdb_attribute_by_lDAPDisplayName(ac->schema, attr_name);
+	if (!attr) {
+		ldb_debug_set(ldb, LDB_DEBUG_FATAL,
+			      "acl_read: %s cannot find attr[%s] in schema\n",
+			      ldb_dn_get_linearized(dn), attr_name);
+		return LDB_ERR_OPERATIONS_ERROR;
+	}
+
+	/*
+	 * If we have no Read Property (RP) rights for a child object, it should
+	 * still appear as a visible object in 'objectClass=*' searches,
+	 * as long as we have List Contents (LC) rights for it.
+	 * This is needed for the acl.py tests (e.g. test_search1()).
+	 * I couldn't find the Windows behaviour documented in the specs, so
+	 * this is a guess, but it seems to only apply to attributes in the
+	 * Public Information Property Set that have the systemOnly flag set to
+	 * TRUE. (This makes sense in a way, as it's not disclosive to find out
+	 * that a child object has a 'objectClass' or 'name' attribute, as every
+	 * object has these attributes).
+	 */
+	if (attr->systemOnly) {
+		struct GUID public_info_guid;
+		NTSTATUS status;
+
+		status = GUID_from_string(PUBLIC_INFO_PROPERTY_SET,
+					  &public_info_guid);
+		if (!NT_STATUS_IS_OK(status)) {
+			ldb_set_errstring(ldb, "Public Info GUID parse error");
+			return LDB_ERR_OPERATIONS_ERROR;
+		}
+
+		if (GUID_compare(&attr->attributeSecurityGUID,
+				 &public_info_guid) == 0) {
+			*is_public_info = true;
+		}
+	}
+
+	access_mask = get_attr_access_mask(attr, ac->sd_flags);
+
+	/* the access-mask should be non-zero. Skip attribute otherwise */
+	if (access_mask == 0) {
+		DBG_ERR("Could not determine access mask for attribute %s\n",
+			attr_name);
+		return LDB_SUCCESS;
+	}
+
+	ret = acl_check_access_on_attribute(ac->module, mem_ctx, sd, sid,
+					    access_mask, attr, objectclass);
+
+	if (ret == LDB_ERR_INSUFFICIENT_ACCESS_RIGHTS) {
+		return ret;
+	}
+
+	if (ret != LDB_SUCCESS) {
+		ldb_debug_set(ldb, LDB_DEBUG_FATAL,
+			      "acl_read: %s check attr[%s] gives %s - %s\n",
+			      ldb_dn_get_linearized(dn), attr_name,
+			      ldb_strerror(ret), ldb_errstring(ldb));
+		return ret;
+	}
+
+	return LDB_SUCCESS;
+}
+
+/*
+ * Returns the attribute name for this particular level of a search operation
+ * parse-tree.
+ */
+static const char * parse_tree_get_attr(struct ldb_parse_tree *tree)
+{
+	const char *attr = NULL;
+
+	switch (tree->operation) {
+	case LDB_OP_EQUALITY:
+	case LDB_OP_GREATER:
+	case LDB_OP_LESS:
+	case LDB_OP_APPROX:
+		attr = tree->u.equality.attr;
+		break;
+	case LDB_OP_SUBSTRING:
+		attr = tree->u.substring.attr;
+		break;
+	case LDB_OP_PRESENT:
+		attr = tree->u.present.attr;
+		break;
+	case LDB_OP_EXTENDED:
+		attr = tree->u.extended.attr;
+		break;
+
+	/* we'll check LDB_OP_AND/_OR/_NOT children later on in the walk */
+	default:
+		break;
+	}
+	return attr;
+}
+
+/*
+ * Checks a single attribute in the search parse-tree to make sure the user has
+ * sufficient rights to view it.
+ */
+static int parse_tree_check_attr_access(struct ldb_parse_tree *tree,
+					void *private_context)
+{
+	struct parse_tree_aclread_ctx *ctx = NULL;
+	const char *attr_name = NULL;
+	bool is_public_info = false;
+	int ret;
+
+	ctx = (struct parse_tree_aclread_ctx *)private_context;
+
+	/*
+	 * we can skip any further checking if we already know that this object
+	 * shouldn't be visible in this user's search
+	 */
+	if (ctx->suppress_result) {
+		return LDB_SUCCESS;
+	}
+
+	/* skip this level of the search-tree if it has no attribute to check */
+	attr_name = parse_tree_get_attr(tree);
+	if (attr_name == NULL) {
+		return LDB_SUCCESS;
+	}
+
+	ret = check_attr_access_rights(ctx->mem_ctx, attr_name, ctx->ac,
+				       ctx->sd, ctx->objectclass, ctx->sid,
+				       ctx->dn, &is_public_info);
+
+	/*
+	 * if the user does not have the rights to view this attribute, then we
+	 * should not return the object as a search result, i.e. act as if the
+	 * object doesn't exist (for this particular user, at least)
+	 */
+	if (ret == LDB_ERR_INSUFFICIENT_ACCESS_RIGHTS) {
+
+		/*
+		 * We make an exception for attribute=* searches involving the
+		 * Public Information property-set. This allows searches like
+		 * objectClass=* to return visible objects, even if the user
+		 * doesn't have Read Property rights on the attribute
+		 */
+		if (tree->operation == LDB_OP_PRESENT && is_public_info) {
+			return LDB_SUCCESS;
+		}
+
+		ctx->suppress_result = true;
+		return LDB_SUCCESS;
+	}
+
+	return ret;
+}
+
+/*
+ * Traverse the search-tree to check that the user has sufficient access rights
+ * to view all the attributes.
+ */
+static int check_search_ops_access(struct aclread_context *ac,
+				   TALLOC_CTX *mem_ctx,
+				   struct security_descriptor *sd,
+				   const struct dsdb_class *objectclass,
+				   struct dom_sid *sid, struct ldb_dn *dn,
+				   bool *suppress_result)
+{
+	int ret;
+	struct parse_tree_aclread_ctx ctx = { 0 };
+	struct ldb_parse_tree *tree = ac->req->op.search.tree;
+
+	ctx.ac = ac;
+	ctx.mem_ctx = mem_ctx;
+	ctx.suppress_result = false;
+	ctx.sid = sid;
+	ctx.dn = dn;
+	ctx.sd = sd;
+	ctx.objectclass = objectclass;
+
+	/* walk the search tree, checking each attribute as we go */
+	ret = ldb_parse_tree_walk(tree, parse_tree_check_attr_access, &ctx);
+
+	/* return whether this search result should be hidden to this user */
+	*suppress_result = ctx.suppress_result;
+	return ret;
+}
+
 static int aclread_callback(struct ldb_request *req, struct ldb_reply *ares)
 {
 	struct ldb_context *ldb;
@@ -275,6 +488,7 @@ static int aclread_callback(struct ldb_request *req, struct ldb_reply *ares)
 	TALLOC_CTX *tmp_ctx;
 	uint32_t instanceType;
 	const struct dsdb_class *objectclass;
+	bool suppress_result = false;
 
 	ac = talloc_get_type(req->context, struct aclread_context);
 	ldb = ldb_module_get_ctx(ac->module);
@@ -436,6 +650,38 @@ static int aclread_callback(struct ldb_request *req, struct ldb_reply *ares)
 				goto fail;
 			}
 		}
+
+		/*
+		 * check access rights for the search attributes, as well as the
+		 * attribute values actually being returned
+		 */
+		ret = check_search_ops_access(ac, tmp_ctx, sd, objectclass, sid,
+					      msg->dn, &suppress_result);
+		if (ret != LDB_SUCCESS) {
+			ldb_debug_set(ldb, LDB_DEBUG_FATAL,
+				      "acl_read: %s check search ops %s - %s\n",
+				      ldb_dn_get_linearized(msg->dn),
+				      ldb_strerror(ret), ldb_errstring(ldb));
+			goto fail;
+		}
+
+		if (suppress_result) {
+
+			/*
+			 * As per the above logic, we strip replPropertyMetaData
+			 * out of the msg so that the dirysync module will do
+			 * what is needed (return just the objectGUID if it's,
+			 * deleted, or remove the object if it is not).
+			 */
+			if (ac->indirsync) {
+				ldb_msg_remove_attr(msg, "replPropertyMetaData");
+				break;
+			} else {
+				talloc_free(tmp_ctx);
+				return LDB_SUCCESS;
+			}
+		}
+
 		for (i=0; i < msg->num_elements; i++) {
 			if (!aclread_is_inaccessible(&msg->elements[i])) {
 				num_of_attrs++;
-- 
2.7.4


From 381977d8f28788fd3a15d14aabbd04438a7c32be Mon Sep 17 00:00:00 2001
From: Tim Beale <timbeale@catalyst.net.nz>
Date: Thu, 26 Jul 2018 10:11:27 +1200
Subject: [PATCH 10/10] TODO: squash into tests

This extends the tests so that they use multiple objects (rather than
just one object). I'm still tidying it up, and will need to rename the
tests in the known fail, etc.

Tests currently pass against Samba and Windows though.
---
 source4/dsdb/tests/python/confidential_attr.py | 723 ++++++++++++++++++-------
 1 file changed, 538 insertions(+), 185 deletions(-)

diff --git a/source4/dsdb/tests/python/confidential_attr.py b/source4/dsdb/tests/python/confidential_attr.py
index 02a4b33..a39cdde 100755
--- a/source4/dsdb/tests/python/confidential_attr.py
+++ b/source4/dsdb/tests/python/confidential_attr.py
@@ -24,6 +24,7 @@ import sys
 sys.path.insert(0, "bin/python")
 
 import samba
+import os
 from samba.tests.subunitrun import SubunitOptions, TestProgram
 import samba.getopt as options
 from ldb import SCOPE_BASE, SCOPE_SUBTREE
@@ -67,14 +68,26 @@ lp = sambaopts.get_loadparm()
 creds = credopts.get_credentials(lp)
 creds.set_gensec_features(creds.get_gensec_features() | gensec.FEATURE_SEAL)
 
+# When a user does not have access rights to view the objects' attributes,
+# Windows and Samba behave slightly differently.
+# A windows DC will always act as if the hidden attribute doesn't exist AT ALL
+# (for an unprivileged user). So, even for a user that lacks access rights,
+# the inverse/'!' queries should return ALL objects. This is similar to the
+# kludgeaclredacted behaviour on Samba.
+# However, on Samba (for implementation simplicity) we never return a matching
+# result for an unprivileged user.
+# Either approach is OK, so long as it gets applied consistently and we don't
+# disclose any sensitive details by varying what gets returned by the search.
+DC_MODE_RETURN_NONE = 0
+DC_MODE_RETURN_ALL = 1
 
 #
 # Tests start here
 #
-class ConfidentialAttrTest(samba.tests.TestCase):
+class ConfidentialAttrCommon(samba.tests.TestCase):
 
     def setUp(self):
-        super(ConfidentialAttrTest, self).setUp()
+        super(ConfidentialAttrCommon, self).setUp()
 
         self.ldb_admin = SamDB(ldaphost, credentials=creds,
                                session_info=system_session(lp), lp=lp)
@@ -83,6 +96,9 @@ class ConfidentialAttrTest(samba.tests.TestCase):
         self.schema_dn = self.ldb_admin.get_schema_basedn()
         self.sd_utils = sd_utils.SDUtils(self.ldb_admin)
 
+        # assume the DC behaves like a Samba DC
+        self.dc_mode = DC_MODE_RETURN_NONE
+
         # the tests work by setting the 'Confidential' 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
@@ -93,25 +109,54 @@ class ConfidentialAttrTest(samba.tests.TestCase):
         # schemaIdGuid for homePostalAddress:
         self.conf_attr_guid = "16775781-47f3-11d1-a9c3-0000f80367c1"
         self.conf_attr_sec_guid = "77b5b886-944a-11d1-aebd-0000f80367c1"
+        self.attr_dn = "{},{}".format(self.conf_attr_cn, self.schema_dn)
+
+        userou = "OU=conf-attr-test"
+        self.ou = "{},{}".format(userou, self.base_dn)
+        self.ldb_admin.create_ou(self.ou)
+
+        # use a common username prefix, so we can use sAMAccountName=CATC-* as
+        # a search filter to only return the users we're interested in
+        self.user_prefix = "catc-"
 
         # add a test object with this attribute set
         self.conf_value = "abcdef"
-        self.conf_user = "confidential-user"
-        self.ldb_admin.newuser(self.conf_user, self.user_pass)
-        self.test_dn = self.get_user_dn(self.conf_user)
-        self.add_attr(self.test_dn, self.conf_attr, self.conf_value)
+        self.conf_user = "{}conf-user".format(self.user_prefix)
+        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)
 
         # add a sneaky user that will try to steal our secrets
-        self.user = "confattr_u1"
-        self.ldb_admin.newuser(self.user, self.user_pass)
+        self.user = "{}sneaky-user".format(self.user_prefix)
+        self.ldb_admin.newuser(self.user, self.user_pass, userou=userou)
         self.ldb_user = self.get_ldb_connection(self.user, self.user_pass)
 
-        self.dirsync = ["dirsync:1:1:1000"]
+        self.all_users = [self.user, self.conf_user]
+
+        # add some other users that also have confidential attributes, so we can
+        # check we don't disclose their details, particularly in '!' searches
+        for i in range(1, 3):
+            username = "{}other-user{}".format(self.user_prefix, i)
+            self.ldb_admin.newuser(username, self.user_pass, userou=userou)
+            userdn = self.get_user_dn(username)
+            self.add_attr(userdn, self.conf_attr, "xyz{}".format(i))
+            self.all_users.append(username)
+
+        # there are 4 users in the OU, plus the OU itself
+        self.test_dn = self.ou
+        self.total_objects = len(self.all_users) + 1
+        self.objects_with_attr = 3
+
+        # 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.assertTrue(int(search_flags) & SEARCH_FLAG_CONFIDENTIAL == 0,
+                        "{} searchFlags already {}".format(self.conf_attr,
+                                                           search_flags))
 
     def tearDown(self):
-        super(ConfidentialAttrTest, self).tearDown()
-        delete_force(self.ldb_admin, self.get_user_dn(self.user))
-        delete_force(self.ldb_admin, self.test_dn)
+        super(ConfidentialAttrCommon, self).tearDown()
+        self.ldb_admin.delete(self.ou, ["tree_delete:1"])
 
     def add_attr(self, dn, attr, value):
         m = Message()
@@ -131,22 +176,46 @@ class ConfidentialAttrTest(samba.tests.TestCase):
         # Windows, at least)
         self.ldb_admin.set_schema_update_now()
 
+    def get_attr_search_flags(self, attr_dn):
+        """Marks the attribute under test as being confidential"""
+        res = self.ldb_admin.search(attr_dn, scope=SCOPE_BASE,
+                                    attrs=['searchFlags'])
+        return res[0]['searchFlags'][0]
+
+
     def make_attr_confidential(self):
         """Marks the attribute under test as being confidential"""
-        attr_dn = "{},{}".format(self.conf_attr_cn, self.schema_dn)
 
         # work out the original 'searchFlags' value before we overwrite it
-        res = self.ldb_admin.search(attr_dn, scope=SCOPE_BASE,
-                                    attrs=['searchFlags'])
-        old_value = res[0]['searchFlags'][0]
+        old_value = self.get_attr_search_flags(self.attr_dn)
 
-        self.set_attr_search_flags(attr_dn, str(SEARCH_FLAG_CONFIDENTIAL))
+        self.set_attr_search_flags(self.attr_dn, str(SEARCH_FLAG_CONFIDENTIAL))
 
         # reset the value after the test completes
-        self.addCleanup(self.set_attr_search_flags, attr_dn, old_value)
+        self.addCleanup(self.set_attr_search_flags, self.attr_dn, old_value)
+
+    def guess_dc_mode(self):
+        # if we're in selftest, assume it's a Samba DC
+        if os.environ.get('SAMBA_SELFTEST') == '1':
+            self.dc_mode = DC_MODE_RETURN_NONE
+        else:
+            searches = self.get_negative_match_all_searches()
+            res = self.ldb_user.search(self.test_dn, expression=searches[0],
+                                       scope=SCOPE_SUBTREE)
+
+            # we default to DC_MODE_RETURN_NONE (samba).Update this if it
+            # looks like we're talking to a Windows DC
+            if len(res) == self.total_objects:
+                self.dc_mode = DC_MODE_RETURN_ALL
+        return self.dc_mode
 
     def get_user_dn(self, name):
-        return "CN={},CN=Users,{}".format(name, self.base_dn)
+        return "CN={},{}".format(name, self.ou)
+
+    def get_user_sid_string(self, username):
+        user_dn = self.get_user_dn(username)
+        user_sid = self.sd_utils.get_object_sid(user_dn)
+        return str(user_sid)
 
     def get_ldb_connection(self, target_username, target_password):
         creds_tmp = Credentials()
@@ -161,31 +230,25 @@ class ConfidentialAttrTest(samba.tests.TestCase):
         ldb_target = SamDB(url=ldaphost, credentials=creds_tmp, lp=lp)
         return ldb_target
 
-    def assert_search_result(self, expected_num, base_dn, expr,
-                             samdb, controls, attrs):
-        # try varying the scope of the search, just in case that has an impact
-        for scope in [SCOPE_SUBTREE, SCOPE_BASE]:
-            res = samdb.search(base_dn, expression=expr, scope=scope,
-                               attrs=attrs, controls=controls)
-            self.assertTrue(len(res) == expected_num,
-                            "%u results, not %u for search %s, attrs %s" %
-                            (len(res), expected_num, expr, str(attrs)))
-
-    def assert_conf_attr_searches(self, expected_num, base_dn=None,
-                                  samdb=None, controls=None, total_objects=1,
-                                  inverse_search=None):
-        """Check searches against the attribute under test work as expected"""
-
-        if base_dn is None:
-            base_dn = self.test_dn
+    def assert_not_in_result(self, res, exclude_dn):
+        for msg in res:
+            self.assertNotEqual(msg.dn, exclude_dn,
+                                "Search revealed object {}".format(exclude_dn))
 
-        if samdb is None:
-            samdb = self.ldb_user
+    def assert_search_result(self, expected_num, expr, samdb):
 
         # try asking for different attributes back: None/all, the confidential
         # attribute itself, and a random unrelated attribute
         attr_filters = [None, [self.conf_attr], ['name']]
+        for attr in attr_filters:
+            res = samdb.search(self.test_dn, expression=expr,
+                               scope=SCOPE_SUBTREE, attrs=attr)
+            self.assertTrue(len(res) == expected_num,
+                            "%u results, not %u for search %s, attr %s" %
+                            (len(res), expected_num, expr, str(attr)))
 
+    # return a selection of searches that match exactly against the test object
+    def get_exact_match_searches(self):
         first_char = self.conf_value[:1]
         last_char = self.conf_value[-1:]
         test_attr = self.conf_attr
@@ -196,46 +259,71 @@ class ConfidentialAttrTest(samba.tests.TestCase):
             "({}={}*)".format(test_attr, first_char),
             "({}=*{})".format(test_attr, last_char),
 
-            # check a full wildcard against the confidential attribute
-            # (which could reveal the attribute's presence/absence)
-            "({}=*)".format(test_attr),
-
             # sanity-check equality against an exact match on value
             "({}={})".format(test_attr, self.conf_value),
 
+            # '~=' searches don't work against Samba
+            # sanity-check an approx search against an exact match on value
+            # "({}~={})".format(test_attr, self.conf_value),
+
             # check wildcard in an AND search...
-            "(&(objectclass=*)({}=*))".format(test_attr),
             "(&({}={}*)(objectclass=*))".format(test_attr, first_char),
 
             # ...an OR search (against another term that will never match)
-            "(|(objectclass=banana)({}=*))".format(test_attr),
-            "(|({}={}*)(objectclass=banana))".format(test_attr, first_char),
+            "(|({}={}*)(objectclass=banana))".format(test_attr, first_char)]
 
-            # '~=' searches don't work against Samba
-            # sanity-check an approx search against an exact match on value
-            # "({}~={})".format(test_attr, self.conf_value),
+        return searches
+
+    # return searches that match any object with the attribute under test
+    def get_match_all_searches(self):
+
+        searches = [
+            # check a full wildcard against the confidential attribute
+            # (which could reveal the attribute's presence/absence)
+            "({}=*)".format(self.conf_attr),
+
+            # check wildcard in an AND search...
+            "(&(objectclass=*)({}=*))".format(self.conf_attr),
+
+            # ...an OR search (against another term that will never match)
+            "(|(objectclass=banana)({}=*))".format(self.conf_attr),
 
             # check <=, and >= expressions that would normally find a match
-            "({}>=0)".format(test_attr),
-            "({}<=ZZZZZZZZZZZ)".format(test_attr)]
+            "({}>=0)".format(self.conf_attr),
+            "({}<=ZZZZZZZZZZZ)".format(self.conf_attr)]
 
-        for search in searches:
-            for attrs in attr_filters:
-                self.assert_search_result(expected_num, base_dn, search, samdb,
-                                          controls, attrs)
+        return searches
 
-        # Check '!' search conditions. This gets a bit more complicated, as
-        # it may return the inverse number of matching objects
-        self.assert_inverse_searches(expected_num, base_dn, samdb, controls,
-                                     total_objects, inverse_search)
+    def assert_conf_attr_searches(self, has_rights_to=0, samdb=None):
+        """Check searches against the attribute under test work as expected"""
 
-    def assert_inverse_searches(self, expected_num, base_dn, samdb, controls,
-                                total_objects, and_expression):
-        """Asserts users cannot guess attr values by using 'NOT' searches"""
+        if samdb is None:
+            samdb = self.ldb_user
+
+        if has_rights_to == "all":
+            has_rights_to = self.objects_with_attr
+
+        # these first few searches we just expect to match against the one
+        # object under test that we're trying to guess the value of
+        expected_num = 1 if has_rights_to > 0 else 0
+
+        for search in self.get_exact_match_searches():
+            self.assert_search_result(expected_num, search, samdb)
+
+        # these next searches will match any objects we have rights to see
+        expected_num = has_rights_to
+
+        for search in self.get_match_all_searches():
+            self.assert_search_result(expected_num, search, samdb)
+
+    def get_negative_match_all_searches(self):
 
         first_char = self.conf_value[:1]
         last_char = self.conf_value[-1:]
+        not_first_char = chr(ord(first_char) + 1)
+        not_last_char = chr(ord(last_char) + 1)
 
+        # TODO:
         # The following are double negative searches (i.e. NOT non-matching-
         # condition) which will therefore match ALL objects, including the test
         # object(s). The NOT search filter makes things a bit more complicated.
@@ -247,69 +335,129 @@ class ConfidentialAttrTest(samba.tests.TestCase):
         # check that the search is non-disclosive, i.e. an unprivileged user
         # either gets consistently told about all objects or no objects, and
         # therefore cannot guess an attribute's value.
-        matching_searches = [
-            "(!({}={}*))".format(self.conf_attr, chr(ord(first_char) + 1)),
-            "(!({}=*{}))".format(self.conf_attr, chr(ord(last_char) + 1))]
-
-        # work out if this user has sufficient access rights
-        insufficient_rights = True if expected_num == 0 else False
-        expected_inverse_num = total_objects - expected_num
-        first_search = True
-        attr_filters = [None, [self.conf_attr], ['name']]
+        searches = [
+            "(!({}={}*))".format(self.conf_attr, not_first_char),
+            "(!({}=*{}))".format(self.conf_attr, not_last_char)]
+
+        return searches
 
-        # these first searches match ALL objects, so regardless of access
-        # rights, a Windows DC will always return all objects
-        expected_num = total_objects
-
-        for search in matching_searches:
-            # workaround for dirsync - we add an extra search clause to filter
-            # out the thousands of non-matching objects we're not interested in
-            if and_expression:
-                search = "(&{}{})".format(search, and_expression)
-
-            # on the first search, check if we're an unprivileged user talking
-            # to a Samba DC, in which case we're told about NO objects. This is
-            # acceptable, as long as the DC returns the same thing every time
-            if first_search and insufficient_rights:
-                res = samdb.search(base_dn, expression=search,
-                                   scope=SCOPE_SUBTREE, controls=controls)
-                if len(res) == 0:
-                    expected_num = 0
-                    expected_inverse_num = 0
-                first_search = False
-
-            for attrs in attr_filters:
-                self.assert_search_result(expected_num, base_dn, search,
-                                          samdb, controls, attrs)
+    def get_inverse_match_searches(self):
+
+        first_char = self.conf_value[:1]
+        last_char = self.conf_value[-1:]
 
         # the following searches will not match against the test object(s). So
         # a user with sufficient rights will see the inverse number of objects.
         # An unprivileged user should still see all objects (Windows) or no
         # objects (Samba)
-        non_matching_searches = [
-            "(!({}=*))".format(self.conf_attr),
+        searches = [
+#            "(!({}=*))".format(self.conf_attr),
             "(!({}={}*))".format(self.conf_attr, first_char),
             "(!({}=*{}))".format(self.conf_attr, last_char)]
 
-        for search in non_matching_searches:
-            # workaround for dirsync - we add an extra search clause to filter
-            # out the thousands of non-matching objects we're not interested in
-            if and_expression:
-                search = "(&{}{})".format(search, and_expression)
-            for attrs in attr_filters:
-                self.assert_search_result(expected_inverse_num, base_dn,
-                                          search, samdb, controls, attrs)
+        return searches
+
+    def expected_results_all_rights(self, total_objects=None):
+
+        if total_objects is None:
+            total_objects = self.total_objects
+
+        expected_results = {}
+
+        # these searches should match ALL objects (including the OU)
+        for search in self.get_negative_match_all_searches():
+            expected_results[search] = total_objects
+
+        # a ! wildcard should only match the objects without the attribute
+        search = "(!({}=*))".format(self.conf_attr)
+        expected_results[search] = total_objects - self.objects_with_attr
+
+        # whereas the inverse searches should match all objects *except* the
+        # one under test
+        for search in self.get_inverse_match_searches():
+            expected_results[search] = total_objects - 1
+
+        return expected_results
+
+    def assert_visible_in_inverse_search(self, samdb=None):
+        """Asserts user with access rights can see objects in '!' searches"""
+
+        if samdb is None:
+            samdb = self.ldb_user
+
+        expected_results = self.expected_results_all_rights()
+
+        for search, expected_num in expected_results.items():
+            self.assert_search_result(expected_num, search, samdb)
+
+    def expected_results_return_all(self, has_rights_to=0, total_objects=None):
+        """Asserts user without rights cannot see objects in '!' searches"""
+
+        if total_objects is None:
+            total_objects = self.total_objects
+
+        expected_results = {}
+
+        # Windows 'hides' objects by always returning all of them, so negative
+        # searches that match all objects will simply return all objects
+        for search in self.get_negative_match_all_searches():
+            expected_results[search] = total_objects
+
+        # if the search is matching on an inverse subset (everything except the
+        # object under test), the
+        inverse_searches = self.get_inverse_match_searches()
+        inverse_searches += ["(!({}=*))".format(self.conf_attr)]
+        for search in inverse_searches:
+            expected_results[search] = total_objects - has_rights_to
+
+        return expected_results
+
+    def expected_results_return_none(self, has_rights_to=0):
+        """Asserts user without rights cannot see objects in '!' searches"""
+
+        expected_results = {}
+
+        # the 'match-all' searches should only return the objects we have
+        # access rights to (if any)
+        for search in self.get_negative_match_all_searches():
+            expected_results[search] = has_rights_to
+
+        # for inverse matches, we should NOT be told about any objects at all
+        inverse_searches = self.get_inverse_match_searches()
+        inverse_searches += ["(!({}=*))".format(self.conf_attr)]
+        for search in inverse_searches:
+            expected_results[search] = 0
+
+        return expected_results
+
+    def assert_inverse_searches(self, has_rights_to=0):
+        """Asserts user without rights cannot see objects in '!' searches"""
+
+        # if it's a Samba DC, we only expect the 'match-all' searches to return
+        # the objects that we have access rights to (all others are hidden).
+        # Whereas Windows 'hides' the objects by always returning all of them
+        if self.dc_mode == DC_MODE_RETURN_NONE:
+            expected_results = self.expected_results_return_none(has_rights_to)
+        else:
+            expected_results = self.expected_results_return_all(has_rights_to)
+
+        for search, expected_num in expected_results.items():
+            self.assert_search_result(expected_num, search, self.ldb_user)
 
     def assert_attr_returned(self, expect_attr, samdb, attrs):
         # does a query that should always return a successful result, and
         # checks whether the confidential attribute is present
-        res = samdb.search(self.test_dn, expression="(objectClass=*)",
+        res = samdb.search(self.conf_dn, expression="(objectClass=*)",
                            scope=SCOPE_SUBTREE, attrs=attrs)
         self.assertTrue(len(res) == 1)
-        if expect_attr:
-            self.assertTrue(self.conf_attr in res[0])
-        else:
-            self.assertTrue(self.conf_attr not in res[0])
+
+        attr_returned = False
+
+        for msg in res:
+            if self.conf_attr in msg:
+                attr_returned = True
+
+        self.assertEqual(expect_attr, attr_returned)
 
     def assert_attr_visible(self, expect_attr, samdb=None):
         if samdb is None:
@@ -318,6 +466,7 @@ class ConfidentialAttrTest(samba.tests.TestCase):
         # sanity-check confidential attribute is/isn't returned as expected
         # based on the filter attributes we ask for
         self.assert_attr_returned(expect_attr, samdb, attrs=None)
+        self.assert_attr_returned(expect_attr, samdb, attrs=["*"])
         self.assert_attr_returned(expect_attr, samdb, attrs=[self.conf_attr])
 
         # filtering on a different attribute should never return the conf_attr
@@ -326,22 +475,28 @@ class ConfidentialAttrTest(samba.tests.TestCase):
 
     def assert_attr_visible_to_admin(self):
         # sanity-check the admin user can always see the confidential attribute
-        self.assert_conf_attr_searches(expected_num=1, samdb=self.ldb_admin)
+        self.assert_conf_attr_searches(has_rights_to="all", samdb=self.ldb_admin)
+        self.assert_visible_in_inverse_search(samdb=self.ldb_admin)
         self.assert_attr_visible(expect_attr=True, samdb=self.ldb_admin)
 
+
+class ConfidentialAttrTest(ConfidentialAttrCommon):
     def test_basic_search(self):
-        """Checks basic confidential attribute search (without any ACEs)"""
+        """Basic test confidential attributes aren't disclosed via searches"""
 
         # check we can see a non-confidential attribute in a basic searches
-        self.assert_conf_attr_searches(expected_num=1)
+        self.assert_conf_attr_searches(has_rights_to="all")
+        self.assert_visible_in_inverse_search()
         self.assert_attr_visible(expect_attr=True)
 
         # now make the attribute confidential. Repeat the tests and check that
         # an ordinary user can't see the attribute, or indirectly match on the
         # attribute via the search expression
         self.make_attr_confidential()
+        self.guess_dc_mode()
 
-        self.assert_conf_attr_searches(expected_num=0)
+        self.assert_conf_attr_searches(has_rights_to=0)
+        self.assert_inverse_searches(has_rights_to=0)
         self.assert_attr_visible(expect_attr=False)
 
         # sanity-check we haven't hidden the attribute from the admin as well
@@ -351,14 +506,19 @@ class ConfidentialAttrTest(samba.tests.TestCase):
         """Checks a ACE with 'CR' rights can override a confidential attr"""
         # make the test attribute confidential and check user can't see it
         self.make_attr_confidential()
-        self.assert_conf_attr_searches(expected_num=0)
+        self.guess_dc_mode()
+
+        self.assert_conf_attr_searches(has_rights_to=0)
+        self.assert_inverse_searches(has_rights_to=0)
         self.assert_attr_visible(expect_attr=False)
 
         # apply the allow ACE to the object under test
-        self.sd_utils.dacl_add_ace(self.test_dn, allow_ace)
+        self.sd_utils.dacl_add_ace(self.conf_dn, allow_ace)
 
-        # the user should now be able to see the attribute
-        self.assert_conf_attr_searches(expected_num=1)
+        # the user should now be able to see the attribute for the one object
+        # we gave it rights to
+        self.assert_conf_attr_searches(has_rights_to=1)
+        self.assert_inverse_searches(has_rights_to=1)
         self.assert_attr_visible(expect_attr=True)
 
         # sanity-check the admin can still see the attribute
@@ -369,9 +529,8 @@ class ConfidentialAttrTest(samba.tests.TestCase):
 
         # set the SEC_ADS_CONTROL_ACCESS bit ('CR') for the user for the
         # attribute under test, so the user can see it once more
-        user_dn = self.get_user_dn(self.user)
-        user_sid = self.sd_utils.get_object_sid(user_dn)
-        ace = "(OA;;CR;{};;{})".format(self.conf_attr_guid, str(user_sid))
+        user_sid = self.get_user_sid_string(self.user)
+        ace = "(OA;;CR;{};;{})".format(self.conf_attr_guid, user_sid)
 
         self._test_search_with_allow_acl(ace)
 
@@ -381,9 +540,8 @@ class ConfidentialAttrTest(samba.tests.TestCase):
         # set the SEC_ADS_CONTROL_ACCESS bit ('CR') for the user for the
         # property-set containing the attribute under test (i.e. the
         # attributeSecurityGuid), so the user can see it once more
-        user_dn = self.get_user_dn(self.user)
-        user_sid = self.sd_utils.get_object_sid(user_dn)
-        ace = "(OA;;CR;{};;{})".format(self.conf_attr_sec_guid, str(user_sid))
+        user_sid = self.get_user_sid_string(self.user)
+        ace = "(OA;;CR;{};;{})".format(self.conf_attr_sec_guid, user_sid)
 
         self._test_search_with_allow_acl(ace)
 
@@ -391,9 +549,8 @@ class ConfidentialAttrTest(samba.tests.TestCase):
         """Make the confidential attr visible via a general 'allow' ACE"""
 
         # set the allow SEC_ADS_CONTROL_ACCESS bit ('CR') for the user
-        user_dn = self.get_user_dn(self.user)
-        user_sid = self.sd_utils.get_object_sid(user_dn)
-        ace = "(A;;CR;;;{})".format(str(user_sid))
+        user_sid = self.get_user_sid_string(self.user)
+        ace = "(A;;CR;;;{})".format(user_sid)
 
         self._test_search_with_allow_acl(ace)
 
@@ -402,9 +559,8 @@ class ConfidentialAttrTest(samba.tests.TestCase):
 
         # this just checks that an Object Access (OA) ACE without a GUID
         # specified will work the same as an 'Access' (A) ACE
-        user_dn = self.get_user_dn(self.user)
-        user_sid = self.sd_utils.get_object_sid(user_dn)
-        ace = "(OA;;CR;;;{})".format(str(user_sid))
+        user_sid = self.get_user_sid_string(self.user)
+        ace = "(OA;;CR;;;{})".format(user_sid)
 
         self._test_search_with_allow_acl(ace)
 
@@ -412,14 +568,17 @@ class ConfidentialAttrTest(samba.tests.TestCase):
         """Checks that a user does NOT gain access via an unrelated ACE"""
         # make the test attribute confidential and check user can't see it
         self.make_attr_confidential()
-        self.assert_conf_attr_searches(expected_num=0)
+        self.guess_dc_mode()
+        self.assert_conf_attr_searches(has_rights_to=0)
+        self.assert_inverse_searches(has_rights_to=0)
         self.assert_attr_visible(expect_attr=False)
 
         # apply the ACE to the object under test
-        self.sd_utils.dacl_add_ace(self.test_dn, neutral_ace)
+        self.sd_utils.dacl_add_ace(self.conf_dn, neutral_ace)
 
         # this should make no difference to the user's ability to see the attr
-        self.assert_conf_attr_searches(expected_num=0)
+        self.assert_conf_attr_searches(has_rights_to=0)
+        self.assert_inverse_searches(has_rights_to=0)
         self.assert_attr_visible(expect_attr=False)
 
         # sanity-check the admin can still see the attribute
@@ -429,43 +588,156 @@ class ConfidentialAttrTest(samba.tests.TestCase):
         """Give the user all rights *except* CR for any attributes"""
 
         # give the user all rights *except* CR and check it makes no difference
-        user_dn = self.get_user_dn(self.user)
-        user_sid = self.sd_utils.get_object_sid(user_dn)
-        ace = "(A;;RPWPCCDCLCLORCWOWDSDDTSW;;;{})".format(str(user_sid))
+        user_sid = self.get_user_sid_string(self.user)
+        ace = "(A;;RPWPCCDCLCLORCWOWDSDDTSW;;;{})".format(user_sid)
         self._test_search_with_neutral_acl(ace)
 
     def test_search_with_neutral_attr_acl(self):
         """Give the user all rights *except* CR for the attribute under test"""
 
-        # giving user object-access rights *except* CR should make no difference
-        user_dn = self.get_user_dn(self.user)
-        user_sid = self.sd_utils.get_object_sid(user_dn)
+        # giving user all OA rights *except* CR should make no difference
+        user_sid = self.get_user_sid_string(self.user)
         rights = "RPWPCCDCLCLORCWOWDSDDTSW"
-        ace = "(OA;;{};{};;{})".format(rights, self.conf_attr_guid,
-                                       str(user_sid))
+        ace = "(OA;;{};{};;{})".format(rights, self.conf_attr_guid, user_sid)
         self._test_search_with_neutral_acl(ace)
 
     def test_search_with_neutral_cr_acl(self):
         """Give the user CR rights for *another* unrelated attribute"""
 
         # giving user object-access CR rights to an unrelated attribute
-        user_dn = self.get_user_dn(self.user)
-        user_sid = self.sd_utils.get_object_sid(user_dn)
+        user_sid = self.get_user_sid_string(self.user)
         # use the GUID for sAMAccountName here (for no particular reason)
         unrelated_attr = "3e0abfd0-126a-11d0-a060-00aa006c33ed"
-        ace = "(OA;;CR;{};;{})".format(unrelated_attr, str(user_sid))
+        ace = "(OA;;CR;{};;{})".format(unrelated_attr, user_sid)
         self._test_search_with_neutral_acl(ace)
 
+# Check that a Deny ACL on an attribute doesn't reveal confidential info
+class ConfidentialAttrTestDenyAcl(ConfidentialAttrCommon):
+
+    def assert_not_in_result(self, res, exclude_dn):
+        for msg in res:
+            self.assertNotEqual(msg.dn, exclude_dn,
+                                "Search revealed object {}".format(exclude_dn))
+
+    # deny ACL tests are slightly different as we are only denying access to
+    # the one object under test (rather than any objects with that attribute).
+    # Therefore we need an extra check that we don't reveal the test object
+    # in the search, if we're not supposed to
+    def assert_search_result(self, expected_num, expr, samdb,
+                             excl_testobj=False):
+
+        # try asking for different attributes back: None/all, the confidential
+        # attribute itself, and a random unrelated attribute
+        attr_filters = [None, [self.conf_attr], ['name']]
+        for attr in attr_filters:
+            res = samdb.search(self.test_dn, expression=expr,
+                               scope=SCOPE_SUBTREE, attrs=attr)
+            self.assertTrue(len(res) == expected_num,
+                           "%u results, not %u for search %s, attr %s" %
+                            (len(res), expected_num, expr, str(attr)))
+
+            # assert we haven't revealed the hidden test-object
+            if excl_testobj:
+                self.assert_not_in_result(res, exclude_dn=self.conf_dn)
+
+    # we make a few tweaks to the regular version of this function to cater to
+    # denying specifically one object via an ACE
+    def assert_conf_attr_searches(self, has_rights_to=0, samdb=None):
+        """Check searches against the attribute under test work as expected"""
+
+        if samdb is None:
+            samdb = self.ldb_user
+
+        # make sure the test object is not returned if we've been denied rights
+        # to it via an ACE
+        excl_testobj = True if has_rights_to == "deny-one" else False
+
+        # these first few searches we just expect to match against the one
+        # object under test that we're trying to guess the value of
+        expected_num = 1 if has_rights_to == "all" else 0
+
+        for search in self.get_exact_match_searches():
+            self.assert_search_result(expected_num, search, samdb,
+                                      excl_testobj)
+
+        # these next searches will match any objects with the attribute that
+        # we have rights to see (i.e. all except the object under test)
+        if has_rights_to == "all":
+            expected_num = self.objects_with_attr
+        elif has_rights_to == "deny-one":
+            expected_num = self.objects_with_attr - 1
+
+        for search in self.get_match_all_searches():
+            self.assert_search_result(expected_num, search, samdb,
+                                      excl_testobj)
+
+    def expected_deny_results_return_none(self):
+        expected_results = {}
+
+        # if it's a Samba DC, we only expect the 'match-all' searches to return
+        # the objects that we have access rights to (all others are hidden).
+        # Whereas Windows 'hides' the objects by always returning all of them
+        searches = self.get_negative_match_all_searches()
+        searches += self.get_inverse_match_searches()
+
+        # for inverse matches on Samba, we should NOT be told about any
+        # objects at all. Whereas Windows filters out any objects TODO
+        expected_num = self.total_objects - 1
+
+        for search in searches:
+            expected_results[search] = self.total_objects - 1
+
+        search = "(!({}=*))".format(self.conf_attr)
+        expected_results[search] = self.total_objects - self.objects_with_attr
+        return expected_results
+
+    def expected_deny_results_return_all(self):
+        expected_results = {}
+
+        # When a user lacks access rights to an object, Windows 'hides' it in
+        # '!' searches by always returning it, regardless of whether it matches
+        searches = self.get_negative_match_all_searches()
+        searches += self.get_inverse_match_searches()
+        for search in searches:
+            expected_results[search] = self.total_objects
+
+        # in the wildcard case, the one object we don't have rights to gets
+        # bundled in with the objects that don't have the attribute at all
+        search = "(!({}=*))".format(self.conf_attr)
+        has_rights_to = self.objects_with_attr - 1
+        expected_results[search] = self.total_objects - has_rights_to
+        return expected_results
+
+    def inverse_search_deny_acl(self):
+        """Checks '!' search behaviour when a deny ACL is applied"""
+
+        # for inverse matches on Samba, we should NOT be told about any
+        # objects at all. Whereas Windows filters out any objects TODO
+        if self.dc_mode == DC_MODE_RETURN_NONE:
+            expected_results = self.expected_deny_results_return_none()
+
+            # assert that the object under test is *never* returned by search
+            excl_testobj = True
+        else:
+            expected_results = self.expected_deny_results_return_all()
+            excl_testobj = False
+
+        for search, expected_num in expected_results.items():
+            self.assert_search_result(expected_num, search, self.ldb_user,
+                                      excl_testobj=excl_testobj)
     def _test_search_with_deny_acl(self, ace):
         # check the user can see the attribute initially
-        self.assert_conf_attr_searches(expected_num=1)
+        self.assert_conf_attr_searches(has_rights_to="all")
+        self.assert_visible_in_inverse_search()
         self.assert_attr_visible(expect_attr=True)
 
         # add the ACE that denies access to the attr under test
-        self.sd_utils.dacl_add_ace(self.test_dn, ace)
+        self.sd_utils.dacl_add_ace(self.conf_dn, ace)
 
         # the user shouldn't be able to see the attribute anymore
-        self.assert_conf_attr_searches(expected_num=0)
+        has_rights_to = self.objects_with_attr - 1
+        self.assert_conf_attr_searches(has_rights_to="deny-one")
+        self.inverse_search_deny_acl()
         self.assert_attr_visible(expect_attr=False)
 
         # sanity-check we haven't hidden the attribute from the admin as well
@@ -476,9 +748,8 @@ class ConfidentialAttrTest(samba.tests.TestCase):
 
         # add an ACE that denies the user Read Property (RP) access to the attr
         # (which is similar to making the attribute confidential)
-        user_dn = self.get_user_dn(self.user)
-        user_sid = self.sd_utils.get_object_sid(user_dn)
-        ace = "(OD;;RP;{};;{})".format(self.conf_attr_guid, str(user_sid))
+        user_sid = self.get_user_sid_string(self.user)
+        ace = "(OD;;RP;{};;{})".format(self.conf_attr_guid, user_sid)
 
         # check the user cannot see the attribute anymore
         self._test_search_with_deny_acl(ace)
@@ -498,9 +769,8 @@ class ConfidentialAttrTest(samba.tests.TestCase):
         """Checks a deny ACE on the attribute's Property-Set"""
 
         # add an blanket deny ACE for Read Property (RP) rights
-        user_dn = self.get_user_dn(self.user)
-        user_sid = self.sd_utils.get_object_sid(user_dn)
-        ace = "(OD;;RP;{};;{})".format(self.conf_attr_sec_guid, str(user_sid))
+        user_sid = self.get_user_sid_string(self.user)
+        ace = "(OD;;RP;{};;{})".format(self.conf_attr_sec_guid, user_sid)
 
         # check the user cannot see the attribute anymore
         self._test_search_with_deny_acl(ace)
@@ -510,59 +780,142 @@ class ConfidentialAttrTest(samba.tests.TestCase):
 
         # this just checks that adding a 'Object Deny' (OD) ACE without
         # specifying a GUID will work the same way as a 'Deny' (D) ACE
-        user_dn = self.get_user_dn(self.user)
-        user_sid = self.sd_utils.get_object_sid(user_dn)
-        ace = "(OD;;RP;;;{})".format(str(user_sid))
+        user_sid = self.get_user_sid_string(self.user)
+        ace = "(OD;;RP;;;{})".format(user_sid)
 
         # check the user cannot see the attribute anymore
         self._test_search_with_deny_acl(ace)
 
-    def assert_dirsync_attr_returned(self, samdb, expect_attr):
-        """Similar to assert_attr_returned() but adapted for dirsync"""
+
+# Check that using the dirsync controls doesn't reveal confidential attributes
+class ConfidentialAttrTestDirsync(ConfidentialAttrCommon):
+
+    def setUp(self):
+        super(ConfidentialAttrTestDirsync, self).setUp()
+        self.dirsync = ["dirsync:1:1:1000"]
+
+    def assert_search_result(self, expected_num, expr, samdb, base_dn=None):
+
+        # Note dirsync must always search on the partition base DN
+        if base_dn is None:
+            base_dn = self.base_dn
+
+# TODO: weird behaviour with self.conf_attr...?
+#        attr_filters = [None, [self.conf_attr], ['name']]
+        attr_filters = [None, ['name']]
+        for attr in attr_filters:
+            res = samdb.search(base_dn, expression=expr, scope=SCOPE_SUBTREE,
+                               attrs=attr, controls=self.dirsync)
+
+#            if (expected_num > 0 and attr == [self.conf_attr] and
+#                len(res) != expected_num):
+#                print("Got %u results, not %u for search %s, attr %s" %
+#                      (len(res), expected_num, expr, str(attr)))
+#                continue
+
+            self.assertTrue(len(res) == expected_num,
+                            "%u results, not %u for search %s, attr %s" %
+                            (len(res), expected_num, expr, str(attr)))
+
+
+    def assert_attr_returned(self, expect_attr, samdb, attrs,
+                             no_result_ok=False):
 
         # When using dirsync, the base DN we search on needs to be a naming
-        # context. Also, the attrs we request just seem to be ignored
+        # context. Add an extra filter to ignore all the objects we aren't
+        # interested in
         expr = "(&(samaccountname={})(!(isDeleted=*)))".format(self.conf_user)
-        res = samdb.search(self.base_dn, expression=expr,
-                           scope=SCOPE_SUBTREE, controls=self.dirsync)
-        self.assertTrue(len(res) == 1)
+        res = samdb.search(self.base_dn, expression=expr, scope=SCOPE_SUBTREE,
+                           attrs=attrs, controls=self.dirsync)
+        self.assertTrue(len(res) == 1 or no_result_ok)
+
+        attr_returned = False
+        for msg in res:
+            if self.conf_attr in msg and len(msg[self.conf_attr]) > 0:
+                attr_returned = True
+        self.assertEqual(expect_attr, attr_returned)
+
+    def assert_attr_visible(self, expect_attr, samdb=None):
+        if samdb is None:
+            samdb = self.ldb_user
+
+        # sanity-check confidential attribute is/isn't returned as expected
+        # based on the filter attributes we ask for
+        self.assert_attr_returned(expect_attr, samdb, attrs=None)
+        self.assert_attr_returned(expect_attr, samdb, attrs=["*"])
 
         if expect_attr:
-            self.assertTrue(self.conf_attr in res[0])
-            self.assertTrue(len(res[0][self.conf_attr]) == 1)
+            self.assert_attr_returned(expect_attr, samdb,
+                                      attrs=[self.conf_attr])
         else:
-            # Windows returns the attribute if it's present (it returns all
-            # attributes) but suppresses its actual value
-            self.assertTrue(self.conf_attr not in res[0] or
-                            len(res[0][self.conf_attr]) == 0)
+            # The behaviour with dirsync when asking solely for an attribute
+            # that you don't have rights to is a bit strange. Samba returns
+            # no result rather than an empty message with just the DN.
+            # Presumably this is due to dirsync module behaviour. It's not
+            # disclosive in that the DC behaves the same way as if you asked
+            # for a garbage/non-existent attribute
+            self.assert_attr_returned(expect_attr, samdb,
+                                      attrs=[self.conf_attr],
+                                      no_result_ok=True)
+            self.assert_attr_returned(expect_attr, samdb,
+                                      attrs=["garbage"], no_result_ok=True)
+
+        # filtering on a different attribute should never return the conf_attr
+        self.assert_attr_returned(expect_attr=False, samdb=samdb,
+                                  attrs=['name'])
+
+    def assert_inverse_dirsync_searches(self, samdb, has_rights_to=0):
+        """Asserts user without rights cannot see objects in '!' searches"""
+
+        # because we need to search on the base DN when using the dirsync
+        # controls, we need an extra filter for the inverse ('!') search,
+        # so we don't get thousands of objects returned
+        extra_filter = \
+            "(samaccountname={}*)(!(isDeleted=*))".format(self.user_prefix)
+
+        # because of the extra filter, the total objects we expect here only
+        # includes the user objects
+        total_objects = len(self.all_users)
+
+        if has_rights_to == "all":
+            expected_results = self.expected_results_all_rights(total_objects)
+
+        # if it's a Samba DC, we only expect the 'match-all' searches to return
+        # the objects that we have access rights to (all others are hidden).
+        # Whereas Windows 'hides' the objects by always returning all of them
+        elif self.dc_mode == DC_MODE_RETURN_NONE:
+            expected_results = self.expected_results_return_none(has_rights_to)
+        else:
+            expected_results = self.expected_results_return_all(has_rights_to,
+                                                                total_objects)
+
+        for search, expected_num in expected_results.items():
+            search = "(&{}{})".format(search, extra_filter)
+
+            self.assert_search_result(expected_num, search, samdb)
 
     def test_search_with_dirsync(self):
         """Checks dirsync controls don't reveal confidential attributes"""
 
-        # try searching using the dirsync controls. Note we need to search on
-        # the base DN when using the dirsync controls. This means, we need an
-        # extra filter for the inverse ('!') search, so we don't get thousands
-        # of objects returned
-        and_expr = "(samaccountname={})(!(isDeleted=*))".format(self.conf_user)
-        self.assert_conf_attr_searches(expected_num=1, base_dn=self.base_dn,
-                                       controls=self.dirsync,
-                                       inverse_search=and_expr)
-        self.assert_dirsync_attr_returned(self.ldb_user, expect_attr=True)
+        self.assert_conf_attr_searches(has_rights_to="all")
+        self.assert_attr_visible(expect_attr=True)
+        self.assert_inverse_dirsync_searches(self.ldb_user,
+                                             has_rights_to="all")
 
         # make the test attribute confidential and check user can't see it,
         # even if they use the dirsync controls
         self.make_attr_confidential()
+        self.guess_dc_mode()
 
-        self.assert_conf_attr_searches(expected_num=0, base_dn=self.base_dn,
-                                       controls=self.dirsync,
-                                       inverse_search=and_expr)
-        self.assert_dirsync_attr_returned(self.ldb_user, expect_attr=False)
+        self.assert_conf_attr_searches(has_rights_to=0)
+        self.assert_attr_visible(expect_attr=False)
+        self.assert_inverse_dirsync_searches(self.ldb_user, has_rights_to=0)
 
         # as a final sanity-check, make sure the admin can still see the attr
-        self.assert_conf_attr_searches(expected_num=1, base_dn=self.base_dn,
-                                       samdb=self.ldb_admin,
-                                       controls=self.dirsync,
-                                       inverse_search=and_expr)
-        self.assert_dirsync_attr_returned(self.ldb_admin, expect_attr=True)
+        self.assert_conf_attr_searches(has_rights_to="all",
+                                       samdb=self.ldb_admin)
+        self.assert_attr_visible(expect_attr=True, samdb=self.ldb_admin)
+        self.assert_inverse_dirsync_searches(self.ldb_admin,
+                                             has_rights_to="all")
 
 TestProgram(module=__name__, opts=subunitopts)
-- 
2.7.4