source4/torture/drs/python/samba_tool_drs.py

   1 # Blackbox tests for "samba-tool drs" command
   2 # Copyright (C) Kamen Mazdrashki <kamenim@samba.org> 2011
   3 # Copyright (C) Andrew Bartlett <abartlet@samba.org> 2017
   4 #
   5 # This program is free software; you can redistribute it and/or modify
   6 # it under the terms of the GNU General Public License as published by
   7 # the Free Software Foundation; either version 3 of the License, or
   8 # (at your option) any later version.
   9 #
  10 # This program is distributed in the hope that it will be useful,
  11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 # GNU General Public License for more details.
  14 #
  15 # You should have received a copy of the GNU General Public License
  16 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17 #
  18
  19 """Blackbox tests for samba-tool drs."""
  20
  21 import samba.tests
  22 import shutil
  23 import os
  24 import ldb
  25 import drs_base
  26
  27 class SambaToolDrsTests(drs_base.DrsBaseTestCase):
  28     """Blackbox test case for samba-tool drs."""
  29
  30     def setUp(self):
  31         super(SambaToolDrsTests, self).setUp()
  32
  33         self.dc1 = samba.tests.env_get_var_value("DC1")
  34         self.dc2 = samba.tests.env_get_var_value("DC2")
  35
  36         creds = self.get_credentials()
  37         self.cmdline_creds = "-U%s/%s%%%s" % (creds.get_domain(),
  38                                               creds.get_username(), creds.get_password())
  39
  40     def tearDown(self):
  41         self._enable_inbound_repl(self.dnsname_dc1)
  42         self._enable_inbound_repl(self.dnsname_dc2)
  43
  44         try:
  45             shutil.rmtree(os.path.join(self.tempdir, "private"))
  46             shutil.rmtree(os.path.join(self.tempdir, "etc"))
  47             shutil.rmtree(os.path.join(self.tempdir, "msg.lock"))
  48             os.remove(os.path.join(self.tempdir, "names.tdb"))
  49             shutil.rmtree(os.path.join(self.tempdir, "state"))
  50             shutil.rmtree(os.path.join(self.tempdir, "bind-dns"))
  51         except Exception:
  52             pass
  53
  54         super(SambaToolDrsTests, self).tearDown()
  55
  56     def _get_rootDSE(self, dc, ldap_only=True):
  57         samdb = samba.tests.connect_samdb(dc, lp=self.get_loadparm(),
  58                                           credentials=self.get_credentials(),
  59                                           ldap_only=ldap_only)
  60         return samdb.search(base="", scope=samba.tests.ldb.SCOPE_BASE)[0]
  61
  62     def test_samba_tool_bind(self):
  63         """Tests 'samba-tool drs bind' command."""
  64
  65         # Output should be like:
  66         #      Extensions supported:
  67         #        <list-of-supported-extensions>
  68         #      Site GUID: <GUID>
  69         #      Repl epoch: 0
  70         out = self.check_output("samba-tool drs bind %s %s" % (self.dc1,
  71                                                                self.cmdline_creds))
  72         self.assertTrue("Site GUID:" in out)
  73         self.assertTrue("Repl epoch:" in out)
  74
  75     def test_samba_tool_kcc(self):
  76         """Tests 'samba-tool drs kcc' command."""
  77
  78         # Output should be like 'Consistency check on <DC> successful.'
  79         out = self.check_output("samba-tool drs kcc %s %s" % (self.dc1,
  80                                                               self.cmdline_creds))
  81         self.assertTrue("Consistency check on" in out)
  82         self.assertTrue("successful" in out)
  83
  84     def test_samba_tool_options(self):
  85         """Tests 'samba-tool drs options' command
  86         """
  87         # Output should be like 'Current DSA options: IS_GC <OTHER_FLAGS>'
  88         out = self.check_output("samba-tool drs options %s %s" % (self.dc1,
  89                                                                   self.cmdline_creds))
  90         self.assertTrue("Current DSA options:" in out)
  91
  92     def test_samba_tool_replicate(self):
  93         """Tests 'samba-tool drs replicate' command."""
  94
  95         # Output should be like 'Replicate from <DC-SRC> to <DC-DEST> was successful.'
  96         nc_name = self._get_rootDSE(self.dc1)["defaultNamingContext"]
  97         out = self.check_output("samba-tool drs replicate %s %s %s %s" % (self.dc1,
  98                                                                           self.dc2,
  99                                                                           nc_name,
 100                                                                           self.cmdline_creds))
 101         self.assertTrue("Replicate from" in out)
 102         self.assertTrue("was successful" in out)
 103
 104     def test_samba_tool_replicate_async(self):
 105         """Tests 'samba-tool drs replicate --async-op' command."""
 106
 107         # Output should be like 'Replicate from <DC-SRC> to <DC-DEST> was started.'
 108         nc_name = self._get_rootDSE(self.dc1)["defaultNamingContext"]
 109         out = self.check_output("samba-tool drs replicate --async-op %s %s %s %s" % (self.dc1,
 110                                                                                      self.dc2,
 111                                                                                      nc_name,
 112                                                                                      self.cmdline_creds))
 113         self.assertTrue("Replicate from" in out)
 114         self.assertTrue("was started" in out)
 115
 116     def test_samba_tool_replicate_local_online(self):
 117         """Tests 'samba-tool drs replicate --local-online' command."""
 118
 119         # Output should be like 'Replicate from <DC-SRC> to <DC-DEST> was successful.'
 120         nc_name = self._get_rootDSE(self.dc1)["defaultNamingContext"]
 121         out = self.check_output("samba-tool drs replicate --local-online %s %s %s" % (self.dc1,
 122                                                                                       self.dc2,
 123                                                                                       nc_name))
 124         self.assertTrue("Replicate from" in out)
 125         self.assertTrue("was successful" in out)
 126
 127     def test_samba_tool_replicate_local_online_async(self):
 128         """Tests 'samba-tool drs replicate --local-online --async-op' command."""
 129
 130         # Output should be like 'Replicate from <DC-SRC> to <DC-DEST> was started.'
 131         nc_name = self._get_rootDSE(self.dc1)["defaultNamingContext"]
 132         out = self.check_output("samba-tool drs replicate --local-online --async-op %s %s %s" % (self.dc1,
 133                                                                                                  self.dc2,
 134                                                                                                  nc_name))
 135         self.assertTrue("Replicate from" in out)
 136         self.assertTrue("was started" in out)
 137
 138     def test_samba_tool_replicate_local_machine_creds(self):
 139         """Tests 'samba-tool drs replicate --local -P' command (uses machine creds)."""
 140
 141         # Output should be like 'Replicate from <DC-SRC> to <DC-DEST> was successful.'
 142         nc_name = self._get_rootDSE(self.dc1)["defaultNamingContext"]
 143         out = self.check_output("samba-tool drs replicate -P --local %s %s %s" % (self.dc1,
 144                                                                                   self.dc2,
 145                                                                                   nc_name))
 146         self.assertTrue("Incremental" in out)
 147         self.assertTrue("was successful" in out)
 148
 149     def test_samba_tool_replicate_local(self):
 150         """Tests 'samba-tool drs replicate --local' command (uses machine creds)."""
 151
 152         # Output should be like 'Replicate from <DC-SRC> to <DC-DEST> was successful.'
 153         nc_name = self._get_rootDSE(self.dc1)["defaultNamingContext"]
 154
 155         def get_num_obj_links(output):
 156             num_objs = None
 157             num_links = None
 158             for word in output.split(" "):
 159                 try:
 160                     int(word)
 161                     if num_objs is None:
 162                         num_objs = int(word)
 163                     elif num_links is None:
 164                         num_links = int(word)
 165                 except ValueError:
 166                     pass
 167
 168             return (num_objs, num_links)
 169
 170         out = self.check_output("samba-tool drs replicate --local --full-sync %s %s %s %s"
 171                                 % (self.dc1, self.dc2, nc_name, self.cmdline_creds))
 172         self.assertTrue("was successful" in out)
 173         self.assertTrue("Full" in out)
 174
 175         (first_obj, _) = get_num_obj_links(out)
 176
 177         out = self.check_output("samba-tool drs replicate --local %s %s %s %s"
 178                                 % (self.dc1, self.dc2, nc_name, self.cmdline_creds))
 179         self.assertTrue("was successful" in out)
 180         self.assertTrue("Incremental" in out)
 181
 182         (second_obj, _) = get_num_obj_links(out)
 183
 184         self.assertTrue(first_obj > second_obj)
 185
 186         server_rootdse = self._get_rootDSE(self.dc1)
 187         server_nc_name = server_rootdse["defaultNamingContext"]
 188         server_ds_name = server_rootdse["dsServiceName"]
 189         server_ldap_service_name = str(server_rootdse["ldapServiceName"][0])
 190         server_realm = server_ldap_service_name.split(":")[0]
 191         creds = self.get_credentials()
 192
 193         # We have to give it a different netbiosname every time
 194         # it runs, otherwise the collision causes strange issues
 195         # to happen. This should be different on different environments.
 196         netbiosname = "test" + self.dc2
 197         if len(netbiosname) > 15:
 198             netbiosname = netbiosname[:15]
 199
 200         out = self.check_output("samba-tool domain join %s dc --server=%s %s --targetdir=%s --option=netbiosname=%s"
 201                                 % (server_realm, self.dc1, self.cmdline_creds, self.tempdir, netbiosname))
 202
 203         new_dc_config_file = "%s/etc/smb.conf" % self.tempdir
 204
 205         self.check_output("samba-tool drs replicate --local %s %s %s %s -s %s"
 206                           % ("invalid", self.dc1, nc_name,
 207                              self.cmdline_creds, new_dc_config_file))
 208
 209         self._disable_inbound_repl(self.dnsname_dc1)
 210         self._disable_inbound_repl(self.dnsname_dc2)
 211
 212         self._net_drs_replicate(DC=self.dnsname_dc1, fromDC=self.dnsname_dc2)
 213
 214         # add an object with link on dc1
 215         group_name = "group-repl-local-%s" % self.dc2
 216         user_name = "user-repl-local-%s" % self.dc2
 217
 218         self.check_output("samba-tool group add %s %s -H ldap://%s"
 219                           % (group_name, self.cmdline_creds, self.dc1))
 220         self.check_output("samba-tool user add %s %s --random-password -H ldap://%s"
 221                           % (user_name, self.cmdline_creds, self.dc1))
 222         self.check_output("samba-tool group addmembers %s %s %s -H ldap://%s"
 223                           % (group_name, user_name, self.cmdline_creds, self.dc1))
 224
 225         self._net_drs_replicate(DC=self.dnsname_dc2, fromDC=self.dnsname_dc1)
 226
 227         # pull that change with --local into local db from dc1: should send link and some objects
 228         out = self.check_output("samba-tool drs replicate --local %s %s %s %s -s %s"
 229                                 % ("invalid", self.dc1, nc_name,
 230                                    self.cmdline_creds, new_dc_config_file))
 231
 232         (obj_1, link_1) = get_num_obj_links(out)
 233
 234         self.assertEqual(obj_1, 2)
 235         self.assertEqual(link_1, 1)
 236
 237         # pull that change with --local into local db from dc2: shouldn't send link or object
 238         # as we sent an up-to-dateness vector showing that we had already synced with DC1
 239         out = self.check_output("samba-tool drs replicate --local %s %s %s %s -s %s"
 240                                 % ("invalid", self.dc2, nc_name,
 241                                    self.cmdline_creds, new_dc_config_file))
 242
 243         (obj_2, link_2) = get_num_obj_links(out)
 244
 245         self.assertEqual(obj_2, 0)
 246         self.assertEqual(link_2, 0)
 247
 248         self.check_output("samba-tool domain demote --remove-other-dead-server=%s -H ldap://%s %s -s %s"
 249                           % (netbiosname, self.dc1, self.cmdline_creds, new_dc_config_file))
 250
 251     def test_samba_tool_replicate_machine_creds_P(self):
 252         """Tests 'samba-tool drs replicate -P' command with machine creds."""
 253
 254         # Output should be like 'Replicate from <DC-SRC> to <DC-DEST> was successful.'
 255         nc_name = self._get_rootDSE(self.dc1)["defaultNamingContext"]
 256         out = self.check_output("samba-tool drs replicate -P %s %s %s" % (self.dc1,
 257                                                                           self.dc2,
 258                                                                           nc_name))
 259         self.assertTrue("Replicate from" in out)
 260         self.assertTrue("was successful" in out)
 261
 262     def test_samba_tool_replicate_machine_creds(self):
 263         """Tests 'samba-tool drs replicate' command with implicit machine creds."""
 264
 265         # Output should be like 'Replicate from <DC-SRC> to <DC-DEST> was successful.'
 266         nc_name = self._get_rootDSE(self.dc1)["defaultNamingContext"]
 267         out = self.check_output("samba-tool drs replicate %s %s %s" % (self.dc1,
 268                                                                        self.dc2,
 269                                                                        nc_name))
 270         self.assertTrue("Replicate from" in out)
 271         self.assertTrue("was successful" in out)
 272
 273     def test_samba_tool_drs_clone_dc(self):
 274         """Tests 'samba-tool drs clone-dc-database' command."""
 275         server_rootdse = self._get_rootDSE(self.dc1)
 276         server_nc_name = server_rootdse["defaultNamingContext"]
 277         server_ds_name = server_rootdse["dsServiceName"]
 278         server_ldap_service_name = str(server_rootdse["ldapServiceName"][0])
 279         server_realm = server_ldap_service_name.split(":")[0]
 280         creds = self.get_credentials()
 281         out = self.check_output("samba-tool drs clone-dc-database %s --server=%s %s --targetdir=%s"
 282                                 % (server_realm,
 283                                    self.dc1,
 284                                    self.cmdline_creds,
 285                                    self.tempdir))
 286         ldb_rootdse = self._get_rootDSE("ldb://" + os.path.join(self.tempdir, "private", "sam.ldb"), ldap_only=False)
 287         nc_name = ldb_rootdse["defaultNamingContext"]
 288         ds_name = ldb_rootdse["dsServiceName"]
 289         ldap_service_name = str(server_rootdse["ldapServiceName"][0])
 290         self.assertEqual(nc_name, server_nc_name)
 291         # The clone should pretend to be the source server
 292         self.assertEqual(ds_name, server_ds_name)
 293         self.assertEqual(ldap_service_name, server_ldap_service_name)
 294
 295         samdb = samba.tests.connect_samdb("ldb://" + os.path.join(self.tempdir, "private", "sam.ldb"),
 296                                           ldap_only=False, lp=self.get_loadparm())
 297         def get_krbtgt_pw():
 298             krbtgt_pw = samdb.searchone("unicodePwd", "cn=krbtgt,CN=users,%s" % nc_name)
 299         self.assertRaises(KeyError, get_krbtgt_pw)
 300
 301         server_dn = samdb.searchone("serverReferenceBL", "cn=%s,ou=domain controllers,%s" % (self.dc2, server_nc_name))
 302         ntds_guid = samdb.searchone("objectGUID", "cn=ntds settings,%s" % server_dn)
 303
 304         res = samdb.search(base=str(server_nc_name),
 305                            expression="(&(objectclass=user)(cn=dns-%s))" % (self.dc2),
 306                            attrs=[], scope=ldb.SCOPE_SUBTREE)
 307         if len(res) == 1:
 308             dns_obj = res[0]
 309         else:
 310             dns_obj = None
 311
 312         # While we have this cloned, try demoting the other server on the clone, by GUID
 313         out = self.check_output("samba-tool domain demote --remove-other-dead-server=%s -H %s/private/sam.ldb"
 314                                 % (ntds_guid,
 315                                    self.tempdir))
 316
 317         # Check some of the objects that should have been removed
 318         def check_machine_obj():
 319             samdb.searchone("CN", "cn=%s,ou=domain controllers,%s" % (self.dc2, server_nc_name))
 320         self.assertRaises(ldb.LdbError, check_machine_obj)
 321
 322         def check_server_obj():
 323             samdb.searchone("CN", server_dn)
 324         self.assertRaises(ldb.LdbError, check_server_obj)
 325
 326         def check_ntds_guid():
 327             samdb.searchone("CN", "<GUID=%s>" % ntds_guid)
 328         self.assertRaises(ldb.LdbError, check_ntds_guid)
 329
 330         if dns_obj is not None:
 331             # Check some of the objects that should have been removed
 332             def check_dns_account_obj():
 333                 samdb.search(base=dns_obj.dn, scope=ldb.SCOPE_BASE,
 334                              attrs=[])
 335             self.assertRaises(ldb.LdbError, check_dns_account_obj)
 336
 337     def test_samba_tool_drs_clone_dc_secrets(self):
 338         """Tests 'samba-tool drs clone-dc-database --include-secrets' command ."""
 339         server_rootdse = self._get_rootDSE(self.dc1)
 340         server_nc_name = server_rootdse["defaultNamingContext"]
 341         server_ds_name = server_rootdse["dsServiceName"]
 342         server_ldap_service_name = str(server_rootdse["ldapServiceName"][0])
 343         server_realm = server_ldap_service_name.split(":")[0]
 344         creds = self.get_credentials()
 345         out = self.check_output("samba-tool drs clone-dc-database %s --server=%s %s --targetdir=%s --include-secrets"
 346                                 % (server_realm,
 347                                    self.dc1,
 348                                    self.cmdline_creds,
 349                                    self.tempdir))
 350         ldb_rootdse = self._get_rootDSE("ldb://" + os.path.join(self.tempdir, "private", "sam.ldb"), ldap_only=False)
 351         nc_name = ldb_rootdse["defaultNamingContext"]
 352         config_nc_name = ldb_rootdse["configurationNamingContext"]
 353         ds_name = ldb_rootdse["dsServiceName"]
 354         ldap_service_name = str(server_rootdse["ldapServiceName"][0])
 355
 356         samdb = samba.tests.connect_samdb("ldb://" + os.path.join(self.tempdir, "private", "sam.ldb"),
 357                                           ldap_only=False, lp=self.get_loadparm())
 358         krbtgt_pw = samdb.searchone("unicodePwd", "cn=krbtgt,CN=users,%s" % nc_name)
 359         self.assertIsNotNone(krbtgt_pw)
 360
 361         self.assertEqual(nc_name, server_nc_name)
 362         # The clone should pretend to be the source server
 363         self.assertEqual(ds_name, server_ds_name)
 364         self.assertEqual(ldap_service_name, server_ldap_service_name)
 365
 366         server_dn = samdb.searchone("serverReferenceBL", "cn=%s,ou=domain controllers,%s" % (self.dc2, server_nc_name))
 367         ntds_guid = samdb.searchone("objectGUID", "cn=ntds settings,%s" % server_dn)
 368
 369         res = samdb.search(base=str(server_nc_name),
 370                            expression="(&(objectclass=user)(cn=dns-%s))" % (self.dc2),
 371                            attrs=[], scope=ldb.SCOPE_SUBTREE)
 372         if len(res) == 1:
 373             dns_obj = res[0]
 374         else:
 375             dns_obj = None
 376
 377         def demote_self():
 378             # While we have this cloned, try demoting the other server on the clone
 379             out = self.check_output("samba-tool domain demote --remove-other-dead-server=%s -H %s/private/sam.ldb"
 380                                     % (self.dc1,
 381                                        self.tempdir))
 382         self.assertRaises(samba.tests.BlackboxProcessError, demote_self)
 383
 384         # While we have this cloned, try demoting the other server on the clone
 385         out = self.check_output("samba-tool domain demote --remove-other-dead-server=%s -H ldb://%s/private/sam.ldb"
 386                                 % (self.dc2,
 387                                    self.tempdir))
 388
 389         # Check some of the objects that should have been removed
 390         def check_machine_obj():
 391             samdb.searchone("CN", "cn=%s,ou=domain controllers,%s" % (self.dc2, server_nc_name))
 392         self.assertRaises(ldb.LdbError, check_machine_obj)
 393
 394         def check_server_obj():
 395             samdb.searchone("CN", server_dn)
 396         self.assertRaises(ldb.LdbError, check_server_obj)
 397
 398         def check_ntds_guid():
 399             samdb.searchone("CN", "<GUID=%s>" % ntds_guid)
 400         self.assertRaises(ldb.LdbError, check_ntds_guid)
 401
 402         if dns_obj is not None:
 403             # Check some of the objects that should have been removed
 404             def check_dns_account_obj():
 405                 samdb.search(base=dns_obj.dn, scope=ldb.SCOPE_BASE,
 406                              attrs=[])
 407             self.assertRaises(ldb.LdbError, check_dns_account_obj)
 408
 409     def test_samba_tool_drs_clone_dc_secrets_without_targetdir(self):
 410         """Tests 'samba-tool drs clone-dc-database' command without --targetdir."""
 411         server_rootdse = self._get_rootDSE(self.dc1)
 412         server_ldap_service_name = str(server_rootdse["ldapServiceName"][0])
 413         server_realm = server_ldap_service_name.split(":")[0]
 414         creds = self.get_credentials()
 415         def attempt_clone():
 416             out = self.check_output("samba-tool drs clone-dc-database %s --server=%s %s"
 417                                     % (server_realm,
 418                                        self.dc1,
 419                                        self.cmdline_creds))
 420         self.assertRaises(samba.tests.BlackboxProcessError, attempt_clone)