2 # -*- coding: utf-8 -*-
4 # Unix SMB/CIFS implementation.
5 # Copyright (C) Kamen Mazdrashki <kamenim@samba.org> 2010
7 # This program is free software; you can redistribute it and/or modify
8 # it under the terms of the GNU General Public License as published by
9 # the Free Software Foundation; either version 3 of the License, or
10 # (at your option) any later version.
12 # This program is distributed in the hope that it will be useful,
13 # but WITHOUT ANY WARRANTY; without even the implied warranty of
14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 # GNU General Public License for more details.
17 # You should have received a copy of the GNU General Public License
18 # along with this program. If not, see <http://www.gnu.org/licenses/>.
23 # export DC1=dc1_dns_name
24 # export DC2=dc2_dns_name
25 # export SUBUNITRUN=$samba4srcdir/scripting/bin/subunitrun
26 # PYTHONPATH="$PYTHONPATH:$samba4srcdir/torture/drs/python" $SUBUNITRUN delete_object -U"$DOMAIN/$DC_USERNAME"%"$DC_PASSWORD"
33 sys
.path
.append("bin/python")
35 from samba
.auth
import system_session
36 from ldb
import SCOPE_BASE
, SCOPE_SUBTREE
37 from samba
.samdb
import SamDB
42 class DrsDeleteObjectTestCase(samba
.tests
.TestCase
):
52 super(DrsDeleteObjectTestCase
, self
).setUp()
54 # connect to DCs singleton
55 if self
.ldb_dc1
is None:
56 DrsDeleteObjectTestCase
.dc1
= get_env_var("DC1")
57 DrsDeleteObjectTestCase
.ldb_dc1
= connect_samdb(self
.dc1
)
58 if self
.ldb_dc2
is None:
59 DrsDeleteObjectTestCase
.dc2
= get_env_var("DC2")
60 DrsDeleteObjectTestCase
.ldb_dc2
= connect_samdb(self
.dc2
)
63 if self
.info_dc1
is None:
65 res
= ldb
.search(base
="", expression
="", scope
=SCOPE_BASE
, attrs
=["*"])
66 self
.assertEquals(len(res
), 1)
67 DrsDeleteObjectTestCase
.info_dc1
= res
[0]
68 if self
.info_dc2
is None:
70 res
= ldb
.search(base
="", expression
="", scope
=SCOPE_BASE
, attrs
=["*"])
71 self
.assertEquals(len(res
), 1)
72 DrsDeleteObjectTestCase
.info_dc2
= res
[0]
74 # cache some of RootDSE props
75 self
.schema_dn
= self
.info_dc1
["schemaNamingContext"][0]
76 self
.domain_dn
= self
.info_dc1
["defaultNamingContext"][0]
77 self
.config_dn
= self
.info_dc1
["configurationNamingContext"][0]
78 self
.forest_level
= int(self
.info_dc1
["forestFunctionality"][0])
80 # we will need DCs DNS names for 'net drs' command
81 self
.dnsname_dc1
= self
.info_dc1
["dnsHostName"][0]
82 self
.dnsname_dc2
= self
.info_dc2
["dnsHostName"][0]
86 super(DrsDeleteObjectTestCase
, self
).tearDown()
88 def _GUID_string(self
, guid
):
89 return self
.ldb_dc1
.schema_format_value("objectGUID", guid
)
91 def _deleted_objects_dn(self
, sam_ldb
):
92 wkdn
= "<WKGUID=18E2EA80684F11D2B9AA00C04F79F805,%s>" % self
.domain_dn
93 res
= sam_ldb
.search(base
=wkdn
,
95 controls
=["show_deleted:1"])
96 self
.assertEquals(len(res
), 1)
97 return str(res
[0]["dn"])
99 def _make_username(self
):
100 return "DrsDelObjUser_" + time
.strftime("%s", time
.gmtime())
102 def _check_user(self
, sam_ldb
, user_orig
, is_deleted
):
103 # search the user by guid as it may be deleted
104 guid_str
= self
._GUID
_string
(user_orig
["objectGUID"][0])
105 expression
= "(objectGUID=%s)" % guid_str
106 res
= sam_ldb
.search(base
=self
.domain_dn
,
107 expression
=expression
,
108 controls
=["show_deleted:1"])
109 self
.assertEquals(len(res
), 1)
111 # Deleted Object base DN
112 dodn
= self
._deleted
_objects
_dn
(sam_ldb
)
113 # now check properties of the user
114 name_orig
= user_orig
["cn"][0]
115 name_cur
= user_cur
["cn"][0]
117 self
.assertEquals(user_cur
["isDeleted"][0],"TRUE")
118 self
.assertTrue(not("objectCategory" in user_cur
))
119 self
.assertTrue(not("sAMAccountType" in user_cur
))
120 self
.assertTrue(dodn
in str(user_cur
["dn"]),
121 "User %s is deleted but it is not located under %s!" % (name_orig
, dodn
))
122 self
.assertEquals(name_cur
, name_orig
+ "\nDEL:" + guid_str
)
124 self
.assertTrue(not("isDeleted" in user_cur
))
125 self
.assertEquals(name_cur
, name_orig
)
126 self
.assertEquals(user_orig
["dn"], user_cur
["dn"])
127 self
.assertTrue(dodn
not in str(user_cur
["dn"]))
130 def _net_drs_replicate(self
, DC
, fromDC
):
131 # find out where is net command
132 net_cmd
= os
.path
.abspath("./bin/net")
133 # make command line credentials string
134 creds
= samba
.tests
.cmdline_credentials
135 cmd_line_auth
= "-U%s/%s%%%s" % (creds
.get_domain(),
136 creds
.get_username(), creds
.get_password())
137 # bin/net drs replicate <Dest_DC_NAME> <Src_DC_NAME> <Naming Context>
138 cmd_line
= "%s drs replicate %s %s %s %s" % (net_cmd
, DC
, fromDC
,
139 self
.domain_dn
, cmd_line_auth
)
140 ret
= os
.system(cmd_line
)
141 self
.assertEquals(ret
, 0, "Replicating %s from %s has failed!" % (DC
, fromDC
))
145 def test_NetReplicateCmd(self
):
146 """Triggers replication from DC1 to DC2
147 and vice versa so both DCs are synchronized
148 before test_ReplicateDeteleteObject test"""
149 # replicate Domain NC on DC2 from DC1
150 self
._net
_drs
_replicate
(DC
=self
.dnsname_dc2
, fromDC
=self
.dnsname_dc1
)
151 # replicate Domain NC on DC1 from DC2
152 self
._net
_drs
_replicate
(DC
=self
.dnsname_dc1
, fromDC
=self
.dnsname_dc2
)
155 def test_ReplicateDeteleteObject(self
):
156 """Verifies how a deleted-object is replicated between two DCs.
157 This test should verify that:
158 - deleted-object is replicated properly
159 TODO: We should verify that after replication,
160 object's state to conform to a deleted-object state
161 or tombstone -object, depending on DC's features
162 It will also be great if check replPropertyMetaData."""
163 # work-out unique username to test with
164 username
= self
._make
_username
()
167 self
.ldb_dc1
.newuser(username
=username
, password
="P@sswOrd!")
168 ldb_res
= self
.ldb_dc1
.search(base
=self
.domain_dn
,
170 expression
="(samAccountName=%s)" % username
)
171 self
.assertEquals(len(ldb_res
), 1)
172 user_orig
= ldb_res
[0]
173 user_dn
= ldb_res
[0]["dn"]
175 # check user info on DC1
176 print "Testing for %s with GUID %s" % (username
, self
._GUID
_string
(user_orig
["objectGUID"][0]))
177 self
._check
_user
(sam_ldb
=self
.ldb_dc1
, user_orig
=user_orig
, is_deleted
=False)
179 # trigger replication from DC1 to DC2
180 self
._net
_drs
_replicate
(DC
=self
.dnsname_dc2
, fromDC
=self
.dnsname_dc1
)
183 self
.ldb_dc1
.delete(user_dn
)
184 # check user info on DC1 - should be deleted
185 self
._check
_user
(sam_ldb
=self
.ldb_dc1
, user_orig
=user_orig
, is_deleted
=True)
186 # check user info on DC2 - should be valid user
188 self
._check
_user
(sam_ldb
=self
.ldb_dc2
, user_orig
=user_orig
, is_deleted
=False)
189 except self
.failureException
:
190 print ("Checking for not isDeleted user on %s failed, "
191 "probably because a replication took place. "
192 "Ideally we should block automatic replications during this test, "
193 "but until then, just ignore the error" % self
.dnsname_dc2
)
195 # trigger replication from DC2 to DC1
196 # to check if deleted object gets restored
197 self
._net
_drs
_replicate
(DC
=self
.dnsname_dc1
, fromDC
=self
.dnsname_dc2
)
198 # check user info on DC1 - should be deleted
199 self
._check
_user
(sam_ldb
=self
.ldb_dc1
, user_orig
=user_orig
, is_deleted
=True)
200 # check user info on DC2 - should be valid user
202 self
._check
_user
(sam_ldb
=self
.ldb_dc2
, user_orig
=user_orig
, is_deleted
=False)
203 except self
.failureException
:
204 print ("Checking for not isDeleted user on %s failed, "
205 "probably because a replication took place. "
206 "Ideally we should block automatic replications during this test, "
207 "but until then, just ignore the error" % self
.dnsname_dc2
)
209 # trigger replication from DC1 to DC2
210 # to check if deleted object is replicated
211 self
._net
_drs
_replicate
(DC
=self
.dnsname_dc2
, fromDC
=self
.dnsname_dc1
)
212 # check user info on DC1 - should be deleted
213 self
._check
_user
(sam_ldb
=self
.ldb_dc1
, user_orig
=user_orig
, is_deleted
=True)
214 # check user info on DC2 - should be deleted
215 self
._check
_user
(sam_ldb
=self
.ldb_dc2
, user_orig
=user_orig
, is_deleted
=True)
220 ########################################################################################
221 def get_env_var(var_name
):
222 if not var_name
in os
.environ
.keys():
223 raise AssertionError("Please supply %s in environment" % var_name
)
224 return os
.environ
[var_name
]
226 def connect_samdb(samdb_url
):
228 if not "://" in samdb_url
:
229 if os
.path
.isfile(samdb_url
):
230 samdb_url
= "tdb://%s" % samdb_url
232 samdb_url
= "ldap://%s" % samdb_url
233 # user 'paged_search' module when connecting remotely
234 ldb_options
= ["modules:paged_searches"]
236 return SamDB(url
=samdb_url
,
237 lp
=samba
.tests
.env_loadparm(),
238 session_info
=system_session(),
239 credentials
=samba
.tests
.cmdline_credentials
,