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