deb-heat/heat/common/policy.py
Jens Rosenboom a99fe6a760 Make import of oslo libraries namespaceless
See https://blueprints.launchpad.net/oslo-incubator/+spec/drop-namespace-packages

Closes-Bug: 1423174
Change-Id: I48680ce6e7ce91005c147ab4388203946171d433
2015-02-18 22:54:10 +01:00

87 lines
3.3 KiB
Python

#
# Copyright (c) 2011 OpenStack Foundation
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
# Based on glance/api/policy.py
"""Policy Engine For Heat"""
from oslo_config import cfg
from heat.common import exception
from heat.openstack.common import policy
CONF = cfg.CONF
DEFAULT_RULES = {
'default': policy.FalseCheck(),
}
class Enforcer(object):
"""Responsible for loading and enforcing rules."""
def __init__(self, scope='heat', exc=exception.Forbidden,
default_rule=DEFAULT_RULES['default'], policy_file=None):
self.scope = scope
self.exc = exc
self.default_rule = default_rule
self.enforcer = policy.Enforcer(
default_rule=default_rule, policy_file=policy_file)
def set_rules(self, rules, overwrite=True):
"""Create a new Rules object based on the provided dict of rules."""
rules_obj = policy.Rules(rules, self.default_rule)
self.enforcer.set_rules(rules_obj, overwrite)
def load_rules(self, force_reload=False):
"""Set the rules found in the json file on disk."""
self.enforcer.load_rules(force_reload)
def _check(self, context, rule, target, exc, *args, **kwargs):
"""Verifies that the action is valid on the target in this context.
:param context: Heat request context
:param rule: String representing the action to be checked
:param target: Dictionary representing the object of the action.
:raises: self.exc (defaults to heat.common.exception.Forbidden)
:returns: A non-False value if access is allowed.
"""
do_raise = False if not exc else True
credentials = context.to_dict()
return self.enforcer.enforce(rule, target, credentials,
do_raise, exc=exc, *args, **kwargs)
def enforce(self, context, action, scope=None, target=None):
"""Verifies that the action is valid on the target in this context.
:param context: Heat request context
:param action: String representing the action to be checked
:param target: Dictionary representing the object of the action.
:raises: self.exc (defaults to heat.common.exception.Forbidden)
:returns: A non-False value if access is allowed.
"""
_action = '%s:%s' % (scope or self.scope, action)
_target = target or {}
return self._check(context, _action, _target, self.exc, action=action)
def check_is_admin(self, context):
"""Whether or not roles contains 'admin' role according to policy.json
:param context: Heat request context
:returns: A non-False value if the user is admin according to policy
"""
return self._check(context, 'context_is_admin', target={}, exc=None)