python/samba/tests/samba_tool/group.py

   1 # Unix SMB/CIFS implementation.
   2 # Copyright (C) Michael Adam 2012
   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 time
  20 import ldb
  21 from samba.tests.samba_tool.base import SambaToolCmdTest
  22 from samba import (
  23         nttime2unix,
  24         dsdb
  25         )
  26
  27
  28 class GroupCmdTestCase(SambaToolCmdTest):
  29     """Tests for samba-tool group subcommands"""
  30     groups = []
  31     samdb = None
  32
  33     def setUp(self):
  34         super(GroupCmdTestCase, self).setUp()
  35         self.samdb = self.getSamDB("-H", "ldap://%s" % os.environ["DC_SERVER"],
  36                                    "-U%s%%%s" % (os.environ["DC_USERNAME"], os.environ["DC_PASSWORD"]))
  37         self.groups = []
  38         self.groups.append(self._randomGroup({"name": "testgroup1"}))
  39         self.groups.append(self._randomGroup({"name": "testgroup2"}))
  40         self.groups.append(self._randomGroup({"name": "testgroup3"}))
  41         self.groups.append(self._randomGroup({"name": "testgroup4"}))
  42         self.groups.append(self._randomPosixGroup({"name": "posixgroup1"}))
  43         self.groups.append(self._randomPosixGroup({"name": "posixgroup2"}))
  44         self.groups.append(self._randomPosixGroup({"name": "posixgroup3"}))
  45         self.groups.append(self._randomPosixGroup({"name": "posixgroup4"}))
  46         self.groups.append(self._randomUnixGroup({"name": "unixgroup1"}))
  47         self.groups.append(self._randomUnixGroup({"name": "unixgroup2"}))
  48         self.groups.append(self._randomUnixGroup({"name": "unixgroup3"}))
  49         self.groups.append(self._randomUnixGroup({"name": "unixgroup4"}))
  50
  51         # setup the 12 groups and ensure they are correct
  52         for group in self.groups:
  53             (result, out, err) = group["createGroupFn"](group)
  54
  55             self.assertCmdSuccess(result, out, err)
  56             self.assertEqual(err, "", "There shouldn't be any error message")
  57
  58             if 'unix' in group["name"]:
  59                 self.assertIn("Modified Group '%s' successfully"
  60                               % group["name"], out)
  61             else:
  62                 self.assertIn("Added group %s" % group["name"], out)
  63
  64             group["checkGroupFn"](group)
  65
  66             found = self._find_group(group["name"])
  67
  68             self.assertIsNotNone(found)
  69
  70             self.assertEqual("%s" % found.get("name"), group["name"])
  71             self.assertEqual("%s" % found.get("description"), group["description"])
  72
  73     def tearDown(self):
  74         super(GroupCmdTestCase, self).tearDown()
  75         # clean up all the left over groups, just in case
  76         for group in self.groups:
  77             if self._find_group(group["name"]):
  78                 self.runsubcmd("group", "delete", group["name"])
  79
  80     def test_newgroup(self):
  81         """This tests the "group add" and "group delete" commands"""
  82         # try to add all the groups again, this should fail
  83         for group in self.groups:
  84             (result, out, err) = self._create_group(group)
  85             self.assertCmdFail(result, "Succeeded to create existing group")
  86             self.assertIn("LDAP error 68 LDAP_ENTRY_ALREADY_EXISTS", err)
  87
  88         # try to delete all the groups we just added
  89         for group in self.groups:
  90             (result, out, err) = self.runsubcmd("group", "delete", group["name"])
  91             self.assertCmdSuccess(result, out, err,
  92                                   "Failed to delete group '%s'" % group["name"])
  93             found = self._find_group(group["name"])
  94             self.assertIsNone(found,
  95                               "Deleted group '%s' still exists" % group["name"])
  96
  97         # test adding groups
  98         for group in self.groups:
  99             (result, out, err) = self.runsubcmd("group", "add", group["name"],
 100                                                 "--description=%s" % group["description"],
 101                                                 "-H", "ldap://%s" % os.environ["DC_SERVER"],
 102                                                 "-U%s%%%s" % (os.environ["DC_USERNAME"],
 103                                                               os.environ["DC_PASSWORD"]))
 104
 105             self.assertCmdSuccess(result, out, err)
 106             self.assertEqual(err, "", "There shouldn't be any error message")
 107             self.assertIn("Added group %s" % group["name"], out)
 108
 109             found = self._find_group(group["name"])
 110
 111             self.assertEqual("%s" % found.get("samaccountname"),
 112                               "%s" % group["name"])
 113
 114     def test_list(self):
 115         (result, out, err) = self.runsubcmd("group", "list",
 116                                             "-H", "ldap://%s" % os.environ["DC_SERVER"],
 117                                             "-U%s%%%s" % (os.environ["DC_USERNAME"],
 118                                                           os.environ["DC_PASSWORD"]))
 119         self.assertCmdSuccess(result, out, err, "Error running list")
 120
 121         search_filter = "(objectClass=group)"
 122
 123         grouplist = self.samdb.search(base=self.samdb.domain_dn(),
 124                                       scope=ldb.SCOPE_SUBTREE,
 125                                       expression=search_filter,
 126                                       attrs=["samaccountname"])
 127
 128         self.assertTrue(len(grouplist) > 0, "no groups found in samdb")
 129
 130         for groupobj in grouplist:
 131             name = str(groupobj.get("samaccountname", idx=0))
 132             found = self.assertMatch(out, name,
 133                                      "group '%s' not found" % name)
 134
 135     def test_list_verbose(self):
 136         (result, out, err) = self.runsubcmd("group", "list", "--verbose",
 137                                             "-H", "ldap://%s" % os.environ["DC_SERVER"],
 138                                             "-U%s%%%s" % (os.environ["DC_USERNAME"],
 139                                                           os.environ["DC_PASSWORD"]))
 140         self.assertCmdSuccess(result, out, err, "Error running list --verbose")
 141
 142         # use the output to build a dictionary, where key=group-name,
 143         # value=num-members
 144         output_memberships = {}
 145
 146         # split the output by line, skipping the first 2 header lines
 147         group_lines = out.split('\n')[2:-1]
 148         for line in group_lines:
 149             # split line by column whitespace (but keep the group name together
 150             # if it contains spaces)
 151             values = line.split("   ")
 152             name = values[0]
 153             num_members = int(values[-1])
 154             output_memberships[name] = num_members
 155
 156         # build up a similar dict using an LDAP search
 157         search_filter = "(objectClass=group)"
 158         grouplist = self.samdb.search(base=self.samdb.domain_dn(),
 159                                       scope=ldb.SCOPE_SUBTREE,
 160                                       expression=search_filter,
 161                                       attrs=["samaccountname", "member"])
 162         self.assertTrue(len(grouplist) > 0, "no groups found in samdb")
 163
 164         ldap_memberships = {}
 165         for groupobj in grouplist:
 166             name = str(groupobj.get("samaccountname", idx=0))
 167             num_members = len(groupobj.get("member", default=[]))
 168             ldap_memberships[name] = num_members
 169
 170         # check the command output matches LDAP
 171         self.assertTrue(output_memberships == ldap_memberships,
 172                         "Command output doesn't match LDAP results.\n" +
 173                         "Command='%s'\nLDAP='%s'" %(output_memberships,
 174                                                     ldap_memberships))
 175
 176     def test_list_full_dn(self):
 177         (result, out, err) = self.runsubcmd("group", "list", "--full-dn",
 178                                             "-H", "ldap://%s" % os.environ["DC_SERVER"],
 179                                             "-U%s%%%s" % (os.environ["DC_USERNAME"],
 180                                                           os.environ["DC_PASSWORD"]))
 181         self.assertCmdSuccess(result, out, err, "Error running list")
 182
 183         search_filter = "(objectClass=group)"
 184
 185         grouplist = self.samdb.search(base=self.samdb.domain_dn(),
 186                                       scope=ldb.SCOPE_SUBTREE,
 187                                       expression=search_filter,
 188                                       attrs=[])
 189
 190         self.assertTrue(len(grouplist) > 0, "no groups found in samdb")
 191
 192         for groupobj in grouplist:
 193             name = str(groupobj.get("dn", idx=0))
 194             found = self.assertMatch(out, name,
 195                                      "group '%s' not found" % name)
 196
 197     def test_list_base_dn(self):
 198         base_dn = "CN=Users"
 199         (result, out, err) = self.runsubcmd("group", "list", "--base-dn", base_dn,
 200                                             "-H", "ldap://%s" % os.environ["DC_SERVER"],
 201                                             "-U%s%%%s" % (os.environ["DC_USERNAME"],
 202                                                           os.environ["DC_PASSWORD"]))
 203         self.assertCmdSuccess(result, out, err, "Error running list")
 204
 205         search_filter = "(objectClass=group)"
 206
 207         grouplist = self.samdb.search(base=self.samdb.normalize_dn_in_domain(base_dn),
 208                                       scope=ldb.SCOPE_SUBTREE,
 209                                       expression=search_filter,
 210                                       attrs=["name"])
 211
 212         self.assertTrue(len(grouplist) > 0, "no groups found in samdb")
 213
 214         for groupobj in grouplist:
 215             name = str(groupobj.get("name", idx=0))
 216             found = self.assertMatch(out, name,
 217                                      "group '%s' not found" % name)
 218
 219     def test_listmembers(self):
 220         (result, out, err) = self.runsubcmd("group", "listmembers", "Domain Users",
 221                                             "-H", "ldap://%s" % os.environ["DC_SERVER"],
 222                                             "-U%s%%%s" % (os.environ["DC_USERNAME"],
 223                                                           os.environ["DC_PASSWORD"]))
 224         self.assertCmdSuccess(result, out, err, "Error running listmembers")
 225
 226         search_filter = "(|(primaryGroupID=513)(memberOf=CN=Domain Users,CN=Users,%s))" % self.samdb.domain_dn()
 227
 228         grouplist = self.samdb.search(base=self.samdb.domain_dn(),
 229                                       scope=ldb.SCOPE_SUBTREE,
 230                                       expression=search_filter,
 231                                       attrs=["samAccountName"])
 232
 233         self.assertTrue(len(grouplist) > 0, "no groups found in samdb")
 234
 235         for groupobj in grouplist:
 236             name = str(groupobj.get("samAccountName", idx=0))
 237             found = self.assertMatch(out, name, "group '%s' not found" % name)
 238
 239
 240     def test_listmembers_full_dn(self):
 241         (result, out, err) = self.runsubcmd("group", "listmembers", "Domain Users",
 242                                             "--full-dn",
 243                                             "-H", "ldap://%s" % os.environ["DC_SERVER"],
 244                                             "-U%s%%%s" % (os.environ["DC_USERNAME"],
 245                                                           os.environ["DC_PASSWORD"]))
 246         self.assertCmdSuccess(result, out, err, "Error running listmembers")
 247
 248         search_filter = "(|(primaryGroupID=513)(memberOf=CN=Domain Users,CN=Users,%s))" % self.samdb.domain_dn()
 249
 250         grouplist = self.samdb.search(base=self.samdb.domain_dn(),
 251                                       scope=ldb.SCOPE_SUBTREE,
 252                                       expression=search_filter,
 253                                       attrs=["dn"])
 254
 255         self.assertTrue(len(grouplist) > 0, "no groups found in samdb")
 256
 257         for groupobj in grouplist:
 258             name = str(groupobj.get("dn", idx=0))
 259             found = self.assertMatch(out, name, "group '%s' not found" % name)
 260
 261
 262     def test_move(self):
 263         full_ou_dn = str(self.samdb.normalize_dn_in_domain("OU=movetest"))
 264         (result, out, err) = self.runsubcmd("ou", "create", full_ou_dn)
 265         self.assertCmdSuccess(result, out, err)
 266         self.assertEqual(err, "", "There shouldn't be any error message")
 267         self.assertIn('Created ou "%s"' % full_ou_dn, out)
 268
 269         for group in self.groups:
 270             (result, out, err) = self.runsubcmd(
 271                 "group", "move", group["name"], full_ou_dn)
 272             self.assertCmdSuccess(result, out, err, "Error running move")
 273             self.assertIn('Moved group "%s" into "%s"' %
 274                           (group["name"], full_ou_dn), out)
 275
 276         # Should fail as groups objects are in OU
 277         (result, out, err) = self.runsubcmd("ou", "delete", full_ou_dn)
 278         self.assertCmdFail(result)
 279         self.assertIn(("subtree_delete: Unable to delete a non-leaf node "
 280                        "(it has %d children)!") % len(self.groups), err)
 281
 282         for group in self.groups:
 283             new_dn = "CN=Users,%s" % self.samdb.domain_dn()
 284             (result, out, err) = self.runsubcmd(
 285                 "group", "move", group["name"], new_dn)
 286             self.assertCmdSuccess(result, out, err, "Error running move")
 287             self.assertIn('Moved group "%s" into "%s"' %
 288                           (group["name"], new_dn), out)
 289
 290         (result, out, err) = self.runsubcmd("ou", "delete", full_ou_dn)
 291         self.assertCmdSuccess(result, out, err,
 292                               "Failed to delete ou '%s'" % full_ou_dn)
 293
 294     def test_show(self):
 295         """Assert that we can show a group correctly."""
 296         (result, out, err) = self.runsubcmd("group", "show", "Domain Users",
 297                                             "-H", "ldap://%s" % os.environ["DC_SERVER"],
 298                                             "-U%s%%%s" % (os.environ["DC_USERNAME"],
 299                                                           os.environ["DC_PASSWORD"]))
 300         self.assertCmdSuccess(result, out, err)
 301         self.assertEqual(err, "", "Shouldn't be any error messages")
 302         self.assertIn("dn: CN=Domain Users,CN=Users,DC=addom,DC=samba,DC=example,DC=com", out)
 303
 304     def _randomGroup(self, base={}):
 305         """create a group with random attribute values, you can specify base
 306  attributes"""
 307         group = {
 308             "name": self.randomName(),
 309             "description": self.randomName(count=100),
 310             "createGroupFn": self._create_group,
 311             "checkGroupFn": self._check_group,
 312         }
 313         group.update(base)
 314         return group
 315
 316     def _randomPosixGroup(self, base={}):
 317         """create a group with random attribute values and additional RFC2307
 318         attributes, you can specify base attributes"""
 319         group = self._randomGroup({})
 320         group.update(base)
 321         posixAttributes = {
 322             "unixdomain": self.randomName(),
 323             "gidNumber": self.randomXid(),
 324             "createGroupFn": self._create_posix_group,
 325             "checkGroupFn": self._check_posix_group,
 326         }
 327         group.update(posixAttributes)
 328         group.update(base)
 329         return group
 330
 331     def _randomUnixGroup(self, base={}):
 332         """create a group with random attribute values and additional RFC2307
 333         attributes, you can specify base attributes"""
 334         group = self._randomGroup({})
 335         group.update(base)
 336         posixAttributes = {
 337             "gidNumber": self.randomXid(),
 338             "createGroupFn": self._create_unix_group,
 339             "checkGroupFn": self._check_unix_group,
 340         }
 341         group.update(posixAttributes)
 342         group.update(base)
 343         return group
 344
 345     def _check_group(self, group):
 346         """ check if a group from SamDB has the same attributes as
 347  its template """
 348         found = self._find_group(group["name"])
 349
 350         self.assertEqual("%s" % found.get("name"), group["name"])
 351         self.assertEqual("%s" % found.get("description"), group["description"])
 352
 353     def _check_posix_group(self, group):
 354         """ check if a posix_group from SamDB has the same attributes as
 355  its template """
 356         found = self._find_group(group["name"])
 357
 358         self.assertEqual("%s" % found.get("gidNumber"), "%s" %
 359                           group["gidNumber"])
 360         self._check_group(group)
 361
 362     def _check_unix_group(self, group):
 363         """ check if a unix_group from SamDB has the same attributes as its
 364 template """
 365         found = self._find_group(group["name"])
 366
 367         self.assertEqual("%s" % found.get("gidNumber"), "%s" %
 368                           group["gidNumber"])
 369         self._check_group(group)
 370
 371     def _create_group(self, group):
 372         return self.runsubcmd("group", "add", group["name"],
 373                               "--description=%s" % group["description"],
 374                               "-H", "ldap://%s" % os.environ["DC_SERVER"],
 375                               "-U%s%%%s" % (os.environ["DC_USERNAME"],
 376                                             os.environ["DC_PASSWORD"]))
 377
 378     def _create_posix_group(self, group):
 379         """ create a new group with RFC2307 attributes """
 380         return self.runsubcmd("group", "add", group["name"],
 381                               "--description=%s" % group["description"],
 382                               "--nis-domain=%s" % group["unixdomain"],
 383                               "--gid-number=%s" % group["gidNumber"],
 384                               "-H", "ldap://%s" % os.environ["DC_SERVER"],
 385                               "-U%s%%%s" % (os.environ["DC_USERNAME"],
 386                               os.environ["DC_PASSWORD"]))
 387
 388     def _create_unix_group(self, group):
 389         """ Add RFC2307 attributes to a group"""
 390         self._create_group(group)
 391         return self.runsubcmd("group", "addunixattrs", group["name"],
 392                               "%s" % group["gidNumber"],
 393                               "-H", "ldap://%s" % os.environ["DC_SERVER"],
 394                               "-U%s%%%s" % (os.environ["DC_USERNAME"],
 395                               os.environ["DC_PASSWORD"]))
 396
 397     def _find_group(self, name):
 398         search_filter = ("(&(sAMAccountName=%s)(objectCategory=%s,%s))" %
 399                          (ldb.binary_encode(name),
 400                           "CN=Group,CN=Schema,CN=Configuration",
 401                           self.samdb.domain_dn()))
 402         grouplist = self.samdb.search(base=self.samdb.domain_dn(),
 403                                       scope=ldb.SCOPE_SUBTREE,
 404                                       expression=search_filter)
 405         if grouplist:
 406             return grouplist[0]
 407         else:
 408             return None
 409
 410     def test_stats(self):
 411         (result, out, err) = self.runsubcmd("group", "stats",
 412                                             "-H", "ldap://%s" % os.environ["DC_SERVER"],
 413                                             "-U%s%%%s" % (os.environ["DC_USERNAME"],
 414                                                           os.environ["DC_PASSWORD"]))
 415         self.assertCmdSuccess(result, out, err, "Error running stats")
 416
 417         # sanity-check the command reports 'total groups' correctly
 418         search_filter = "(objectClass=group)"
 419         grouplist = self.samdb.search(base=self.samdb.domain_dn(),
 420                                       scope=ldb.SCOPE_SUBTREE,
 421                                       expression=search_filter,
 422                                       attrs=[])
 423
 424         total_groups = len(grouplist)
 425         self.assertTrue("Total groups: {0}".format(total_groups) in out,
 426                         "Total groups not reported correctly")