ceph/src/pybind/mgr/dashboard/tests/test_access_control.py

   1 # -*- coding: utf-8 -*-
   2 # pylint: disable=dangerous-default-value,too-many-public-methods
   3 from __future__ import absolute_import
   4
   5 import errno
   6 import json
   7 import time
   8 import unittest
   9
  10 from datetime import datetime, timedelta
  11
  12 from . import CmdException, CLICommandTestMixin
  13 from .. import mgr
  14 from ..security import Scope, Permission
  15 from ..services.access_control import load_access_control_db, \
  16                                       password_hash, AccessControlDB, \
  17                                       SYSTEM_ROLES, PasswordPolicy
  18 from ..settings import Settings
  19
  20
  21 class AccessControlTest(unittest.TestCase, CLICommandTestMixin):
  22
  23     @classmethod
  24     def setUpClass(cls):
  25         cls.mock_kv_store()
  26         mgr.ACCESS_CONTROL_DB = None
  27
  28     def setUp(self):
  29         self.CONFIG_KEY_DICT.clear()
  30         load_access_control_db()
  31
  32     def load_persistent_db(self):
  33         config_key = AccessControlDB.accessdb_config_key()
  34         self.assertIn(config_key, self.CONFIG_KEY_DICT)
  35         db_json = self.CONFIG_KEY_DICT[config_key]
  36         db = json.loads(db_json)
  37         return db
  38
  39     # The DB is written to persistent storage the first time it is saved.
  40     # However, should an operation fail due to <reasons>, we may end up in
  41     # a state where we have a completely empty CONFIG_KEY_DICT (our mock
  42     # equivalent to the persistent state). While this works for most of the
  43     # tests in this class, that would prevent us from testing things like
  44     # "run a command that is expected to fail, and then ensure nothing
  45     # happened", because we'd be asserting in `load_persistent_db()` due to
  46     # the map being empty.
  47     #
  48     # This function will therefore force state to be written to our mock
  49     # persistent state. We could have added this extra step to
  50     # `load_persistent_db()` directly, but that would conflict with the
  51     # upgrade tests. This way, we can selectively enforce this requirement
  52     # where we believe it to be necessary; generically speaking, this should
  53     # not be needed unless we're testing very specific behaviors.
  54     #
  55     def setup_and_load_persistent_db(self):
  56         mgr.ACCESS_CTRL_DB.save()
  57         self.load_persistent_db()
  58
  59     def validate_persistent_role(self, rolename, scopes_permissions,
  60                                  description=None):
  61         db = self.load_persistent_db()
  62         self.assertIn('roles', db)
  63         self.assertIn(rolename, db['roles'])
  64         self.assertEqual(db['roles'][rolename]['name'], rolename)
  65         self.assertEqual(db['roles'][rolename]['description'], description)
  66         self.assertDictEqual(db['roles'][rolename]['scopes_permissions'],
  67                              scopes_permissions)
  68
  69     def validate_persistent_no_role(self, rolename):
  70         db = self.load_persistent_db()
  71         self.assertIn('roles', db)
  72         self.assertNotIn(rolename, db['roles'])
  73
  74     def validate_persistent_user(self, username, roles, password=None,
  75                                  name=None, email=None, last_update=None,
  76                                  enabled=True, pwdExpirationDate=None):
  77         db = self.load_persistent_db()
  78         self.assertIn('users', db)
  79         self.assertIn(username, db['users'])
  80         self.assertEqual(db['users'][username]['username'], username)
  81         self.assertListEqual(db['users'][username]['roles'], roles)
  82         if password:
  83             self.assertEqual(db['users'][username]['password'], password)
  84         if name:
  85             self.assertEqual(db['users'][username]['name'], name)
  86         if email:
  87             self.assertEqual(db['users'][username]['email'], email)
  88         if last_update:
  89             self.assertEqual(db['users'][username]['lastUpdate'], last_update)
  90         if pwdExpirationDate:
  91             self.assertEqual(db['users'][username]['pwdExpirationDate'], pwdExpirationDate)
  92         self.assertEqual(db['users'][username]['enabled'], enabled)
  93
  94     def validate_persistent_no_user(self, username):
  95         db = self.load_persistent_db()
  96         self.assertIn('users', db)
  97         self.assertNotIn(username, db['users'])
  98
  99     def test_create_role(self):
 100         role = self.exec_cmd('ac-role-create', rolename='test_role')
 101         self.assertDictEqual(role, {'name': 'test_role', 'description': None,
 102                                     'scopes_permissions': {}})
 103         self.validate_persistent_role('test_role', {})
 104
 105     def test_create_role_with_desc(self):
 106         role = self.exec_cmd('ac-role-create', rolename='test_role',
 107                              description='Test Role')
 108         self.assertDictEqual(role, {'name': 'test_role',
 109                                     'description': 'Test Role',
 110                                     'scopes_permissions': {}})
 111         self.validate_persistent_role('test_role', {}, 'Test Role')
 112
 113     def test_create_duplicate_role(self):
 114         self.test_create_role()
 115
 116         with self.assertRaises(CmdException) as ctx:
 117             self.exec_cmd('ac-role-create', rolename='test_role')
 118
 119         self.assertEqual(ctx.exception.retcode, -errno.EEXIST)
 120         self.assertEqual(str(ctx.exception), "Role 'test_role' already exists")
 121
 122     def test_delete_role(self):
 123         self.test_create_role()
 124         out = self.exec_cmd('ac-role-delete', rolename='test_role')
 125         self.assertEqual(out, "Role 'test_role' deleted")
 126         self.validate_persistent_no_role('test_role')
 127
 128     def test_delete_nonexistent_role(self):
 129         with self.assertRaises(CmdException) as ctx:
 130             self.exec_cmd('ac-role-delete', rolename='test_role')
 131
 132         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 133         self.assertEqual(str(ctx.exception), "Role 'test_role' does not exist")
 134
 135     def test_show_single_role(self):
 136         self.test_create_role()
 137         role = self.exec_cmd('ac-role-show', rolename='test_role')
 138         self.assertDictEqual(role, {'name': 'test_role', 'description': None,
 139                                     'scopes_permissions': {}})
 140
 141     def test_show_nonexistent_role(self):
 142         with self.assertRaises(CmdException) as ctx:
 143             self.exec_cmd('ac-role-show', rolename='test_role')
 144
 145         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 146         self.assertEqual(str(ctx.exception), "Role 'test_role' does not exist")
 147
 148     def test_show_system_roles(self):
 149         roles = self.exec_cmd('ac-role-show')
 150         self.assertEqual(len(roles), len(SYSTEM_ROLES))
 151         for role in roles:
 152             self.assertIn(role, SYSTEM_ROLES)
 153
 154     def test_show_system_role(self):
 155         role = self.exec_cmd('ac-role-show', rolename="read-only")
 156         self.assertEqual(role['name'], 'read-only')
 157         self.assertEqual(role['description'], 'Read-Only')
 158
 159     def test_delete_system_role(self):
 160         with self.assertRaises(CmdException) as ctx:
 161             self.exec_cmd('ac-role-delete', rolename='administrator')
 162
 163         self.assertEqual(ctx.exception.retcode, -errno.EPERM)
 164         self.assertEqual(str(ctx.exception),
 165                          "Cannot delete system role 'administrator'")
 166
 167     def test_add_role_scope_perms(self):
 168         self.test_create_role()
 169         self.exec_cmd('ac-role-add-scope-perms', rolename='test_role',
 170                       scopename=Scope.POOL,
 171                       permissions=[Permission.READ, Permission.DELETE])
 172         role = self.exec_cmd('ac-role-show', rolename='test_role')
 173         self.assertDictEqual(role, {'name': 'test_role',
 174                                     'description': None,
 175                                     'scopes_permissions': {
 176                                         Scope.POOL: [Permission.DELETE,
 177                                                      Permission.READ]
 178                                     }})
 179         self.validate_persistent_role('test_role', {
 180             Scope.POOL: [Permission.DELETE, Permission.READ]
 181         })
 182
 183     def test_del_role_scope_perms(self):
 184         self.test_add_role_scope_perms()
 185         self.exec_cmd('ac-role-add-scope-perms', rolename='test_role',
 186                       scopename=Scope.MONITOR,
 187                       permissions=[Permission.READ, Permission.CREATE])
 188         self.validate_persistent_role('test_role', {
 189             Scope.POOL: [Permission.DELETE, Permission.READ],
 190             Scope.MONITOR: [Permission.CREATE, Permission.READ]
 191         })
 192         self.exec_cmd('ac-role-del-scope-perms', rolename='test_role',
 193                       scopename=Scope.POOL)
 194         role = self.exec_cmd('ac-role-show', rolename='test_role')
 195         self.assertDictEqual(role, {'name': 'test_role',
 196                                     'description': None,
 197                                     'scopes_permissions': {
 198                                         Scope.MONITOR: [Permission.CREATE,
 199                                                         Permission.READ]
 200                                     }})
 201         self.validate_persistent_role('test_role', {
 202             Scope.MONITOR: [Permission.CREATE, Permission.READ]
 203         })
 204
 205     def test_add_role_scope_perms_nonexistent_role(self):
 206
 207         with self.assertRaises(CmdException) as ctx:
 208             self.exec_cmd('ac-role-add-scope-perms', rolename='test_role',
 209                           scopename='pool',
 210                           permissions=['read', 'delete'])
 211
 212         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 213         self.assertEqual(str(ctx.exception), "Role 'test_role' does not exist")
 214
 215     def test_add_role_invalid_scope_perms(self):
 216         self.test_create_role()
 217
 218         with self.assertRaises(CmdException) as ctx:
 219             self.exec_cmd('ac-role-add-scope-perms', rolename='test_role',
 220                           scopename='invalidscope',
 221                           permissions=['read', 'delete'])
 222
 223         self.assertEqual(ctx.exception.retcode, -errno.EINVAL)
 224         self.assertEqual(str(ctx.exception),
 225                          "Scope 'invalidscope' is not valid\n Possible values: "
 226                          "{}".format(Scope.all_scopes()))
 227
 228     def test_add_role_scope_invalid_perms(self):
 229         self.test_create_role()
 230
 231         with self.assertRaises(CmdException) as ctx:
 232             self.exec_cmd('ac-role-add-scope-perms', rolename='test_role',
 233                           scopename='pool', permissions=['invalidperm'])
 234
 235         self.assertEqual(ctx.exception.retcode, -errno.EINVAL)
 236         self.assertEqual(str(ctx.exception),
 237                          "Permission 'invalidperm' is not valid\n Possible "
 238                          "values: {}".format(Permission.all_permissions()))
 239
 240     def test_del_role_scope_perms_nonexistent_role(self):
 241
 242         with self.assertRaises(CmdException) as ctx:
 243             self.exec_cmd('ac-role-del-scope-perms', rolename='test_role',
 244                           scopename='pool')
 245
 246         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 247         self.assertEqual(str(ctx.exception), "Role 'test_role' does not exist")
 248
 249     def test_del_role_nonexistent_scope_perms(self):
 250         self.test_add_role_scope_perms()
 251
 252         with self.assertRaises(CmdException) as ctx:
 253             self.exec_cmd('ac-role-del-scope-perms', rolename='test_role',
 254                           scopename='nonexistentscope')
 255
 256         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 257         self.assertEqual(str(ctx.exception),
 258                          "There are no permissions for scope 'nonexistentscope' "
 259                          "in role 'test_role'")
 260
 261     def test_not_permitted_add_role_scope_perms(self):
 262         with self.assertRaises(CmdException) as ctx:
 263             self.exec_cmd('ac-role-add-scope-perms', rolename='read-only',
 264                           scopename='pool', permissions=['read', 'delete'])
 265
 266         self.assertEqual(ctx.exception.retcode, -errno.EPERM)
 267         self.assertEqual(str(ctx.exception),
 268                          "Cannot update system role 'read-only'")
 269
 270     def test_not_permitted_del_role_scope_perms(self):
 271         with self.assertRaises(CmdException) as ctx:
 272             self.exec_cmd('ac-role-del-scope-perms', rolename='read-only',
 273                           scopename='pool')
 274
 275         self.assertEqual(ctx.exception.retcode, -errno.EPERM)
 276         self.assertEqual(str(ctx.exception),
 277                          "Cannot update system role 'read-only'")
 278
 279     def test_create_user(self, username='admin', rolename=None, enabled=True,
 280                          pwdExpirationDate=None):
 281         user = self.exec_cmd('ac-user-create', username=username,
 282                              rolename=rolename, password='admin',
 283                              name='{} User'.format(username),
 284                              email='{}@user.com'.format(username),
 285                              enabled=enabled, force_password=True,
 286                              pwd_expiration_date=pwdExpirationDate)
 287
 288         pass_hash = password_hash('admin', user['password'])
 289         self.assertDictEqual(user, {
 290             'username': username,
 291             'password': pass_hash,
 292             'pwdExpirationDate': pwdExpirationDate,
 293             'pwdUpdateRequired': False,
 294             'lastUpdate': user['lastUpdate'],
 295             'name': '{} User'.format(username),
 296             'email': '{}@user.com'.format(username),
 297             'roles': [rolename] if rolename else [],
 298             'enabled': enabled
 299         })
 300         self.validate_persistent_user(username, [rolename] if rolename else [],
 301                                       pass_hash, '{} User'.format(username),
 302                                       '{}@user.com'.format(username),
 303                                       user['lastUpdate'], enabled)
 304         return user
 305
 306     def test_create_disabled_user(self):
 307         self.test_create_user(enabled=False)
 308
 309     def test_create_user_pwd_expiration_date(self):
 310         expiration_date = datetime.utcnow() + timedelta(days=10)
 311         expiration_date = int(time.mktime(expiration_date.timetuple()))
 312         self.test_create_user(pwdExpirationDate=expiration_date)
 313
 314     def test_create_user_with_role(self):
 315         self.test_add_role_scope_perms()
 316         self.test_create_user(rolename='test_role')
 317
 318     def test_create_user_with_system_role(self):
 319         self.test_create_user(rolename='administrator')
 320
 321     def test_delete_user(self):
 322         self.test_create_user()
 323         out = self.exec_cmd('ac-user-delete', username='admin')
 324         self.assertEqual(out, "User 'admin' deleted")
 325         users = self.exec_cmd('ac-user-show')
 326         self.assertEqual(len(users), 0)
 327         self.validate_persistent_no_user('admin')
 328
 329     def test_create_duplicate_user(self):
 330         self.test_create_user()
 331
 332         with self.assertRaises(CmdException) as ctx:
 333             self.exec_cmd('ac-user-create', username='admin', password='admin',
 334                           force_password=True)
 335
 336         self.assertEqual(ctx.exception.retcode, -errno.EEXIST)
 337         self.assertEqual(str(ctx.exception), "User 'admin' already exists")
 338
 339     def test_create_users_with_dne_role(self):
 340         # one time call to setup our persistent db
 341         self.setup_and_load_persistent_db()
 342
 343         # create a user with a role that does not exist; expect a failure
 344         try:
 345             self.exec_cmd('ac-user-create', username='foo',
 346                           rolename='dne_role', password='foopass',
 347                           name='foo User', email='foo@user.com',
 348                           force_password=True)
 349         except CmdException as e:
 350             self.assertEqual(e.retcode, -errno.ENOENT)
 351
 352         db = self.load_persistent_db()
 353         if 'users' in db:
 354             self.assertNotIn('foo', db['users'])
 355
 356         # We could just finish our test here, given we ensured that the user
 357         # with a non-existent role is not in persistent storage. However,
 358         # we're going to test the database's consistency, making sure that
 359         # side-effects are not written to persistent storage once we commit
 360         # an unrelated operation. To ensure this, we'll issue another
 361         # operation that is sharing the same code path, and will check whether
 362         # the next operation commits dirty state.
 363
 364         # create a role (this will be 'test_role')
 365         self.test_create_role()
 366         self.exec_cmd('ac-user-create', username='bar',
 367                       rolename='test_role', password='barpass',
 368                       name='bar User', email='bar@user.com',
 369                       force_password=True)
 370
 371         # validate db:
 372         #   user 'foo' should not exist
 373         #   user 'bar' should exist and have role 'test_role'
 374         self.validate_persistent_user('bar', ['test_role'])
 375
 376         db = self.load_persistent_db()
 377         self.assertIn('users', db)
 378         self.assertNotIn('foo', db['users'])
 379
 380     def test_delete_nonexistent_user(self):
 381         with self.assertRaises(CmdException) as ctx:
 382             self.exec_cmd('ac-user-delete', username='admin')
 383
 384         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 385         self.assertEqual(str(ctx.exception), "User 'admin' does not exist")
 386
 387     def test_add_user_roles(self, username='admin',
 388                             roles=['pool-manager', 'block-manager']):
 389         user_orig = self.test_create_user(username)
 390         uroles = []
 391         for role in roles:
 392             uroles.append(role)
 393             uroles.sort()
 394             user = self.exec_cmd('ac-user-add-roles', username=username,
 395                                  roles=[role])
 396             self.assertDictContainsSubset({'roles': uroles}, user)
 397         self.validate_persistent_user(username, uroles)
 398         self.assertGreaterEqual(user['lastUpdate'], user_orig['lastUpdate'])
 399
 400     def test_add_user_roles2(self):
 401         user_orig = self.test_create_user()
 402         user = self.exec_cmd('ac-user-add-roles', username="admin",
 403                              roles=['pool-manager', 'block-manager'])
 404         self.assertDictContainsSubset(
 405             {'roles': ['block-manager', 'pool-manager']}, user)
 406         self.validate_persistent_user('admin', ['block-manager',
 407                                                 'pool-manager'])
 408         self.assertGreaterEqual(user['lastUpdate'], user_orig['lastUpdate'])
 409
 410     def test_add_user_roles_not_existent_user(self):
 411         with self.assertRaises(CmdException) as ctx:
 412             self.exec_cmd('ac-user-add-roles', username="admin",
 413                           roles=['pool-manager', 'block-manager'])
 414
 415         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 416         self.assertEqual(str(ctx.exception), "User 'admin' does not exist")
 417
 418     def test_add_user_roles_not_existent_role(self):
 419         self.test_create_user()
 420         with self.assertRaises(CmdException) as ctx:
 421             self.exec_cmd('ac-user-add-roles', username="admin",
 422                           roles=['Invalid Role'])
 423
 424         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 425         self.assertEqual(str(ctx.exception),
 426                          "Role 'Invalid Role' does not exist")
 427
 428     def test_set_user_roles(self):
 429         user_orig = self.test_create_user()
 430         user = self.exec_cmd('ac-user-add-roles', username="admin",
 431                              roles=['pool-manager'])
 432         self.assertDictContainsSubset(
 433             {'roles': ['pool-manager']}, user)
 434         self.validate_persistent_user('admin', ['pool-manager'])
 435         self.assertGreaterEqual(user['lastUpdate'], user_orig['lastUpdate'])
 436         user2 = self.exec_cmd('ac-user-set-roles', username="admin",
 437                               roles=['rgw-manager', 'block-manager'])
 438         self.assertDictContainsSubset(
 439             {'roles': ['block-manager', 'rgw-manager']}, user2)
 440         self.validate_persistent_user('admin', ['block-manager',
 441                                                 'rgw-manager'])
 442         self.assertGreaterEqual(user2['lastUpdate'], user['lastUpdate'])
 443
 444     def test_set_user_roles_not_existent_user(self):
 445         with self.assertRaises(CmdException) as ctx:
 446             self.exec_cmd('ac-user-set-roles', username="admin",
 447                           roles=['pool-manager', 'block-manager'])
 448
 449         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 450         self.assertEqual(str(ctx.exception), "User 'admin' does not exist")
 451
 452     def test_set_user_roles_not_existent_role(self):
 453         self.test_create_user()
 454         with self.assertRaises(CmdException) as ctx:
 455             self.exec_cmd('ac-user-set-roles', username="admin",
 456                           roles=['Invalid Role'])
 457
 458         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 459         self.assertEqual(str(ctx.exception),
 460                          "Role 'Invalid Role' does not exist")
 461
 462     def test_del_user_roles(self):
 463         self.test_add_user_roles()
 464         user = self.exec_cmd('ac-user-del-roles', username="admin",
 465                              roles=['pool-manager'])
 466         self.assertDictContainsSubset(
 467             {'roles': ['block-manager']}, user)
 468         self.validate_persistent_user('admin', ['block-manager'])
 469
 470     def test_del_user_roles_not_existent_user(self):
 471         with self.assertRaises(CmdException) as ctx:
 472             self.exec_cmd('ac-user-del-roles', username="admin",
 473                           roles=['pool-manager', 'block-manager'])
 474
 475         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 476         self.assertEqual(str(ctx.exception), "User 'admin' does not exist")
 477
 478     def test_del_user_roles_not_existent_role(self):
 479         self.test_create_user()
 480         with self.assertRaises(CmdException) as ctx:
 481             self.exec_cmd('ac-user-del-roles', username="admin",
 482                           roles=['Invalid Role'])
 483
 484         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 485         self.assertEqual(str(ctx.exception),
 486                          "Role 'Invalid Role' does not exist")
 487
 488     def test_del_user_roles_not_associated_role(self):
 489         self.test_create_user()
 490         with self.assertRaises(CmdException) as ctx:
 491             self.exec_cmd('ac-user-del-roles', username="admin",
 492                           roles=['rgw-manager'])
 493
 494         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 495         self.assertEqual(str(ctx.exception),
 496                          "Role 'rgw-manager' is not associated with user "
 497                          "'admin'")
 498
 499     def test_show_user(self):
 500         self.test_add_user_roles()
 501         user = self.exec_cmd('ac-user-show', username='admin')
 502         pass_hash = password_hash('admin', user['password'])
 503         self.assertDictEqual(user, {
 504             'username': 'admin',
 505             'lastUpdate': user['lastUpdate'],
 506             'password': pass_hash,
 507             'pwdExpirationDate': None,
 508             'pwdUpdateRequired': False,
 509             'name': 'admin User',
 510             'email': 'admin@user.com',
 511             'roles': ['block-manager', 'pool-manager'],
 512             'enabled': True
 513         })
 514
 515     def test_show_nonexistent_user(self):
 516         with self.assertRaises(CmdException) as ctx:
 517             self.exec_cmd('ac-user-show', username='admin')
 518
 519         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 520         self.assertEqual(str(ctx.exception), "User 'admin' does not exist")
 521
 522     def test_show_all_users(self):
 523         self.test_add_user_roles('admin', ['administrator'])
 524         self.test_add_user_roles('guest', ['read-only'])
 525         users = self.exec_cmd('ac-user-show')
 526         self.assertEqual(len(users), 2)
 527         for user in users:
 528             self.assertIn(user, ['admin', 'guest'])
 529
 530     def test_del_role_associated_with_user(self):
 531         self.test_create_role()
 532         self.test_add_user_roles('guest', ['test_role'])
 533
 534         with self.assertRaises(CmdException) as ctx:
 535             self.exec_cmd('ac-role-delete', rolename='test_role')
 536
 537         self.assertEqual(ctx.exception.retcode, -errno.EPERM)
 538         self.assertEqual(str(ctx.exception),
 539                          "Role 'test_role' is still associated with user "
 540                          "'guest'")
 541
 542     def test_set_user_info(self):
 543         user_orig = self.test_create_user()
 544         user = self.exec_cmd('ac-user-set-info', username='admin',
 545                              name='Admin Name', email='admin@admin.com')
 546         pass_hash = password_hash('admin', user['password'])
 547         self.assertDictEqual(user, {
 548             'username': 'admin',
 549             'password': pass_hash,
 550             'pwdExpirationDate': None,
 551             'pwdUpdateRequired': False,
 552             'name': 'Admin Name',
 553             'email': 'admin@admin.com',
 554             'lastUpdate': user['lastUpdate'],
 555             'roles': [],
 556             'enabled': True
 557         })
 558         self.validate_persistent_user('admin', [], pass_hash, 'Admin Name',
 559                                       'admin@admin.com')
 560         self.assertEqual(user['lastUpdate'], user_orig['lastUpdate'])
 561
 562     def test_set_user_info_nonexistent_user(self):
 563         with self.assertRaises(CmdException) as ctx:
 564             self.exec_cmd('ac-user-set-info', username='admin',
 565                           name='Admin Name', email='admin@admin.com')
 566
 567         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 568         self.assertEqual(str(ctx.exception), "User 'admin' does not exist")
 569
 570     def test_set_user_password(self):
 571         user_orig = self.test_create_user()
 572         user = self.exec_cmd('ac-user-set-password', username='admin',
 573                              password='newpass', force_password=True)
 574         pass_hash = password_hash('newpass', user['password'])
 575         self.assertDictEqual(user, {
 576             'username': 'admin',
 577             'password': pass_hash,
 578             'pwdExpirationDate': None,
 579             'pwdUpdateRequired': False,
 580             'name': 'admin User',
 581             'email': 'admin@user.com',
 582             'lastUpdate': user['lastUpdate'],
 583             'roles': [],
 584             'enabled': True
 585         })
 586         self.validate_persistent_user('admin', [], pass_hash, 'admin User',
 587                                       'admin@user.com')
 588         self.assertGreaterEqual(user['lastUpdate'], user_orig['lastUpdate'])
 589
 590     def test_set_user_password_nonexistent_user(self):
 591         with self.assertRaises(CmdException) as ctx:
 592             self.exec_cmd('ac-user-set-password', username='admin',
 593                           password='newpass', force_password=True)
 594
 595         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 596         self.assertEqual(str(ctx.exception), "User 'admin' does not exist")
 597
 598     def test_set_user_password_hash(self):
 599         user_orig = self.test_create_user()
 600         user = self.exec_cmd('ac-user-set-password-hash', username='admin',
 601                              hashed_password='$2b$12$Pt3Vq/rDt2y9glTPSV.'
 602                                              'VFegiLkQeIpddtkhoFetNApYmIJOY8gau2')
 603         pass_hash = password_hash('newpass', user['password'])
 604         self.assertDictEqual(user, {
 605             'username': 'admin',
 606             'password': pass_hash,
 607             'pwdExpirationDate': None,
 608             'pwdUpdateRequired': False,
 609             'name': 'admin User',
 610             'email': 'admin@user.com',
 611             'lastUpdate': user['lastUpdate'],
 612             'roles': [],
 613             'enabled': True
 614         })
 615         self.validate_persistent_user('admin', [], pass_hash, 'admin User',
 616                                       'admin@user.com')
 617         self.assertGreaterEqual(user['lastUpdate'], user_orig['lastUpdate'])
 618
 619     def test_set_user_password_hash_nonexistent_user(self):
 620         with self.assertRaises(CmdException) as ctx:
 621             self.exec_cmd('ac-user-set-password-hash', username='admin',
 622                           hashed_password='$2b$12$Pt3Vq/rDt2y9glTPSV.'
 623                                           'VFegiLkQeIpddtkhoFetNApYmIJOY8gau2')
 624
 625         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 626         self.assertEqual(str(ctx.exception), "User 'admin' does not exist")
 627
 628     def test_set_user_password_hash_broken_hash(self):
 629         self.test_create_user()
 630         with self.assertRaises(CmdException) as ctx:
 631             self.exec_cmd('ac-user-set-password-hash', username='admin',
 632                           hashed_password='')
 633
 634         self.assertEqual(ctx.exception.retcode, -errno.EINVAL)
 635         self.assertEqual(str(ctx.exception), 'Invalid password hash')
 636
 637     def test_set_login_credentials(self):
 638         self.exec_cmd('set-login-credentials', username='admin',
 639                       password='admin')
 640         user = self.exec_cmd('ac-user-show', username='admin')
 641         pass_hash = password_hash('admin', user['password'])
 642         self.assertDictEqual(user, {
 643             'username': 'admin',
 644             'password': pass_hash,
 645             'pwdExpirationDate': None,
 646             'pwdUpdateRequired': False,
 647             'name': None,
 648             'email': None,
 649             'lastUpdate': user['lastUpdate'],
 650             'roles': ['administrator'],
 651             'enabled': True,
 652         })
 653         self.validate_persistent_user('admin', ['administrator'], pass_hash,
 654                                       None, None)
 655
 656     def test_set_login_credentials_for_existing_user(self):
 657         self.test_add_user_roles('admin', ['read-only'])
 658         self.exec_cmd('set-login-credentials', username='admin',
 659                       password='admin2')
 660         user = self.exec_cmd('ac-user-show', username='admin')
 661         pass_hash = password_hash('admin2', user['password'])
 662         self.assertDictEqual(user, {
 663             'username': 'admin',
 664             'password': pass_hash,
 665             'pwdExpirationDate': None,
 666             'pwdUpdateRequired': False,
 667             'name': 'admin User',
 668             'email': 'admin@user.com',
 669             'lastUpdate': user['lastUpdate'],
 670             'roles': ['read-only'],
 671             'enabled': True
 672         })
 673         self.validate_persistent_user('admin', ['read-only'], pass_hash,
 674                                       'admin User', 'admin@user.com')
 675
 676     def test_load_v1(self):
 677         self.CONFIG_KEY_DICT['accessdb_v1'] = '''
 678             {{
 679                 "users": {{
 680                     "admin": {{
 681                         "username": "admin",
 682                         "password":
 683                 "$2b$12$sd0Az7mm3FaJl8kN3b/xwOuztaN0sWUwC1SJqjM4wcDw/s5cmGbLK",
 684                         "roles": ["block-manager", "test_role"],
 685                         "name": "admin User",
 686                         "email": "admin@user.com",
 687                         "lastUpdate": {}
 688                     }}
 689                 }},
 690                 "roles": {{
 691                     "test_role": {{
 692                         "name": "test_role",
 693                         "description": "Test Role",
 694                         "scopes_permissions": {{
 695                             "{}": ["{}", "{}"],
 696                             "{}": ["{}"]
 697                         }}
 698                     }}
 699                 }},
 700                 "version": 1
 701             }}
 702         '''.format(int(round(time.time())), Scope.ISCSI, Permission.READ,
 703                    Permission.UPDATE, Scope.POOL, Permission.CREATE)
 704
 705         load_access_control_db()
 706         role = self.exec_cmd('ac-role-show', rolename="test_role")
 707         self.assertDictEqual(role, {
 708             'name': 'test_role',
 709             'description': "Test Role",
 710             'scopes_permissions': {
 711                 Scope.ISCSI: [Permission.READ, Permission.UPDATE],
 712                 Scope.POOL: [Permission.CREATE]
 713             }
 714         })
 715         user = self.exec_cmd('ac-user-show', username="admin")
 716         self.assertDictEqual(user, {
 717             'username': 'admin',
 718             'lastUpdate': user['lastUpdate'],
 719             'password':
 720                 "$2b$12$sd0Az7mm3FaJl8kN3b/xwOuztaN0sWUwC1SJqjM4wcDw/s5cmGbLK",
 721             'pwdExpirationDate': None,
 722             'pwdUpdateRequired': False,
 723             'name': 'admin User',
 724             'email': 'admin@user.com',
 725             'roles': ['block-manager', 'test_role'],
 726             'enabled': True
 727         })
 728
 729     def test_load_v2(self):
 730         self.CONFIG_KEY_DICT['accessdb_v2'] = '''
 731             {{
 732                 "users": {{
 733                     "admin": {{
 734                         "username": "admin",
 735                         "password":
 736                 "$2b$12$sd0Az7mm3FaJl8kN3b/xwOuztaN0sWUwC1SJqjM4wcDw/s5cmGbLK",
 737                         "pwdExpirationDate": null,
 738                         "pwdUpdateRequired": false,
 739                         "roles": ["block-manager", "test_role"],
 740                         "name": "admin User",
 741                         "email": "admin@user.com",
 742                         "lastUpdate": {},
 743                         "enabled": true
 744                     }}
 745                 }},
 746                 "roles": {{
 747                     "test_role": {{
 748                         "name": "test_role",
 749                         "description": "Test Role",
 750                         "scopes_permissions": {{
 751                             "{}": ["{}", "{}"],
 752                             "{}": ["{}"]
 753                         }}
 754                     }}
 755                 }},
 756                 "version": 2
 757             }}
 758         '''.format(int(round(time.time())), Scope.ISCSI, Permission.READ,
 759                    Permission.UPDATE, Scope.POOL, Permission.CREATE)
 760
 761         load_access_control_db()
 762         role = self.exec_cmd('ac-role-show', rolename="test_role")
 763         self.assertDictEqual(role, {
 764             'name': 'test_role',
 765             'description': "Test Role",
 766             'scopes_permissions': {
 767                 Scope.ISCSI: [Permission.READ, Permission.UPDATE],
 768                 Scope.POOL: [Permission.CREATE]
 769             }
 770         })
 771         user = self.exec_cmd('ac-user-show', username="admin")
 772         self.assertDictEqual(user, {
 773             'username': 'admin',
 774             'lastUpdate': user['lastUpdate'],
 775             'password':
 776                 "$2b$12$sd0Az7mm3FaJl8kN3b/xwOuztaN0sWUwC1SJqjM4wcDw/s5cmGbLK",
 777             'pwdExpirationDate': None,
 778             'pwdUpdateRequired': False,
 779             'name': 'admin User',
 780             'email': 'admin@user.com',
 781             'roles': ['block-manager', 'test_role'],
 782             'enabled': True
 783         })
 784
 785     def test_update_from_previous_version_v1(self):
 786         self.CONFIG_KEY_DICT['username'] = 'admin'
 787         self.CONFIG_KEY_DICT['password'] = \
 788             '$2b$12$sd0Az7mm3FaJl8kN3b/xwOuztaN0sWUwC1SJqjM4wcDw/s5cmGbLK'
 789         load_access_control_db()
 790         user = self.exec_cmd('ac-user-show', username="admin")
 791         self.assertDictEqual(user, {
 792             'username': 'admin',
 793             'lastUpdate': user['lastUpdate'],
 794             'password':
 795                 "$2b$12$sd0Az7mm3FaJl8kN3b/xwOuztaN0sWUwC1SJqjM4wcDw/s5cmGbLK",
 796             'pwdExpirationDate': None,
 797             'pwdUpdateRequired': False,
 798             'name': None,
 799             'email': None,
 800             'roles': ['administrator'],
 801             'enabled': True
 802         })
 803
 804     def test_password_policy_pw_length(self):
 805         Settings.PWD_POLICY_CHECK_LENGTH_ENABLED = True
 806         Settings.PWD_POLICY_MIN_LENGTH = 3
 807         pw_policy = PasswordPolicy('foo')
 808         self.assertTrue(pw_policy.check_password_length())
 809
 810     def test_password_policy_pw_length_fail(self):
 811         Settings.PWD_POLICY_CHECK_LENGTH_ENABLED = True
 812         pw_policy = PasswordPolicy('bar')
 813         self.assertFalse(pw_policy.check_password_length())
 814
 815     def test_password_policy_credits_too_weak(self):
 816         Settings.PWD_POLICY_CHECK_COMPLEXITY_ENABLED = True
 817         pw_policy = PasswordPolicy('foo')
 818         pw_credits = pw_policy.check_password_complexity()
 819         self.assertEqual(pw_credits, 3)
 820
 821     def test_password_policy_credits_weak(self):
 822         Settings.PWD_POLICY_CHECK_COMPLEXITY_ENABLED = True
 823         pw_policy = PasswordPolicy('mypassword1')
 824         pw_credits = pw_policy.check_password_complexity()
 825         self.assertEqual(pw_credits, 11)
 826
 827     def test_password_policy_credits_ok(self):
 828         Settings.PWD_POLICY_CHECK_COMPLEXITY_ENABLED = True
 829         pw_policy = PasswordPolicy('mypassword1!@')
 830         pw_credits = pw_policy.check_password_complexity()
 831         self.assertEqual(pw_credits, 17)
 832
 833     def test_password_policy_credits_strong(self):
 834         Settings.PWD_POLICY_CHECK_COMPLEXITY_ENABLED = True
 835         pw_policy = PasswordPolicy('testpassword0047!@')
 836         pw_credits = pw_policy.check_password_complexity()
 837         self.assertEqual(pw_credits, 22)
 838
 839     def test_password_policy_credits_very_strong(self):
 840         Settings.PWD_POLICY_CHECK_COMPLEXITY_ENABLED = True
 841         pw_policy = PasswordPolicy('testpassword#!$!@$')
 842         pw_credits = pw_policy.check_password_complexity()
 843         self.assertEqual(pw_credits, 30)
 844
 845     def test_password_policy_forbidden_words(self):
 846         Settings.PWD_POLICY_CHECK_EXCLUSION_LIST_ENABLED = True
 847         pw_policy = PasswordPolicy('!@$testdashboard#!$')
 848         self.assertTrue(pw_policy.check_if_contains_forbidden_words())
 849
 850     def test_password_policy_forbidden_words_custom(self):
 851         Settings.PWD_POLICY_CHECK_EXCLUSION_LIST_ENABLED = True
 852         Settings.PWD_POLICY_EXCLUSION_LIST = 'foo,bar'
 853         pw_policy = PasswordPolicy('foo123bar')
 854         self.assertTrue(pw_policy.check_if_contains_forbidden_words())
 855
 856     def test_password_policy_sequential_chars(self):
 857         Settings.PWD_POLICY_CHECK_SEQUENTIAL_CHARS_ENABLED = True
 858         pw_policy = PasswordPolicy('!@$test123#!$')
 859         self.assertTrue(pw_policy.check_if_sequential_characters())
 860
 861     def test_password_policy_repetitive_chars(self):
 862         Settings.PWD_POLICY_CHECK_REPETITIVE_CHARS_ENABLED = True
 863         pw_policy = PasswordPolicy('!@$testfooo#!$')
 864         self.assertTrue(pw_policy.check_if_repetitive_characters())
 865
 866     def test_password_policy_contain_username(self):
 867         Settings.PWD_POLICY_CHECK_USERNAME_ENABLED = True
 868         pw_policy = PasswordPolicy('%admin135)', 'admin')
 869         self.assertTrue(pw_policy.check_if_contains_username())
 870
 871     def test_password_policy_is_old_pwd(self):
 872         Settings.PWD_POLICY_CHECK_OLDPWD_ENABLED = True
 873         pw_policy = PasswordPolicy('foo', old_password='foo')
 874         self.assertTrue(pw_policy.check_is_old_password())