¡@
OpenStack Study: policy.py**** CubicPower OpenStack Study **** def reset(): global _POLICY_PATH global _POLICY_CACHE _POLICY_PATH = None _POLICY_CACHE = {} policy.reset()
**** CubicPower OpenStack Study **** def init(): global _POLICY_PATH global _POLICY_CACHE if not _POLICY_PATH: _POLICY_PATH = utils.find_config(CONF.policy_file) utils.read_cached_file(_POLICY_PATH, _POLICY_CACHE, reload_func=_set_brain)
**** CubicPower OpenStack Study **** def _set_brain(data): default_rule = CONF.policy_default_rule policy.set_brain(policy.Brain.load_json(data, default_rule))
**** CubicPower OpenStack Study **** def enforce_action(context, action): """Checks that the action can be done by the given context. Applies a check to ensure the context's project_id and user_id can be applied to the given action using the policy enforcement api. """ target = { 'project_id': context.project_id, 'user_id': context.user_id, } enforce(context, action, target)
**** CubicPower OpenStack Study **** def enforce(context, action, target): """Verifies that the action is valid on the target in this context. :param context: cinder context :param action: string representing the action to be checked this should be colon separated for clarity. i.e. ``compute:create_instance``, ``compute:attach_volume``, ``volume:attach_volume`` :param object: dictionary representing the object of the action for object creation this should be a dictionary representing the location of the object e.g. ``{'project_id': context.project_id}`` :raises cinder.exception.PolicyNotAuthorized: if verification fails. """ init() match_list = ('rule:%s' % action,) credentials = context.to_dict() policy.enforce(match_list, target, credentials, exception.PolicyNotAuthorized, action=action)
**** CubicPower OpenStack Study **** def check_is_admin(roles): """Whether or not roles contains 'admin' role according to policy setting. """ init() action = 'context_is_admin' match_list = ('rule:%s' % action,) # include project_id on target to avoid KeyError if context_is_admin # policy definition is missing, and default admin_or_owner rule # attempts to apply. Since our credentials dict does not include a # project_id, this target can never match as a generic rule. target = {'project_id': ''} credentials = {'roles': roles} return policy.enforce(match_list, target, credentials)
|