python/samba/tests/domain_backup_offline.py

   1 # Unix SMB/CIFS implementation.
   2 # Copyright (C) Andrew Bartlett <abartlet@samba.org>
   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 import os
  19 import shutil
  20 import tempfile
  21 from samba.tests import BlackboxTestCase, BlackboxProcessError
  22 from samba.netcmd import CommandError
  23 from samba.param import LoadParm
  24 from samba import safe_tarfile as tarfile
  25
  26
  27 # The backup tests require that a completely clean LoadParm object gets used
  28 # for the restore. Otherwise the same global LP gets re-used, and the LP
  29 # settings can bleed from one test case to another.
  30 # To do this, these tests should use check_output(), which executes the command
  31 # in a separate process (as opposed to runcmd(), runsubcmd()).
  32 # So although this is a samba-tool test, we don't inherit from SambaToolCmdTest
  33 # so that we never inadvertently use .runcmd() by accident.
  34 class DomainBackupOfflineCmp(BlackboxTestCase):
  35
  36     def test_domain_backup_offline_nested_tdb(self):
  37         self.nested_testcase('tdb')
  38
  39     def test_domain_backup_offline_nested_mdb(self):
  40         self.nested_testcase('mdb')
  41
  42     def nested_testcase(self, backend):
  43         self.prov_dir = self.provision(backend)
  44         self.extract_dir = None
  45
  46         src = os.path.join(self.prov_dir, "private")
  47         dst = os.path.join(self.prov_dir, "state", "private")
  48
  49         # Move private directory inside state directory
  50         shutil.move(src, dst)
  51
  52         smbconf = os.path.join(self.prov_dir, "etc", "smb.conf")
  53
  54         # Update the conf file
  55         lp = LoadParm(filename_for_non_global_lp=smbconf)
  56         lp.set("private dir", dst)
  57         lp.dump(False, smbconf)
  58
  59         backup_file = self.backup(self.prov_dir)
  60
  61         # Ensure each file is only present once in the tar file
  62         tf = tarfile.open(backup_file)
  63         names = tf.getnames()
  64         self.assertEqual(len(names), len(set(names)))
  65
  66     def test_domain_backup_offline_join_restore_tdb(self):
  67         self.join_restore_testcase('tdb')
  68
  69     def test_domain_backup_offline_join_restore_mdb(self):
  70         self.join_restore_testcase('mdb')
  71
  72     def join_restore_testcase(self, backend):
  73         self.prov_dir = self.join(backend)
  74         self.extract_dir = None
  75
  76         try:
  77             backup_file = self.backup(self.prov_dir)
  78         except BlackboxProcessError as e:
  79             self.fail(e)
  80
  81         self.extract_dir = self.restore(backup_file)
  82
  83     def test_domain_backup_offline_hard_link_tdb(self):
  84         self.hard_link_testcase('tdb')
  85
  86     def test_domain_backup_offline_hard_link_mdb(self):
  87         self.hard_link_testcase('mdb')
  88
  89     def hard_link_testcase(self, backend):
  90         self.prov_dir = self.provision(backend)
  91         self.extract_dir = None
  92
  93         # Create hard links in the private and state directories
  94         os.link(os.path.join(self.prov_dir, "private", "krb5.conf"),
  95                 os.path.join(self.prov_dir, "state", "krb5.conf"))
  96
  97         backup_file = self.backup(self.prov_dir)
  98
  99         # Extract the backup
 100         self.extract_dir = tempfile.mkdtemp(dir=self.tempdir)
 101         tf = tarfile.open(backup_file)
 102         tf.extractall(self.extract_dir)
 103
 104         # Ensure that the hard link in the private directory was backed up,
 105         # while the one in the state directory was not.
 106         self.assertTrue(os.path.exists(os.path.join(self.extract_dir,
 107                                                     "private", "krb5.conf")))
 108         self.assertFalse(os.path.exists(os.path.join(self.extract_dir,
 109                                                     "statedir", "krb5.conf")))
 110
 111     def test_domain_backup_offline_untar_tdb(self):
 112         self.untar_testcase('tdb')
 113
 114     def test_domain_backup_offline_untar_mdb(self):
 115         self.untar_testcase('mdb')
 116
 117     def test_domain_backup_offline_restore_tdb(self):
 118         self.restore_testcase('tdb')
 119
 120     def test_domain_backup_offline_restore_mdb(self):
 121         self.restore_testcase('mdb')
 122
 123     def restore_testcase(self, backend):
 124         self.prov_dir = self.provision(backend)
 125         self.extract_dir = None
 126         backup_file = self.backup(self.prov_dir)
 127
 128         self.extract_dir = self.restore(backup_file)
 129
 130         # attrs that are altered by the restore process
 131         ignore_attrs = ["servicePrincipalName", "lastLogonTimestamp",
 132                         "rIDAllocationPool", "rIDAvailablePool", "rIDUsedPool",
 133                         "localPolicyFlags", "operatingSystem", "displayName",
 134                         "dnsRecord", "dNSTombstoned",
 135                         "msDS-NC-Replica-Locations", "msDS-HasInstantiatedNCs",
 136                         "interSiteTopologyGenerator", "msKds-DomainID"]
 137         filter_arg = "--filter=" + ",".join(ignore_attrs)
 138         args = ["--two", filter_arg]
 139         self.ldapcmp(self.prov_dir, self.extract_dir, args)
 140
 141     def untar_testcase(self, backend):
 142         self.prov_dir = self.provision(backend)
 143         self.extract_dir = None
 144         backup_file = self.backup(self.prov_dir)
 145
 146         self.extract_dir = tempfile.mkdtemp(dir=self.tempdir)
 147         tf = tarfile.open(backup_file)
 148         tf.extractall(self.extract_dir)
 149
 150         self.ldapcmp(self.prov_dir, self.extract_dir)
 151
 152     def ldapcmp(self, prov_dir, ex_dir, args=None):
 153         if args is None:
 154             args = []
 155         sam_fn = os.path.join("private", "sam.ldb")
 156         url1 = "tdb://" + os.path.join(os.path.realpath(prov_dir), sam_fn)
 157         url2 = "tdb://" + os.path.join(os.path.realpath(ex_dir), sam_fn)
 158
 159         # Compare the restored sam.ldb with the old one
 160         for partition in ["domain", "configuration", "schema",
 161                           "dnsdomain", "dnsforest"]:
 162             cmd = "samba-tool ldapcmp " + " ".join([url1, url2, partition] + args)
 163             self.check_output(cmd)
 164
 165     # Test the "samba-tool domain backup" command with ldapcmp
 166     def provision(self, backend):
 167         target = tempfile.mkdtemp(dir=self.tempdir)
 168
 169         # Provision domain.  Use fake ACLs and store xattrs in tdbs so that
 170         # NTACL backup will work inside the testenv.
 171         # host-name option must be given because if this test runs on a
 172         # system with a very long hostname, it will be shortened in certain
 173         # circumstances, causing the ldapcmp to fail.
 174         prov_cmd = "samba-tool domain provision " +\
 175                    "--domain FOO --realm foo.example.com " +\
 176                    "--targetdir {target} " +\
 177                    "--backend-store {backend} " +\
 178                    "--host-name OLDSERVER "+\
 179                    "--option=\"vfs objects=dfs_samba4 acl_xattr fake_acls xattr_tdb\""
 180         prov_cmd = prov_cmd.format(target=target, backend=backend)
 181         self.check_output(prov_cmd)
 182
 183         return target
 184
 185     def join(self, backend):
 186         target = tempfile.mkdtemp(dir=self.tempdir)
 187
 188         new_dc_name = "offlinebackupdc"
 189
 190         join_cmd = "samba-tool domain join {domain} DC " +\
 191                    "--server {server} " +\
 192                    "--realm {realm} " +\
 193                    "--username {username}%{password} " +\
 194                    "--targetdir {target} " +\
 195                    "--backend-store {backend} " +\
 196                    "--option='netbios name = {new_dc_name}' " +\
 197                    "--option=\"vfs objects=dfs_samba4 acl_xattr fake_acls xattr_tdb\""
 198         join_cmd = join_cmd.format(server=os.environ["DC_SERVER"],
 199                                    domain=os.environ["DOMAIN"],
 200                                    realm=os.environ["REALM"],
 201                                    username=os.environ["USERNAME"],
 202                                    password=os.environ["PASSWORD"],
 203                                    target=target,
 204                                    backend=backend,
 205                                    new_dc_name=new_dc_name)
 206         self.check_output(join_cmd)
 207
 208         demote_cmd = "samba-tool domain demote " +\
 209                    "--server {server} " +\
 210                    "--username {username}%{password} " +\
 211                    "--remove-other-dead-server={new_dc_name}"
 212
 213         demote_cmd = demote_cmd.format(server=os.environ["DC_SERVER"],
 214                                        username=os.environ["USERNAME"],
 215                                        password=os.environ["PASSWORD"],
 216                                        new_dc_name=new_dc_name)
 217         self.check_output(demote_cmd)
 218
 219         return target
 220
 221     def backup(self, prov_dir):
 222         # Run the backup and check we got one backup tar file
 223         cmd = ("samba-tool domain backup offline --targetdir={prov_dir} "
 224                "--configfile={prov_dir}/etc/smb.conf").format(prov_dir=prov_dir)
 225         self.check_output(cmd)
 226
 227         tar_files = [fn for fn in os.listdir(prov_dir)
 228                      if fn.startswith("samba-backup-") and
 229                      fn.endswith(".tar.bz2")]
 230         if len(tar_files) != 1:
 231             raise CommandError("expected domain backup to create one tar" +
 232                                " file but got {0}".format(len(tar_files)))
 233
 234         backup_file = os.path.join(prov_dir, tar_files[0])
 235         return backup_file
 236
 237     def restore(self, backup_file):
 238         # Restore from a backup file
 239         extract_dir = tempfile.mkdtemp(dir=self.tempdir)
 240         cmd = ("samba-tool domain backup restore --backup-file={f}"
 241                " --targetdir={d} "
 242                "--newservername=NEWSERVER").format(f=backup_file,
 243                                                    d=extract_dir)
 244         self.check_output(cmd)
 245
 246         return extract_dir
 247
 248     def tearDown(self):
 249         # Remove temporary directories
 250         shutil.rmtree(self.prov_dir)
 251         if self.extract_dir:
 252             shutil.rmtree(self.extract_dir)