python/samba/tests/samba_tool/user.py

   1 # Unix SMB/CIFS implementation.
   2 # Copyright (C) Sean Dague <sdague@linux.vnet.ibm.com> 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 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 class UserCmdTestCase(SambaToolCmdTest):
  28     """Tests for samba-tool user subcommands"""
  29     users = []
  30     samdb = None
  31
  32     def setUp(self):
  33         super(UserCmdTestCase, self).setUp()
  34         self.samdb = self.getSamDB("-H", "ldap://%s" % os.environ["DC_SERVER"],
  35             "-U%s%%%s" % (os.environ["DC_USERNAME"], os.environ["DC_PASSWORD"]))
  36         self.users = []
  37         self.users.append(self._randomUser({"name": "sambatool1", "company": "comp1"}))
  38         self.users.append(self._randomUser({"name": "sambatool2", "company": "comp1"}))
  39         self.users.append(self._randomUser({"name": "sambatool3", "company": "comp2"}))
  40         self.users.append(self._randomUser({"name": "sambatool4", "company": "comp2"}))
  41         self.users.append(self._randomPosixUser({"name": "posixuser1"}))
  42         self.users.append(self._randomPosixUser({"name": "posixuser2"}))
  43         self.users.append(self._randomPosixUser({"name": "posixuser3"}))
  44         self.users.append(self._randomPosixUser({"name": "posixuser4"}))
  45
  46         # setup the 8 users and ensure they are correct
  47         for user in self.users:
  48             (result, out, err) = user["createUserFn"](user)
  49
  50             self.assertCmdSuccess(result)
  51             self.assertEquals(err,"","Shouldn't be any error messages")
  52             self.assertIn("User '%s' created successfully" % user["name"], out)
  53
  54             user["checkUserFn"](user)
  55
  56
  57     def tearDown(self):
  58         super(UserCmdTestCase, self).tearDown()
  59         # clean up all the left over users, just in case
  60         for user in self.users:
  61             if self._find_user(user["name"]):
  62                 self.runsubcmd("user", "delete", user["name"])
  63
  64
  65     def test_newuser(self):
  66         # try to add all the users again, this should fail
  67         for user in self.users:
  68             (result, out, err) = self._create_user(user)
  69             self.assertCmdFail(result, "Ensure that create user fails")
  70             self.assertIn("LDAP error 68 LDAP_ENTRY_ALREADY_EXISTS", err)
  71
  72         # try to delete all the 4 users we just added
  73         for user in self.users:
  74             (result, out, err) = self.runsubcmd("user", "delete", user["name"])
  75             self.assertCmdSuccess(result, "Can we delete users")
  76             found = self._find_user(user["name"])
  77             self.assertIsNone(found)
  78
  79         # test adding users with --use-username-as-cn
  80         for user in self.users:
  81             (result, out, err) =  self.runsubcmd("user", "add", user["name"], user["password"],
  82                                                  "--use-username-as-cn",
  83                                                  "--surname=%s" % user["surname"],
  84                                                  "--given-name=%s" % user["given-name"],
  85                                                  "--job-title=%s" % user["job-title"],
  86                                                  "--department=%s" % user["department"],
  87                                                  "--description=%s" % user["description"],
  88                                                  "--company=%s" % user["company"],
  89                                                  "-H", "ldap://%s" % os.environ["DC_SERVER"],
  90                                                  "-U%s%%%s" % (os.environ["DC_USERNAME"], os.environ["DC_PASSWORD"]))
  91
  92             self.assertCmdSuccess(result)
  93             self.assertEquals(err,"","Shouldn't be any error messages")
  94             self.assertIn("User '%s' created successfully" % user["name"], out)
  95
  96             found = self._find_user(user["name"])
  97
  98             self.assertEquals("%s" % found.get("cn"), "%(name)s" % user)
  99             self.assertEquals("%s" % found.get("name"), "%(name)s" % user)
 100
 101
 102
 103     def test_setpassword(self):
 104         for user in self.users:
 105             newpasswd = self.randomPass()
 106             (result, out, err) = self.runsubcmd("user", "setpassword",
 107                                                 user["name"],
 108                                                 "--newpassword=%s" % newpasswd,
 109                                                 "-H", "ldap://%s" % os.environ["DC_SERVER"],
 110                                                 "-U%s%%%s" % (os.environ["DC_USERNAME"], os.environ["DC_PASSWORD"]))
 111             # self.assertCmdSuccess(result, "Ensure setpassword runs")
 112             self.assertEquals(err,"","setpassword with url")
 113             self.assertMatch(out, "Changed password OK", "setpassword with url")
 114
 115         for user in self.users:
 116             newpasswd = self.randomPass()
 117             (result, out, err) = self.runsubcmd("user", "setpassword",
 118                                                 user["name"],
 119                                                 "--newpassword=%s" % newpasswd)
 120             # self.assertCmdSuccess(result, "Ensure setpassword runs")
 121             self.assertEquals(err,"","setpassword without url")
 122             self.assertMatch(out, "Changed password OK", "setpassword without url")
 123
 124         for user in self.users:
 125             newpasswd = self.randomPass()
 126             (result, out, err) = self.runsubcmd("user", "setpassword",
 127                                                 user["name"],
 128                                                 "--newpassword=%s" % newpasswd,
 129                                                 "--must-change-at-next-login",
 130                                                 "-H", "ldap://%s" % os.environ["DC_SERVER"],
 131                                                 "-U%s%%%s" % (os.environ["DC_USERNAME"], os.environ["DC_PASSWORD"]))
 132             # self.assertCmdSuccess(result, "Ensure setpassword runs")
 133             self.assertEquals(err,"","setpassword with forced change")
 134             self.assertMatch(out, "Changed password OK", "setpassword with forced change")
 135
 136
 137
 138
 139     def test_setexpiry(self):
 140         twodays = time.time() + (2 * 24 * 60 * 60)
 141
 142         for user in self.users:
 143             (result, out, err) = self.runsubcmd("user", "setexpiry", user["name"],
 144                                                 "--days=2",
 145                                                 "-H", "ldap://%s" % os.environ["DC_SERVER"],
 146                                                 "-U%s%%%s" % (os.environ["DC_USERNAME"], os.environ["DC_PASSWORD"]))
 147             self.assertCmdSuccess(result, "Can we run setexpiry with names")
 148             self.assertIn("Expiry for user '%s' set to 2 days." % user["name"], out)
 149
 150         for user in self.users:
 151             found = self._find_user(user["name"])
 152
 153             expires = nttime2unix(int("%s" % found.get("accountExpires")))
 154             self.assertWithin(expires, twodays, 5, "Ensure account expires is within 5 seconds of the expected time")
 155
 156         # TODO: renable this after the filter case is sorted out
 157         if "filters are broken, bail now":
 158             return
 159
 160         # now run the expiration based on a filter
 161         fourdays = time.time() + (4 * 24 * 60 * 60)
 162         (result, out, err) = self.runsubcmd("user", "setexpiry",
 163                                                 "--filter", "(&(objectClass=user)(company=comp2))",
 164                                                 "--days=4",
 165                                                 "-H", "ldap://%s" % os.environ["DC_SERVER"],
 166                                                 "-U%s%%%s" % (os.environ["DC_USERNAME"], os.environ["DC_PASSWORD"]))
 167         self.assertCmdSuccess(result, "Can we run setexpiry with a filter")
 168
 169         for user in self.users:
 170             found = self._find_user(user["name"])
 171             if ("%s" % found.get("company")) == "comp2":
 172                 expires = nttime2unix(int("%s" % found.get("accountExpires")))
 173                 self.assertWithin(expires, fourdays, 5, "Ensure account expires is within 5 seconds of the expected time")
 174             else:
 175                 expires = nttime2unix(int("%s" % found.get("accountExpires")))
 176                 self.assertWithin(expires, twodays, 5, "Ensure account expires is within 5 seconds of the expected time")
 177
 178
 179     def test_list(self):
 180         (result, out, err) = self.runsubcmd("user", "list",
 181                                             "-H", "ldap://%s" % os.environ["DC_SERVER"],
 182                                             "-U%s%%%s" % (os.environ["DC_USERNAME"],
 183                                                           os.environ["DC_PASSWORD"]))
 184         self.assertCmdSuccess(result, "Error running list")
 185
 186         search_filter = ("(&(objectClass=user)(userAccountControl:%s:=%u))" %
 187                          (ldb.OID_COMPARATOR_AND, dsdb.UF_NORMAL_ACCOUNT))
 188
 189         userlist = self.samdb.search(base=self.samdb.domain_dn(),
 190                                      scope=ldb.SCOPE_SUBTREE,
 191                                      expression=search_filter,
 192                                      attrs=["samaccountname"])
 193
 194         self.assertTrue(len(userlist) > 0, "no users found in samdb")
 195
 196         for userobj in userlist:
 197             name = userobj.get("samaccountname", idx=0)
 198             found = self.assertMatch(out, name,
 199                                      "user '%s' not found" % name)
 200     def test_getpwent(self):
 201         try:
 202             import pwd
 203         except ImportError:
 204             self.skipTest("Skipping getpwent test, no 'pwd' module available")
 205             return
 206
 207         # get the current user's data for the test
 208         uid = os.geteuid()
 209         try:
 210             u = pwd.getpwuid(uid)
 211         except KeyError:
 212             self.skipTest("Skipping getpwent test, current EUID not found in NSS")
 213             return
 214
 215
 216 # samba-tool user add command didn't support users with empty gecos if none is
 217 # specified on the command line and the user hasn't one in the passwd file it
 218 # will fail, so let's add some contents
 219
 220         gecos = u[4]
 221         if (gecos is None or len(gecos) == 0):
 222             gecos = "Foo GECOS"
 223         user = self._randomPosixUser({
 224                         "name": u[0],
 225                         "uid": u[0],
 226                         "uidNumber": u[2],
 227                         "gidNumber": u[3],
 228                         "gecos": gecos,
 229                         "loginShell": u[6],
 230                         })
 231         # check if --rfc2307-from-nss sets the same values as we got from pwd.getpwuid()
 232         (result, out, err) = self.runsubcmd("user", "add", user["name"], user["password"],
 233                                                 "--surname=%s" % user["surname"],
 234                                                 "--given-name=%s" % user["given-name"],
 235                                                 "--job-title=%s" % user["job-title"],
 236                                                 "--department=%s" % user["department"],
 237                                                 "--description=%s" % user["description"],
 238                                                 "--company=%s" % user["company"],
 239                                                 "--gecos=%s" % user["gecos"],
 240                                                 "--rfc2307-from-nss",
 241                                                 "-H", "ldap://%s" % os.environ["DC_SERVER"],
 242                                                 "-U%s%%%s" % (os.environ["DC_USERNAME"], os.environ["DC_PASSWORD"]))
 243
 244         msg = "command should return %s" % err
 245         self.assertCmdSuccess(result, msg)
 246         self.assertEquals(err,"","Shouldn't be any error messages")
 247         self.assertIn("User '%s' created successfully" % user["name"], out)
 248
 249         self._check_posix_user(user)
 250         self.runsubcmd("user", "delete", user["name"])
 251
 252         # Check if overriding the attributes from NSS with explicit values works
 253         #
 254         # get a user with all random posix attributes
 255         user = self._randomPosixUser({"name": u[0]})
 256         # create a user with posix attributes from nss but override all of them with the
 257         # random ones just obtained
 258         (result, out, err) = self.runsubcmd("user", "add", user["name"], user["password"],
 259                                                 "--surname=%s" % user["surname"],
 260                                                 "--given-name=%s" % user["given-name"],
 261                                                 "--job-title=%s" % user["job-title"],
 262                                                 "--department=%s" % user["department"],
 263                                                 "--description=%s" % user["description"],
 264                                                 "--company=%s" % user["company"],
 265                                                 "--rfc2307-from-nss",
 266                                                 "--gecos=%s" % user["gecos"],
 267                                                 "--login-shell=%s" % user["loginShell"],
 268                                                 "--uid=%s" % user["uid"],
 269                                                 "--uid-number=%s" % user["uidNumber"],
 270                                                 "--gid-number=%s" % user["gidNumber"],
 271                                                 "-H", "ldap://%s" % os.environ["DC_SERVER"],
 272                                                 "-U%s%%%s" % (os.environ["DC_USERNAME"], os.environ["DC_PASSWORD"]))
 273
 274         msg = "command should return %s" % err
 275         self.assertCmdSuccess(result, msg)
 276         self.assertEquals(err,"","Shouldn't be any error messages")
 277         self.assertIn("User '%s' created successfully" % user["name"], out)
 278
 279         self._check_posix_user(user)
 280         self.runsubcmd("user", "delete", user["name"])
 281
 282     def _randomUser(self, base={}):
 283         """create a user with random attribute values, you can specify base attributes"""
 284         user = {
 285             "name": self.randomName(),
 286             "password": self.randomPass(),
 287             "surname": self.randomName(),
 288             "given-name": self.randomName(),
 289             "job-title": self.randomName(),
 290             "department": self.randomName(),
 291             "company": self.randomName(),
 292             "description": self.randomName(count=100),
 293             "createUserFn": self._create_user,
 294             "checkUserFn": self._check_user,
 295             }
 296         user.update(base)
 297         return user
 298
 299     def _randomPosixUser(self, base={}):
 300         """create a user with random attribute values and additional RFC2307
 301         attributes, you can specify base attributes"""
 302         user = self._randomUser({})
 303         user.update(base)
 304         posixAttributes = {
 305             "uid": self.randomName(),
 306             "loginShell": self.randomName(),
 307             "gecos": self.randomName(),
 308             "uidNumber": self.randomXid(),
 309             "gidNumber": self.randomXid(),
 310             "createUserFn": self._create_posix_user,
 311             "checkUserFn": self._check_posix_user,
 312         }
 313         user.update(posixAttributes)
 314         user.update(base)
 315         return user
 316
 317     def _check_user(self, user):
 318         """ check if a user from SamDB has the same attributes as its template """
 319         found = self._find_user(user["name"])
 320
 321         self.assertEquals("%s" % found.get("name"), "%(given-name)s %(surname)s" % user)
 322         self.assertEquals("%s" % found.get("title"), user["job-title"])
 323         self.assertEquals("%s" % found.get("company"), user["company"])
 324         self.assertEquals("%s" % found.get("description"), user["description"])
 325         self.assertEquals("%s" % found.get("department"), user["department"])
 326
 327     def _check_posix_user(self, user):
 328         """ check if a posix_user from SamDB has the same attributes as its template """
 329         found = self._find_user(user["name"])
 330
 331         self.assertEquals("%s" % found.get("loginShell"), user["loginShell"])
 332         self.assertEquals("%s" % found.get("gecos"), user["gecos"])
 333         self.assertEquals("%s" % found.get("uidNumber"), "%s" % user["uidNumber"])
 334         self.assertEquals("%s" % found.get("gidNumber"), "%s" % user["gidNumber"])
 335         self.assertEquals("%s" % found.get("uid"), user["uid"])
 336         self._check_user(user)
 337
 338     def _create_user(self, user):
 339         return self.runsubcmd("user", "add", user["name"], user["password"],
 340                                                 "--surname=%s" % user["surname"],
 341                                                 "--given-name=%s" % user["given-name"],
 342                                                 "--job-title=%s" % user["job-title"],
 343                                                 "--department=%s" % user["department"],
 344                                                 "--description=%s" % user["description"],
 345                                                 "--company=%s" % user["company"],
 346                                                 "-H", "ldap://%s" % os.environ["DC_SERVER"],
 347                                                 "-U%s%%%s" % (os.environ["DC_USERNAME"], os.environ["DC_PASSWORD"]))
 348     def _create_posix_user(self, user):
 349         """ create a new user with RFC2307 attributes """
 350         return self.runsubcmd("user", "create", user["name"], user["password"],
 351                                                 "--surname=%s" % user["surname"],
 352                                                 "--given-name=%s" % user["given-name"],
 353                                                 "--job-title=%s" % user["job-title"],
 354                                                 "--department=%s" % user["department"],
 355                                                 "--description=%s" % user["description"],
 356                                                 "--company=%s" % user["company"],
 357                                                 "--gecos=%s" % user["gecos"],
 358                                                 "--login-shell=%s" % user["loginShell"],
 359                                                 "--uid=%s" % user["uid"],
 360                                                 "--uid-number=%s" % user["uidNumber"],
 361                                                 "--gid-number=%s" % user["gidNumber"],
 362                                                 "-H", "ldap://%s" % os.environ["DC_SERVER"],
 363                                                 "-U%s%%%s" % (os.environ["DC_USERNAME"], os.environ["DC_PASSWORD"]))
 364
 365     def _find_user(self, name):
 366         search_filter = "(&(sAMAccountName=%s)(objectCategory=%s,%s))" % (ldb.binary_encode(name), "CN=Person,CN=Schema,CN=Configuration", self.samdb.domain_dn())
 367         userlist = self.samdb.search(base=self.samdb.domain_dn(),
 368                                   scope=ldb.SCOPE_SUBTREE,
 369                                   expression=search_filter, attrs=[])
 370         if userlist:
 371             return userlist[0]
 372         else:
 373             return None