{
"context_is_admin": "role:admin",
"admin_or_owner": "rule:context_is_admin or tenant_id:%(tenant_id)s",
- "admin_or_network_owner": "rule:context_is_admin or tenant_id:%(network_tenant_id)s",
+ "admin_or_network_owner": "rule:context_is_admin or tenant_id:%(network:tenant_id)s",
"admin_only": "rule:context_is_admin",
"regular_user": "",
"shared": "field:networks:shared=True",
}
}
-# Associates to each resource its own parent resource
-# Resources without parents, such as networks, are not in this list
+# Identify the attribute used by a resource to reference another resource
-RESOURCE_HIERARCHY_MAP = {
- PORTS: {'parent': NETWORKS, 'identified_by': 'network_id'},
- SUBNETS: {'parent': NETWORKS, 'identified_by': 'network_id'}
+RESOURCE_FOREIGN_KEYS = {
+ NETWORKS: 'network_id'
}
PLURALS = {NETWORKS: NETWORK,
# TODO(salvatore-orlando): bp/make-authz-ortogonal
# The body of the action request should be included
# in the info passed to the policy engine
- # Enforce policy, if any, for this action
# It is ok to raise a 403 because accessibility to the
# object was checked earlier in this method
- policy.enforce(request.context, name, resource,
- plugin=self._plugin)
+ policy.enforce(request.context, name, resource)
return getattr(self._plugin, name)(*arg_list, **kwargs)
return _handle_action
else:
# FIXME(salvatore-orlando): obj_getter might return references to
# other resources. Must check authZ on them too.
if do_authz:
- policy.enforce(request.context, action, obj, plugin=self._plugin)
+ policy.enforce(request.context, action, obj)
return obj
def _send_dhcp_notification(self, context, data, methodname):
item[self._resource])
policy.enforce(request.context,
action,
- item[self._resource],
- plugin=self._plugin)
+ item[self._resource])
try:
tenant_id = item[self._resource]['tenant_id']
count = quota.QUOTAS.count(request.context, self._resource,
try:
policy.enforce(request.context,
action,
- obj,
- plugin=self._plugin)
+ obj)
except exceptions.PolicyNotAuthorized:
# To avoid giving away information, pretend that it
# doesn't exist
try:
policy.enforce(request.context,
action,
- orig_obj,
- plugin=self._plugin)
+ orig_obj)
except exceptions.PolicyNotAuthorized:
# To avoid giving away information, pretend that it
# doesn't exist
message = _("Requested rule:%(rule)s cannot be found")
+class PolicyInitError(QuantumException):
+ message = _("Failed to init policy %(policy)s because %(reason)s")
+
+
+class PolicyCheckError(QuantumException):
+ message = _("Failed to check policy %(policy)s because %(reason)s")
+
+
class StateInvalid(BadRequest):
message = _("Unsupported port state: %(port_state)s")
plugin = manager.QuantumManager.get_plugin()
policy.enforce(request.context,
"get_%s" % DHCP_NETS,
- {},
- plugin=plugin)
+ {})
return plugin.list_networks_on_dhcp_agent(
request.context, kwargs['agent_id'])
plugin = manager.QuantumManager.get_plugin()
policy.enforce(request.context,
"create_%s" % DHCP_NET,
- {},
- plugin=plugin)
+ {})
return plugin.add_network_to_dhcp_agent(
request.context, kwargs['agent_id'], body['network_id'])
plugin = manager.QuantumManager.get_plugin()
policy.enforce(request.context,
"delete_%s" % DHCP_NET,
- {},
- plugin=plugin)
+ {})
return plugin.remove_network_from_dhcp_agent(
request.context, kwargs['agent_id'], id)
plugin = manager.QuantumManager.get_plugin()
policy.enforce(request.context,
"get_%s" % L3_ROUTERS,
- {},
- plugin=plugin)
+ {})
return plugin.list_routers_on_l3_agent(
request.context, kwargs['agent_id'])
plugin = manager.QuantumManager.get_plugin()
policy.enforce(request.context,
"create_%s" % L3_ROUTER,
- {},
- plugin=plugin)
+ {})
return plugin.add_router_to_l3_agent(
request.context,
kwargs['agent_id'],
plugin = manager.QuantumManager.get_plugin()
policy.enforce(request.context,
"delete_%s" % L3_ROUTER,
- {},
- plugin=plugin)
+ {})
return plugin.remove_router_from_l3_agent(
request.context, kwargs['agent_id'], id)
plugin = manager.QuantumManager.get_plugin()
policy.enforce(request.context,
"get_%s" % DHCP_AGENTS,
- {},
- plugin=plugin)
+ {})
return plugin.list_dhcp_agents_hosting_network(
request.context, kwargs['network_id'])
plugin = manager.QuantumManager.get_plugin()
policy.enforce(request.context,
"get_%s" % L3_AGENTS,
- {},
- plugin=plugin)
+ {})
return plugin.list_l3_agents_hosting_router(
request.context, kwargs['router_id'])
Policy engine for quantum. Largely copied from nova.
"""
import itertools
+import re
from oslo.config import cfg
from quantum.api.v2 import attributes
from quantum.common import exceptions
import quantum.common.utils as utils
+from quantum import manager
+from quantum.openstack.common import importutils
from quantum.openstack.common import log as logging
from quantum.openstack.common import policy
target[attribute_name] != resource[attribute_name]['default'])
-def _build_target(action, original_target, plugin, context):
- """Augment dictionary of target attributes for policy engine.
-
- This routine adds to the dictionary attributes belonging to the
- "parent" resource of the targeted one.
- """
- target = original_target.copy()
- resource, _a = get_resource_and_action(action)
- hierarchy_info = attributes.RESOURCE_HIERARCHY_MAP.get(resource, None)
- if hierarchy_info and plugin:
- # use the 'singular' version of the resource name
- parent_resource = hierarchy_info['parent'][:-1]
- parent_id = hierarchy_info['identified_by']
- f = getattr(plugin, 'get_%s' % parent_resource)
- # f *must* exist, if not found it is better to let quantum explode
- # Note: we do not use admin context
- data = f(context, target[parent_id], fields=['tenant_id'])
- target['%s_tenant_id' % parent_resource] = data['tenant_id']
- return target
-
-
def _build_match_rule(action, target):
"""Create the rule to match for a given action.
return match_rule
+# This check is registered as 'tenant_id' so that it can override
+# GenericCheck which was used for validating parent resource ownership.
+# This will prevent us from having to handling backward compatibility
+# for policy.json
+# TODO(salv-orlando): Reinstate GenericCheck for simple tenant_id checks
+@policy.register('tenant_id')
+class OwnerCheck(policy.Check):
+ """Resource ownership check.
+
+ This check verifies the owner of the current resource, or of another
+ resource referenced by the one under analysis.
+ In the former case it falls back to a regular GenericCheck, whereas
+ in the latter case it leverages the plugin to load the referenced
+ resource and perform the check.
+ """
+ def __init__(self, kind, match):
+ # Process the match
+ try:
+ self.target_field = re.findall('^\%\((.*)\)s$',
+ match)[0]
+ except IndexError:
+ err_reason = (_("Unable to identify a target field from:%s."
+ "match should be in the form %%(<field_name>)s"),
+ match)
+ LOG.exception(err_reason)
+ raise exceptions.PolicyInitError(
+ policy="%s:%s" % (kind, match),
+ reason=err_reason)
+ super(OwnerCheck, self).__init__(kind, match)
+
+ def __call__(self, target, creds):
+ if self.target_field not in target:
+ # policy needs a plugin check
+ # target field is in the form resource:field
+ # however if they're not separated by a colon, use an underscore
+ # as a separator for backward compatibility
+
+ def do_split(separator):
+ parent_res, parent_field = self.target_field.split(
+ separator, 1)
+ return parent_res, parent_field
+
+ for separator in (':', '_'):
+ try:
+ parent_res, parent_field = do_split(separator)
+ break
+ except ValueError:
+ LOG.debug(_("Unable to find ':' as separator in %s."),
+ self.target_field)
+ else:
+ # If we are here split failed with both separators
+ err_reason = (_("Unable to find resource name in %s") %
+ self.target_field)
+ LOG.exception(err_reason)
+ raise exceptions.PolicyCheckError(
+ policy="%s:%s" % (self.kind, self.match),
+ reason=err_reason)
+ parent_foreign_key = attributes.RESOURCE_FOREIGN_KEYS.get(
+ "%ss" % parent_res, None)
+ if not parent_foreign_key:
+ err_reason = (_("Unable to verify match:%(match)s as the "
+ "parent resource: %(res)s was not found") %
+ {'match': self.match, 'res': parent_res})
+ LOG.exception(err_reason)
+ raise exceptions.PolicyCheckError(
+ policy="%s:%s" % (self.kind, self.match),
+ reason=err_reason)
+ # NOTE(salv-orlando): This check currently assumes the parent
+ # resource is handled by the core plugin. It might be worth
+ # having a way to map resources to plugins so to make this
+ # check more general
+ f = getattr(manager.QuantumManager.get_instance().plugin,
+ 'get_%s' % parent_res)
+ # f *must* exist, if not found it is better to let quantum
+ # explode. Check will be performed with admin context
+ context = importutils.import_module('quantum.context')
+ data = f(context.get_admin_context(),
+ target[parent_foreign_key],
+ fields=[parent_field])
+ target[self.target_field] = data[parent_field]
+ match = self.match % target
+ if self.kind in creds:
+ return match == unicode(creds[self.kind])
+ return False
+
+
@policy.register('field')
class FieldCheck(policy.Check):
def __init__(self, kind, match):
{'field': self.field,
'target_dict': target_dict})
return False
-
return target_value == self.value
-def _prepare_check(context, action, target, plugin=None):
+def _prepare_check(context, action, target):
"""Prepare rule, target, and credentials for the policy engine."""
init()
# Compare with None to distinguish case in which target is {}
if target is None:
target = {}
- # Update target only if plugin is provided
- if plugin:
- target = _build_target(action, target, plugin, context)
match_rule = _build_match_rule(action, target)
credentials = context.to_dict()
return match_rule, target, credentials
: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 plugin: quantum plugin used to retrieve information required
- for augmenting the target
+ :param plugin: currently unused and deprecated.
+ Kept for backward compatibility.
:return: Returns True if access is permitted else False.
"""
- return policy.check(*(_prepare_check(context, action, target, plugin)))
+ return policy.check(*(_prepare_check(context, action, target)))
def check_if_exists(context, action, target):
: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 plugin: quantum plugin used to retrieve information required
- for augmenting the target
+ :param plugin: currently unused and deprecated.
+ Kept for backward compatibility.
:raises quantum.exceptions.PolicyNotAllowed: if verification fails.
"""
init()
- # Compare with None to distinguish case in which target is {}
- if target is None:
- target = {}
- real_target = _build_target(action, target, plugin, context)
- match_rule = _build_match_rule(action, real_target)
- credentials = context.to_dict()
- return policy.check(match_rule, real_target, credentials,
- exceptions.PolicyNotAuthorized, action=action)
+ rule, target, credentials = _prepare_check(context, action, target)
+ return policy.check(rule, target, credentials,
+ exc=exceptions.PolicyNotAuthorized, action=action)
def check_is_admin(context):
import quantum
from quantum.common import exceptions
from quantum import context
+from quantum import manager
from quantum.openstack.common import importutils
from quantum.openstack.common import policy as common_policy
from quantum import policy
self.rules = dict((k, common_policy.parse_rule(v)) for k, v in {
"context_is_admin": "role:admin",
"admin_or_network_owner": "rule:context_is_admin or "
- "tenant_id:%(network_tenant_id)s",
+ "tenant_id:%(network:tenant_id)s",
"admin_or_owner": ("rule:context_is_admin or "
"tenant_id:%(tenant_id)s"),
"admin_only": "rule:context_is_admin",
self.context = context.Context('fake', 'fake', roles=['user'])
plugin_klass = importutils.import_class(
"quantum.db.db_base_plugin_v2.QuantumDbPluginV2")
- self.plugin = plugin_klass()
+ self.manager_patcher = mock.patch('quantum.manager.QuantumManager')
+ fake_manager = self.manager_patcher.start()
+ fake_manager_instance = fake_manager.return_value
+ fake_manager_instance.plugin = plugin_klass()
+ self.addCleanup(self.manager_patcher.stop)
def _test_action_on_attr(self, context, action, attr, value,
exception=None):
target = {'tenant_id': 'the_owner', attr: value}
if exception:
self.assertRaises(exception, policy.enforce,
- context, action, target, None)
+ context, action, target)
else:
- result = policy.enforce(context, action, target, None)
+ result = policy.enforce(context, action, target)
self.assertEqual(result, True)
def _test_nonadmin_action_on_attr(self, action, attr, value,
def _test_enforce_adminonly_attribute(self, action):
admin_context = context.get_admin_context()
target = {'shared': True}
- result = policy.enforce(admin_context, action, target, None)
+ result = policy.enforce(admin_context, action, target)
self.assertEqual(result, True)
def test_enforce_adminonly_attribute_create(self):
action = "create_network"
target = {'shared': True, 'tenant_id': 'somebody_else'}
self.assertRaises(exceptions.PolicyNotAuthorized, policy.enforce,
- self.context, action, target, None)
+ self.context, action, target)
def test_enforce_adminonly_nonadminctx_no_ctx_is_admin_policy_403(self):
del self.rules[policy.ADMIN_CTX_POLICY]
action = "create_network"
target = {'shared': True, 'tenant_id': 'somebody_else'}
self.assertRaises(exceptions.PolicyNotAuthorized, policy.enforce,
- self.context, action, target, None)
+ self.context, action, target)
def test_enforce_regularuser_on_read(self):
action = "get_network"
target = {'shared': True, 'tenant_id': 'somebody_else'}
- result = policy.enforce(self.context, action, target, None)
+ result = policy.enforce(self.context, action, target)
self.assertTrue(result)
- def test_enforce_parentresource_owner(self):
+ def test_enforce_tenant_id_check(self):
+ # Trigger a policy with rule admin_or_owner
+ action = "create_network"
+ target = {'tenant_id': 'fake'}
+ result = policy.enforce(self.context, action, target)
+ self.assertTrue(result)
+
+ def test_enforce_tenant_id_check_parent_resource(self):
+
+ def fakegetnetwork(*args, **kwargs):
+ return {'tenant_id': 'fake'}
+
+ action = "create_port:mac"
+ with mock.patch.object(manager.QuantumManager.get_instance().plugin,
+ 'get_network', new=fakegetnetwork):
+ target = {'network_id': 'whatever'}
+ result = policy.enforce(self.context, action, target)
+ self.assertTrue(result)
+
+ def test_enforce_tenant_id_check_parent_resource_bw_compatibility(self):
def fakegetnetwork(*args, **kwargs):
return {'tenant_id': 'fake'}
+ del self.rules['admin_or_network_owner']
+ self.rules['admin_or_network_owner'] = common_policy.parse_rule(
+ "role:admin or tenant_id:%(network_tenant_id)s")
action = "create_port:mac"
- with mock.patch.object(self.plugin, 'get_network', new=fakegetnetwork):
+ with mock.patch.object(manager.QuantumManager.get_instance().plugin,
+ 'get_network', new=fakegetnetwork):
target = {'network_id': 'whatever'}
- result = policy.enforce(self.context, action, target, self.plugin)
+ result = policy.enforce(self.context, action, target)
self.assertTrue(result)
+ def test_tenant_id_check_no_target_field_raises(self):
+ # Try and add a bad rule
+ self.assertRaises(
+ exceptions.PolicyInitError,
+ common_policy.parse_rule,
+ 'tenant_id:(wrong_stuff)')
+
+ def _test_enforce_tenant_id_raises(self, bad_rule):
+ self.rules['admin_or_owner'] = common_policy.parse_rule(bad_rule)
+ # Trigger a policy with rule admin_or_owner
+ action = "create_network"
+ target = {'tenant_id': 'fake'}
+ self.assertRaises(exceptions.PolicyCheckError,
+ policy.enforce,
+ self.context, action, target)
+
+ def test_enforce_tenant_id_check_malformed_target_field_raises(self):
+ self._test_enforce_tenant_id_raises('tenant_id:%(malformed_field)s')
+
+ def test_enforce_tenant_id_check_invalid_parent_resource_raises(self):
+ self._test_enforce_tenant_id_raises('tenant_id:%(foobaz_tenant_id)s')
+
def test_get_roles_context_is_admin_rule_missing(self):
rules = dict((k, common_policy.parse_rule(v)) for k, v in {
"some_other_rule": "role:admin",
Code Review / openstack-build / neutron-build.git / commitdiff