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 . import CmdException, CLICommandTestMixin
  11 from .. import mgr
  12 from ..security import Scope, Permission
  13 from ..services.access_control import load_access_control_db, \
  14                                       password_hash, AccessControlDB, \
  15                                       SYSTEM_ROLES
  16
  17
  18 class AccessControlTest(unittest.TestCase, CLICommandTestMixin):
  19
  20     @classmethod
  21     def setUpClass(cls):
  22         cls.mock_kv_store()
  23         mgr.ACCESS_CONTROL_DB = None
  24
  25     def setUp(self):
  26         self.CONFIG_KEY_DICT.clear()
  27         load_access_control_db()
  28
  29     def load_persistent_db(self):
  30         config_key = AccessControlDB.accessdb_config_key()
  31         self.assertIn(config_key, self.CONFIG_KEY_DICT)
  32         db_json = self.CONFIG_KEY_DICT[config_key]
  33         db = json.loads(db_json)
  34         return db
  35
  36     # The DB is written to persistent storage the first time it is saved.
  37     # However, should an operation fail due to <reasons>, we may end up in
  38     # a state where we have a completely empty CONFIG_KEY_DICT (our mock
  39     # equivalent to the persistent state). While this works for most of the
  40     # tests in this class, that would prevent us from testing things like
  41     # "run a command that is expected to fail, and then ensure nothing
  42     # happened", because we'd be asserting in `load_persistent_db()` due to
  43     # the map being empty.
  44     #
  45     # This function will therefore force state to be written to our mock
  46     # persistent state. We could have added this extra step to
  47     # `load_persistent_db()` directly, but that would conflict with the
  48     # upgrade tests. This way, we can selectively enforce this requirement
  49     # where we believe it to be necessary; generically speaking, this should
  50     # not be needed unless we're testing very specific behaviors.
  51     #
  52     def setup_and_load_persistent_db(self):
  53         mgr.ACCESS_CTRL_DB.save()
  54         self.load_persistent_db()
  55
  56     def validate_persistent_role(self, rolename, scopes_permissions,
  57                                  description=None):
  58         db = self.load_persistent_db()
  59         self.assertIn('roles', db)
  60         self.assertIn(rolename, db['roles'])
  61         self.assertEqual(db['roles'][rolename]['name'], rolename)
  62         self.assertEqual(db['roles'][rolename]['description'], description)
  63         self.assertDictEqual(db['roles'][rolename]['scopes_permissions'],
  64                              scopes_permissions)
  65
  66     def validate_persistent_no_role(self, rolename):
  67         db = self.load_persistent_db()
  68         self.assertIn('roles', db)
  69         self.assertNotIn(rolename, db['roles'])
  70
  71     def validate_persistent_user(self, username, roles, password=None,
  72                                  name=None, email=None, lastUpdate=None):
  73         db = self.load_persistent_db()
  74         self.assertIn('users', db)
  75         self.assertIn(username, db['users'])
  76         self.assertEqual(db['users'][username]['username'], username)
  77         self.assertListEqual(db['users'][username]['roles'], roles)
  78         if password:
  79             self.assertEqual(db['users'][username]['password'], password)
  80         if name:
  81             self.assertEqual(db['users'][username]['name'], name)
  82         if email:
  83             self.assertEqual(db['users'][username]['email'], email)
  84         if lastUpdate:
  85             self.assertEqual(db['users'][username]['lastUpdate'], lastUpdate)
  86
  87     def validate_persistent_no_user(self, username):
  88         db = self.load_persistent_db()
  89         self.assertIn('users', db)
  90         self.assertNotIn(username, db['users'])
  91
  92     def test_create_role(self):
  93         role = self.exec_cmd('ac-role-create', rolename='test_role')
  94         self.assertDictEqual(role, {'name': 'test_role', 'description': None,
  95                                     'scopes_permissions': {}})
  96         self.validate_persistent_role('test_role', {})
  97
  98     def test_create_role_with_desc(self):
  99         role = self.exec_cmd('ac-role-create', rolename='test_role',
 100                              description='Test Role')
 101         self.assertDictEqual(role, {'name': 'test_role',
 102                                     'description': 'Test Role',
 103                                     'scopes_permissions': {}})
 104         self.validate_persistent_role('test_role', {}, 'Test Role')
 105
 106     def test_create_duplicate_role(self):
 107         self.test_create_role()
 108
 109         with self.assertRaises(CmdException) as ctx:
 110             self.exec_cmd('ac-role-create', rolename='test_role')
 111
 112         self.assertEqual(ctx.exception.retcode, -errno.EEXIST)
 113         self.assertEqual(str(ctx.exception), "Role 'test_role' already exists")
 114
 115     def test_delete_role(self):
 116         self.test_create_role()
 117         out = self.exec_cmd('ac-role-delete', rolename='test_role')
 118         self.assertEqual(out, "Role 'test_role' deleted")
 119         self.validate_persistent_no_role('test_role')
 120
 121     def test_delete_nonexistent_role(self):
 122         with self.assertRaises(CmdException) as ctx:
 123             self.exec_cmd('ac-role-delete', rolename='test_role')
 124
 125         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 126         self.assertEqual(str(ctx.exception), "Role 'test_role' does not exist")
 127
 128     def test_show_single_role(self):
 129         self.test_create_role()
 130         role = self.exec_cmd('ac-role-show', rolename='test_role')
 131         self.assertDictEqual(role, {'name': 'test_role', 'description': None,
 132                                     'scopes_permissions': {}})
 133
 134     def test_show_nonexistent_role(self):
 135         with self.assertRaises(CmdException) as ctx:
 136             self.exec_cmd('ac-role-show', rolename='test_role')
 137
 138         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 139         self.assertEqual(str(ctx.exception), "Role 'test_role' does not exist")
 140
 141     def test_show_system_roles(self):
 142         roles = self.exec_cmd('ac-role-show')
 143         self.assertEqual(len(roles), len(SYSTEM_ROLES))
 144         for role in roles:
 145             self.assertIn(role, SYSTEM_ROLES)
 146
 147     def test_show_system_role(self):
 148         role = self.exec_cmd('ac-role-show', rolename="read-only")
 149         self.assertEqual(role['name'], 'read-only')
 150         self.assertEqual(role['description'], 'Read-Only')
 151
 152     def test_delete_system_role(self):
 153         with self.assertRaises(CmdException) as ctx:
 154             self.exec_cmd('ac-role-delete', rolename='administrator')
 155
 156         self.assertEqual(ctx.exception.retcode, -errno.EPERM)
 157         self.assertEqual(str(ctx.exception),
 158                          "Cannot delete system role 'administrator'")
 159
 160     def test_add_role_scope_perms(self):
 161         self.test_create_role()
 162         self.exec_cmd('ac-role-add-scope-perms', rolename='test_role',
 163                       scopename=Scope.POOL,
 164                       permissions=[Permission.READ, Permission.DELETE])
 165         role = self.exec_cmd('ac-role-show', rolename='test_role')
 166         self.assertDictEqual(role, {'name': 'test_role',
 167                                     'description': None,
 168                                     'scopes_permissions': {
 169                                         Scope.POOL: [Permission.DELETE,
 170                                                      Permission.READ]
 171                                     }})
 172         self.validate_persistent_role('test_role', {
 173             Scope.POOL: [Permission.DELETE, Permission.READ]
 174         })
 175
 176     def test_del_role_scope_perms(self):
 177         self.test_add_role_scope_perms()
 178         self.exec_cmd('ac-role-add-scope-perms', rolename='test_role',
 179                       scopename=Scope.MONITOR,
 180                       permissions=[Permission.READ, Permission.CREATE])
 181         self.validate_persistent_role('test_role', {
 182             Scope.POOL: [Permission.DELETE, Permission.READ],
 183             Scope.MONITOR: [Permission.CREATE, Permission.READ]
 184         })
 185         self.exec_cmd('ac-role-del-scope-perms', rolename='test_role',
 186                       scopename=Scope.POOL)
 187         role = self.exec_cmd('ac-role-show', rolename='test_role')
 188         self.assertDictEqual(role, {'name': 'test_role',
 189                                     'description': None,
 190                                     'scopes_permissions': {
 191                                         Scope.MONITOR: [Permission.CREATE,
 192                                                         Permission.READ]
 193                                     }})
 194         self.validate_persistent_role('test_role', {
 195             Scope.MONITOR: [Permission.CREATE, Permission.READ]
 196         })
 197
 198     def test_add_role_scope_perms_nonexistent_role(self):
 199
 200         with self.assertRaises(CmdException) as ctx:
 201             self.exec_cmd('ac-role-add-scope-perms', rolename='test_role',
 202                           scopename='pool',
 203                           permissions=['read', 'delete'])
 204
 205         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 206         self.assertEqual(str(ctx.exception), "Role 'test_role' does not exist")
 207
 208     def test_add_role_invalid_scope_perms(self):
 209         self.test_create_role()
 210
 211         with self.assertRaises(CmdException) as ctx:
 212             self.exec_cmd('ac-role-add-scope-perms', rolename='test_role',
 213                           scopename='invalidscope',
 214                           permissions=['read', 'delete'])
 215
 216         self.assertEqual(ctx.exception.retcode, -errno.EINVAL)
 217         self.assertEqual(str(ctx.exception),
 218                          "Scope 'invalidscope' is not valid\n Possible values: "
 219                          "{}".format(Scope.all_scopes()))
 220
 221     def test_add_role_scope_invalid_perms(self):
 222         self.test_create_role()
 223
 224         with self.assertRaises(CmdException) as ctx:
 225             self.exec_cmd('ac-role-add-scope-perms', rolename='test_role',
 226                           scopename='pool', permissions=['invalidperm'])
 227
 228         self.assertEqual(ctx.exception.retcode, -errno.EINVAL)
 229         self.assertEqual(str(ctx.exception),
 230                          "Permission 'invalidperm' is not valid\n Possible "
 231                          "values: {}".format(Permission.all_permissions()))
 232
 233     def test_del_role_scope_perms_nonexistent_role(self):
 234
 235         with self.assertRaises(CmdException) as ctx:
 236             self.exec_cmd('ac-role-del-scope-perms', rolename='test_role',
 237                           scopename='pool')
 238
 239         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 240         self.assertEqual(str(ctx.exception), "Role 'test_role' does not exist")
 241
 242     def test_del_role_nonexistent_scope_perms(self):
 243         self.test_add_role_scope_perms()
 244
 245         with self.assertRaises(CmdException) as ctx:
 246             self.exec_cmd('ac-role-del-scope-perms', rolename='test_role',
 247                           scopename='nonexistentscope')
 248
 249         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 250         self.assertEqual(str(ctx.exception),
 251                          "There are no permissions for scope 'nonexistentscope' "
 252                          "in role 'test_role'")
 253
 254     def test_not_permitted_add_role_scope_perms(self):
 255         with self.assertRaises(CmdException) as ctx:
 256             self.exec_cmd('ac-role-add-scope-perms', rolename='read-only',
 257                           scopename='pool', permissions=['read', 'delete'])
 258
 259         self.assertEqual(ctx.exception.retcode, -errno.EPERM)
 260         self.assertEqual(str(ctx.exception),
 261                          "Cannot update system role 'read-only'")
 262
 263     def test_not_permitted_del_role_scope_perms(self):
 264         with self.assertRaises(CmdException) as ctx:
 265             self.exec_cmd('ac-role-del-scope-perms', rolename='read-only',
 266                           scopename='pool')
 267
 268         self.assertEqual(ctx.exception.retcode, -errno.EPERM)
 269         self.assertEqual(str(ctx.exception),
 270                          "Cannot update system role 'read-only'")
 271
 272     def test_create_user(self, username='admin', rolename=None):
 273         user = self.exec_cmd('ac-user-create', username=username,
 274                              rolename=rolename, password='admin',
 275                              name='{} User'.format(username),
 276                              email='{}@user.com'.format(username))
 277
 278         pass_hash = password_hash('admin', user['password'])
 279         self.assertDictEqual(user, {
 280             'username': username,
 281             'password': pass_hash,
 282             'lastUpdate': user['lastUpdate'],
 283             'name': '{} User'.format(username),
 284             'email': '{}@user.com'.format(username),
 285             'roles': [rolename] if rolename else []
 286         })
 287         self.validate_persistent_user(username, [rolename] if rolename else [],
 288                                       pass_hash, '{} User'.format(username),
 289                                       '{}@user.com'.format(username),
 290                                       user['lastUpdate'])
 291         return user
 292
 293     def test_create_user_with_role(self):
 294         self.test_add_role_scope_perms()
 295         self.test_create_user(rolename='test_role')
 296
 297     def test_create_user_with_system_role(self):
 298         self.test_create_user(rolename='administrator')
 299
 300     def test_delete_user(self):
 301         self.test_create_user()
 302         out = self.exec_cmd('ac-user-delete', username='admin')
 303         self.assertEqual(out, "User 'admin' deleted")
 304         users = self.exec_cmd('ac-user-show')
 305         self.assertEqual(len(users), 0)
 306         self.validate_persistent_no_user('admin')
 307
 308     def test_create_duplicate_user(self):
 309         self.test_create_user()
 310
 311         with self.assertRaises(CmdException) as ctx:
 312             self.exec_cmd('ac-user-create', username='admin', password='admin')
 313
 314         self.assertEqual(ctx.exception.retcode, -errno.EEXIST)
 315         self.assertEqual(str(ctx.exception), "User 'admin' already exists")
 316
 317     def test_create_users_with_dne_role(self):
 318         # one time call to setup our persistent db
 319         self.setup_and_load_persistent_db()
 320
 321         # create a user with a role that does not exist; expect a failure
 322         try:
 323             self.exec_cmd('ac-user-create', username='foo',
 324                           rolename='dne_role', password='foopass',
 325                           name='foo User', email='foo@user.com')
 326         except CmdException as e:
 327             self.assertEqual(e.retcode, -errno.ENOENT)
 328
 329         db = self.load_persistent_db()
 330         if 'users' in db:
 331             self.assertNotIn('foo', db['users'])
 332
 333         # We could just finish our test here, given we ensured that the user
 334         # with a non-existent role is not in persistent storage. However,
 335         # we're going to test the database's consistency, making sure that
 336         # side-effects are not written to persistent storage once we commit
 337         # an unrelated operation. To ensure this, we'll issue another
 338         # operation that is sharing the same code path, and will check whether
 339         # the next operation commits dirty state.
 340
 341         # create a role (this will be 'test_role')
 342         self.test_create_role()
 343         self.exec_cmd('ac-user-create', username='bar',
 344                       rolename='test_role', password='barpass',
 345                       name='bar User', email='bar@user.com')
 346
 347         # validate db:
 348         #   user 'foo' should not exist
 349         #   user 'bar' should exist and have role 'test_role'
 350         self.validate_persistent_user('bar', ['test_role'])
 351
 352         db = self.load_persistent_db()
 353         self.assertIn('users', db)
 354         self.assertNotIn('foo', db['users'])
 355
 356     def test_delete_nonexistent_user(self):
 357         with self.assertRaises(CmdException) as ctx:
 358             self.exec_cmd('ac-user-delete', username='admin')
 359
 360         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 361         self.assertEqual(str(ctx.exception), "User 'admin' does not exist")
 362
 363     def test_add_user_roles(self, username='admin',
 364                             roles=['pool-manager', 'block-manager']):
 365         user_orig = self.test_create_user(username)
 366         uroles = []
 367         for role in roles:
 368             uroles.append(role)
 369             uroles.sort()
 370             user = self.exec_cmd('ac-user-add-roles', username=username,
 371                                  roles=[role])
 372             self.assertDictContainsSubset({'roles': uroles}, user)
 373         self.validate_persistent_user(username, uroles)
 374         self.assertGreaterEqual(user['lastUpdate'], user_orig['lastUpdate'])
 375
 376     def test_add_user_roles2(self):
 377         user_orig = self.test_create_user()
 378         user = self.exec_cmd('ac-user-add-roles', username="admin",
 379                              roles=['pool-manager', 'block-manager'])
 380         self.assertDictContainsSubset(
 381             {'roles': ['block-manager', 'pool-manager']}, user)
 382         self.validate_persistent_user('admin', ['block-manager',
 383                                                 'pool-manager'])
 384         self.assertGreaterEqual(user['lastUpdate'], user_orig['lastUpdate'])
 385
 386     def test_add_user_roles_not_existent_user(self):
 387         with self.assertRaises(CmdException) as ctx:
 388             self.exec_cmd('ac-user-add-roles', username="admin",
 389                           roles=['pool-manager', 'block-manager'])
 390
 391         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 392         self.assertEqual(str(ctx.exception), "User 'admin' does not exist")
 393
 394     def test_add_user_roles_not_existent_role(self):
 395         self.test_create_user()
 396         with self.assertRaises(CmdException) as ctx:
 397             self.exec_cmd('ac-user-add-roles', username="admin",
 398                           roles=['Invalid Role'])
 399
 400         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 401         self.assertEqual(str(ctx.exception),
 402                          "Role 'Invalid Role' does not exist")
 403
 404     def test_set_user_roles(self):
 405         user_orig = self.test_create_user()
 406         user = self.exec_cmd('ac-user-add-roles', username="admin",
 407                              roles=['pool-manager'])
 408         self.assertDictContainsSubset(
 409             {'roles': ['pool-manager']}, user)
 410         self.validate_persistent_user('admin', ['pool-manager'])
 411         self.assertGreaterEqual(user['lastUpdate'], user_orig['lastUpdate'])
 412         user2 = self.exec_cmd('ac-user-set-roles', username="admin",
 413                               roles=['rgw-manager', 'block-manager'])
 414         self.assertDictContainsSubset(
 415             {'roles': ['block-manager', 'rgw-manager']}, user2)
 416         self.validate_persistent_user('admin', ['block-manager',
 417                                                 'rgw-manager'])
 418         self.assertGreaterEqual(user2['lastUpdate'], user['lastUpdate'])
 419
 420     def test_set_user_roles_not_existent_user(self):
 421         with self.assertRaises(CmdException) as ctx:
 422             self.exec_cmd('ac-user-set-roles', username="admin",
 423                           roles=['pool-manager', 'block-manager'])
 424
 425         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 426         self.assertEqual(str(ctx.exception), "User 'admin' does not exist")
 427
 428     def test_set_user_roles_not_existent_role(self):
 429         self.test_create_user()
 430         with self.assertRaises(CmdException) as ctx:
 431             self.exec_cmd('ac-user-set-roles', username="admin",
 432                           roles=['Invalid Role'])
 433
 434         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 435         self.assertEqual(str(ctx.exception),
 436                          "Role 'Invalid Role' does not exist")
 437
 438     def test_del_user_roles(self):
 439         self.test_add_user_roles()
 440         user = self.exec_cmd('ac-user-del-roles', username="admin",
 441                              roles=['pool-manager'])
 442         self.assertDictContainsSubset(
 443             {'roles': ['block-manager']}, user)
 444         self.validate_persistent_user('admin', ['block-manager'])
 445
 446     def test_del_user_roles_not_existent_user(self):
 447         with self.assertRaises(CmdException) as ctx:
 448             self.exec_cmd('ac-user-del-roles', username="admin",
 449                           roles=['pool-manager', 'block-manager'])
 450
 451         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 452         self.assertEqual(str(ctx.exception), "User 'admin' does not exist")
 453
 454     def test_del_user_roles_not_existent_role(self):
 455         self.test_create_user()
 456         with self.assertRaises(CmdException) as ctx:
 457             self.exec_cmd('ac-user-del-roles', username="admin",
 458                           roles=['Invalid Role'])
 459
 460         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 461         self.assertEqual(str(ctx.exception),
 462                          "Role 'Invalid Role' does not exist")
 463
 464     def test_del_user_roles_not_associated_role(self):
 465         self.test_create_user()
 466         with self.assertRaises(CmdException) as ctx:
 467             self.exec_cmd('ac-user-del-roles', username="admin",
 468                           roles=['rgw-manager'])
 469
 470         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 471         self.assertEqual(str(ctx.exception),
 472                          "Role 'rgw-manager' is not associated with user "
 473                          "'admin'")
 474
 475     def test_show_user(self):
 476         self.test_add_user_roles()
 477         user = self.exec_cmd('ac-user-show', username='admin')
 478         pass_hash = password_hash('admin', user['password'])
 479         self.assertDictEqual(user, {
 480             'username': 'admin',
 481             'lastUpdate': user['lastUpdate'],
 482             'password': pass_hash,
 483             'name': 'admin User',
 484             'email': 'admin@user.com',
 485             'roles': ['block-manager', 'pool-manager']
 486         })
 487
 488     def test_show_nonexistent_user(self):
 489         with self.assertRaises(CmdException) as ctx:
 490             self.exec_cmd('ac-user-show', username='admin')
 491
 492         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 493         self.assertEqual(str(ctx.exception), "User 'admin' does not exist")
 494
 495     def test_show_all_users(self):
 496         self.test_add_user_roles('admin', ['administrator'])
 497         self.test_add_user_roles('guest', ['read-only'])
 498         users = self.exec_cmd('ac-user-show')
 499         self.assertEqual(len(users), 2)
 500         for user in users:
 501             self.assertIn(user, ['admin', 'guest'])
 502
 503     def test_del_role_associated_with_user(self):
 504         self.test_create_role()
 505         self.test_add_user_roles('guest', ['test_role'])
 506
 507         with self.assertRaises(CmdException) as ctx:
 508             self.exec_cmd('ac-role-delete', rolename='test_role')
 509
 510         self.assertEqual(ctx.exception.retcode, -errno.EPERM)
 511         self.assertEqual(str(ctx.exception),
 512                          "Role 'test_role' is still associated with user "
 513                          "'guest'")
 514
 515     def test_set_user_info(self):
 516         user_orig = self.test_create_user()
 517         user = self.exec_cmd('ac-user-set-info', username='admin',
 518                              name='Admin Name', email='admin@admin.com')
 519         pass_hash = password_hash('admin', user['password'])
 520         self.assertDictEqual(user, {
 521             'username': 'admin',
 522             'password': pass_hash,
 523             'name': 'Admin Name',
 524             'email': 'admin@admin.com',
 525             'lastUpdate': user['lastUpdate'],
 526             'roles': []
 527         })
 528         self.validate_persistent_user('admin', [], pass_hash, 'Admin Name',
 529                                       'admin@admin.com')
 530         self.assertEqual(user['lastUpdate'], user_orig['lastUpdate'])
 531
 532     def test_set_user_info_nonexistent_user(self):
 533         with self.assertRaises(CmdException) as ctx:
 534             self.exec_cmd('ac-user-set-info', username='admin',
 535                           name='Admin Name', email='admin@admin.com')
 536
 537         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 538         self.assertEqual(str(ctx.exception), "User 'admin' does not exist")
 539
 540     def test_set_user_password(self):
 541         user_orig = self.test_create_user()
 542         user = self.exec_cmd('ac-user-set-password', username='admin',
 543                              password='newpass')
 544         pass_hash = password_hash('newpass', user['password'])
 545         self.assertDictEqual(user, {
 546             'username': 'admin',
 547             'password': pass_hash,
 548             'name': 'admin User',
 549             'email': 'admin@user.com',
 550             'lastUpdate': user['lastUpdate'],
 551             'roles': []
 552         })
 553         self.validate_persistent_user('admin', [], pass_hash, 'admin User',
 554                                       'admin@user.com')
 555         self.assertGreaterEqual(user['lastUpdate'], user_orig['lastUpdate'])
 556
 557     def test_set_user_password_nonexistent_user(self):
 558         with self.assertRaises(CmdException) as ctx:
 559             self.exec_cmd('ac-user-set-password', username='admin',
 560                           password='newpass')
 561
 562         self.assertEqual(ctx.exception.retcode, -errno.ENOENT)
 563         self.assertEqual(str(ctx.exception), "User 'admin' does not exist")
 564
 565     def test_set_login_credentials(self):
 566         self.exec_cmd('set-login-credentials', username='admin',
 567                       password='admin')
 568         user = self.exec_cmd('ac-user-show', username='admin')
 569         pass_hash = password_hash('admin', user['password'])
 570         self.assertDictEqual(user, {
 571             'username': 'admin',
 572             'password': pass_hash,
 573             'name': None,
 574             'email': None,
 575             'lastUpdate': user['lastUpdate'],
 576             'roles': ['administrator']
 577         })
 578         self.validate_persistent_user('admin', ['administrator'], pass_hash,
 579                                       None, None)
 580
 581     def test_set_login_credentials_for_existing_user(self):
 582         self.test_add_user_roles('admin', ['read-only'])
 583         self.exec_cmd('set-login-credentials', username='admin',
 584                       password='admin2')
 585         user = self.exec_cmd('ac-user-show', username='admin')
 586         pass_hash = password_hash('admin2', user['password'])
 587         self.assertDictEqual(user, {
 588             'username': 'admin',
 589             'password': pass_hash,
 590             'name': 'admin User',
 591             'email': 'admin@user.com',
 592             'lastUpdate': user['lastUpdate'],
 593             'roles': ['read-only']
 594         })
 595         self.validate_persistent_user('admin', ['read-only'], pass_hash,
 596                                       'admin User', 'admin@user.com')
 597
 598     def test_load_v1(self):
 599         self.CONFIG_KEY_DICT['accessdb_v1'] = '''
 600             {{
 601                 "users": {{
 602                     "admin": {{
 603                         "username": "admin",
 604                         "password":
 605                 "$2b$12$sd0Az7mm3FaJl8kN3b/xwOuztaN0sWUwC1SJqjM4wcDw/s5cmGbLK",
 606                         "roles": ["block-manager", "test_role"],
 607                         "name": "admin User",
 608                         "email": "admin@user.com",
 609                         "lastUpdate": {}
 610                     }}
 611                 }},
 612                 "roles": {{
 613                     "test_role": {{
 614                         "name": "test_role",
 615                         "description": "Test Role",
 616                         "scopes_permissions": {{
 617                             "{}": ["{}", "{}"],
 618                             "{}": ["{}"]
 619                         }}
 620                     }}
 621                 }},
 622                 "version": 1
 623             }}
 624         '''.format(int(round(time.time())), Scope.ISCSI, Permission.READ,
 625                    Permission.UPDATE, Scope.POOL, Permission.CREATE)
 626
 627         load_access_control_db()
 628         role = self.exec_cmd('ac-role-show', rolename="test_role")
 629         self.assertDictEqual(role, {
 630             'name': 'test_role',
 631             'description': "Test Role",
 632             'scopes_permissions': {
 633                 Scope.ISCSI: [Permission.READ, Permission.UPDATE],
 634                 Scope.POOL: [Permission.CREATE]
 635             }
 636         })
 637         user = self.exec_cmd('ac-user-show', username="admin")
 638         self.assertDictEqual(user, {
 639             'username': 'admin',
 640             'lastUpdate': user['lastUpdate'],
 641             'password':
 642                 "$2b$12$sd0Az7mm3FaJl8kN3b/xwOuztaN0sWUwC1SJqjM4wcDw/s5cmGbLK",
 643             'name': 'admin User',
 644             'email': 'admin@user.com',
 645             'roles': ['block-manager', 'test_role']
 646         })
 647
 648     def test_update_from_previous_version_v1(self):
 649         self.CONFIG_KEY_DICT['username'] = 'admin'
 650         self.CONFIG_KEY_DICT['password'] = \
 651             '$2b$12$sd0Az7mm3FaJl8kN3b/xwOuztaN0sWUwC1SJqjM4wcDw/s5cmGbLK'
 652         load_access_control_db()
 653         user = self.exec_cmd('ac-user-show', username="admin")
 654         self.assertDictEqual(user, {
 655             'username': 'admin',
 656             'lastUpdate': user['lastUpdate'],
 657             'password':
 658                 "$2b$12$sd0Az7mm3FaJl8kN3b/xwOuztaN0sWUwC1SJqjM4wcDw/s5cmGbLK",
 659             'name': None,
 660             'email': None,
 661             'roles': ['administrator']
 662         })