python/samba/tests/samba_tool/gpo.py

   1 # Unix SMB/CIFS implementation.
   2 # Copyright (C) Andrew Bartlett 2012
   3 #
   4 # based on time.py:
   5 # Copyright (C) Sean Dague <sdague@linux.vnet.ibm.com> 2011
   6 #
   7 # This program is free software; you can redistribute it and/or modify
   8 # it under the terms of the GNU General Public License as published by
   9 # the Free Software Foundation; either version 3 of the License, or
  10 # (at your option) any later version.
  11 #
  12 # This program is distributed in the hope that it will be useful,
  13 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15 # GNU General Public License for more details.
  16 #
  17 # You should have received a copy of the GNU General Public License
  18 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  19 #
  20
  21 import os
  22 import samba
  23 from samba.tests.samba_tool.base import SambaToolCmdTest
  24 import shutil
  25
  26 def has_difference(path1, path2, binary=True, xml=True, sortlines=False):
  27     """Use this function to determine if the GPO backup differs from another.
  28
  29     xml=True checks whether any xml files are equal
  30     binary=True checks whether any .SAMBABACKUP files are equal
  31     """
  32     if os.path.isfile(path1):
  33         if sortlines:
  34             file1 = open(path1).readlines()
  35             file1.sort()
  36             file2 = open(path1).readlines()
  37             file2.sort()
  38             if file1 != file2:
  39                 return path1
  40
  41         elif open(path1).read() != open(path2).read():
  42             return path1
  43
  44         return None
  45
  46     l_dirs = [ path1 ]
  47     r_dirs = [ path2 ]
  48     while l_dirs:
  49         l_dir = l_dirs.pop()
  50         r_dir = r_dirs.pop()
  51
  52         dirlist = os.listdir(l_dir)
  53         dirlist_other = os.listdir(r_dir)
  54
  55         dirlist.sort()
  56         dirlist_other.sort()
  57         if dirlist != dirlist_other:
  58             return dirlist
  59
  60         for e in dirlist:
  61             l_name = os.path.join(l_dir, e)
  62             r_name = os.path.join(r_dir, e)
  63
  64             if os.path.isdir(l_name):
  65                 l_dirs.append(l_name)
  66                 r_dirs.append(r_name)
  67             else:
  68                 if (l_name.endswith('.xml') and xml or
  69                     l_name.endswith('.SAMBABACKUP') and binary):
  70                     if open(l_name).read() != open(r_name).read():
  71                         return l_name
  72
  73     return None
  74
  75 class GpoCmdTestCase(SambaToolCmdTest):
  76     """Tests for samba-tool time subcommands"""
  77
  78     gpo_name = "testgpo"
  79
  80     # This exists in the source tree to be restored
  81     backup_gpo_guid = "{1E1DC8EA-390C-4800-B327-98B56A0AEA5D}"
  82
  83     def test_gpo_list(self):
  84         """Run gpo list against the server and make sure it looks accurate"""
  85         (result, out, err) = self.runsubcmd("gpo", "listall", "-H", "ldap://%s" % os.environ["SERVER"])
  86         self.assertCmdSuccess(result, out, err, "Ensuring gpo listall ran successfully")
  87
  88     def test_fetchfail(self):
  89         """Run against a non-existent GPO, and make sure it fails (this hard-coded UUID is very unlikely to exist"""
  90         (result, out, err) = self.runsubcmd("gpo", "fetch", "c25cac17-a02a-4151-835d-fae17446ee43", "-H", "ldap://%s" % os.environ["SERVER"])
  91         self.assertCmdFail(result, "check for result code")
  92
  93     def test_fetch(self):
  94         """Run against a real GPO, and make sure it passes"""
  95         (result, out, err) = self.runsubcmd("gpo", "fetch", self.gpo_guid, "-H", "ldap://%s" % os.environ["SERVER"], "--tmpdir", self.tempdir)
  96         self.assertCmdSuccess(result, out, err, "Ensuring gpo fetched successfully")
  97         shutil.rmtree(os.path.join(self.tempdir, "policy"))
  98
  99     def test_show(self):
 100         """Show a real GPO, and make sure it passes"""
 101         (result, out, err) = self.runsubcmd("gpo", "show", self.gpo_guid, "-H", "ldap://%s" % os.environ["SERVER"])
 102         self.assertCmdSuccess(result, out, err, "Ensuring gpo fetched successfully")
 103
 104     def test_show_as_admin(self):
 105         """Show a real GPO, and make sure it passes"""
 106         (result, out, err) = self.runsubcmd("gpo", "show", self.gpo_guid, "-H", "ldap://%s" % os.environ["SERVER"], "-U%s%%%s" % (os.environ["USERNAME"], os.environ["PASSWORD"]))
 107         self.assertCmdSuccess(result, out, err, "Ensuring gpo fetched successfully")
 108
 109     def test_aclcheck(self):
 110         """Check all the GPOs on the remote server have correct ACLs"""
 111         (result, out, err) = self.runsubcmd("gpo", "aclcheck", "-H", "ldap://%s" % os.environ["SERVER"], "-U%s%%%s" % (os.environ["USERNAME"], os.environ["PASSWORD"]))
 112         self.assertCmdSuccess(result, out, err, "Ensuring gpo checked successfully")
 113
 114     def test_backup_restore_compare_binary(self):
 115         """Restore from a static backup and compare the binary contents"""
 116
 117         static_path = os.path.join(self.backup_path, 'policy',
 118                                    self.backup_gpo_guid)
 119
 120         temp_path = os.path.join(self.tempdir, 'temp')
 121         os.mkdir(temp_path)
 122
 123         new_path = os.path.join(self.tempdir, 'new')
 124         os.mkdir(new_path)
 125
 126         gpo_guid = None
 127         try:
 128             (result, out, err) = self.runsubcmd("gpo", "restore", "BACKUP_RESTORE1",
 129                                                 static_path,
 130                                                 "-H", "ldap://%s" %
 131                                                 os.environ["SERVER"], "--tmpdir",
 132                                                 temp_path, "--entities",
 133                                                 self.entity_file, "-U%s%%%s" %
 134                                                 (os.environ["USERNAME"],
 135                                                  os.environ["PASSWORD"]))
 136
 137             gpo_guid = "{%s}" % out.split("{")[1].split("}")[0]
 138
 139             (result, out, err) = self.runsubcmd("gpo", "backup", gpo_guid,
 140                                                 "-H", "ldap://%s" %
 141                                                 os.environ["SERVER"],
 142                                                 "--tmpdir", new_path)
 143
 144             self.assertCmdSuccess(result, out, err, "Ensuring gpo fetched successfully")
 145
 146             # Compare the directories
 147             self.assertIsNone(has_difference(os.path.join(new_path, 'policy',
 148                                                          gpo_guid),
 149                                             static_path, binary=True,
 150                                             xml=False))
 151         finally:
 152             if gpo_guid:
 153                 (result, out, err) = self.runsubcmd("gpo", "del", gpo_guid,
 154                                                     "-H", "ldap://%s" %
 155                                                     os.environ["SERVER"],
 156                                                     "-U%s%%%s" %
 157                                                     (os.environ["USERNAME"],
 158                                                      os.environ["PASSWORD"]))
 159                 self.assertCmdSuccess(result, out, err, "Ensuring gpo deleted successfully")
 160
 161             shutil.rmtree(temp_path)
 162             shutil.rmtree(new_path)
 163
 164     def test_backup_restore_no_entities_compare_binary(self):
 165         """Restore from a static backup (and use no entity file, resulting in
 166         copy-restore fallback), and compare the binary contents"""
 167
 168         static_path = os.path.join(self.backup_path, 'policy',
 169                                    self.backup_gpo_guid)
 170
 171         temp_path = os.path.join(self.tempdir, 'temp')
 172         os.mkdir(temp_path)
 173
 174         new_path = os.path.join(self.tempdir, 'new')
 175         os.mkdir(new_path)
 176
 177         gpo_guid = None
 178         gpo_guid1 = None
 179         gpo_guid2 = None
 180         try:
 181             (result, out, err) = self.runsubcmd("gpo", "restore", "BACKUP_RESTORE1",
 182                                                 static_path,
 183                                                 "-H", "ldap://%s" %
 184                                                 os.environ["SERVER"], "--tmpdir",
 185                                                 temp_path, "--entities",
 186                                                 self.entity_file, "-U%s%%%s" %
 187                                                 (os.environ["USERNAME"],
 188                                                  os.environ["PASSWORD"]))
 189
 190             gpo_guid = "{%s}" % out.split("{")[1].split("}")[0]
 191             gpo_guid1 = gpo_guid
 192
 193             # Do not output entities file
 194             (result, out, err) = self.runsubcmd("gpo", "backup", gpo_guid,
 195                                                 "-H", "ldap://%s" %
 196                                                 os.environ["SERVER"],
 197                                                 "--tmpdir", new_path,
 198                                                 "--generalize")
 199
 200             self.assertCmdSuccess(result, out, err, "Ensuring gpo fetched successfully")
 201
 202             # Do not use an entities file
 203             (result, out, err) = self.runsubcmd("gpo", "restore", "BACKUP_RESTORE2",
 204                                                 os.path.join(new_path, 'policy', gpo_guid1),
 205                                                 "-H", "ldap://%s" %
 206                                                 os.environ["SERVER"], "--tmpdir",
 207                                                 temp_path, "-U%s%%%s" %
 208                                                 (os.environ["USERNAME"],
 209                                                  os.environ["PASSWORD"]))
 210
 211             gpo_guid = "{%s}" % out.split("{")[1].split("}")[0]
 212             gpo_guid2 = gpo_guid
 213
 214             self.assertCmdSuccess(result, out, err, "Ensuring gpo restored successfully")
 215
 216             (result, out, err) = self.runsubcmd("gpo", "backup", gpo_guid,
 217                                                 "-H", "ldap://%s" %
 218                                                 os.environ["SERVER"],
 219                                                 "--tmpdir", new_path)
 220
 221             # Compare the directories
 222             self.assertIsNone(has_difference(os.path.join(new_path, 'policy',
 223                                                          gpo_guid1),
 224                                             os.path.join(new_path, 'policy',
 225                                                          gpo_guid2),
 226                                             binary=True, xml=False))
 227         finally:
 228             if gpo_guid1:
 229                 (result, out, err) = self.runsubcmd("gpo", "del", gpo_guid1,
 230                                                     "-H", "ldap://%s" %
 231                                                     os.environ["SERVER"],
 232                                                     "-U%s%%%s" %
 233                                                     (os.environ["USERNAME"],
 234                                                      os.environ["PASSWORD"]))
 235                 self.assertCmdSuccess(result, out, err, "Ensuring gpo deleted successfully")
 236
 237             if gpo_guid2:
 238                 (result, out, err) = self.runsubcmd("gpo", "del", gpo_guid2,
 239                                                     "-H", "ldap://%s" %
 240                                                     os.environ["SERVER"],
 241                                                     "-U%s%%%s" %
 242                                                     (os.environ["USERNAME"],
 243                                                      os.environ["PASSWORD"]))
 244                 self.assertCmdSuccess(result, out, err, "Ensuring gpo deleted successfully")
 245
 246             shutil.rmtree(temp_path)
 247             shutil.rmtree(new_path)
 248
 249     def test_backup_restore_backup_compare_XML(self):
 250         """Restore from a static backup and backup to compare XML"""
 251         static_path = os.path.join(self.backup_path, 'policy',
 252                                    self.backup_gpo_guid)
 253
 254         temp_path = os.path.join(self.tempdir, 'temp')
 255         os.mkdir(temp_path)
 256
 257         new_path = os.path.join(self.tempdir, 'new')
 258         os.mkdir(new_path)
 259
 260         gpo_guid = None
 261         gpo_guid1 = None
 262         gpo_guid2 = None
 263         try:
 264             (result, out, err) = self.runsubcmd("gpo", "restore", "BACKUP_RESTORE1",
 265                                                 static_path,
 266                                                 "-H", "ldap://%s" %
 267                                                 os.environ["SERVER"], "--tmpdir",
 268                                                 temp_path, "--entities",
 269                                                 self.entity_file, "-U%s%%%s" %
 270                                                 (os.environ["USERNAME"],
 271                                                  os.environ["PASSWORD"]))
 272
 273             gpo_guid = "{%s}" % out.split("{")[1].split("}")[0]
 274             gpo_guid1 = gpo_guid
 275
 276             (result, out, err) = self.runsubcmd("gpo", "backup", gpo_guid,
 277                                                 "-H", "ldap://%s" %
 278                                                 os.environ["SERVER"],
 279                                                 "--tmpdir", new_path)
 280
 281             self.assertCmdSuccess(result, out, err, "Ensuring gpo fetched successfully")
 282
 283             (result, out, err) = self.runsubcmd("gpo", "restore", "BACKUP_RESTORE2",
 284                                                 os.path.join(new_path, 'policy', gpo_guid1),
 285                                                 "-H", "ldap://%s" %
 286                                                 os.environ["SERVER"], "--tmpdir",
 287                                                 temp_path, "--entities",
 288                                                 self.entity_file, "-U%s%%%s" %
 289                                                 (os.environ["USERNAME"],
 290                                                  os.environ["PASSWORD"]))
 291
 292             gpo_guid = "{%s}" % out.split("{")[1].split("}")[0]
 293             gpo_guid2 = gpo_guid
 294
 295             self.assertCmdSuccess(result, out, err, "Ensuring gpo restored successfully")
 296
 297             (result, out, err) = self.runsubcmd("gpo", "backup", gpo_guid,
 298                                                 "-H", "ldap://%s" %
 299                                                 os.environ["SERVER"],
 300                                                 "--tmpdir", new_path)
 301
 302             # Compare the directories
 303             self.assertIsNone(has_difference(os.path.join(new_path, 'policy',
 304                                                          gpo_guid1),
 305                                             os.path.join(new_path, 'policy',
 306                                                          gpo_guid2),
 307                                             binary=True, xml=True))
 308         finally:
 309             if gpo_guid1:
 310                 (result, out, err) = self.runsubcmd("gpo", "del", gpo_guid1,
 311                                                     "-H", "ldap://%s" %
 312                                                     os.environ["SERVER"],
 313                                                     "-U%s%%%s" %
 314                                                     (os.environ["USERNAME"],
 315                                                      os.environ["PASSWORD"]))
 316                 self.assertCmdSuccess(result, out, err, "Ensuring gpo deleted successfully")
 317
 318             if gpo_guid2:
 319                 (result, out, err) = self.runsubcmd("gpo", "del", gpo_guid2,
 320                                                     "-H", "ldap://%s" %
 321                                                     os.environ["SERVER"],
 322                                                     "-U%s%%%s" %
 323                                                     (os.environ["USERNAME"],
 324                                                      os.environ["PASSWORD"]))
 325                 self.assertCmdSuccess(result, out, err, "Ensuring gpo deleted successfully")
 326
 327             shutil.rmtree(temp_path)
 328             shutil.rmtree(new_path)
 329
 330     def test_backup_restore_generalize(self):
 331         """Restore from a static backup with different entities, generalize it
 332         again, and compare the XML"""
 333         static_path = os.path.join(self.backup_path, 'policy',
 334                                    self.backup_gpo_guid)
 335
 336         temp_path = os.path.join(self.tempdir, 'temp')
 337         os.mkdir(temp_path)
 338
 339         new_path = os.path.join(self.tempdir, 'new')
 340         os.mkdir(new_path)
 341
 342         alt_entity_file = os.path.join(new_path, 'entities')
 343         with open(alt_entity_file, 'wb') as f:
 344             f.write('''<!ENTITY SAMBA__NETWORK_PATH__82419dafed126a07d6b96c66fc943735__ "\\\\samdom.example.com">
 345 <!ENTITY SAMBA__NETWORK_PATH__0484cd41ded45a0728333a9c5e5ef619__ "\\\\samdom">
 346 <!ENTITY SAMBA____SDDL_ACL____4ce8277be3f630300cbcf80a80e21cf4__ "D:PAR(A;CI;KA;;;BA)(A;CIIO;KA;;;CO)(A;CI;KA;;;SY)(A;CI;KR;;;S-1-16-0)">
 347 <!ENTITY SAMBA____USER_ID_____d0970f5a1e19cb803f916c203d5c39c4__ "*S-1-5-113">
 348 <!ENTITY SAMBA____USER_ID_____7b7bc2512ee1fedcd76bdc68926d4f7b__ "Administrator">
 349 <!ENTITY SAMBA____USER_ID_____a3069f5a7a6530293ad8df6abd32af3d__ "Foobaz">
 350 <!ENTITY SAMBA____USER_ID_____fdf60b2473b319c8c341de5f62479a7d__ "*S-1-5-32-545">
 351 <!ENTITY SAMBA____USER_ID_____adb831a7fdd83dd1e2a309ce7591dff8__ "Guest">
 352 <!ENTITY SAMBA____USER_ID_____9fa835214b4fc8b6102c991f7d97c2f8__ "*S-1-5-32-547">
 353 <!ENTITY SAMBA____USER_ID_____bf8caafa94a19a6262bad2e8b6d4bce6__ "*S-1-5-32-546">
 354 <!ENTITY SAMBA____USER_ID_____a45da96d0bf6575970f2d27af22be28a__ "System">
 355 <!ENTITY SAMBA____USER_ID_____171d33a63ebd67f856552940ed491ad3__ "s-1-5-32-545">
 356 <!ENTITY SAMBA____USER_ID_____7140932fff16ce85cc64d3caab588d0d__ "s-1-1-0">
 357 ''')
 358
 359         gen_entity_file = os.path.join(temp_path, 'entities')
 360
 361         gpo_guid = None
 362         try:
 363             (result, out, err) = self.runsubcmd("gpo", "restore", "BACKUP_RESTORE1",
 364                                                 static_path,
 365                                                 "-H", "ldap://%s" %
 366                                                 os.environ["SERVER"], "--tmpdir",
 367                                                 temp_path, "--entities",
 368                                                 alt_entity_file, "-U%s%%%s" %
 369                                                 (os.environ["USERNAME"],
 370                                                  os.environ["PASSWORD"]))
 371
 372             self.assertCmdSuccess(result, out, err, "Ensuring gpo restored successfully")
 373
 374             gpo_guid = "{%s}" % out.split("{")[1].split("}")[0]
 375
 376             (result, out, err) = self.runsubcmd("gpo", "backup", gpo_guid,
 377                                                 "-H", "ldap://%s" %
 378                                                 os.environ["SERVER"],
 379                                                 "--tmpdir", new_path,
 380                                                 "--generalize", "--entities",
 381                                                 gen_entity_file)
 382
 383             self.assertCmdSuccess(result, out, err, "Ensuring gpo fetched successfully")
 384
 385             # Assert entity files are identical (except for line order)
 386             self.assertIsNone(has_difference(alt_entity_file,
 387                                              gen_entity_file,
 388                                              sortlines=True))
 389
 390             # Compare the directories (XML)
 391             self.assertIsNone(has_difference(os.path.join(new_path, 'policy',
 392                                                          gpo_guid),
 393                                             static_path, binary=False,
 394                                             xml=True))
 395         finally:
 396             if gpo_guid:
 397                 (result, out, err) = self.runsubcmd("gpo", "del", gpo_guid,
 398                                                     "-H", "ldap://%s" %
 399                                                     os.environ["SERVER"],
 400                                                     "-U%s%%%s" %
 401                                                     (os.environ["USERNAME"],
 402                                                      os.environ["PASSWORD"]))
 403                 self.assertCmdSuccess(result, out, err, "Ensuring gpo deleted successfully")
 404
 405             shutil.rmtree(temp_path)
 406             shutil.rmtree(new_path)
 407
 408     def setUp(self):
 409         """set up a temporary GPO to work with"""
 410         super(GpoCmdTestCase, self).setUp()
 411         (result, out, err) = self.runsubcmd("gpo", "create", self.gpo_name,
 412                                             "-H", "ldap://%s" % os.environ["SERVER"],
 413                                             "-U%s%%%s" % (os.environ["USERNAME"], os.environ["PASSWORD"]),
 414                                             "--tmpdir", self.tempdir)
 415         self.assertCmdSuccess(result, out, err, "Ensuring gpo created successfully")
 416         shutil.rmtree(os.path.join(self.tempdir, "policy"))
 417         try:
 418             self.gpo_guid = "{%s}" % out.split("{")[1].split("}")[0]
 419         except IndexError:
 420             self.fail("Failed to find GUID in output: %s" % out)
 421
 422         self.backup_path = os.path.join(samba.source_tree_topdir(), 'source4',
 423                                         'selftest', 'provisions',
 424                                         'generalized-gpo-backup')
 425
 426         self.entity_file = os.path.join(self.backup_path, 'entities')
 427
 428     def tearDown(self):
 429         """remove the temporary GPO to work with"""
 430         (result, out, err) = self.runsubcmd("gpo", "del", self.gpo_guid, "-H", "ldap://%s" % os.environ["SERVER"], "-U%s%%%s" % (os.environ["USERNAME"], os.environ["PASSWORD"]))
 431         self.assertCmdSuccess(result, out, err, "Ensuring gpo deleted successfully")
 432         super(GpoCmdTestCase, self).tearDown()