import jwt
from . import ApiController, RESTController
-from .. import logger
+from .. import logger, mgr
from ..exceptions import DashboardException
from ..services.auth import AuthManager, JwtManager
from ..services.access_control import UserDoesNotExist
-from ..services.sso import SSO_DB
@ApiController('/auth', secure=False)
token = JwtManager.get_token_from_header()
JwtManager.blacklist_token(token)
redirect_url = '#/login'
- if SSO_DB.protocol == 'saml2':
+ if mgr.SSO_DB.protocol == 'saml2':
redirect_url = 'auth/saml2/slo'
return {
'redirect_url': redirect_url
}
def _get_login_url(self):
- if SSO_DB.protocol == 'saml2':
+ if mgr.SSO_DB.protocol == 'saml2':
return 'auth/saml2/login'
return '#/login'
import cherrypy
from . import ApiController, RESTController, UiApiController
+from .. import mgr
from ..exceptions import RoleDoesNotExist, DashboardException,\
RoleIsAssociatedWithUser, RoleAlreadyExists
from ..security import Scope as SecurityScope, Permission
-from ..services.access_control import ACCESS_CTRL_DB, SYSTEM_ROLES
+from ..services.access_control import SYSTEM_ROLES
@ApiController('/role', SecurityScope.USER)
role.set_scope_permissions(scope, permissions)
def list(self):
- roles = dict(ACCESS_CTRL_DB.roles)
+ roles = dict(mgr.ACCESS_CTRL_DB.roles)
roles.update(SYSTEM_ROLES)
roles = sorted(roles.values(), key=lambda role: role.name)
return [Role._role_to_dict(r) for r in roles]
role = SYSTEM_ROLES.get(name)
if not role:
try:
- role = ACCESS_CTRL_DB.get_role(name)
+ role = mgr.ACCESS_CTRL_DB.get_role(name)
except RoleDoesNotExist:
raise cherrypy.HTTPError(404)
return Role._role_to_dict(role)
component='role')
Role._validate_permissions(scopes_permissions)
try:
- role = ACCESS_CTRL_DB.create_role(name, description)
+ role = mgr.ACCESS_CTRL_DB.create_role(name, description)
except RoleAlreadyExists:
raise DashboardException(msg='Role already exists',
code='role_already_exists',
component='role')
Role._set_permissions(role, scopes_permissions)
- ACCESS_CTRL_DB.save()
+ mgr.ACCESS_CTRL_DB.save()
return Role._role_to_dict(role)
def set(self, name, description=None, scopes_permissions=None):
try:
- role = ACCESS_CTRL_DB.get_role(name)
+ role = mgr.ACCESS_CTRL_DB.get_role(name)
except RoleDoesNotExist:
if name in SYSTEM_ROLES:
raise DashboardException(msg='Cannot update system role',
Role._validate_permissions(scopes_permissions)
Role._set_permissions(role, scopes_permissions)
role.description = description
- ACCESS_CTRL_DB.update_users_with_roles(role)
- ACCESS_CTRL_DB.save()
+ mgr.ACCESS_CTRL_DB.update_users_with_roles(role)
+ mgr.ACCESS_CTRL_DB.save()
return Role._role_to_dict(role)
def delete(self, name):
try:
- ACCESS_CTRL_DB.delete_role(name)
+ mgr.ACCESS_CTRL_DB.delete_role(name)
except RoleDoesNotExist:
if name in SYSTEM_ROLES:
raise DashboardException(msg='Cannot delete system role',
raise DashboardException(msg='Role is associated with user',
code='role_is_associated_with_user',
component='role')
- ACCESS_CTRL_DB.save()
+ mgr.ACCESS_CTRL_DB.save()
@UiApiController('/scope', SecurityScope.USER)
from .. import mgr, logger
from ..exceptions import UserDoesNotExist
from ..services.auth import JwtManager
-from ..services.access_control import ACCESS_CTRL_DB
-from ..services.sso import SSO_DB
from ..tools import prepare_url_prefix
from . import Controller, Endpoint, BaseController
raise cherrypy.HTTPError(400,
'Required library not found: `{}`'.format(python_saml_name))
try:
- OneLogin_Saml2_Settings(SSO_DB.saml2.onelogin_settings)
+ OneLogin_Saml2_Settings(mgr.SSO_DB.saml2.onelogin_settings)
except OneLogin_Saml2_Error:
raise cherrypy.HTTPError(400, 'Single Sign-On is not configured.')
def auth_response(self, **kwargs):
Saml2._check_python_saml()
req = Saml2._build_req(self._request, kwargs)
- auth = OneLogin_Saml2_Auth(req, SSO_DB.saml2.onelogin_settings)
+ auth = OneLogin_Saml2_Auth(req, mgr.SSO_DB.saml2.onelogin_settings)
auth.process_response()
errors = auth.get_errors()
if auth.is_authenticated():
JwtManager.reset_user()
- username_attribute = auth.get_attribute(SSO_DB.saml2.get_username_attribute())
+ username_attribute = auth.get_attribute(mgr.SSO_DB.saml2.get_username_attribute())
if username_attribute is None:
raise cherrypy.HTTPError(400,
'SSO error - `{}` not found in auth attributes. '
'Received attributes: {}'
.format(
- SSO_DB.saml2.get_username_attribute(),
+ mgr.SSO_DB.saml2.get_username_attribute(),
auth.get_attributes()))
username = username_attribute[0]
try:
- ACCESS_CTRL_DB.get_user(username)
+ mgr.ACCESS_CTRL_DB.get_user(username)
except UserDoesNotExist:
raise cherrypy.HTTPError(400,
'SSO error - Username `{}` does not exist.'
@Endpoint(xml=True)
def metadata(self):
Saml2._check_python_saml()
- saml_settings = OneLogin_Saml2_Settings(SSO_DB.saml2.onelogin_settings)
+ saml_settings = OneLogin_Saml2_Settings(mgr.SSO_DB.saml2.onelogin_settings)
return saml_settings.get_sp_metadata()
@Endpoint(json_response=False)
def login(self):
Saml2._check_python_saml()
req = Saml2._build_req(self._request, {})
- auth = OneLogin_Saml2_Auth(req, SSO_DB.saml2.onelogin_settings)
+ auth = OneLogin_Saml2_Auth(req, mgr.SSO_DB.saml2.onelogin_settings)
raise cherrypy.HTTPRedirect(auth.login())
@Endpoint(json_response=False)
def slo(self):
Saml2._check_python_saml()
req = Saml2._build_req(self._request, {})
- auth = OneLogin_Saml2_Auth(req, SSO_DB.saml2.onelogin_settings)
+ auth = OneLogin_Saml2_Auth(req, mgr.SSO_DB.saml2.onelogin_settings)
raise cherrypy.HTTPRedirect(auth.logout())
@Endpoint(json_response=False)
import cherrypy
from . import ApiController, RESTController
+from .. import mgr
from ..exceptions import DashboardException, UserAlreadyExists, \
UserDoesNotExist
from ..security import Scope
-from ..services.access_control import ACCESS_CTRL_DB, SYSTEM_ROLES
+from ..services.access_control import SYSTEM_ROLES
from ..services.auth import JwtManager
@staticmethod
def _get_user_roles(roles):
- all_roles = dict(ACCESS_CTRL_DB.roles)
+ all_roles = dict(mgr.ACCESS_CTRL_DB.roles)
all_roles.update(SYSTEM_ROLES)
try:
return [all_roles[rolename] for rolename in roles]
component='user')
def list(self):
- users = ACCESS_CTRL_DB.users
+ users = mgr.ACCESS_CTRL_DB.users
result = [User._user_to_dict(u) for _, u in users.items()]
return result
def get(self, username):
try:
- user = ACCESS_CTRL_DB.get_user(username)
+ user = mgr.ACCESS_CTRL_DB.get_user(username)
except UserDoesNotExist:
raise cherrypy.HTTPError(404)
return User._user_to_dict(user)
if roles:
user_roles = User._get_user_roles(roles)
try:
- user = ACCESS_CTRL_DB.create_user(username, password, name, email)
+ user = mgr.ACCESS_CTRL_DB.create_user(username, password, name, email)
except UserAlreadyExists:
raise DashboardException(msg='Username already exists',
code='username_already_exists',
component='user')
if user_roles:
user.set_roles(user_roles)
- ACCESS_CTRL_DB.save()
+ mgr.ACCESS_CTRL_DB.save()
return User._user_to_dict(user)
def delete(self, username):
code='cannot_delete_current_user',
component='user')
try:
- ACCESS_CTRL_DB.delete_user(username)
+ mgr.ACCESS_CTRL_DB.delete_user(username)
except UserDoesNotExist:
raise cherrypy.HTTPError(404)
- ACCESS_CTRL_DB.save()
+ mgr.ACCESS_CTRL_DB.save()
def set(self, username, password=None, name=None, email=None, roles=None):
try:
- user = ACCESS_CTRL_DB.get_user(username)
+ user = mgr.ACCESS_CTRL_DB.get_user(username)
except UserDoesNotExist:
raise cherrypy.HTTPError(404)
user_roles = []
user.name = name
user.email = email
user.set_roles(user_roles)
- ACCESS_CTRL_DB.save()
+ mgr.ACCESS_CTRL_DB.save()
return User._user_to_dict(user)
self._stopping = threading.Event()
self.shutdown_event = threading.Event()
+ self.ACCESS_CTRL_DB = None
+ self.SSO_DB = None
+
@classmethod
def can_run(cls):
if cherrypy is None:
return cls(db['version'], users, roles)
-ACCESS_CTRL_DB = None
-
-
def load_access_control_db():
- # pylint: disable=W0603
- global ACCESS_CTRL_DB
- ACCESS_CTRL_DB = AccessControlDB.load()
+ mgr.ACCESS_CTRL_DB = AccessControlDB.load()
# CLI dashboard access control scope commands
'Set the login credentials')
def set_login_credentials_cmd(_, username, password):
try:
- user = ACCESS_CTRL_DB.get_user(username)
+ user = mgr.ACCESS_CTRL_DB.get_user(username)
user.set_password(password)
except UserDoesNotExist:
- user = ACCESS_CTRL_DB.create_user(username, password, None, None)
+ user = mgr.ACCESS_CTRL_DB.create_user(username, password, None, None)
user.set_roles([ADMIN_ROLE])
- ACCESS_CTRL_DB.save()
+ mgr.ACCESS_CTRL_DB.save()
return 0, '''\
******************************************************************
'Show role info')
def ac_role_show_cmd(_, rolename=None):
if not rolename:
- roles = dict(ACCESS_CTRL_DB.roles)
+ roles = dict(mgr.ACCESS_CTRL_DB.roles)
roles.update(SYSTEM_ROLES)
roles_list = [name for name, _ in roles.items()]
return 0, json.dumps(roles_list), ''
try:
- role = ACCESS_CTRL_DB.get_role(rolename)
+ role = mgr.ACCESS_CTRL_DB.get_role(rolename)
except RoleDoesNotExist as ex:
if rolename not in SYSTEM_ROLES:
return -errno.ENOENT, '', str(ex)
'Create a new access control role')
def ac_role_create_cmd(_, rolename, description=None):
try:
- role = ACCESS_CTRL_DB.create_role(rolename, description)
- ACCESS_CTRL_DB.save()
+ role = mgr.ACCESS_CTRL_DB.create_role(rolename, description)
+ mgr.ACCESS_CTRL_DB.save()
return 0, json.dumps(role.to_dict()), ''
except RoleAlreadyExists as ex:
return -errno.EEXIST, '', str(ex)
'Delete an access control role')
def ac_role_delete_cmd(_, rolename):
try:
- ACCESS_CTRL_DB.delete_role(rolename)
- ACCESS_CTRL_DB.save()
+ mgr.ACCESS_CTRL_DB.delete_role(rolename)
+ mgr.ACCESS_CTRL_DB.save()
return 0, "Role '{}' deleted".format(rolename), ""
except RoleDoesNotExist as ex:
if rolename in SYSTEM_ROLES:
'Add the scope permissions for a role')
def ac_role_add_scope_perms_cmd(_, rolename, scopename, permissions):
try:
- role = ACCESS_CTRL_DB.get_role(rolename)
+ role = mgr.ACCESS_CTRL_DB.get_role(rolename)
perms_array = [perm.strip() for perm in permissions]
role.set_scope_permissions(scopename, perms_array)
- ACCESS_CTRL_DB.update_users_with_roles(role)
- ACCESS_CTRL_DB.save()
+ mgr.ACCESS_CTRL_DB.update_users_with_roles(role)
+ mgr.ACCESS_CTRL_DB.save()
return 0, json.dumps(role.to_dict()), ''
except RoleDoesNotExist as ex:
if rolename in SYSTEM_ROLES:
'Delete the scope permissions for a role')
def ac_role_del_scope_perms_cmd(_, rolename, scopename):
try:
- role = ACCESS_CTRL_DB.get_role(rolename)
+ role = mgr.ACCESS_CTRL_DB.get_role(rolename)
role.del_scope_permissions(scopename)
- ACCESS_CTRL_DB.update_users_with_roles(role)
- ACCESS_CTRL_DB.save()
+ mgr.ACCESS_CTRL_DB.update_users_with_roles(role)
+ mgr.ACCESS_CTRL_DB.save()
return 0, json.dumps(role.to_dict()), ''
except RoleDoesNotExist as ex:
if rolename in SYSTEM_ROLES:
'Show user info')
def ac_user_show_cmd(_, username=None):
if not username:
- users = ACCESS_CTRL_DB.users
+ users = mgr.ACCESS_CTRL_DB.users
users_list = [name for name, _ in users.items()]
return 0, json.dumps(users_list), ''
try:
- user = ACCESS_CTRL_DB.get_user(username)
+ user = mgr.ACCESS_CTRL_DB.get_user(username)
return 0, json.dumps(user.to_dict()), ''
except UserDoesNotExist as ex:
return -errno.ENOENT, '', str(ex)
def ac_user_create_cmd(_, username, password=None, rolename=None, name=None,
email=None):
try:
- role = ACCESS_CTRL_DB.get_role(rolename) if rolename else None
+ role = mgr.ACCESS_CTRL_DB.get_role(rolename) if rolename else None
except RoleDoesNotExist as ex:
if rolename not in SYSTEM_ROLES:
return -errno.ENOENT, '', str(ex)
role = SYSTEM_ROLES[rolename]
try:
- user = ACCESS_CTRL_DB.create_user(username, password, name, email)
+ user = mgr.ACCESS_CTRL_DB.create_user(username, password, name, email)
except UserAlreadyExists as ex:
return -errno.EEXIST, '', str(ex)
if role:
user.set_roles([role])
- ACCESS_CTRL_DB.save()
+ mgr.ACCESS_CTRL_DB.save()
return 0, json.dumps(user.to_dict()), ''
'Delete user')
def ac_user_delete_cmd(_, username):
try:
- ACCESS_CTRL_DB.delete_user(username)
- ACCESS_CTRL_DB.save()
+ mgr.ACCESS_CTRL_DB.delete_user(username)
+ mgr.ACCESS_CTRL_DB.save()
return 0, "User '{}' deleted".format(username), ""
except UserDoesNotExist as ex:
return -errno.ENOENT, '', str(ex)
roles = []
for rolename in rolesname:
try:
- roles.append(ACCESS_CTRL_DB.get_role(rolename))
+ roles.append(mgr.ACCESS_CTRL_DB.get_role(rolename))
except RoleDoesNotExist as ex:
if rolename not in SYSTEM_ROLES:
return -errno.ENOENT, '', str(ex)
roles.append(SYSTEM_ROLES[rolename])
try:
- user = ACCESS_CTRL_DB.get_user(username)
+ user = mgr.ACCESS_CTRL_DB.get_user(username)
user.set_roles(roles)
- ACCESS_CTRL_DB.save()
+ mgr.ACCESS_CTRL_DB.save()
return 0, json.dumps(user.to_dict()), ''
except UserDoesNotExist as ex:
return -errno.ENOENT, '', str(ex)
roles = []
for rolename in rolesname:
try:
- roles.append(ACCESS_CTRL_DB.get_role(rolename))
+ roles.append(mgr.ACCESS_CTRL_DB.get_role(rolename))
except RoleDoesNotExist as ex:
if rolename not in SYSTEM_ROLES:
return -errno.ENOENT, '', str(ex)
roles.append(SYSTEM_ROLES[rolename])
try:
- user = ACCESS_CTRL_DB.get_user(username)
+ user = mgr.ACCESS_CTRL_DB.get_user(username)
user.add_roles(roles)
- ACCESS_CTRL_DB.save()
+ mgr.ACCESS_CTRL_DB.save()
return 0, json.dumps(user.to_dict()), ''
except UserDoesNotExist as ex:
return -errno.ENOENT, '', str(ex)
roles = []
for rolename in rolesname:
try:
- roles.append(ACCESS_CTRL_DB.get_role(rolename))
+ roles.append(mgr.ACCESS_CTRL_DB.get_role(rolename))
except RoleDoesNotExist as ex:
if rolename not in SYSTEM_ROLES:
return -errno.ENOENT, '', str(ex)
roles.append(SYSTEM_ROLES[rolename])
try:
- user = ACCESS_CTRL_DB.get_user(username)
+ user = mgr.ACCESS_CTRL_DB.get_user(username)
user.del_roles(roles)
- ACCESS_CTRL_DB.save()
+ mgr.ACCESS_CTRL_DB.save()
return 0, json.dumps(user.to_dict()), ''
except UserDoesNotExist as ex:
return -errno.ENOENT, '', str(ex)
'Set user password')
def ac_user_set_password(_, username, password):
try:
- user = ACCESS_CTRL_DB.get_user(username)
+ user = mgr.ACCESS_CTRL_DB.get_user(username)
user.set_password(password)
- ACCESS_CTRL_DB.save()
+ mgr.ACCESS_CTRL_DB.save()
return 0, json.dumps(user.to_dict()), ''
except UserDoesNotExist as ex:
return -errno.ENOENT, '', str(ex)
'Set user info')
def ac_user_set_info(_, username, name, email):
try:
- user = ACCESS_CTRL_DB.get_user(username)
+ user = mgr.ACCESS_CTRL_DB.get_user(username)
if name:
user.name = name
if email:
user.email = email
- ACCESS_CTRL_DB.save()
+ mgr.ACCESS_CTRL_DB.save()
return 0, json.dumps(user.to_dict()), ''
except UserDoesNotExist as ex:
return -errno.ENOENT, '', str(ex)
load_access_control_db()
def get_user(self, username):
- return ACCESS_CTRL_DB.get_user(username)
+ return mgr.ACCESS_CTRL_DB.get_user(username)
def authenticate(self, username, password):
try:
- user = ACCESS_CTRL_DB.get_user(username)
+ user = mgr.ACCESS_CTRL_DB.get_user(username)
if user.password:
pass_hash = password_hash(password, user.password)
if pass_hash == user.password:
return None
def authorize(self, username, scope, permissions):
- user = ACCESS_CTRL_DB.get_user(username)
+ user = mgr.ACCESS_CTRL_DB.get_user(username)
return user.authorize(scope, permissions)
return cls(db['version'], db.get('protocol'), Saml2.from_dict(db.get('saml2')))
-SSO_DB = None
-
-
def load_sso_db():
- # pylint: disable=W0603
- global SSO_DB
- SSO_DB = SsoDB.load()
+ mgr.SSO_DB = SsoDB.load()
SSO_COMMANDS = [
if cmd['prefix'] == 'dashboard sso enable saml2':
try:
- OneLogin_Saml2_Settings(SSO_DB.saml2.onelogin_settings)
+ OneLogin_Saml2_Settings(mgr.SSO_DB.saml2.onelogin_settings)
except OneLogin_Saml2_Error:
return -errno.EPERM, '', 'Single Sign-On is not configured: ' \
'use `ceph dashboard sso setup saml2`'
- SSO_DB.protocol = 'saml2'
- SSO_DB.save()
+ mgr.SSO_DB.protocol = 'saml2'
+ mgr.SSO_DB.save()
return 0, 'SSO is "enabled" with "SAML2" protocol.', ''
if cmd['prefix'] == 'dashboard sso disable':
- SSO_DB.protocol = ''
- SSO_DB.save()
+ mgr.SSO_DB.protocol = ''
+ mgr.SSO_DB.save()
return 0, 'SSO is "disabled".', ''
if cmd['prefix'] == 'dashboard sso status':
- if SSO_DB.protocol == 'saml2':
+ if mgr.SSO_DB.protocol == 'saml2':
return 0, 'SSO is "enabled" with "SAML2" protocol.', ''
return 0, 'SSO is "disabled".', ''
if cmd['prefix'] == 'dashboard sso show saml2':
- return 0, json.dumps(SSO_DB.saml2.to_dict()), ''
+ return 0, json.dumps(mgr.SSO_DB.saml2.to_dict()), ''
if cmd['prefix'] == 'dashboard sso setup saml2':
ceph_dashboard_base_url = cmd['ceph_dashboard_base_url']
}
}
settings = OneLogin_Saml2_IdPMetadataParser.merge_settings(settings, idp_settings)
- SSO_DB.saml2.onelogin_settings = settings
- SSO_DB.protocol = 'saml2'
- SSO_DB.save()
- return 0, json.dumps(SSO_DB.saml2.onelogin_settings), ''
+ mgr.SSO_DB.saml2.onelogin_settings = settings
+ mgr.SSO_DB.protocol = 'saml2'
+ mgr.SSO_DB.save()
+ return 0, json.dumps(mgr.SSO_DB.saml2.onelogin_settings), ''
return -errno.ENOSYS, '', ''
# kludge below
mgr.set_store.side_effect = cls.mock_set_module_option
mgr.get_store.side_effect = cls.mock_get_module_option
+ mgr.ACCESS_CONTROL_DB = None
def setUp(self):
self.CONFIG_KEY_DICT.clear()
# not be needed unless we're testing very specific behaviors.
#
def setup_and_load_persistent_db(self):
- from ..services.access_control import ACCESS_CTRL_DB
- ACCESS_CTRL_DB.save()
+ mgr.ACCESS_CTRL_DB.save()
self.load_persistent_db()
def validate_persistent_role(self, rolename, scopes_permissions,
projects / ceph-ci.git / commitdiff