selftest: Extend further getnc_unpriv tests to pass against windows 2012R2
[nivanova/samba-autobuild/.git] / source4 / torture / drs / python / getnc_unpriv.py
1 #!/usr/bin/env python
2 # -*- coding: utf-8 -*-
3 #
4 # Tests replication scenarios with different user privileges.
5 # We want to test every replication scenario we can think of against:
6 # - users with only GET_CHANGES privileges
7 # - users with only GET_ALL_CHANGES privileges
8 # - users with both GET_CHANGES and GET_ALL_CHANGES privileges
9 # - users with no privileges
10 #
11 # Copyright (C) Kamen Mazdrashki <kamenim@samba.org> 2011
12 # Copyright (C) Andrew Bartlett <abartlet@samba.org> 2017
13 #
14 # This program is free software; you can redistribute it and/or modify
15 # it under the terms of the GNU General Public License as published by
16 # the Free Software Foundation; either version 3 of the License, or
17 # (at your option) any later version.
18 #
19 # This program is distributed in the hope that it will be useful,
20 # but WITHOUT ANY WARRANTY; without even the implied warranty of
21 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
22 # GNU General Public License for more details.
23 #
24 # You should have received a copy of the GNU General Public License
25 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
26 #
27
28 #
29 # Usage:
30 #  export DC1=dc1_dns_name
31 #  export DC2=dc2_dns_name
32 #  export SUBUNITRUN=$samba4srcdir/scripting/bin/subunitrun
33 #  PYTHONPATH="$PYTHONPATH:$samba4srcdir/torture/drs/python" $SUBUNITRUN getnc_unpriv -U"$DOMAIN/$DC_USERNAME"%"$DC_PASSWORD"
34 #
35
36 import drs_base
37 import samba.tests
38 from samba import werror, WERRORError
39
40 from samba import sd_utils
41 import ldb
42 from ldb import SCOPE_BASE
43
44 from samba.dcerpc import drsuapi, security
45 from samba.credentials import DONT_USE_KERBEROS
46
47 class DrsReplicaSyncUnprivTestCase(drs_base.DrsBaseTestCase):
48     """Confirm the behaviour of DsGetNCChanges for unprivileged users"""
49
50     def setUp(self):
51         super(DrsReplicaSyncUnprivTestCase, self).setUp()
52         self.get_changes_user = "get-changes-user"
53         self.base_dn = self.ldb_dc1.get_default_basedn()
54         self.ou = "OU=test_getncchanges,%s" % self.base_dn
55         self.user_pass = samba.generate_random_password(12, 16)
56         self.ldb_dc1.add({
57             "dn": self.ou,
58             "objectclass": "organizationalUnit"})
59         self.ldb_dc1.newuser(self.get_changes_user, self.user_pass,
60                              userou="OU=test_getncchanges")
61         (self.drs, self.drs_handle) = self._ds_bind(self.dnsname_dc1)
62
63         self.sd_utils = sd_utils.SDUtils(self.ldb_dc1)
64         self.user_dn = "cn=%s,%s" % (self.get_changes_user, self.ou)
65         user_sid = self.sd_utils.get_object_sid(self.user_dn)
66         self.acl_mod_get_changes = "(OA;;CR;%s;;%s)" % (security.GUID_DRS_GET_CHANGES,
67                                                         str(user_sid))
68         self.acl_mod_get_all_changes = "(OA;;CR;%s;;%s)" % (security.GUID_DRS_GET_ALL_CHANGES,
69                                                             str(user_sid))
70         self.desc_sddl = self.sd_utils.get_sd_as_sddl(self.base_dn)
71
72         # We set DONT_USE_KERBEROS to avoid a race with getting the
73         # user replicated to our selected KDC
74         self.user_creds = self.insta_creds(template=self.get_credentials(),
75                                            username=self.get_changes_user,
76                                            userpass=self.user_pass,
77                                            kerberos_state=DONT_USE_KERBEROS)
78         (self.user_drs, self.user_drs_handle) = self._ds_bind(self.dnsname_dc1,
79                                                               self.user_creds)
80
81     def tearDown(self):
82         self.sd_utils.modify_sd_on_dn(self.base_dn, self.desc_sddl)
83         try:
84             self.ldb_dc1.delete(self.ou, ["tree_delete:1"])
85         except ldb.LdbError as (enum, string):
86             if enum == ldb.ERR_NO_SUCH_OBJECT:
87                 pass
88         super(DrsReplicaSyncUnprivTestCase, self).tearDown()
89
90     def _test_repl_exop(self, exop, repl_obj, expected_error, dest_dsa=None,
91                         partial_attribute_set=None):
92         """
93         Common function to send a replication request and check the result
94         matches what's expected.
95         """
96         req8 = self._exop_req8(dest_dsa=dest_dsa,
97                                invocation_id=self.ldb_dc1.get_invocation_id(),
98                                nc_dn_str=repl_obj,
99                                exop=exop,
100                                replica_flags=drsuapi.DRSUAPI_DRS_WRIT_REP,
101                                partial_attribute_set=partial_attribute_set)
102
103         if expected_error is None:
104             # user is OK, request should be accepted without throwing an error
105             (level, ctr) = self.user_drs.DsGetNCChanges(self.user_drs_handle,
106                                                         8, req8)
107         else:
108             # check the request is rejected (with the error we're expecting)
109             try:
110                 (level, ctr) = self.user_drs.DsGetNCChanges(self.user_drs_handle,
111                                                             8, req8)
112                 self.fail("Should have failed with user denied access")
113             except WERRORError as (enum, estr):
114                 self.assertEquals(enum, expected_error,
115                                   "Got unexpected error: %s" % estr)
116
117     def _test_repl_single_obj(self, repl_obj, expected_error,
118                               partial_attribute_set=None):
119         """
120         Checks that replication on a single object either succeeds or fails as
121         expected (based on the user's access rights)
122         """
123         self._test_repl_exop(exop=drsuapi.DRSUAPI_EXOP_REPL_OBJ,
124                              repl_obj=repl_obj,
125                              expected_error=expected_error,
126                              partial_attribute_set=partial_attribute_set)
127
128     def _test_repl_secret(self, repl_obj, expected_error, dest_dsa=None):
129         """
130         Checks that REPL_SECRET on an object either succeeds or fails as
131         expected (based on the user's access rights)
132         """
133         self._test_repl_exop(exop=drsuapi.DRSUAPI_EXOP_REPL_SECRET,
134                              repl_obj=repl_obj,
135                              expected_error=expected_error,
136                              dest_dsa=dest_dsa)
137
138     def _test_repl_full(self, expected_error, partial_attribute_set=None):
139         """
140         Checks that a full replication either succeeds or fails as expected
141         (based on the user's access rights)
142         """
143         self._test_repl_exop(exop=drsuapi.DRSUAPI_EXOP_NONE,
144                              repl_obj=self.ldb_dc1.get_default_basedn(),
145                              expected_error=expected_error,
146                              partial_attribute_set=partial_attribute_set)
147
148     def _test_repl_full_on_ou(self, expected_error):
149         """
150         Full replication on a specific OU should always fail (it should be done
151         against a base NC). The error may vary based on the user's access rights
152         """
153         # Just try against the OU created in the test setup
154         self._test_repl_exop(exop=drsuapi.DRSUAPI_EXOP_NONE,
155                              repl_obj=self.ou,
156                              expected_error=expected_error)
157
158     def test_repl_getchanges_userpriv(self):
159         """
160         Tests various replication requests made by a user with only GET_CHANGES
161         rights. Some requests will be accepted, but most will be rejected.
162         """
163
164         # Assign the user GET_CHANGES rights
165         self.sd_utils.dacl_add_ace(self.base_dn, self.acl_mod_get_changes)
166
167         self._test_repl_single_obj(repl_obj=self.ou,
168                                    expected_error=werror.WERR_DS_DRA_ACCESS_DENIED)
169
170         self._test_repl_secret(repl_obj=self.ou,
171                                expected_error=werror.WERR_DS_DRA_ACCESS_DENIED)
172         self._test_repl_secret(repl_obj=self.user_dn,
173                                expected_error=werror.WERR_DS_DRA_ACCESS_DENIED)
174         self._test_repl_secret(repl_obj=self.user_dn,
175                                dest_dsa=self.ldb_dc1.get_ntds_GUID(),
176                                expected_error=werror.WERR_DS_DRA_ACCESS_DENIED)
177
178         self._test_repl_full(expected_error=werror.WERR_DS_DRA_ACCESS_DENIED)
179         self._test_repl_full_on_ou(expected_error=werror.WERR_DS_CANT_FIND_EXPECTED_NC)
180
181         # Partial Attribute Sets don't require GET_ALL_CHANGES rights, so we
182         # expect the following to succeed
183         self._test_repl_single_obj(repl_obj=self.ou,
184                                    expected_error=None,
185                                    partial_attribute_set=self.get_partial_attribute_set())
186         self._test_repl_full(expected_error=None,
187                              partial_attribute_set=self.get_partial_attribute_set())
188
189     def test_repl_getallchanges_userpriv(self):
190         """
191         Tests various replication requests made by a user with only
192         GET_ALL_CHANGES rights. Note that assigning these rights is possible,
193         but doesn't make a lot of sense. We test it anyway for consistency.
194         """
195
196         # Assign the user GET_ALL_CHANGES rights
197         self.sd_utils.dacl_add_ace(self.base_dn, self.acl_mod_get_all_changes)
198
199         # We can expect to get the same responses as an unprivileged user,
200         # i.e. we have permission to see the results, but don't have permission
201         # to ask
202         self.test_repl_no_userpriv()
203
204     def test_repl_both_userpriv(self):
205         """
206         Tests various replication requests made by a privileged user (i.e. has
207         both GET_CHANGES and GET_ALL_CHANGES). We expect any valid requests
208         to be accepted.
209         """
210
211         # Assign the user both GET_CHANGES and GET_ALL_CHANGES rights
212         both_rights = self.acl_mod_get_changes + self.acl_mod_get_all_changes
213         self.sd_utils.dacl_add_ace(self.base_dn, both_rights)
214
215         self._test_repl_single_obj(repl_obj=self.ou,
216                                    expected_error=None)
217
218         self._test_repl_secret(repl_obj=self.ou,
219                                expected_error=werror.WERR_DS_DRA_DB_ERROR)
220         self._test_repl_secret(repl_obj=self.user_dn,
221                                expected_error=werror.WERR_DS_DRA_DB_ERROR)
222         self._test_repl_secret(repl_obj=self.user_dn,
223                                dest_dsa=self.ldb_dc1.get_ntds_GUID(),
224                                expected_error=None)
225
226         self._test_repl_full(expected_error=None)
227         self._test_repl_full_on_ou(expected_error=werror.WERR_DS_CANT_FIND_EXPECTED_NC)
228
229         self._test_repl_single_obj(repl_obj=self.ou,
230                                    expected_error=None,
231                                    partial_attribute_set=self.get_partial_attribute_set())
232         self._test_repl_full(expected_error=None,
233                              partial_attribute_set=self.get_partial_attribute_set())
234
235     def test_repl_no_userpriv(self):
236         """
237         Tests various replication requests made by a unprivileged user.
238         We expect all these requests to be rejected.
239         """
240
241         self._test_repl_single_obj(repl_obj=self.ou,
242                                    expected_error=werror.WERR_DS_DRA_BAD_DN)
243
244         self._test_repl_secret(repl_obj=self.ou,
245                                expected_error=werror.WERR_DS_DRA_BAD_DN)
246         self._test_repl_secret(repl_obj=self.user_dn,
247                                expected_error=werror.WERR_DS_DRA_BAD_DN)
248         self._test_repl_secret(repl_obj=self.user_dn,
249                                dest_dsa=self.ldb_dc1.get_ntds_GUID(),
250                                expected_error=werror.WERR_DS_DRA_BAD_DN)
251
252         self._test_repl_full(expected_error=werror.WERR_DS_DRA_ACCESS_DENIED)
253         self._test_repl_full_on_ou(expected_error=werror.WERR_DS_DRA_BAD_DN)
254
255         self._test_repl_single_obj(repl_obj=self.ou,
256                                    expected_error=werror.WERR_DS_DRA_BAD_DN,
257                                    partial_attribute_set=self.get_partial_attribute_set())
258         self._test_repl_full(expected_error=werror.WERR_DS_DRA_ACCESS_DENIED,
259                              partial_attribute_set=self.get_partial_attribute_set())
260
261