Implemented a policy enforcement engine
Under monasca_common/policy an enforcement engine is added for using oslo.policy in the other monasca projects. There are same methods as in enforcement engings currently used in nova and keystone projects. Also added unit tests under tests/policy for testing implemented methods. Task: 6105 Story: 2001233 Change-Id: Ic5402ba0986416c9386c1dc3fc1559f148ea9625 Signed-off-by: Amir Mofakhar <amofakhar@op5.com>
This commit is contained in:
Amir Mofakhar
parent
5cacaceae6
commit
4250756a1f
|
|
@ -0,0 +1,46 @@
|
|||
# Copyright 2014 IBM Corp.
|
||||
#
|
||||
# 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.
|
||||
|
||||
"""oslo.i18n integration module.
|
||||
|
||||
See http://docs.openstack.org/developer/oslo.i18n/usage.html .
|
||||
|
||||
"""
|
||||
|
||||
import oslo_i18n
|
||||
|
||||
DOMAIN = 'monasca'
|
||||
|
||||
_translators = oslo_i18n.TranslatorFactory(domain=DOMAIN)
|
||||
|
||||
# The primary translation function using the well-known name "_"
|
||||
_ = _translators.primary
|
||||
|
||||
# Translators for log levels.
|
||||
#
|
||||
# The abbreviated names are meant to reflect the usual use of a short
|
||||
# name like '_'. The "L" is for "log" and the other letter comes from
|
||||
# the level.
|
||||
_LI = _translators.log_info
|
||||
_LW = _translators.log_warning
|
||||
_LE = _translators.log_error
|
||||
_LC = _translators.log_critical
|
||||
|
||||
|
||||
def translate(value, user_locale):
|
||||
return oslo_i18n.translate(value, user_locale)
|
||||
|
||||
|
||||
def get_available_languages():
|
||||
return oslo_i18n.get_available_languages(DOMAIN)
|
||||
|
|
@ -0,0 +1,248 @@
|
|||
# Copyright 2017 OP5 AB
|
||||
# Copyright 2017 FUJITSU LIMITED
|
||||
# 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.
|
||||
|
||||
|
||||
import copy
|
||||
import re
|
||||
import sys
|
||||
|
||||
import logging
|
||||
from oslo_config import cfg
|
||||
from oslo_policy import policy
|
||||
|
||||
from monasca_common.policy.i18n import _LW
|
||||
|
||||
CONF = cfg.CONF
|
||||
LOG = logging.getLogger(__name__)
|
||||
POLICIES = None
|
||||
USER_BASED_RESOURCES = ['os-keypairs']
|
||||
KEY_EXPR = re.compile(r'%\((\w+)\)s')
|
||||
|
||||
|
||||
_ENFORCER = None
|
||||
# oslo_policy will read the policy configuration file again when the file
|
||||
# is changed in runtime so the old policy rules will be saved to
|
||||
# saved_file_rules and used to compare with new rules to determine
|
||||
# whether the rules were updated.
|
||||
saved_file_rules = []
|
||||
|
||||
|
||||
def reset():
|
||||
"""Reset Enforcer class."""
|
||||
global _ENFORCER
|
||||
if _ENFORCER:
|
||||
_ENFORCER.clear()
|
||||
_ENFORCER = None
|
||||
|
||||
|
||||
def init(policy_file=None, rules=None, default_rule=None, use_conf=True):
|
||||
"""Init an Enforcer class.
|
||||
|
||||
:param policy_file: Custom policy file to use, if none is specified,
|
||||
`CONF.policy_file` will be used.
|
||||
:param rules: Default dictionary / Rules to use. It will be
|
||||
considered just in the first instantiation.
|
||||
:param default_rule: Default rule to use, CONF.default_rule will
|
||||
be used if none is specified.
|
||||
:param use_conf: Whether to load rules from config file.
|
||||
"""
|
||||
|
||||
global _ENFORCER
|
||||
global saved_file_rules
|
||||
|
||||
if not _ENFORCER:
|
||||
_ENFORCER = policy.Enforcer(CONF,
|
||||
policy_file=policy_file,
|
||||
rules=rules,
|
||||
default_rule=default_rule,
|
||||
use_conf=use_conf
|
||||
)
|
||||
register_rules(_ENFORCER)
|
||||
_ENFORCER.load_rules()
|
||||
# Only the rules which are loaded from file may be changed
|
||||
current_file_rules = _ENFORCER.file_rules
|
||||
current_file_rules = _serialize_rules(current_file_rules)
|
||||
|
||||
if saved_file_rules != current_file_rules:
|
||||
_warning_for_deprecated_user_based_rules(current_file_rules)
|
||||
saved_file_rules = copy.deepcopy(current_file_rules)
|
||||
|
||||
|
||||
def _serialize_rules(rules):
|
||||
"""Serialize all the Rule object as string.
|
||||
|
||||
New string is used to compare the rules list.
|
||||
"""
|
||||
result = [(rule_name, str(rule)) for rule_name, rule in rules.items()]
|
||||
return sorted(result, key=lambda rule: rule[0])
|
||||
|
||||
|
||||
def _warning_for_deprecated_user_based_rules(rules):
|
||||
"""Warning user based policy enforcement used in the rule but the rule
|
||||
doesn't support it.
|
||||
"""
|
||||
for rule in rules:
|
||||
# We will skip the warning for the resources which support user based
|
||||
# policy enforcement.
|
||||
if [resource for resource in USER_BASED_RESOURCES
|
||||
if resource in rule[0]]:
|
||||
continue
|
||||
if 'user_id' in KEY_EXPR.findall(rule[1]):
|
||||
LOG.warning(_LW("The user_id attribute isn't supported in the "
|
||||
"rule '%s'. All the user_id based policy "
|
||||
"enforcement will be removed in the "
|
||||
"future."), rule[0])
|
||||
|
||||
|
||||
def register_rules(enforcer):
|
||||
"""Register default policy rules."""
|
||||
rules = POLICIES.list_rules()
|
||||
enforcer.register_defaults(rules)
|
||||
|
||||
|
||||
def authorize(context, action, target, do_raise=True):
|
||||
"""Verify that the action is valid on the target in this context.
|
||||
|
||||
:param context: monasca project context
|
||||
:param action: String representing the action to be checked. This
|
||||
should be colon separated for clarity.
|
||||
:param target: 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'}``
|
||||
:param do_raise: if True (the default), raises PolicyNotAuthorized,
|
||||
if False returns False
|
||||
:type context: object
|
||||
:type action: str
|
||||
:type target: dict
|
||||
:type do_raise: bool
|
||||
:return: returns a non-False value (not necessarily True) if authorized,
|
||||
and the False if not authorized and do_raise if False
|
||||
|
||||
:raises oslo_policy.policy.PolicyNotAuthorized: if verification fails
|
||||
"""
|
||||
init()
|
||||
credentials = context.to_policy_values()
|
||||
try:
|
||||
result = _ENFORCER.authorize(action, target, credentials,
|
||||
do_raise=do_raise, action=action)
|
||||
return result
|
||||
except policy.PolicyNotRegistered:
|
||||
LOG.exception('Policy not registered')
|
||||
raise
|
||||
except Exception:
|
||||
LOG.debug('Policy check for %(action)s failed with credentials '
|
||||
'%(credentials)s',
|
||||
{'action': action, 'credentials': credentials})
|
||||
raise
|
||||
|
||||
|
||||
def check_is_admin(context):
|
||||
"""Check if roles contains 'admin' role according to policy settings."""
|
||||
init()
|
||||
credentials = context.to_policy_values()
|
||||
target = credentials
|
||||
return _ENFORCER.authorize('admin_required', target, credentials)
|
||||
|
||||
|
||||
def set_rules(rules, overwrite=True, use_conf=False): # pragma: no cover
|
||||
"""Set rules based on the provided dict of rules.
|
||||
|
||||
Note:
|
||||
Used in tests only.
|
||||
|
||||
:param rules: New rules to use. It should be an instance of dict
|
||||
:param overwrite: Whether to overwrite current rules or update them
|
||||
with the new rules.
|
||||
:param use_conf: Whether to reload rules from config file.
|
||||
"""
|
||||
init(use_conf=False)
|
||||
_ENFORCER.set_rules(rules, overwrite, use_conf)
|
||||
|
||||
|
||||
def verify_deprecated_policy(old_policy, new_policy, default_rule, context):
|
||||
"""Check the rule of the deprecated policy action
|
||||
|
||||
If the current rule of the deprecated policy action is set to a non-default
|
||||
value, then a warning message is logged stating that the new policy
|
||||
action should be used to dictate permissions as the old policy action is
|
||||
being deprecated.
|
||||
|
||||
:param old_policy: policy action that is being deprecated
|
||||
:param new_policy: policy action that is replacing old_policy
|
||||
:param default_rule: the old_policy action default rule value
|
||||
:param context: the monasca context
|
||||
"""
|
||||
|
||||
if _ENFORCER:
|
||||
current_rule = str(_ENFORCER.rules[old_policy])
|
||||
else:
|
||||
current_rule = None
|
||||
|
||||
if current_rule != default_rule:
|
||||
LOG.warning("Start using the new action '{0}'. The existing "
|
||||
"action '{1}' is being deprecated and will be "
|
||||
"removed in future release.".format(new_policy,
|
||||
old_policy))
|
||||
target = {'project_id': context.project_id,
|
||||
'user_id': context.user_id}
|
||||
|
||||
return authorize(context=context, action=old_policy, target=target)
|
||||
else:
|
||||
return False
|
||||
|
||||
|
||||
def get_rules():
|
||||
if _ENFORCER:
|
||||
return _ENFORCER.rules
|
||||
|
||||
|
||||
def get_enforcer():
|
||||
# This method is for use by oslopolicy CLI scripts. Those scripts need the
|
||||
# 'output-file' and 'namespace' options, but having those in sys.argv means
|
||||
# loading the project config options will fail as those are not expected to
|
||||
# be present. So we pass in an arg list with those stripped out.
|
||||
conf_args = []
|
||||
# Start at 1 because cfg.CONF expects the equivalent of sys.argv[1:]
|
||||
i = 1
|
||||
while i < len(sys.argv):
|
||||
if sys.argv[i].strip('-') in ['namespace', 'output-file']:
|
||||
i += 2
|
||||
continue
|
||||
conf_args.append(sys.argv[i])
|
||||
i += 1
|
||||
|
||||
cfg.CONF(conf_args, project='monasca')
|
||||
init()
|
||||
return _ENFORCER
|
||||
|
||||
|
||||
@policy.register('is_admin')
|
||||
class IsAdminCheck(policy.Check):
|
||||
"""An explicit check for is_admin."""
|
||||
|
||||
def __init__(self, kind, match):
|
||||
"""Initialize the check."""
|
||||
|
||||
self.expected = (match.lower() == 'true')
|
||||
|
||||
super(IsAdminCheck, self).__init__(kind, str(self.expected))
|
||||
|
||||
def __call__(self, target, creds, enforcer):
|
||||
"""Determine whether is_admin matches the requested value."""
|
||||
|
||||
return creds['is_admin'] == self.expected
|
||||
|
|
@ -0,0 +1,102 @@
|
|||
# Copyright 2017 OP5 AB
|
||||
# Copyright 2010 United States Government as represented by the
|
||||
# Administrator of the National Aeronautics and Space Administration.
|
||||
# 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.
|
||||
|
||||
"""Base classes for policy unit tests."""
|
||||
|
||||
import os
|
||||
|
||||
import fixtures
|
||||
|
||||
from oslo_config import cfg
|
||||
from oslo_config import fixture as config_fixture
|
||||
from oslo_policy import opts as policy_opts
|
||||
from oslo_serialization import jsonutils
|
||||
from oslotest import base
|
||||
|
||||
from monasca_common.policy import policy_engine
|
||||
|
||||
CONF = cfg.CONF
|
||||
|
||||
|
||||
class FakePolicy(object):
|
||||
def list_rules(self):
|
||||
return []
|
||||
|
||||
|
||||
class ConfigFixture(config_fixture.Config):
|
||||
|
||||
def setUp(self):
|
||||
super(ConfigFixture, self).setUp()
|
||||
CONF(args=[],
|
||||
prog='common',
|
||||
project='monasca',
|
||||
version=0,
|
||||
description='Testing monasca-common')
|
||||
policy_opts.set_defaults(CONF)
|
||||
|
||||
|
||||
class BaseTestCase(base.BaseTestCase):
|
||||
def setUp(self):
|
||||
super(BaseTestCase, self).setUp()
|
||||
self.useFixture(ConfigFixture(CONF))
|
||||
self.useFixture(EmptyPolicyFixture())
|
||||
|
||||
@staticmethod
|
||||
def conf_override(**kw):
|
||||
"""Override flag variables for a test."""
|
||||
group = kw.pop('group', None)
|
||||
for k, v in kw.items():
|
||||
CONF.set_override(k, v, group)
|
||||
|
||||
|
||||
class EmptyPolicyFixture(fixtures.Fixture):
|
||||
"""Override the policy with an empty policy file.
|
||||
|
||||
This overrides the policy with a completely fake and synthetic
|
||||
policy file.
|
||||
|
||||
"""
|
||||
def setUp(self):
|
||||
super(EmptyPolicyFixture, self).setUp()
|
||||
self._prepare_policy()
|
||||
policy_engine.POLICIES = FakePolicy()
|
||||
policy_engine.reset()
|
||||
policy_engine.init()
|
||||
self.addCleanup(policy_engine.reset)
|
||||
|
||||
def _prepare_policy(self):
|
||||
|
||||
policy_dir = self.useFixture(fixtures.TempDir())
|
||||
policy_file = os.path.join(policy_dir.path, 'policy.yaml')
|
||||
|
||||
policy_rules = jsonutils.loads('{}')
|
||||
|
||||
self.add_missing_default_rules(policy_rules)
|
||||
|
||||
with open(policy_file, 'w') as f:
|
||||
jsonutils.dump(policy_rules, f)
|
||||
|
||||
BaseTestCase.conf_override(policy_file=policy_file,
|
||||
group='oslo_policy')
|
||||
BaseTestCase.conf_override(policy_dirs=[], group='oslo_policy')
|
||||
|
||||
def add_missing_default_rules(self, rules):
|
||||
policies = FakePolicy()
|
||||
|
||||
for rule in policies.list_rules():
|
||||
if rule.name not in rules:
|
||||
rules[rule.name] = rule.check_str
|
||||
|
|
@ -0,0 +1,275 @@
|
|||
# Copyright 2017 OP5 AB
|
||||
# Copyright 2011 Piston Cloud Computing, Inc.
|
||||
# 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.
|
||||
|
||||
import mock
|
||||
import requests_mock
|
||||
|
||||
from oslo_context import context
|
||||
from oslo_policy import policy as os_policy
|
||||
|
||||
from monasca_common.policy import policy_engine
|
||||
from monasca_common.tests.policy import base
|
||||
|
||||
|
||||
class PolicyFileTestCase(base.BaseTestCase):
|
||||
def setUp(self):
|
||||
super(PolicyFileTestCase, self).setUp()
|
||||
self.context = context.RequestContext(user='fake',
|
||||
tenant='fake',
|
||||
is_admin=False)
|
||||
self.target = {}
|
||||
|
||||
def test_modified_policy_reloads(self):
|
||||
tmp_file = \
|
||||
self.create_tempfiles(files=[('policies', '{}')], ext='.yaml')[0]
|
||||
base.BaseTestCase.conf_override(policy_file=tmp_file,
|
||||
group='oslo_policy')
|
||||
|
||||
policy_engine.reset()
|
||||
policy_engine.init()
|
||||
|
||||
action = 'example:test'
|
||||
rule = os_policy.RuleDefault(action, '')
|
||||
policy_engine._ENFORCER.register_defaults([rule])
|
||||
|
||||
with open(tmp_file, 'w') as policy_file:
|
||||
policy_file.write('{"example:test": ""}')
|
||||
policy_engine.authorize(self.context, action, self.target)
|
||||
|
||||
with open(tmp_file, 'w') as policy_file:
|
||||
policy_file.write('{"example:test": "!"}')
|
||||
policy_engine._ENFORCER.load_rules(True)
|
||||
self.assertRaises(os_policy.PolicyNotAuthorized,
|
||||
policy_engine.authorize,
|
||||
self.context, action, self.target)
|
||||
|
||||
|
||||
|
||||
class PolicyTestCase(base.BaseTestCase):
|
||||
def setUp(self):
|
||||
super(PolicyTestCase, self).setUp()
|
||||
rules = [
|
||||
os_policy.RuleDefault("true", "@"),
|
||||
os_policy.RuleDefault("example:allowed", "@"),
|
||||
os_policy.RuleDefault("example:denied", "!"),
|
||||
os_policy.RuleDefault("old_action_not_default", "@"),
|
||||
os_policy.RuleDefault("new_action", "@"),
|
||||
os_policy.RuleDefault("old_action_default", "rule:admin_api"),
|
||||
os_policy.RuleDefault("example:lowercase_admin",
|
||||
"role:admin or role:sysadmin"),
|
||||
os_policy.RuleDefault("example:uppercase_admin",
|
||||
"role:ADMIN or role:sysadmin"),
|
||||
os_policy.RuleDefault("example:get_http",
|
||||
"http://www.example.com"),
|
||||
os_policy.RuleDefault("example:my_file",
|
||||
"role:compute_admin or "
|
||||
"project_id:%(project_id)s"),
|
||||
os_policy.RuleDefault("example:early_and_fail", "! and @"),
|
||||
os_policy.RuleDefault("example:early_or_success", "@ or !"),
|
||||
]
|
||||
policy_engine.reset()
|
||||
policy_engine.init()
|
||||
|
||||
self.context = context.RequestContext(user='fake',
|
||||
tenant='fake',
|
||||
is_admin=False)
|
||||
policy_engine._ENFORCER.register_defaults(rules)
|
||||
self.target = {}
|
||||
|
||||
def test_authorize_nonexistent_action_throws(self):
|
||||
|
||||
action = 'example:noexists'
|
||||
self.assertRaises(os_policy.PolicyNotRegistered, policy_engine.authorize,
|
||||
self.context, action, self.target)
|
||||
|
||||
def test_authorize_bad_action_throws(self):
|
||||
action = 'example:denied'
|
||||
self.assertRaises(os_policy.PolicyNotAuthorized, policy_engine.authorize,
|
||||
self.context, action, self.target)
|
||||
|
||||
def test_authorize_bad_action_noraise(self):
|
||||
action = "example:denied"
|
||||
result = policy_engine.authorize(self.context, action, self.target, False)
|
||||
self.assertFalse(result)
|
||||
|
||||
def test_authorize_good_action(self):
|
||||
action = "example:allowed"
|
||||
result = policy_engine.authorize(self.context, action, self.target)
|
||||
self.assertTrue(result)
|
||||
|
||||
@requests_mock.mock()
|
||||
def test_authorize_http_true(self, req_mock):
|
||||
req_mock.post('http://www.example.com/',
|
||||
text='True')
|
||||
action = "example:get_http"
|
||||
target = {}
|
||||
result = policy_engine.authorize(self.context, action, target)
|
||||
self.assertTrue(result)
|
||||
|
||||
@requests_mock.mock()
|
||||
def test_authorize_http_false(self, req_mock):
|
||||
req_mock.post('http://www.example.com/',
|
||||
text='False')
|
||||
action = "example:get_http"
|
||||
target = {}
|
||||
self.assertRaises(os_policy.PolicyNotAuthorized, policy_engine.authorize,
|
||||
self.context, action, target)
|
||||
|
||||
def test_templatized_authorization(self):
|
||||
target_mine = {'project_id': 'fake'}
|
||||
target_not_mine = {'project_id': 'another'}
|
||||
action = "example:my_file"
|
||||
policy_engine.authorize(self.context, action, target_mine)
|
||||
self.assertRaises(os_policy.PolicyNotAuthorized, policy_engine.authorize,
|
||||
self.context, action, target_not_mine)
|
||||
|
||||
def test_early_AND_authorization(self):
|
||||
action = "example:early_and_fail"
|
||||
self.assertRaises(os_policy.PolicyNotAuthorized, policy_engine.authorize,
|
||||
self.context, action, self.target)
|
||||
|
||||
def test_early_OR_authorization(self):
|
||||
action = "example:early_or_success"
|
||||
policy_engine.authorize(self.context, action, self.target)
|
||||
|
||||
def test_ignore_case_role_check(self):
|
||||
lowercase_action = "example:lowercase_admin"
|
||||
uppercase_action = "example:uppercase_admin"
|
||||
# NOTE(dprince) we mix case in the Admin role here to ensure
|
||||
# case is ignored
|
||||
admin_context = context.RequestContext('admin',
|
||||
'fake',
|
||||
roles=['AdMiN'])
|
||||
policy_engine.authorize(admin_context, lowercase_action, self.target)
|
||||
policy_engine.authorize(admin_context, uppercase_action, self.target)
|
||||
|
||||
@mock.patch.object(policy_engine.LOG, 'warning')
|
||||
def test_warning_when_deprecated_user_based_rule_used(self, mock_warning):
|
||||
policy_engine._warning_for_deprecated_user_based_rules(
|
||||
[("os_compute_api:servers:index",
|
||||
"project_id:%(project_id)s or user_id:%(user_id)s")])
|
||||
mock_warning.assert_called_once_with(
|
||||
u"The user_id attribute isn't supported in the rule "
|
||||
"'%s'. All the user_id based policy enforcement will be removed "
|
||||
"in the future.", "os_compute_api:servers:index")
|
||||
|
||||
@mock.patch.object(policy_engine.LOG, 'warning')
|
||||
def test_no_warning_for_user_based_resource(self, mock_warning):
|
||||
policy_engine._warning_for_deprecated_user_based_rules(
|
||||
[("os_compute_api:os-keypairs:index",
|
||||
"user_id:%(user_id)s")])
|
||||
mock_warning.assert_not_called()
|
||||
|
||||
@mock.patch.object(policy_engine.LOG, 'warning')
|
||||
def test_no_warning_for_no_user_based_rule(self, mock_warning):
|
||||
policy_engine._warning_for_deprecated_user_based_rules(
|
||||
[("os_compute_api:servers:index",
|
||||
"project_id:%(project_id)s")])
|
||||
mock_warning.assert_not_called()
|
||||
|
||||
@mock.patch.object(policy_engine.LOG, 'warning')
|
||||
def test_verify_deprecated_policy_using_old_action(self, mock_warning):
|
||||
policy_engine._ENFORCER.load_rules(True)
|
||||
old_policy = "old_action_not_default"
|
||||
new_policy = "new_action"
|
||||
default_rule = "rule:admin_api"
|
||||
|
||||
using_old_action = policy_engine.verify_deprecated_policy(
|
||||
old_policy, new_policy, default_rule, self.context)
|
||||
|
||||
mock_warning.assert_called_once_with(
|
||||
"Start using the new action '{0}'. The existing action '{1}' is "
|
||||
"being deprecated and will be removed in "
|
||||
"future release.".format(new_policy, old_policy))
|
||||
self.assertTrue(using_old_action)
|
||||
|
||||
def test_verify_deprecated_policy_using_new_action(self):
|
||||
policy_engine._ENFORCER.load_rules(True)
|
||||
old_policy = "old_action_default"
|
||||
new_policy = "new_action"
|
||||
default_rule = "rule:admin_api"
|
||||
|
||||
using_old_action = policy_engine.verify_deprecated_policy(
|
||||
old_policy, new_policy, default_rule, self.context)
|
||||
|
||||
self.assertFalse(using_old_action)
|
||||
|
||||
|
||||
class IsAdminCheckTestCase(base.BaseTestCase):
|
||||
def setUp(self):
|
||||
super(IsAdminCheckTestCase, self).setUp()
|
||||
policy_engine.init()
|
||||
|
||||
def test_init_true(self):
|
||||
check = policy_engine.IsAdminCheck('is_admin', 'True')
|
||||
|
||||
self.assertEqual(check.kind, 'is_admin')
|
||||
self.assertEqual(check.match, 'True')
|
||||
self.assertTrue(check.expected)
|
||||
|
||||
def test_init_false(self):
|
||||
check = policy_engine.IsAdminCheck('is_admin', 'nottrue')
|
||||
|
||||
self.assertEqual(check.kind, 'is_admin')
|
||||
self.assertEqual(check.match, 'False')
|
||||
self.assertFalse(check.expected)
|
||||
|
||||
def test_call_true(self):
|
||||
check = policy_engine.IsAdminCheck('is_admin', 'True')
|
||||
|
||||
self.assertTrue(check('target', dict(is_admin=True),
|
||||
policy_engine._ENFORCER))
|
||||
self.assertFalse(check('target', dict(is_admin=False),
|
||||
policy_engine._ENFORCER))
|
||||
|
||||
def test_call_false(self):
|
||||
check = policy_engine.IsAdminCheck('is_admin', 'False')
|
||||
|
||||
self.assertFalse(check('target', dict(is_admin=True),
|
||||
policy_engine._ENFORCER))
|
||||
self.assertTrue(check('target', dict(is_admin=False),
|
||||
policy_engine._ENFORCER))
|
||||
|
||||
|
||||
class AdminRolePolicyTestCase(base.BaseTestCase):
|
||||
def setUp(self):
|
||||
super(AdminRolePolicyTestCase, self).setUp()
|
||||
self.noadmin_context = context.RequestContext('fake', 'fake',
|
||||
roles=['member'])
|
||||
self.admin_context = context.RequestContext('fake', 'fake',
|
||||
roles=['admin'])
|
||||
|
||||
admin_rule = [
|
||||
os_policy.RuleDefault('example.admin', 'role:admin'),
|
||||
]
|
||||
policy_engine.reset()
|
||||
policy_engine.init(policy_file=None)
|
||||
policy_engine._ENFORCER.register_defaults(admin_rule)
|
||||
policy_engine._ENFORCER.load_rules(True)
|
||||
self.target = {}
|
||||
|
||||
def test_authorize_admin_actions_with_admin_context(self):
|
||||
for action in policy_engine.get_rules().keys():
|
||||
policy_engine.authorize(self.admin_context, action, self.target)
|
||||
|
||||
def test_authorize_admin_actions_with_nonadmin_context_throws(self):
|
||||
"""Check if non-admin context passed to admin actions throws
|
||||
Policy not authorized exception
|
||||
"""
|
||||
for action in policy_engine.get_rules().keys():
|
||||
self.assertRaises(os_policy.PolicyNotAuthorized,
|
||||
policy_engine.authorize,
|
||||
self.noadmin_context, action, self.target)
|
||||
|
|
@ -6,6 +6,7 @@ kazoo>=2.2 # Apache-2.0
|
|||
pykafka>=2.5.0 # Apache 2.0 License
|
||||
PyMySQL>=0.7.6 # MIT License
|
||||
oslo.config>=5.1.0 # Apache-2.0
|
||||
oslo.policy>=1.30.0 # Apache-2.0
|
||||
pbr!=2.1.0,>=2.0.0 # Apache-2.0
|
||||
pyparsing>=2.1.0 # MIT
|
||||
ujson>=1.35 # BSD
|
||||
|
|
|
|||
|
|
@ -10,8 +10,10 @@ fixtures>=3.0.0 # Apache-2.0/BSD
|
|||
httplib2>=0.9.1 # MIT
|
||||
mock>=2.0.0 # BSD
|
||||
mox>=0.5.3 # Apache-2.0
|
||||
oslo.context>=2.19.2 # Apache-2.0
|
||||
oslotest>=1.10.0 # Apache-2.0
|
||||
os-testr>=1.0.0 # Apache-2.0
|
||||
requests-mock>=1.1.0 # Apache-2.0
|
||||
testrepository>=0.0.18 # Apache-2.0/BSD
|
||||
testscenarios>=0.4 # Apache-2.0/BSD
|
||||
testtools>=2.2.0 # MIT
|
||||
|
|
|
|||
Loading…
Reference in New Issue