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
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
|
#!/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 samba.auth import system_session
from ldb import SCOPE_BASE
from samba.samdb import SamDB
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.sleep_time = 5
# connect to DCs singleton
if self.ldb_dc1 is None:
DrsFsmoTestCase.dc1 = get_env_var("DC1")
DrsFsmoTestCase.ldb_dc1 = connect_samdb(self.dc1)
if self.ldb_dc2 is None:
DrsFsmoTestCase.dc2 = get_env_var("DC2")
DrsFsmoTestCase.ldb_dc2 = connect_samdb(self.dc2)
# 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]
pass
def tearDown(self):
super(DrsFsmoTestCase, self).tearDown()
def _net_fsmo_role_transfer(self, DC, role):
# find out where is net command
net_cmd = os.path.abspath("./bin/net")
# make command line credentials string
creds = samba.tests.cmdline_credentials
cmd_line_auth = "-U%s/%s%%%s" % (creds.get_domain(),
creds.get_username(), creds.get_password())
# bin/net 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, "Transfering schema to %s has failed!" % (DC))
pass
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 schema master 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, but wait a little first so the getncchanges can pass
time.sleep(self.sleep_time)
res = self.ldb_dc2.search(role_dn,
scope=SCOPE_BASE, attrs=["fSMORoleOwner"])
assert len(res) == 1
self.master = res[0]["fSMORoleOwner"][0]
self.assertEquals(self.master, self.dsServiceName_dc2,
"Transfering %s role to %s has failed, master is: %s!"%(role, self.dsServiceName_dc2,self.master))
# dc1 gets back the schema master 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
time.sleep(self.sleep_time)
res = self.ldb_dc1.search(role_dn,
scope=SCOPE_BASE, attrs=["fSMORoleOwner"])
assert len(res) == 1
self.master = res[0]["fSMORoleOwner"][0]
self.assertEquals(self.master, self.dsServiceName_dc1,
"Transfering %s role to %s has failed, master is %s"%(role, self.dsServiceName_dc1, self.master))
pass
def test_SchemaMasterTransfer(self):
self._role_transfer(role="schema", role_dn=self.schema_dn)
pass
def test_InfrastructureMasterTransfer(self):
self._role_transfer(role="infrastructure", role_dn=self.infrastructure_dn)
pass
def test_PDCMasterTransfer(self):
self._role_transfer(role="pdc", role_dn=self.domain_dn)
pass
def test_RIDMasterTransfer(self):
self._role_transfer(role="rid", role_dn=self.rid_dn)
pass
def test_NamingMasterTransfer(self):
self._role_transfer(role="naming", role_dn=self.naming_dn)
pass
########################################################################################
def get_env_var(var_name):
if not var_name in os.environ.keys():
raise AssertionError("Please supply %s in environment" % var_name)
return os.environ[var_name]
def connect_samdb(samdb_url):
ldb_options = []
if not "://" in samdb_url:
if os.path.isfile(samdb_url):
samdb_url = "tdb://%s" % samdb_url
else:
samdb_url = "ldap://%s:389" % samdb_url
# use 'paged_search' module when connecting remotely
if samdb_url.lower().startswith("ldap://"):
ldb_options = ["modules:paged_searches"]
return SamDB(url=samdb_url,
lp=samba.tests.env_loadparm(),
session_info=system_session(),
credentials=samba.tests.cmdline_credentials,
options=ldb_options)
|
