python/samba/tests/blackbox/samba_tool_drs.py

   1 # Blackbox tests for "samba-tool drs" command
   2 # Copyright (C) Kamen Mazdrashki <kamenim@samba.org> 2011
   3 #
   4 # This program is free software; you can redistribute it and/or modify
   5 # it under the terms of the GNU General Public License as published by
   6 # the Free Software Foundation; either version 3 of the License, or
   7 # (at your option) any later version.
   8 #
   9 # This program is distributed in the hope that it will be useful,
  10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12 # GNU General Public License for more details.
  13 #
  14 # You should have received a copy of the GNU General Public License
  15 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16 #
  17
  18 """Blackbox tests for samba-tool drs."""
  19
  20 import samba.tests
  21 import shutil
  22 import os
  23 import ldb
  24
  25 class SambaToolDrsTests(samba.tests.BlackboxTestCase):
  26     """Blackbox test case for samba-tool drs."""
  27
  28     def setUp(self):
  29         super(SambaToolDrsTests, self).setUp()
  30
  31         self.dc1 = samba.tests.env_get_var_value("DC1")
  32         self.dc2 = samba.tests.env_get_var_value("DC2")
  33
  34         creds = self.get_credentials()
  35         self.cmdline_creds = "-U%s/%s%%%s" % (creds.get_domain(),
  36                                               creds.get_username(), creds.get_password())
  37
  38     def _get_rootDSE(self, dc, ldap_only=True):
  39         samdb = samba.tests.connect_samdb(dc, lp=self.get_loadparm(),
  40                                           credentials=self.get_credentials(),
  41                                           ldap_only=ldap_only)
  42         return samdb.search(base="", scope=samba.tests.ldb.SCOPE_BASE)[0]
  43
  44     def test_samba_tool_bind(self):
  45         """Tests 'samba-tool drs bind' command."""
  46
  47         # Output should be like:
  48         #      Extensions supported:
  49         #        <list-of-supported-extensions>
  50         #      Site GUID: <GUID>
  51         #      Repl epoch: 0
  52         out = self.check_output("samba-tool drs bind %s %s" % (self.dc1,
  53                                                                self.cmdline_creds))
  54         self.assertTrue("Site GUID:" in out)
  55         self.assertTrue("Repl epoch:" in out)
  56
  57     def test_samba_tool_kcc(self):
  58         """Tests 'samba-tool drs kcc' command."""
  59
  60         # Output should be like 'Consistency check on <DC> successful.'
  61         out = self.check_output("samba-tool drs kcc %s %s" % (self.dc1,
  62                                                               self.cmdline_creds))
  63         self.assertTrue("Consistency check on" in out)
  64         self.assertTrue("successful" in out)
  65
  66     def test_samba_tool_showrepl(self):
  67         """Tests 'samba-tool drs showrepl' command.
  68         """
  69         # Output should be like:
  70         #      <site-name>/<domain-name>
  71         #      DSA Options: <hex-options>
  72         #      DSA object GUID: <DSA-object-GUID>
  73         #      DSA invocationId: <DSA-invocationId>
  74         #      <Inbound-connections-list>
  75         #      <Outbound-connections-list>
  76         #      <KCC-objects>
  77         #      ...
  78         #   TODO: Perhaps we should check at least for
  79         #         DSA's objectGUDI and invocationId
  80         out = self.check_output("samba-tool drs showrepl %s %s" % (self.dc1,
  81                                                                    self.cmdline_creds))
  82         self.assertTrue("DSA Options:" in out)
  83         self.assertTrue("DSA object GUID:" in out)
  84         self.assertTrue("DSA invocationId:" in out)
  85
  86     def test_samba_tool_options(self):
  87         """Tests 'samba-tool drs options' command
  88         """
  89         # Output should be like 'Current DSA options: IS_GC <OTHER_FLAGS>'
  90         out = self.check_output("samba-tool drs options %s %s" % (self.dc1,
  91                                                                   self.cmdline_creds))
  92         self.assertTrue("Current DSA options:" in out)
  93
  94     def test_samba_tool_replicate(self):
  95         """Tests 'samba-tool drs replicate' command."""
  96
  97         # Output should be like 'Replicate from <DC-SRC> to <DC-DEST> was successful.'
  98         nc_name = self._get_rootDSE(self.dc1)["defaultNamingContext"]
  99         out = self.check_output("samba-tool drs replicate %s %s %s %s" % (self.dc1,
 100                                                                           self.dc2,
 101                                                                           nc_name,
 102                                                                           self.cmdline_creds))
 103         self.assertTrue("Replicate from" in out)
 104         self.assertTrue("was successful" in out)
 105
 106     def test_samba_tool_replicate_local_online(self):
 107         """Tests 'samba-tool drs replicate --local-online' command."""
 108
 109         # Output should be like 'Replicate from <DC-SRC> to <DC-DEST> was successful.'
 110         nc_name = self._get_rootDSE(self.dc1)["defaultNamingContext"]
 111         out = self.check_output("samba-tool drs replicate --local-online %s %s %s" % (self.dc1,
 112                                                                                       self.dc2,
 113                                                                                       nc_name))
 114         self.assertTrue("Replicate from" in out)
 115         self.assertTrue("was successful" in out)
 116
 117     def test_samba_tool_replicate_local_machine_creds(self):
 118         """Tests 'samba-tool drs replicate --local -P' command (uses machine creds)."""
 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 -P --local %s %s %s" % (self.dc1,
 123                                                                                   self.dc2,
 124                                                                                   nc_name))
 125         self.assertTrue("Replicate from" in out)
 126         self.assertTrue("was successful" in out)
 127
 128     def test_samba_tool_replicate_local(self):
 129         """Tests 'samba-tool drs replicate --local' command (uses machine creds)."""
 130
 131         # Output should be like 'Replicate from <DC-SRC> to <DC-DEST> was successful.'
 132         nc_name = self._get_rootDSE(self.dc1)["defaultNamingContext"]
 133         out = self.check_output("samba-tool drs replicate --local %s %s %s %s" % (self.dc1,
 134                                                                                   self.dc2,
 135                                                                                   nc_name,
 136                                                                                   self.cmdline_creds))
 137         self.assertTrue("Replicate from" in out)
 138         self.assertTrue("was successful" in out)
 139
 140     def test_samba_tool_replicate_machine_creds_P(self):
 141         """Tests 'samba-tool drs replicate -P' command with machine creds."""
 142
 143         # Output should be like 'Replicate from <DC-SRC> to <DC-DEST> was successful.'
 144         nc_name = self._get_rootDSE(self.dc1)["defaultNamingContext"]
 145         out = self.check_output("samba-tool drs replicate -P %s %s %s" % (self.dc1,
 146                                                                           self.dc2,
 147                                                                           nc_name))
 148         self.assertTrue("Replicate from" in out)
 149         self.assertTrue("was successful" in out)
 150
 151     def test_samba_tool_replicate_machine_creds(self):
 152         """Tests 'samba-tool drs replicate' command with implicit machine creds."""
 153
 154         # Output should be like 'Replicate from <DC-SRC> to <DC-DEST> was successful.'
 155         nc_name = self._get_rootDSE(self.dc1)["defaultNamingContext"]
 156         out = self.check_output("samba-tool drs replicate %s %s %s" % (self.dc1,
 157                                                                        self.dc2,
 158                                                                        nc_name))
 159         self.assertTrue("Replicate from" in out)
 160         self.assertTrue("was successful" in out)
 161
 162     def test_samba_tool_drs_clone_dc(self):
 163         """Tests 'samba-tool drs clone-dc-database' command."""
 164         server_rootdse = self._get_rootDSE(self.dc1)
 165         server_nc_name = server_rootdse["defaultNamingContext"]
 166         server_ds_name = server_rootdse["dsServiceName"]
 167         server_ldap_service_name = str(server_rootdse["ldapServiceName"][0])
 168         server_realm = server_ldap_service_name.split(":")[0]
 169         creds = self.get_credentials()
 170         out = self.check_output("samba-tool drs clone-dc-database %s --server=%s %s --targetdir=%s"
 171                                 % (server_realm,
 172                                    self.dc1,
 173                                    self.cmdline_creds,
 174                                    self.tempdir))
 175         ldb_rootdse = self._get_rootDSE("tdb://" + os.path.join(self.tempdir, "private", "sam.ldb"), ldap_only=False)
 176         nc_name = ldb_rootdse["defaultNamingContext"]
 177         ds_name = ldb_rootdse["dsServiceName"]
 178         ldap_service_name = str(server_rootdse["ldapServiceName"][0])
 179         self.assertEqual(nc_name, server_nc_name)
 180         # The clone should pretend to be the source server
 181         self.assertEqual(ds_name, server_ds_name)
 182         self.assertEqual(ldap_service_name, server_ldap_service_name)
 183
 184         samdb = samba.tests.connect_samdb("tdb://" + os.path.join(self.tempdir, "private", "sam.ldb"),
 185                                           ldap_only=False, lp=self.get_loadparm())
 186         def get_krbtgt_pw():
 187             krbtgt_pw = samdb.searchone("unicodePwd", "cn=krbtgt,CN=users,%s" % nc_name)
 188         self.assertRaises(KeyError, get_krbtgt_pw)
 189
 190         server_dn = samdb.searchone("serverReferenceBL", "cn=%s,ou=domain controllers,%s" % (self.dc2, server_nc_name))
 191         ntds_guid = samdb.searchone("objectGUID", "cn=ntds settings,%s" % server_dn)
 192
 193         res = samdb.search(base=str(server_nc_name),
 194                            expression="(&(objectclass=user)(cn=dns-%s))" % (self.dc2),
 195                            attrs=[], scope=ldb.SCOPE_SUBTREE)
 196         if len(res) == 1:
 197             dns_obj = res[0]
 198         else:
 199             dns_obj = None
 200
 201         # While we have this cloned, try demoting the other server on the clone, by GUID
 202         out = self.check_output("samba-tool domain demote --remove-other-dead-server=%s -H %s/private/sam.ldb"
 203                                 % (ntds_guid,
 204                                    self.tempdir))
 205
 206         # Check some of the objects that should have been removed
 207         def check_machine_obj():
 208             samdb.searchone("CN", "cn=%s,ou=domain controllers,%s" % (self.dc2, server_nc_name))
 209         self.assertRaises(ldb.LdbError, check_machine_obj)
 210
 211         def check_server_obj():
 212             samdb.searchone("CN", server_dn)
 213         self.assertRaises(ldb.LdbError, check_server_obj)
 214
 215         def check_ntds_guid():
 216             samdb.searchone("CN", "<GUID=%s>" % ntds_guid)
 217         self.assertRaises(ldb.LdbError, check_ntds_guid)
 218
 219         if dns_obj is not None:
 220             # Check some of the objects that should have been removed
 221             def check_dns_account_obj():
 222                 samdb.search(base=dns_obj.dn, scope=ldb.SCOPE_BASE,
 223                              attrs=[])
 224             self.assertRaises(ldb.LdbError, check_dns_account_obj)
 225
 226         shutil.rmtree(os.path.join(self.tempdir, "private"))
 227         shutil.rmtree(os.path.join(self.tempdir, "etc"))
 228         shutil.rmtree(os.path.join(self.tempdir, "msg.lock"))
 229         os.remove(os.path.join(self.tempdir, "names.tdb"))
 230         shutil.rmtree(os.path.join(self.tempdir, "state"))
 231
 232     def test_samba_tool_drs_clone_dc_secrets(self):
 233         """Tests 'samba-tool drs clone-dc-database --include-secrets' command ."""
 234         server_rootdse = self._get_rootDSE(self.dc1)
 235         server_nc_name = server_rootdse["defaultNamingContext"]
 236         server_ds_name = server_rootdse["dsServiceName"]
 237         server_ldap_service_name = str(server_rootdse["ldapServiceName"][0])
 238         server_realm = server_ldap_service_name.split(":")[0]
 239         creds = self.get_credentials()
 240         out = self.check_output("samba-tool drs clone-dc-database %s --server=%s %s --targetdir=%s --include-secrets"
 241                                 % (server_realm,
 242                                    self.dc1,
 243                                    self.cmdline_creds,
 244                                    self.tempdir))
 245         ldb_rootdse = self._get_rootDSE("tdb://" + os.path.join(self.tempdir, "private", "sam.ldb"), ldap_only=False)
 246         nc_name = ldb_rootdse["defaultNamingContext"]
 247         config_nc_name = ldb_rootdse["configurationNamingContext"]
 248         ds_name = ldb_rootdse["dsServiceName"]
 249         ldap_service_name = str(server_rootdse["ldapServiceName"][0])
 250
 251         samdb = samba.tests.connect_samdb("tdb://" + os.path.join(self.tempdir, "private", "sam.ldb"),
 252                                           ldap_only=False, lp=self.get_loadparm())
 253         krbtgt_pw = samdb.searchone("unicodePwd", "cn=krbtgt,CN=users,%s" % nc_name)
 254         self.assertIsNotNone(krbtgt_pw)
 255
 256         self.assertEqual(nc_name, server_nc_name)
 257         # The clone should pretend to be the source server
 258         self.assertEqual(ds_name, server_ds_name)
 259         self.assertEqual(ldap_service_name, server_ldap_service_name)
 260
 261         server_dn = samdb.searchone("serverReferenceBL", "cn=%s,ou=domain controllers,%s" % (self.dc2, server_nc_name))
 262         ntds_guid = samdb.searchone("objectGUID", "cn=ntds settings,%s" % server_dn)
 263
 264         res = samdb.search(base=str(server_nc_name),
 265                            expression="(&(objectclass=user)(cn=dns-%s))" % (self.dc2),
 266                            attrs=[], scope=ldb.SCOPE_SUBTREE)
 267         if len(res) == 1:
 268             dns_obj = res[0]
 269         else:
 270             dns_obj = None
 271
 272         def demote_self():
 273             # While we have this cloned, try demoting the other server on the clone
 274             out = self.check_output("samba-tool domain demote --remove-other-dead-server=%s -H %s/private/sam.ldb"
 275                                 % (self.dc1,
 276                                    self.tempdir))
 277         self.assertRaises(samba.tests.BlackboxProcessError, demote_self)
 278
 279         # While we have this cloned, try demoting the other server on the clone
 280         out = self.check_output("samba-tool domain demote --remove-other-dead-server=%s -H %s/private/sam.ldb"
 281                                 % (self.dc2,
 282                                    self.tempdir))
 283
 284         # Check some of the objects that should have been removed
 285         def check_machine_obj():
 286             samdb.searchone("CN", "cn=%s,ou=domain controllers,%s" % (self.dc2, server_nc_name))
 287         self.assertRaises(ldb.LdbError, check_machine_obj)
 288
 289         def check_server_obj():
 290             samdb.searchone("CN", server_dn)
 291         self.assertRaises(ldb.LdbError, check_server_obj)
 292
 293         def check_ntds_guid():
 294             samdb.searchone("CN", "<GUID=%s>" % ntds_guid)
 295         self.assertRaises(ldb.LdbError, check_ntds_guid)
 296
 297         if dns_obj is not None:
 298             # Check some of the objects that should have been removed
 299             def check_dns_account_obj():
 300                 samdb.search(base=dns_obj.dn, scope=ldb.SCOPE_BASE,
 301                              attrs=[])
 302             self.assertRaises(ldb.LdbError, check_dns_account_obj)
 303
 304         shutil.rmtree(os.path.join(self.tempdir, "private"))
 305         shutil.rmtree(os.path.join(self.tempdir, "etc"))
 306         shutil.rmtree(os.path.join(self.tempdir, "msg.lock"))
 307         os.remove(os.path.join(self.tempdir, "names.tdb"))
 308         shutil.rmtree(os.path.join(self.tempdir, "state"))
 309
 310     def test_samba_tool_drs_clone_dc_secrets_without_targetdir(self):
 311         """Tests 'samba-tool drs clone-dc-database' command without --targetdir."""
 312         server_rootdse = self._get_rootDSE(self.dc1)
 313         server_ldap_service_name = str(server_rootdse["ldapServiceName"][0])
 314         server_realm = server_ldap_service_name.split(":")[0]
 315         creds = self.get_credentials()
 316         def attempt_clone():
 317             out = self.check_output("samba-tool drs clone-dc-database %s --server=%s %s"
 318                                     % (server_realm,
 319                                        self.dc1,
 320                                        self.cmdline_creds))
 321         self.assertRaises(samba.tests.BlackboxProcessError, attempt_clone)