1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
|
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Unix SMB/CIFS implementation.
# Copyright (C) Anatoliy Atanasov <anatoliy.atanasov@postpath.com> 2010
#
# 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/>.
#
#
# Usage:
# export DC1=dc1_dns_name
# export DC2=dc2_dns_name
# export SUBUNITRUN=$samba4srcdir/scripting/bin/subunitrun
# PYTHONPATH="$PYTHONPATH:$samba4srcdir/torture/drs/python" $SUBUNITRUN fsmo -U"$DOMAIN/$DC_USERNAME"%"$DC_PASSWORD"
#
import sys
import time
import os
sys.path.append("bin/python")
from ldb import SCOPE_BASE
import samba.tests
class DrsFsmoTestCase(samba.tests.TestCase):
# RootDSE msg for DC1
info_dc1 = None
ldb_dc1 = None
# RootDSE msg for DC1
info_dc2 = None
ldb_dc2 = None
def setUp(self):
super(DrsFsmoTestCase, self).setUp()
# we have to wait for the replication before we make the check
self.fsmo_wait_max_time = 20
self.fsmo_wait_sleep_time = 0.2
# connect to DCs singleton
if self.ldb_dc1 is None:
DrsFsmoTestCase.dc1 = samba.tests.env_get_var_value("DC1")
DrsFsmoTestCase.ldb_dc1 = samba.tests.connect_samdb(self.dc1, ldap_only=True)
if self.ldb_dc2 is None:
DrsFsmoTestCase.dc2 = samba.tests.env_get_var_value("DC2")
DrsFsmoTestCase.ldb_dc2 = samba.tests.connect_samdb(self.dc2, ldap_only=True)
# fetch rootDSEs
if self.info_dc1 is None:
ldb = self.ldb_dc1
res = ldb.search(base="", expression="", scope=SCOPE_BASE, attrs=["*"])
self.assertEquals(len(res), 1)
DrsFsmoTestCase.info_dc1 = res[0]
if self.info_dc2 is None:
ldb = self.ldb_dc2
res = ldb.search(base="", expression="", scope=SCOPE_BASE, attrs=["*"])
self.assertEquals(len(res), 1)
DrsFsmoTestCase.info_dc2 = res[0]
# cache some of RootDSE props
self.schema_dn = self.info_dc1["schemaNamingContext"][0]
self.domain_dn = self.info_dc1["defaultNamingContext"][0]
self.config_dn = self.info_dc1["configurationNamingContext"][0]
self.dsServiceName_dc1 = self.info_dc1["dsServiceName"][0]
self.dsServiceName_dc2 = self.info_dc2["dsServiceName"][0]
self.infrastructure_dn = "CN=Infrastructure," + self.domain_dn
self.naming_dn = "CN=Partitions," + self.config_dn
self.rid_dn = "CN=RID Manager$,CN=System," + self.domain_dn
# we will need DCs DNS names for 'net fsmo' command
self.dnsname_dc1 = self.info_dc1["dnsHostName"][0]
self.dnsname_dc2 = self.info_dc2["dnsHostName"][0]
def tearDown(self):
super(DrsFsmoTestCase, self).tearDown()
def _net_fsmo_role_transfer(self, DC, role):
# find out where is samba-tool command
net_cmd = os.path.abspath("./bin/samba-tool")
# make command line credentials string
creds = self.get_credentials()
cmd_line_auth = "-U%s/%s%%%s" % (creds.get_domain(),
creds.get_username(), creds.get_password())
# bin/samba-tool fsmo transfer --role=role --host=ldap://DC:389
cmd_line = "%s fsmo transfer --role=%s --host=ldap://%s:389 %s" % (net_cmd, role, DC,
cmd_line_auth)
ret = os.system(cmd_line)
self.assertEquals(ret, 0, "Transferring role %s to %s has failed!" % (role, DC))
def _wait_for_role_transfer(self, ldb_dc, role_dn, master):
"""Wait for role transfer for certain amount of time
:return: (Result=True|False, CurrentMasterDnsName) tuple
"""
cur_master = ''
retries = int(self.fsmo_wait_max_time / self.fsmo_wait_sleep_time) + 1
for i in range(0, retries):
# check if master has been transfered
res = ldb_dc.search(role_dn,
scope=SCOPE_BASE, attrs=["fSMORoleOwner"])
assert len(res) == 1, "Only one fSMORoleOwner value expected!"
cur_master = res[0]["fSMORoleOwner"][0]
if master == cur_master:
return (True, cur_master)
# skip last sleep, if no need to wait anymore
if i != (retries - 1):
# wait a little bit before next retry
time.sleep(self.fsmo_wait_sleep_time)
return (False, cur_master)
def _role_transfer(self, role, role_dn):
"""Triggers transfer of role from DC1 to DC2
and vice versa so the role goes back to the original dc"""
# dc2 gets the role from dc1
print "Testing for %s role transfer from %s to %s" % (role, self.dnsname_dc1, self.dnsname_dc2)
self._net_fsmo_role_transfer(DC=self.dnsname_dc2, role=role)
# check if the role is transfered
(res, master) = self._wait_for_role_transfer(ldb_dc=self.ldb_dc2,
role_dn=role_dn,
master=self.dsServiceName_dc2)
self.assertTrue(res,
"Transferring %s role to %s has failed, master is: %s!"%(role, self.dsServiceName_dc2, master))
# dc1 gets back the role from dc2
print "Testing for %s role transfer from %s to %s" % (role, self.dnsname_dc2, self.dnsname_dc1)
self._net_fsmo_role_transfer(DC=self.dnsname_dc1, role=role)
# check if the role is transfered
(res, master) = self._wait_for_role_transfer(ldb_dc=self.ldb_dc1,
role_dn=role_dn,
master=self.dsServiceName_dc1)
self.assertTrue(res,
"Transferring %s role to %s has failed, master is: %s!"%(role, self.dsServiceName_dc1, master))
def test_SchemaMasterTransfer(self):
self._role_transfer(role="schema", role_dn=self.schema_dn)
def test_InfrastructureMasterTransfer(self):
self._role_transfer(role="infrastructure", role_dn=self.infrastructure_dn)
def test_PDCMasterTransfer(self):
self._role_transfer(role="pdc", role_dn=self.domain_dn)
def test_RIDMasterTransfer(self):
self._role_transfer(role="rid", role_dn=self.rid_dn)
def test_NamingMasterTransfer(self):
self._role_transfer(role="naming", role_dn=self.naming_dn)
|
