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_drs_clone_dc(self):
 107         """Tests 'samba-tool drs clone-dc-database' command."""
 108         server_rootdse = self._get_rootDSE(self.dc1)
 109         server_nc_name = server_rootdse["defaultNamingContext"]
 110         server_ds_name = server_rootdse["dsServiceName"]
 111         server_ldap_service_name = str(server_rootdse["ldapServiceName"][0])
 112         server_realm = server_ldap_service_name.split(":")[0]
 113         creds = self.get_credentials()
 114         out = self.check_output("samba-tool drs clone-dc-database %s --server=%s %s --targetdir=%s"
 115                                 % (server_realm,
 116                                    self.dc1,
 117                                    self.cmdline_creds,
 118                                    self.tempdir))
 119         ldb_rootdse = self._get_rootDSE("tdb://" + os.path.join(self.tempdir, "private", "sam.ldb"), ldap_only=False)
 120         nc_name = ldb_rootdse["defaultNamingContext"]
 121         ds_name = ldb_rootdse["dsServiceName"]
 122         ldap_service_name = str(server_rootdse["ldapServiceName"][0])
 123         self.assertEqual(nc_name, server_nc_name)
 124         # The clone should pretend to be the source server
 125         self.assertEqual(ds_name, server_ds_name)
 126         self.assertEqual(ldap_service_name, server_ldap_service_name)
 127
 128         samdb = samba.tests.connect_samdb("tdb://" + os.path.join(self.tempdir, "private", "sam.ldb"),
 129                                           ldap_only=False, lp=self.get_loadparm())
 130         def get_krbtgt_pw():
 131             krbtgt_pw = samdb.searchone("unicodePwd", "cn=krbtgt,CN=users,%s" % nc_name)
 132         self.assertRaises(KeyError, get_krbtgt_pw)
 133
 134         server_dn = samdb.searchone("serverReferenceBL", "cn=%s,ou=domain controllers,%s" % (self.dc2, server_nc_name))
 135         ntds_guid = samdb.searchone("objectGUID", "cn=ntds settings,%s" % server_dn)
 136
 137         res = samdb.search(base=str(server_nc_name),
 138                            expression="(&(objectclass=user)(cn=dns-%s))" % (self.dc2),
 139                            attrs=[], scope=ldb.SCOPE_SUBTREE)
 140         if len(res) == 1:
 141             dns_obj = res[0]
 142         else:
 143             dns_obj = None
 144
 145         # While we have this cloned, try demoting the other server on the clone, by GUID
 146         out = self.check_output("samba-tool domain demote --remove-other-dead-server=%s -H %s/private/sam.ldb"
 147                                 % (ntds_guid,
 148                                    self.tempdir))
 149
 150         # Check some of the objects that should have been removed
 151         def check_machine_obj():
 152             samdb.searchone("CN", "cn=%s,ou=domain controllers,%s" % (self.dc2, server_nc_name))
 153         self.assertRaises(ldb.LdbError, check_machine_obj)
 154
 155         def check_server_obj():
 156             samdb.searchone("CN", server_dn)
 157         self.assertRaises(ldb.LdbError, check_server_obj)
 158
 159         def check_ntds_guid():
 160             samdb.searchone("CN", "<GUID=%s>" % ntds_guid)
 161         self.assertRaises(ldb.LdbError, check_ntds_guid)
 162
 163         if dns_obj is not None:
 164             # Check some of the objects that should have been removed
 165             def check_dns_account_obj():
 166                 samdb.search(base=dns_obj.dn, scope=ldb.SCOPE_BASE,
 167                              attrs=[])
 168             self.assertRaises(ldb.LdbError, check_dns_account_obj)
 169
 170         shutil.rmtree(os.path.join(self.tempdir, "private"))
 171         shutil.rmtree(os.path.join(self.tempdir, "etc"))
 172         shutil.rmtree(os.path.join(self.tempdir, "msg.lock"))
 173         os.remove(os.path.join(self.tempdir, "names.tdb"))
 174         shutil.rmtree(os.path.join(self.tempdir, "state"))
 175
 176     def test_samba_tool_drs_clone_dc_secrets(self):
 177         """Tests 'samba-tool drs clone-dc-database --include-secrets' command ."""
 178         server_rootdse = self._get_rootDSE(self.dc1)
 179         server_nc_name = server_rootdse["defaultNamingContext"]
 180         server_ds_name = server_rootdse["dsServiceName"]
 181         server_ldap_service_name = str(server_rootdse["ldapServiceName"][0])
 182         server_realm = server_ldap_service_name.split(":")[0]
 183         creds = self.get_credentials()
 184         out = self.check_output("samba-tool drs clone-dc-database %s --server=%s %s --targetdir=%s --include-secrets"
 185                                 % (server_realm,
 186                                    self.dc1,
 187                                    self.cmdline_creds,
 188                                    self.tempdir))
 189         ldb_rootdse = self._get_rootDSE("tdb://" + os.path.join(self.tempdir, "private", "sam.ldb"), ldap_only=False)
 190         nc_name = ldb_rootdse["defaultNamingContext"]
 191         config_nc_name = ldb_rootdse["configurationNamingContext"]
 192         ds_name = ldb_rootdse["dsServiceName"]
 193         ldap_service_name = str(server_rootdse["ldapServiceName"][0])
 194
 195         samdb = samba.tests.connect_samdb("tdb://" + os.path.join(self.tempdir, "private", "sam.ldb"),
 196                                           ldap_only=False, lp=self.get_loadparm())
 197         krbtgt_pw = samdb.searchone("unicodePwd", "cn=krbtgt,CN=users,%s" % nc_name)
 198         self.assertIsNotNone(krbtgt_pw)
 199
 200         self.assertEqual(nc_name, server_nc_name)
 201         # The clone should pretend to be the source server
 202         self.assertEqual(ds_name, server_ds_name)
 203         self.assertEqual(ldap_service_name, server_ldap_service_name)
 204
 205         server_dn = samdb.searchone("serverReferenceBL", "cn=%s,ou=domain controllers,%s" % (self.dc2, server_nc_name))
 206         ntds_guid = samdb.searchone("objectGUID", "cn=ntds settings,%s" % server_dn)
 207
 208         res = samdb.search(base=str(server_nc_name),
 209                            expression="(&(objectclass=user)(cn=dns-%s))" % (self.dc2),
 210                            attrs=[], scope=ldb.SCOPE_SUBTREE)
 211         if len(res) == 1:
 212             dns_obj = res[0]
 213         else:
 214             dns_obj = None
 215
 216         def demote_self():
 217             # While we have this cloned, try demoting the other server on the clone
 218             out = self.check_output("samba-tool domain demote --remove-other-dead-server=%s -H %s/private/sam.ldb"
 219                                 % (self.dc1,
 220                                    self.tempdir))
 221         self.assertRaises(samba.tests.BlackboxProcessError, demote_self)
 222
 223         # While we have this cloned, try demoting the other server on the clone
 224         out = self.check_output("samba-tool domain demote --remove-other-dead-server=%s -H %s/private/sam.ldb"
 225                                 % (self.dc2,
 226                                    self.tempdir))
 227
 228         # Check some of the objects that should have been removed
 229         def check_machine_obj():
 230             samdb.searchone("CN", "cn=%s,ou=domain controllers,%s" % (self.dc2, server_nc_name))
 231         self.assertRaises(ldb.LdbError, check_machine_obj)
 232
 233         def check_server_obj():
 234             samdb.searchone("CN", server_dn)
 235         self.assertRaises(ldb.LdbError, check_server_obj)
 236
 237         def check_ntds_guid():
 238             samdb.searchone("CN", "<GUID=%s>" % ntds_guid)
 239         self.assertRaises(ldb.LdbError, check_ntds_guid)
 240
 241         if dns_obj is not None:
 242             # Check some of the objects that should have been removed
 243             def check_dns_account_obj():
 244                 samdb.search(base=dns_obj.dn, scope=ldb.SCOPE_BASE,
 245                              attrs=[])
 246             self.assertRaises(ldb.LdbError, check_dns_account_obj)
 247
 248         shutil.rmtree(os.path.join(self.tempdir, "private"))
 249         shutil.rmtree(os.path.join(self.tempdir, "etc"))
 250         shutil.rmtree(os.path.join(self.tempdir, "msg.lock"))
 251         os.remove(os.path.join(self.tempdir, "names.tdb"))
 252         shutil.rmtree(os.path.join(self.tempdir, "state"))
 253
 254     def test_samba_tool_drs_clone_dc_secrets_without_targetdir(self):
 255         """Tests 'samba-tool drs clone-dc-database' command without --targetdir."""
 256         server_rootdse = self._get_rootDSE(self.dc1)
 257         server_ldap_service_name = str(server_rootdse["ldapServiceName"][0])
 258         server_realm = server_ldap_service_name.split(":")[0]
 259         creds = self.get_credentials()
 260         def attempt_clone():
 261             out = self.check_output("samba-tool drs clone-dc-database %s --server=%s %s"
 262                                     % (server_realm,
 263                                        self.dc1,
 264                                        self.cmdline_creds))
 265         self.assertRaises(samba.tests.BlackboxProcessError, attempt_clone)