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
  24 class SambaToolDrsTests(samba.tests.BlackboxTestCase):
  25     """Blackbox test case for samba-tool drs."""
  26
  27     def setUp(self):
  28         super(SambaToolDrsTests, self).setUp()
  29
  30         self.dc1 = samba.tests.env_get_var_value("DC1")
  31         self.dc2 = samba.tests.env_get_var_value("DC2")
  32
  33         creds = self.get_credentials()
  34         self.cmdline_creds = "-U%s/%s%%%s" % (creds.get_domain(),
  35                                               creds.get_username(), creds.get_password())
  36
  37     def _get_rootDSE(self, dc, ldap_only=True):
  38         samdb = samba.tests.connect_samdb(dc, lp=self.get_loadparm(),
  39                                           credentials=self.get_credentials(),
  40                                           ldap_only=ldap_only)
  41         return samdb.search(base="", scope=samba.tests.ldb.SCOPE_BASE)[0]
  42
  43     def test_samba_tool_bind(self):
  44         """Tests 'samba-tool drs bind' command."""
  45
  46         # Output should be like:
  47         #      Extensions supported:
  48         #        <list-of-supported-extensions>
  49         #      Site GUID: <GUID>
  50         #      Repl epoch: 0
  51         out = self.check_output("samba-tool drs bind %s %s" % (self.dc1,
  52                                                                self.cmdline_creds))
  53         self.assertTrue("Site GUID:" in out)
  54         self.assertTrue("Repl epoch:" in out)
  55
  56     def test_samba_tool_kcc(self):
  57         """Tests 'samba-tool drs kcc' command."""
  58
  59         # Output should be like 'Consistency check on <DC> successful.'
  60         out = self.check_output("samba-tool drs kcc %s %s" % (self.dc1,
  61                                                               self.cmdline_creds))
  62         self.assertTrue("Consistency check on" in out)
  63         self.assertTrue("successful" in out)
  64
  65     def test_samba_tool_showrepl(self):
  66         """Tests 'samba-tool drs showrepl' command.
  67         """
  68         # Output should be like:
  69         #      <site-name>/<domain-name>
  70         #      DSA Options: <hex-options>
  71         #      DSA object GUID: <DSA-object-GUID>
  72         #      DSA invocationId: <DSA-invocationId>
  73         #      <Inbound-connections-list>
  74         #      <Outbound-connections-list>
  75         #      <KCC-objects>
  76         #      ...
  77         #   TODO: Perhaps we should check at least for
  78         #         DSA's objectGUDI and invocationId
  79         out = self.check_output("samba-tool drs showrepl %s %s" % (self.dc1,
  80                                                                    self.cmdline_creds))
  81         self.assertTrue("DSA Options:" in out)
  82         self.assertTrue("DSA object GUID:" in out)
  83         self.assertTrue("DSA invocationId:" 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("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("Replicate from" in out)
 103         self.assertTrue("was successful" in out)
 104
 105     def test_samba_tool_drs_clone_dc(self):
 106         """Tests 'samba-tool drs clone-dc-database' command."""
 107         server_rootdse = self._get_rootDSE(self.dc1)
 108         server_nc_name = server_rootdse["defaultNamingContext"]
 109         server_ds_name = server_rootdse["dsServiceName"]
 110         server_ldap_service_name = str(server_rootdse["ldapServiceName"][0])
 111         server_realm = server_ldap_service_name.split(":")[0]
 112         creds = self.get_credentials()
 113         out = self.check_output("samba-tool drs clone-dc-database %s --server=%s %s --targetdir=%s"
 114                                 % (server_realm,
 115                                    self.dc1,
 116                                    self.cmdline_creds,
 117                                    self.tempdir))
 118         ldb_rootdse = self._get_rootDSE("tdb://" + os.path.join(self.tempdir, "private", "sam.ldb"), ldap_only=False)
 119         nc_name = ldb_rootdse["defaultNamingContext"]
 120         ds_name = ldb_rootdse["dsServiceName"]
 121         ldap_service_name = str(server_rootdse["ldapServiceName"][0])
 122         self.assertEqual(nc_name, server_nc_name)
 123         # The clone should pretend to be the source server
 124         self.assertEqual(ds_name, server_ds_name)
 125         self.assertEqual(ldap_service_name, server_ldap_service_name)
 126
 127         samdb = samba.tests.connect_samdb("tdb://" + os.path.join(self.tempdir, "private", "sam.ldb"),
 128                                           ldap_only=False, lp=self.get_loadparm())
 129         def get_krbtgt_pw():
 130             krbtgt_pw = samdb.searchone("unicodePwd", "cn=krbtgt,CN=users,%s" % nc_name)
 131         self.assertRaises(KeyError, get_krbtgt_pw)
 132         shutil.rmtree(os.path.join(self.tempdir, "private"))
 133         shutil.rmtree(os.path.join(self.tempdir, "etc"))
 134         shutil.rmtree(os.path.join(self.tempdir, "msg.lock"))
 135         os.remove(os.path.join(self.tempdir, "names.tdb"))
 136         shutil.rmtree(os.path.join(self.tempdir, "state"))
 137
 138     def test_samba_tool_drs_clone_dc_secrets(self):
 139         """Tests 'samba-tool drs clone-dc-database --include-secrets' command ."""
 140         server_rootdse = self._get_rootDSE(self.dc1)
 141         server_nc_name = server_rootdse["defaultNamingContext"]
 142         server_ds_name = server_rootdse["dsServiceName"]
 143         server_ldap_service_name = str(server_rootdse["ldapServiceName"][0])
 144         server_realm = server_ldap_service_name.split(":")[0]
 145         creds = self.get_credentials()
 146         out = self.check_output("samba-tool drs clone-dc-database %s --server=%s %s --targetdir=%s --include-secrets"
 147                                 % (server_realm,
 148                                    self.dc1,
 149                                    self.cmdline_creds,
 150                                    self.tempdir))
 151         ldb_rootdse = self._get_rootDSE("tdb://" + os.path.join(self.tempdir, "private", "sam.ldb"), ldap_only=False)
 152         nc_name = ldb_rootdse["defaultNamingContext"]
 153         ds_name = ldb_rootdse["dsServiceName"]
 154         ldap_service_name = str(server_rootdse["ldapServiceName"][0])
 155
 156         samdb = samba.tests.connect_samdb("tdb://" + os.path.join(self.tempdir, "private", "sam.ldb"),
 157                                           ldap_only=False, lp=self.get_loadparm())
 158         krbtgt_pw = samdb.searchone("unicodePwd", "cn=krbtgt,CN=users,%s" % nc_name)
 159         self.assertIsNotNone(krbtgt_pw)
 160
 161         self.assertEqual(nc_name, server_nc_name)
 162         # The clone should pretend to be the source server
 163         self.assertEqual(ds_name, server_ds_name)
 164         self.assertEqual(ldap_service_name, server_ldap_service_name)
 165
 166         # While we have this cloned, try demoting the other server on the clone
 167         out = self.check_output("samba-tool domain demote --remove-other-dead-server=%s -H %s/private/sam.ldb"
 168                                 % (self.dc2,
 169                                    self.tempdir))
 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_without_targetdir(self):
 177         """Tests 'samba-tool drs clone-dc-database' command without --targetdir."""
 178         server_rootdse = self._get_rootDSE(self.dc1)
 179         server_ldap_service_name = str(server_rootdse["ldapServiceName"][0])
 180         server_realm = server_ldap_service_name.split(":")[0]
 181         creds = self.get_credentials()
 182         def attempt_clone():
 183             out = self.check_output("samba-tool drs clone-dc-database %s --server=%s %s"
 184                                     % (server_realm,
 185                                        self.dc1,
 186                                        self.cmdline_creds))
 187         self.assertRaises(samba.tests.BlackboxProcessError, attempt_clone)