tests/ridalloc_exop: Add a new suite of tests for RID allocation
[nivanova/samba-autobuild/.git] / source4 / torture / drs / python / ridalloc_exop.py
1 #!/usr/bin/env python
2 # -*- coding: utf-8 -*-
3 #
4 # Tests various RID allocation scenarios
5 #
6 # Copyright (C) Kamen Mazdrashki <kamenim@samba.org> 2011
7 # Copyright (C) Andrew Bartlett <abartlet@samba.org> 2016
8 # Copyright (C) Catalyst IT Ltd. 2016
9 #
10 # This program is free software; you can redistribute it and/or modify
11 # it under the terms of the GNU General Public License as published by
12 # the Free Software Foundation; either version 3 of the License, or
13 # (at your option) any later version.
14 #
15 # This program is distributed in the hope that it will be useful,
16 # but WITHOUT ANY WARRANTY; without even the implied warranty of
17 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
18 # GNU General Public License for more details.
19 #
20 # You should have received a copy of the GNU General Public License
21 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
22 #
23
24 #
25 # Usage:
26 #  export DC1=dc1_dns_name
27 #  export DC2=dc2_dns_name
28 #  export SUBUNITRUN=$samba4srcdir/scripting/bin/subunitrun
29 #  PYTHONPATH="$PYTHONPATH:$samba4srcdir/torture/drs/python" $SUBUNITRUN ridalloc_exop -U"$DOMAIN/$DC_USERNAME"%"$DC_PASSWORD"
30 #
31
32 import drs_base
33 import samba.tests
34
35 import ldb
36 from ldb import SCOPE_BASE
37
38 from samba.dcerpc import drsuapi, misc
39 from samba.drs_utils import drs_DsBind
40 from samba.samdb import SamDB
41
42 import shutil, tempfile, os
43 from samba.netcmd.main import cmd_sambatool
44 from samba.auth import system_session, admin_session
45 from samba.dbchecker import dbcheck
46 from samba.ndr import ndr_pack
47 from samba.dcerpc import security
48
49 class ExopBaseTest:
50     def _exop_req8(self, dest_dsa, invocation_id, nc_dn_str, exop,
51                    replica_flags=0, max_objects=0, partial_attribute_set=None,
52                    partial_attribute_set_ex=None, mapping_ctr=None):
53         req8 = drsuapi.DsGetNCChangesRequest8()
54
55         req8.destination_dsa_guid = misc.GUID(dest_dsa) if dest_dsa else misc.GUID()
56         req8.source_dsa_invocation_id = misc.GUID(invocation_id)
57         req8.naming_context = drsuapi.DsReplicaObjectIdentifier()
58         req8.naming_context.dn = unicode(nc_dn_str)
59         req8.highwatermark = drsuapi.DsReplicaHighWaterMark()
60         req8.highwatermark.tmp_highest_usn = 0
61         req8.highwatermark.reserved_usn = 0
62         req8.highwatermark.highest_usn = 0
63         req8.uptodateness_vector = None
64         req8.replica_flags = replica_flags
65         req8.max_object_count = max_objects
66         req8.max_ndr_size = 402116
67         req8.extended_op = exop
68         req8.fsmo_info = 0
69         req8.partial_attribute_set = partial_attribute_set
70         req8.partial_attribute_set_ex = partial_attribute_set_ex
71         if mapping_ctr:
72             req8.mapping_ctr = mapping_ctr
73         else:
74             req8.mapping_ctr.num_mappings = 0
75             req8.mapping_ctr.mappings = None
76
77         return req8
78
79     def _ds_bind(self, server_name):
80         binding_str = "ncacn_ip_tcp:%s[seal]" % server_name
81
82         drs = drsuapi.drsuapi(binding_str, self.get_loadparm(), self.get_credentials())
83         (drs_handle, supported_extensions) = drs_DsBind(drs)
84         return (drs, drs_handle)
85
86
87 class DrsReplicaSyncTestCase(drs_base.DrsBaseTestCase, ExopBaseTest):
88     """Intended as a semi-black box test case for DsGetNCChanges
89        implementation for extended operations. It should be testing
90        how DsGetNCChanges handles different input params (mostly invalid).
91        Final goal is to make DsGetNCChanges as binary compatible to
92        Windows implementation as possible"""
93
94     def setUp(self):
95         super(DrsReplicaSyncTestCase, self).setUp()
96
97     def tearDown(self):
98         super(DrsReplicaSyncTestCase, self).tearDown()
99
100     def _determine_fSMORoleOwner(self, fsmo_obj_dn):
101         """Returns (owner, not_owner) pair where:
102              owner: dns name for FSMO owner
103              not_owner: dns name for DC not owning the FSMO"""
104         # collect info to return later
105         fsmo_info_1 = {"dns_name": self.dnsname_dc1,
106                        "invocation_id": self.ldb_dc1.get_invocation_id(),
107                        "ntds_guid": self.ldb_dc1.get_ntds_GUID(),
108                        "server_dn": self.ldb_dc1.get_serverName()}
109         fsmo_info_2 = {"dns_name": self.dnsname_dc2,
110                        "invocation_id": self.ldb_dc2.get_invocation_id(),
111                        "ntds_guid": self.ldb_dc2.get_ntds_GUID(),
112                        "server_dn": self.ldb_dc2.get_serverName()}
113
114         msgs = self.ldb_dc1.search(scope=ldb.SCOPE_BASE, base=fsmo_info_1["server_dn"], attrs=["serverReference"])
115         fsmo_info_1["server_acct_dn"] = ldb.Dn(self.ldb_dc1, msgs[0]["serverReference"][0])
116         fsmo_info_1["rid_set_dn"] = ldb.Dn(self.ldb_dc1, "CN=RID Set") + fsmo_info_1["server_acct_dn"]
117
118         msgs = self.ldb_dc2.search(scope=ldb.SCOPE_BASE, base=fsmo_info_2["server_dn"], attrs=["serverReference"])
119         fsmo_info_2["server_acct_dn"] = ldb.Dn(self.ldb_dc2, msgs[0]["serverReference"][0])
120         fsmo_info_2["rid_set_dn"] = ldb.Dn(self.ldb_dc2, "CN=RID Set") + fsmo_info_2["server_acct_dn"]
121
122         # determine the owner dc
123         res = self.ldb_dc1.search(fsmo_obj_dn,
124                                   scope=SCOPE_BASE, attrs=["fSMORoleOwner"])
125         assert len(res) == 1, "Only one fSMORoleOwner value expected for %s!"%fsmo_obj_dn
126         fsmo_owner = res[0]["fSMORoleOwner"][0]
127         if fsmo_owner == self.info_dc1["dsServiceName"][0]:
128             return (fsmo_info_1, fsmo_info_2)
129         return (fsmo_info_2, fsmo_info_1)
130
131     def _check_exop_failed(self, ctr6, expected_failure):
132         self.assertEqual(ctr6.extended_ret, expected_failure)
133         #self.assertEqual(ctr6.object_count, 0)
134         #self.assertEqual(ctr6.first_object, None)
135         self.assertEqual(ctr6.more_data, False)
136         self.assertEqual(ctr6.nc_object_count, 0)
137         self.assertEqual(ctr6.nc_linked_attributes_count, 0)
138         self.assertEqual(ctr6.linked_attributes_count, 0)
139         self.assertEqual(ctr6.linked_attributes, [])
140         self.assertEqual(ctr6.drs_error[0], 0)
141
142     def test_InvalidDestDSA_ridalloc(self):
143         """Test RID allocation with invalid destination DSA guid"""
144         fsmo_dn = ldb.Dn(self.ldb_dc1, "CN=RID Manager$,CN=System," + self.ldb_dc1.domain_dn())
145         (fsmo_owner, fsmo_not_owner) = self._determine_fSMORoleOwner(fsmo_dn)
146
147         req8 = self._exop_req8(dest_dsa="9c637462-5b8c-4467-aef2-bdb1f57bc4ef",
148                                invocation_id=fsmo_owner["invocation_id"],
149                                nc_dn_str=fsmo_dn,
150                                exop=drsuapi.DRSUAPI_EXOP_FSMO_RID_ALLOC)
151
152         (drs, drs_handle) = self._ds_bind(fsmo_owner["dns_name"])
153         (level, ctr) = drs.DsGetNCChanges(drs_handle, 8, req8)
154         self.assertEqual(level, 6, "Expected level 6 response!")
155         self._check_exop_failed(ctr, drsuapi.DRSUAPI_EXOP_ERR_UNKNOWN_CALLER)
156         self.assertEqual(ctr.source_dsa_guid, misc.GUID(fsmo_owner["ntds_guid"]))
157         self.assertEqual(ctr.source_dsa_invocation_id, misc.GUID(fsmo_owner["invocation_id"]))
158
159     def test_do_ridalloc(self):
160         """Test doing a RID allocation with a valid destination DSA guid"""
161         fsmo_dn = ldb.Dn(self.ldb_dc1, "CN=RID Manager$,CN=System," + self.ldb_dc1.domain_dn())
162         (fsmo_owner, fsmo_not_owner) = self._determine_fSMORoleOwner(fsmo_dn)
163
164         req8 = self._exop_req8(dest_dsa=fsmo_not_owner["ntds_guid"],
165                                invocation_id=fsmo_owner["invocation_id"],
166                                nc_dn_str=fsmo_dn,
167                                exop=drsuapi.DRSUAPI_EXOP_FSMO_RID_ALLOC)
168
169         (drs, drs_handle) = self._ds_bind(fsmo_owner["dns_name"])
170         (level, ctr) = drs.DsGetNCChanges(drs_handle, 8, req8)
171         self.assertEqual(level, 6, "Expected level 6 response!")
172         self.assertEqual(ctr.source_dsa_guid, misc.GUID(fsmo_owner["ntds_guid"]))
173         self.assertEqual(ctr.source_dsa_invocation_id, misc.GUID(fsmo_owner["invocation_id"]))
174         ctr6 = ctr
175         self.assertEqual(ctr6.extended_ret, drsuapi.DRSUAPI_EXOP_ERR_SUCCESS)
176         self.assertEqual(ctr6.object_count, 3)
177         self.assertNotEqual(ctr6.first_object, None)
178         self.assertEqual(ldb.Dn(self.ldb_dc1, ctr6.first_object.object.identifier.dn), fsmo_dn)
179         self.assertNotEqual(ctr6.first_object.next_object, None)
180         self.assertNotEqual(ctr6.first_object.next_object.next_object, None)
181         second_object = ctr6.first_object.next_object.object
182         self.assertEqual(ldb.Dn(self.ldb_dc1, second_object.identifier.dn), fsmo_not_owner["rid_set_dn"])
183         third_object = ctr6.first_object.next_object.next_object.object
184         self.assertEqual(ldb.Dn(self.ldb_dc1, third_object.identifier.dn), fsmo_not_owner["server_acct_dn"])
185
186         self.assertEqual(ctr6.more_data, False)
187         self.assertEqual(ctr6.nc_object_count, 0)
188         self.assertEqual(ctr6.nc_linked_attributes_count, 0)
189         self.assertEqual(ctr6.drs_error[0], 0)
190         # We don't check the linked_attributes_count as if the domain
191         # has an RODC, it can gain links on the server account object
192
193     def test_do_ridalloc_get_anc(self):
194         """Test doing a RID allocation with a valid destination DSA guid and GET_ANC flag"""
195         fsmo_dn = ldb.Dn(self.ldb_dc1, "CN=RID Manager$,CN=System," + self.ldb_dc1.domain_dn())
196         (fsmo_owner, fsmo_not_owner) = self._determine_fSMORoleOwner(fsmo_dn)
197
198         req8 = self._exop_req8(dest_dsa=fsmo_not_owner["ntds_guid"],
199                                invocation_id=fsmo_owner["invocation_id"],
200                                nc_dn_str=fsmo_dn,
201                                exop=drsuapi.DRSUAPI_EXOP_FSMO_RID_ALLOC,
202                                replica_flags=drsuapi.DRSUAPI_DRS_GET_ANC)
203
204         (drs, drs_handle) = self._ds_bind(fsmo_owner["dns_name"])
205         (level, ctr) = drs.DsGetNCChanges(drs_handle, 8, req8)
206         self.assertEqual(level, 6, "Expected level 6 response!")
207         self.assertEqual(ctr.source_dsa_guid, misc.GUID(fsmo_owner["ntds_guid"]))
208         self.assertEqual(ctr.source_dsa_invocation_id, misc.GUID(fsmo_owner["invocation_id"]))
209         ctr6 = ctr
210         self.assertEqual(ctr6.extended_ret, drsuapi.DRSUAPI_EXOP_ERR_SUCCESS)
211         self.assertEqual(ctr6.object_count, 3)
212         self.assertNotEqual(ctr6.first_object, None)
213         self.assertEqual(ldb.Dn(self.ldb_dc1, ctr6.first_object.object.identifier.dn), fsmo_dn)
214         self.assertNotEqual(ctr6.first_object.next_object, None)
215         self.assertNotEqual(ctr6.first_object.next_object.next_object, None)
216         second_object = ctr6.first_object.next_object.object
217         self.assertEqual(ldb.Dn(self.ldb_dc1, second_object.identifier.dn), fsmo_not_owner["rid_set_dn"])
218         third_object = ctr6.first_object.next_object.next_object.object
219         self.assertEqual(ldb.Dn(self.ldb_dc1, third_object.identifier.dn), fsmo_not_owner["server_acct_dn"])
220         self.assertEqual(ctr6.more_data, False)
221         self.assertEqual(ctr6.nc_object_count, 0)
222         self.assertEqual(ctr6.nc_linked_attributes_count, 0)
223         self.assertEqual(ctr6.drs_error[0], 0)
224         # We don't check the linked_attributes_count as if the domain
225         # has an RODC, it can gain links on the server account object
226
227     def test_edit_rid_master(self):
228         """Test doing a RID allocation after changing the RID master from the original one.
229            This should set rIDNextRID to 0 on the new RID master."""
230         # 1. a. Transfer role to non-RID master
231         #    b. Check that it succeeds correctly
232         #
233         # 2. a. Call the RID alloc against the former master.
234         #    b. Check that it succeeds.
235         fsmo_dn = ldb.Dn(self.ldb_dc1, "CN=RID Manager$,CN=System," + self.ldb_dc1.domain_dn())
236         (fsmo_owner, fsmo_not_owner) = self._determine_fSMORoleOwner(fsmo_dn)
237
238         # 1. Swap RID master role
239         m = ldb.Message()
240         m.dn = ldb.Dn(self.ldb_dc1, "")
241         m["becomeRidMaster"] = ldb.MessageElement("1", ldb.FLAG_MOD_REPLACE,
242                                                   "becomeRidMaster")
243
244         # Make sure that ldb_dc1 == RID Master
245
246         server_dn = str(ldb.Dn(self.ldb_dc1, self.ldb_dc1.get_dsServiceName()).parent())
247
248         # self.ldb_dc1 == LOCALDC
249         if server_dn == fsmo_owner['server_dn']:
250             # ldb_dc1 == VAMPIREDC
251             ldb_dc1, ldb_dc2 = self.ldb_dc2, self.ldb_dc1
252         else:
253             # Otherwise switch the two
254             ldb_dc1, ldb_dc2 = self.ldb_dc1, self.ldb_dc2
255
256         try:
257             # ldb_dc1 is now RID MASTER (as VAMPIREDC)
258             ldb_dc1.modify(m)
259         except ldb.LdbError, (num, msg):
260             self.fail("Failed to reassign RID Master " +  msg)
261
262         try:
263             # 2. Perform a RID alloc
264             req8 = self._exop_req8(dest_dsa=fsmo_owner["ntds_guid"],
265                     invocation_id=fsmo_not_owner["invocation_id"],
266                     nc_dn_str=fsmo_dn,
267                     exop=drsuapi.DRSUAPI_EXOP_FSMO_RID_ALLOC)
268
269             (drs, drs_handle) = self._ds_bind(fsmo_not_owner["dns_name"])
270             # 3. Make sure the allocation succeeds
271             try:
272                 (level, ctr) = drs.DsGetNCChanges(drs_handle, 8, req8)
273             except RuntimeError, e:
274                 self.fail("RID allocation failed: " + str(e))
275
276             fsmo_dn = ldb.Dn(self.ldb_dc1, "CN=RID Manager$,CN=System," + self.ldb_dc1.domain_dn())
277
278             self.assertEqual(level, 6, "Expected level 6 response!")
279             self.assertEqual(ctr.source_dsa_guid, misc.GUID(fsmo_not_owner["ntds_guid"]))
280             self.assertEqual(ctr.source_dsa_invocation_id, misc.GUID(fsmo_not_owner["invocation_id"]))
281             ctr6 = ctr
282             self.assertEqual(ctr6.extended_ret, drsuapi.DRSUAPI_EXOP_ERR_SUCCESS)
283             self.assertEqual(ctr6.object_count, 3)
284             self.assertNotEqual(ctr6.first_object, None)
285             self.assertEqual(ldb.Dn(ldb_dc2, ctr6.first_object.object.identifier.dn), fsmo_dn)
286             self.assertNotEqual(ctr6.first_object.next_object, None)
287             self.assertNotEqual(ctr6.first_object.next_object.next_object, None)
288             second_object = ctr6.first_object.next_object.object
289             self.assertEqual(ldb.Dn(self.ldb_dc1, second_object.identifier.dn), fsmo_owner["rid_set_dn"])
290             third_object = ctr6.first_object.next_object.next_object.object
291             self.assertEqual(ldb.Dn(self.ldb_dc1, third_object.identifier.dn), fsmo_owner["server_acct_dn"])
292         finally:
293             # Swap the RID master back for other tests
294             m = ldb.Message()
295             m.dn = ldb.Dn(ldb_dc2, "")
296             m["becomeRidMaster"] = ldb.MessageElement("1", ldb.FLAG_MOD_REPLACE, "becomeRidMaster")
297             try:
298                 ldb_dc2.modify(m)
299             except ldb.LdbError, (num, msg):
300                 self.fail("Failed to restore RID Master " +  msg)
301
302     def test_offline_samba_tool_seized_ridalloc(self):
303         """Perform a join against the non-RID manager and then seize the RID Manager role"""
304
305         fsmo_dn = ldb.Dn(self.ldb_dc1, "CN=RID Manager$,CN=System," + self.ldb_dc1.domain_dn())
306         (fsmo_owner, fsmo_not_owner) = self._determine_fSMORoleOwner(fsmo_dn)
307
308         targetdir = self._test_join(fsmo_not_owner['dns_name'], "RIDALLOCTEST1")
309         try:
310             # Connect to the database
311             ldb_url = "tdb://%s" % os.path.join(targetdir, "private/sam.ldb")
312             smbconf = os.path.join(targetdir, "etc/smb.conf")
313
314             lp = self.get_loadparm()
315             new_ldb = SamDB(ldb_url, credentials=self.get_credentials(),
316                             session_info=system_session(lp), lp=lp)
317
318             # 1. Get server name
319             res = new_ldb.search(base=ldb.Dn(new_ldb, new_ldb.get_serverName()),
320                                  scope=ldb.SCOPE_BASE, attrs=["serverReference"])
321             # 2. Get server reference
322             server_ref_dn = ldb.Dn(new_ldb, res[0]['serverReference'][0])
323
324             # Assert that no RID Set has been set
325             res = new_ldb.search(base=server_ref_dn,
326                                  scope=ldb.SCOPE_BASE, attrs=['rIDSetReferences'])
327
328             self.assertFalse("rIDSetReferences" in res[0])
329
330             (result, out, err) = self.runsubcmd("fsmo", "seize", "--role", "rid", "-H", ldb_url, "-s", smbconf, "--force")
331             self.assertCmdSuccess(result, out, err)
332             self.assertEquals(err,"","Shouldn't be any error messages")
333
334             # 3. Assert we get the RID Set
335             res = new_ldb.search(base=server_ref_dn,
336                                  scope=ldb.SCOPE_BASE, attrs=['rIDSetReferences'])
337
338             self.assertTrue("rIDSetReferences" in res[0])
339         finally:
340             shutil.rmtree(targetdir, ignore_errors=True)
341             self._test_force_demote(fsmo_not_owner['dns_name'], "RIDALLOCTEST1")
342
343     def _test_join(self, server, netbios_name):
344         tmpdir = os.path.join(self.tempdir, "targetdir")
345         creds = self.get_credentials()
346         cmd = cmd_sambatool.subcommands['domain'].subcommands['join']
347         result = cmd._run("samba-tool domain join",
348                           creds.get_realm(),
349                           "dc", "-U%s%%%s" % (creds.get_username(),
350                                               creds.get_password()),
351                           '--targetdir=%s' % tmpdir,
352                           '--server=%s' % server,
353                           "--option=netbios name = %s" % netbios_name)
354         return tmpdir
355
356     def _test_force_demote(self, server, netbios_name):
357         creds = self.get_credentials()
358         cmd = cmd_sambatool.subcommands['domain'].subcommands['demote']
359         result = cmd._run("samba-tool domain demote",
360                           "-U%s%%%s" % (creds.get_username(),
361                                               creds.get_password()),
362                           '--server=%s' % server,
363                           "--remove-other-dead-server=%s" % netbios_name)
364
365     def test_offline_manual_seized_ridalloc_with_dbcheck(self):
366         """Peform the same actions as test_offline_samba_tool_seized_ridalloc,
367         but do not create the RID set. Confirm that dbcheck correctly creates
368         the RID Set.
369
370         Also check
371         """
372         fsmo_dn = ldb.Dn(self.ldb_dc1, "CN=RID Manager$,CN=System," + self.ldb_dc1.domain_dn())
373         (fsmo_owner, fsmo_not_owner) = self._determine_fSMORoleOwner(fsmo_dn)
374
375         targetdir = self._test_join(fsmo_not_owner['dns_name'], "RIDALLOCTEST2")
376         try:
377             # Connect to the database
378             ldb_url = "tdb://%s" % os.path.join(targetdir, "private/sam.ldb")
379             lp = self.get_loadparm()
380
381             new_ldb = SamDB(ldb_url, credentials=self.get_credentials(),
382                             session_info=system_session(lp), lp=lp)
383
384             serviceName = new_ldb.get_dsServiceName()
385             m = ldb.Message()
386             m.dn = fsmo_dn
387             m["fSMORoleOwner"] = ldb.MessageElement(serviceName,
388                                                    ldb.FLAG_MOD_REPLACE,
389                                                    "fSMORoleOwner")
390             new_ldb.modify(m)
391
392             # 1. Get server name
393             res = new_ldb.search(base=ldb.Dn(new_ldb, new_ldb.get_serverName()),
394                                  scope=ldb.SCOPE_BASE, attrs=["serverReference"])
395             # 2. Get server reference
396             server_ref_dn = ldb.Dn(new_ldb, res[0]['serverReference'][0])
397
398             # Assert that no RID Set has been set
399             res = new_ldb.search(base=server_ref_dn,
400                                  scope=ldb.SCOPE_BASE, attrs=['rIDSetReferences'])
401
402             self.assertFalse("rIDSetReferences" in res[0])
403
404             smbconf = os.path.join(targetdir, "etc/smb.conf")
405
406             chk = dbcheck(new_ldb, verbose=False, fix=True, yes=True, quiet=True)
407
408             self.assertEqual(chk.check_database(DN=server_ref_dn, scope=ldb.SCOPE_BASE), 1, "Should have fixed one error (missing RID Set)")
409
410             # 3. Assert we get the RID Set
411             res = new_ldb.search(base=server_ref_dn,
412                                  scope=ldb.SCOPE_BASE, attrs=['rIDSetReferences'])
413
414             self.assertTrue("rIDSetReferences" in res[0])
415         finally:
416             self._test_force_demote(fsmo_not_owner['dns_name'], "RIDALLOCTEST2")
417             shutil.rmtree(targetdir, ignore_errors=True)
418
419     def test_offline_manual_seized_ridalloc_add_user(self):
420         """Peform the same actions as test_offline_samba_tool_seized_ridalloc,
421         but do not create the RID set. Confirm that user-add correctly creates
422         the RID Set."""
423         fsmo_dn = ldb.Dn(self.ldb_dc1, "CN=RID Manager$,CN=System," + self.ldb_dc1.domain_dn())
424         (fsmo_owner, fsmo_not_owner) = self._determine_fSMORoleOwner(fsmo_dn)
425
426         targetdir = self._test_join(fsmo_not_owner['dns_name'], "RIDALLOCTEST3")
427         try:
428             # Connect to the database
429             ldb_url = "tdb://%s" % os.path.join(targetdir, "private/sam.ldb")
430             lp = self.get_loadparm()
431
432             new_ldb = SamDB(ldb_url, credentials=self.get_credentials(),
433                             session_info=system_session(lp), lp=lp)
434
435             serviceName = new_ldb.get_dsServiceName()
436             m = ldb.Message()
437             m.dn = fsmo_dn
438             m["fSMORoleOwner"] = ldb.MessageElement(serviceName,
439                                                    ldb.FLAG_MOD_REPLACE,
440                                                    "fSMORoleOwner")
441             new_ldb.modify(m)
442
443             # 1. Get server name
444             res = new_ldb.search(base=ldb.Dn(new_ldb, new_ldb.get_serverName()),
445                                  scope=ldb.SCOPE_BASE, attrs=["serverReference"])
446             # 2. Get server reference
447             server_ref_dn = ldb.Dn(new_ldb, res[0]['serverReference'][0])
448
449             # Assert that no RID Set has been set
450             res = new_ldb.search(base=server_ref_dn,
451                                  scope=ldb.SCOPE_BASE, attrs=['rIDSetReferences'])
452
453             self.assertFalse("rIDSetReferences" in res[0])
454
455             smbconf = os.path.join(targetdir, "etc/smb.conf")
456
457             new_ldb.newuser("ridalloctestuser", "P@ssword!")
458
459             # 3. Assert we get the RID Set
460             res = new_ldb.search(base=server_ref_dn,
461                                  scope=ldb.SCOPE_BASE, attrs=['rIDSetReferences'])
462
463             self.assertTrue("rIDSetReferences" in res[0])
464
465         finally:
466             self._test_force_demote(fsmo_not_owner['dns_name'], "RIDALLOCTEST3")
467             shutil.rmtree(targetdir, ignore_errors=True)
468
469     def test_offline_manual_seized_ridalloc_add_user_as_admin(self):
470         """Peform the same actions as test_offline_samba_tool_seized_ridalloc,
471         but do not create the RID set. Confirm that user-add correctly creates
472         the RID Set."""
473         fsmo_dn = ldb.Dn(self.ldb_dc1, "CN=RID Manager$,CN=System," + self.ldb_dc1.domain_dn())
474         (fsmo_owner, fsmo_not_owner) = self._determine_fSMORoleOwner(fsmo_dn)
475
476         targetdir = self._test_join(fsmo_not_owner['dns_name'], "RIDALLOCTEST4")
477         try:
478             # Connect to the database
479             ldb_url = "tdb://%s" % os.path.join(targetdir, "private/sam.ldb")
480             lp = self.get_loadparm()
481
482             new_ldb = SamDB(ldb_url, credentials=self.get_credentials(),
483                             session_info=admin_session(lp, self.ldb_dc1.get_domain_sid()), lp=lp)
484
485             serviceName = new_ldb.get_dsServiceName()
486             m = ldb.Message()
487             m.dn = fsmo_dn
488             m["fSMORoleOwner"] = ldb.MessageElement(serviceName,
489                                                    ldb.FLAG_MOD_REPLACE,
490                                                    "fSMORoleOwner")
491             new_ldb.modify(m)
492
493             # 1. Get server name
494             res = new_ldb.search(base=ldb.Dn(new_ldb, new_ldb.get_serverName()),
495                                  scope=ldb.SCOPE_BASE, attrs=["serverReference"])
496             # 2. Get server reference
497             server_ref_dn = ldb.Dn(new_ldb, res[0]['serverReference'][0])
498
499             # Assert that no RID Set has been set
500             res = new_ldb.search(base=server_ref_dn,
501                                  scope=ldb.SCOPE_BASE, attrs=['rIDSetReferences'])
502
503             self.assertFalse("rIDSetReferences" in res[0])
504
505             smbconf = os.path.join(targetdir, "etc/smb.conf")
506
507             # Create a user to allocate a RID Set for itself (the RID master)
508             new_ldb.newuser("ridalloctestuser", "P@ssword!")
509
510             # 3. Assert we get the RID Set
511             res = new_ldb.search(base=server_ref_dn,
512                                  scope=ldb.SCOPE_BASE, attrs=['rIDSetReferences'])
513
514             self.assertTrue("rIDSetReferences" in res[0])
515
516         finally:
517             self._test_force_demote(fsmo_not_owner['dns_name'], "RIDALLOCTEST4")
518             shutil.rmtree(targetdir, ignore_errors=True)
519
520     def test_join_time_ridalloc(self):
521         """Perform a join against the RID manager and assert we have a RID Set"""
522
523         fsmo_dn = ldb.Dn(self.ldb_dc1, "CN=RID Manager$,CN=System," + self.ldb_dc1.domain_dn())
524         (fsmo_owner, fsmo_not_owner) = self._determine_fSMORoleOwner(fsmo_dn)
525
526         targetdir = self._test_join(fsmo_owner['dns_name'], "RIDALLOCTEST5")
527         try:
528             # Connect to the database
529             ldb_url = "tdb://%s" % os.path.join(targetdir, "private/sam.ldb")
530             smbconf = os.path.join(targetdir, "etc/smb.conf")
531
532             lp = self.get_loadparm()
533             new_ldb = SamDB(ldb_url, credentials=self.get_credentials(),
534                             session_info=system_session(lp), lp=lp)
535
536             # 1. Get server name
537             res = new_ldb.search(base=ldb.Dn(new_ldb, new_ldb.get_serverName()),
538                                  scope=ldb.SCOPE_BASE, attrs=["serverReference"])
539             # 2. Get server reference
540             server_ref_dn = ldb.Dn(new_ldb, res[0]['serverReference'][0])
541
542             # 3. Assert we get the RID Set
543             res = new_ldb.search(base=server_ref_dn,
544                                  scope=ldb.SCOPE_BASE, attrs=['rIDSetReferences'])
545
546             self.assertTrue("rIDSetReferences" in res[0])
547         finally:
548             self._test_force_demote(fsmo_owner['dns_name'], "RIDALLOCTEST5")
549             shutil.rmtree(targetdir, ignore_errors=True)
550
551     def test_rid_set_dbcheck(self):
552         """Perform a join against the RID manager and assert we have a RID Set.
553         Using dbcheck, we assert that we can detect out of range users."""
554
555         fsmo_dn = ldb.Dn(self.ldb_dc1, "CN=RID Manager$,CN=System," + self.ldb_dc1.domain_dn())
556         (fsmo_owner, fsmo_not_owner) = self._determine_fSMORoleOwner(fsmo_dn)
557
558         targetdir = self._test_join(fsmo_owner['dns_name'], "RIDALLOCTEST6")
559         try:
560             # Connect to the database
561             ldb_url = "tdb://%s" % os.path.join(targetdir, "private/sam.ldb")
562             smbconf = os.path.join(targetdir, "etc/smb.conf")
563
564             lp = self.get_loadparm()
565             new_ldb = SamDB(ldb_url, credentials=self.get_credentials(),
566                             session_info=system_session(lp), lp=lp)
567
568             # 1. Get server name
569             res = new_ldb.search(base=ldb.Dn(new_ldb, new_ldb.get_serverName()),
570                                  scope=ldb.SCOPE_BASE, attrs=["serverReference"])
571             # 2. Get server reference
572             server_ref_dn = ldb.Dn(new_ldb, res[0]['serverReference'][0])
573
574             # 3. Assert we get the RID Set
575             res = new_ldb.search(base=server_ref_dn,
576                                  scope=ldb.SCOPE_BASE, attrs=['rIDSetReferences'])
577
578             self.assertTrue("rIDSetReferences" in res[0])
579             rid_set_dn = ldb.Dn(new_ldb, res[0]["rIDSetReferences"][0])
580
581             # 4. Add a new user (triggers RID set work)
582             new_ldb.newuser("ridalloctestuser", "P@ssword!")
583
584             # 5. Now fetch the RID SET
585             rid_set_res = new_ldb.search(base=rid_set_dn,
586                                          scope=ldb.SCOPE_BASE, attrs=['rIDNextRid',
587                                                                       'rIDAllocationPool'])
588             next_pool = int(rid_set_res[0]["rIDAllocationPool"][0])
589             last_rid = (0xFFFFFFFF00000000 & next_pool) >> 32
590
591             # 6. Add user above the ridNextRid and at mid-range.
592             #
593             # We can do this with safety because this is an offline DB that will be
594             # destroyed.
595             m = ldb.Message()
596             m.dn = ldb.Dn(new_ldb, "CN=ridsettestuser1,CN=Users")
597             m.dn.add_base(new_ldb.get_default_basedn())
598             m['objectClass'] = ldb.MessageElement('user', ldb.FLAG_MOD_ADD, 'objectClass')
599             m['objectSid'] = ldb.MessageElement(ndr_pack(security.dom_sid(str(new_ldb.get_domain_sid()) + "-%d" % (last_rid - 10))),
600                                                 ldb.FLAG_MOD_ADD,
601                                                 'objectSid')
602             new_ldb.add(m, controls=["relax:0"])
603
604             # 7. Check the RID Set
605             chk = dbcheck(new_ldb, verbose=False, fix=True, yes=True, quiet=True)
606
607             # Should have one error (wrong rIDNextRID)
608             self.assertEqual(chk.check_database(DN=rid_set_dn, scope=ldb.SCOPE_BASE), 1)
609
610             # 8. Assert we get didn't show any other errors
611             chk = dbcheck(new_ldb, verbose=False, fix=False, quiet=True)
612
613             rid_set_res = new_ldb.search(base=rid_set_dn,
614                                          scope=ldb.SCOPE_BASE, attrs=['rIDNextRid',
615                                                                       'rIDAllocationPool'])
616             last_allocated_rid = int(rid_set_res[0]["rIDNextRid"][0])
617             self.assertEquals(last_allocated_rid, last_rid - 10)
618
619             # 9. Assert that the range wasn't thrown away
620
621             next_pool = int(rid_set_res[0]["rIDAllocationPool"][0])
622             self.assertEqual(last_rid, (0xFFFFFFFF00000000 & next_pool) >> 32, "rid pool should have changed")
623         finally:
624             self._test_force_demote(fsmo_owner['dns_name'], "RIDALLOCTEST6")
625             shutil.rmtree(targetdir, ignore_errors=True)
626
627
628     def test_rid_set_dbcheck_after_seize(self):
629         """Perform a join against the RID manager and assert we have a RID Set.
630         We seize the RID master role, then using dbcheck, we assert that we can
631         detect out of range users (and then bump the RID set as required)."""
632
633         fsmo_dn = ldb.Dn(self.ldb_dc1, "CN=RID Manager$,CN=System," + self.ldb_dc1.domain_dn())
634         (fsmo_owner, fsmo_not_owner) = self._determine_fSMORoleOwner(fsmo_dn)
635
636         targetdir = self._test_join(fsmo_owner['dns_name'], "RIDALLOCTEST7")
637         try:
638             # Connect to the database
639             ldb_url = "tdb://%s" % os.path.join(targetdir, "private/sam.ldb")
640             smbconf = os.path.join(targetdir, "etc/smb.conf")
641
642             lp = self.get_loadparm()
643             new_ldb = SamDB(ldb_url, credentials=self.get_credentials(),
644                             session_info=system_session(lp), lp=lp)
645
646             # 1. Get server name
647             res = new_ldb.search(base=ldb.Dn(new_ldb, new_ldb.get_serverName()),
648                                  scope=ldb.SCOPE_BASE, attrs=["serverReference"])
649             # 2. Get server reference
650             server_ref_dn = ldb.Dn(new_ldb, res[0]['serverReference'][0])
651
652             # 3. Assert we get the RID Set
653             res = new_ldb.search(base=server_ref_dn,
654                                  scope=ldb.SCOPE_BASE, attrs=['rIDSetReferences'])
655
656             self.assertTrue("rIDSetReferences" in res[0])
657             rid_set_dn = ldb.Dn(new_ldb, res[0]["rIDSetReferences"][0])
658
659             # 4. Seize the RID Manager role
660             (result, out, err) = self.runsubcmd("fsmo", "seize", "--role", "rid", "-H", ldb_url, "-s", smbconf, "--force")
661             self.assertCmdSuccess(result, out, err)
662             self.assertEquals(err,"","Shouldn't be any error messages")
663
664             # 5. Add a new user (triggers RID set work)
665             new_ldb.newuser("ridalloctestuser", "P@ssword!")
666
667             # 6. Now fetch the RID SET
668             rid_set_res = new_ldb.search(base=rid_set_dn,
669                                          scope=ldb.SCOPE_BASE, attrs=['rIDNextRid',
670                                                                       'rIDAllocationPool'])
671             next_pool = int(rid_set_res[0]["rIDAllocationPool"][0])
672             last_rid = (0xFFFFFFFF00000000 & next_pool) >> 32
673
674             # 7. Add user above the ridNextRid and at almost the end of the range.
675             #
676             m = ldb.Message()
677             m.dn = ldb.Dn(new_ldb, "CN=ridsettestuser2,CN=Users")
678             m.dn.add_base(new_ldb.get_default_basedn())
679             m['objectClass'] = ldb.MessageElement('user', ldb.FLAG_MOD_ADD, 'objectClass')
680             m['objectSid'] = ldb.MessageElement(ndr_pack(security.dom_sid(str(new_ldb.get_domain_sid()) + "-%d" % (last_rid - 3))),
681                                                 ldb.FLAG_MOD_ADD,
682                                                 'objectSid')
683             new_ldb.add(m, controls=["relax:0"])
684
685             # 8. Add user above the ridNextRid and at the end of the range
686             m = ldb.Message()
687             m.dn = ldb.Dn(new_ldb, "CN=ridsettestuser3,CN=Users")
688             m.dn.add_base(new_ldb.get_default_basedn())
689             m['objectClass'] = ldb.MessageElement('user', ldb.FLAG_MOD_ADD, 'objectClass')
690             m['objectSid'] = ldb.MessageElement(ndr_pack(security.dom_sid(str(new_ldb.get_domain_sid()) + "-%d" % last_rid)),
691                                                 ldb.FLAG_MOD_ADD,
692                                                 'objectSid')
693             new_ldb.add(m, controls=["relax:0"])
694
695             chk = dbcheck(new_ldb, verbose=False, fix=True, yes=True, quiet=True)
696
697             # Should have fixed two errors (wrong ridNextRid)
698             self.assertEqual(chk.check_database(DN=rid_set_dn, scope=ldb.SCOPE_BASE), 2)
699
700             # 9. Assert we get didn't show any other errors
701             chk = dbcheck(new_ldb, verbose=False, fix=False, quiet=True)
702
703             # 10. Add another user (checks RID rollover)
704             # We have seized the role, so we can do that.
705             new_ldb.newuser("ridalloctestuser3", "P@ssword!")
706
707             rid_set_res = new_ldb.search(base=rid_set_dn,
708                                          scope=ldb.SCOPE_BASE, attrs=['rIDNextRid',
709                                                                       'rIDAllocationPool'])
710             next_pool = int(rid_set_res[0]["rIDAllocationPool"][0])
711             self.assertNotEqual(last_rid, (0xFFFFFFFF00000000 & next_pool) >> 32, "rid pool should have changed")
712         finally:
713             self._test_force_demote(fsmo_owner['dns_name'], "RIDALLOCTEST7")
714             shutil.rmtree(targetdir, ignore_errors=True)