{
"admin_or_owner": [["role:admin"], ["tenant_id:%(tenant_id)s"]],
+ "admin_or_network_owner": [["role:admin"], ["tenant_id:%(network_tenant_id)s"]],
+ "admin_only": [["role:admin"]],
+ "regular_user": [],
"default": [["rule:admin_or_owner"]],
- "admin_api": [["role:admin"]],
- "extension:provider_network:view": [["rule:admin_api"]],
- "extension:provider_network:set": [["rule:admin_api"]],
+ "extension:provider_network:view": [["rule:admin_only"]],
+ "extension:provider_network:set": [["rule:admin_only"]],
- "create_subnet": [],
+ "networks:private:read": [["rule:admin_or_owner"]],
+ "networks:private:write": [["rule:admin_or_owner"]],
+ "networks:shared:read": [["rule:regular_user"]],
+ "networks:shared:write": [["rule:admin_only"]],
+
+ "create_subnet": [["rule:admin_or_network_owner"]],
"get_subnet": [["rule:admin_or_owner"]],
- "update_subnet": [["rule:admin_or_owner"]],
- "delete_subnet": [["rule:admin_or_owner"]],
+ "update_subnet": [["rule:admin_or_network_owner"]],
+ "delete_subnet": [["rule:admin_or_network_owner"]],
"create_network": [],
- "get_network": [["rule:admin_or_owner"]],
- "update_network": [["rule:admin_or_owner"]],
- "delete_network": [["rule:admin_or_owner"]],
+ "get_network": [],
+ "create_network:shared": [["rule:admin_only"]],
+ "update_network": [],
+ "update_network:shared": [["rule:admin_only"]],
+ "delete_network": [],
"create_port": [],
+ "create_port:mac_address": [["rule:admin_or_network_owner"]],
+ "create_port:host_routes": [["rule:admin_or_network_owner"]],
+ "create_port:fixed_ips": [["rule:admin_or_network_owner"]],
"get_port": [["rule:admin_or_owner"]],
"update_port": [["rule:admin_or_owner"]],
+ "update_port:host_routes": [["rule:admin_or_network_owner"]],
+ "update_port:fixed_ips": [["rule:admin_or_network_owner"]],
"delete_port": [["rule:admin_or_owner"]]
}
# limitations under the License.
ATTR_NOT_SPECIFIED = object()
-
-# Note: a default of ATTR_NOT_SPECIFIED indicates that an
-# attribute is not required, but will be generated by the plugin
-# if it is not specified. Particularly, a value of ATTR_NOT_SPECIFIED
-# is different from an attribute that has been specified with a value of
-# None. For example, if 'gateway_ip' is ommitted in a request to
-# create a subnet, the plugin will receive ATTR_NOT_SPECIFIED
-# and the default gateway_ip will be generated.
-# However, if gateway_ip is specified as None, this means that
-# the subnet does not have a gateway IP.
+# Defining a constant to avoid repeating string literal in several modules
+SHARED = 'shared'
import logging
import netaddr
# and the default gateway_ip will be generated.
# However, if gateway_ip is specified as None, this means that
# the subnet does not have a gateway IP.
-# Some of the following attributes are used by the policy engine.
-# They are explicitly marked with the required_by_policy flag to ensure
-# they are always returned by a plugin for policy processing, even if
-# they are not specified in the 'fields' query param
+# The following is a short reference for understanding attribute info:
+# default: default value of the attribute (if missing, the attribute
+# becomes mandatory.
+# allow_post: the attribute can be used on POST requests.
+# allow_put: the attribute can be used on PUT requests.
+# validate: specifies rules for validating data in the attribute.
+# convert_to: transformation to apply to the value before it is returned
+# is_visible: the attribute is returned in GET responses.
+# required_by_policy: the attribute is required by the policy engine and
+# should therefore be filled by the API layer even if not present in
+# request body.
+# enforce_policy: the attribute is actively part of the policy enforcing
+# mechanism, ie: there might be rules which refer to this attribute.
+
RESOURCE_ATTRIBUTE_MAP = {
'networks': {
'id': {'allow_post': False, 'allow_put': False,
'is_visible': True},
'tenant_id': {'allow_post': True, 'allow_put': False,
'required_by_policy': True,
- 'is_visible': True}
+ 'is_visible': True},
+ SHARED: {'allow_post': True,
+ 'allow_put': True,
+ 'default': False,
+ 'convert_to': convert_to_boolean,
+ 'validate': {'type:boolean': None},
+ 'is_visible': True,
+ 'required_by_policy': True,
+ 'enforce_policy': True},
},
'ports': {
'id': {'allow_post': False, 'allow_put': False,
'mac_address': {'allow_post': True, 'allow_put': False,
'default': ATTR_NOT_SPECIFIED,
'validate': {'type:mac_address': None},
+ 'enforce_policy': True,
'is_visible': True},
'fixed_ips': {'allow_post': True, 'allow_put': True,
'default': ATTR_NOT_SPECIFIED,
+ 'enforce_policy': True,
'is_visible': True},
'host_routes': {'allow_post': True, 'allow_put': True,
'default': ATTR_NOT_SPECIFIED,
+ 'enforce_policy': True,
'is_visible': False},
'device_id': {'allow_post': True, 'allow_put': True,
'default': '',
'is_visible': True},
}
}
+
+# Associates to each resource its own parent resource
+# Resources without parents, such as networks, are not in this list
+
+RESOURCE_HIERARCHY_MAP = {
+ 'ports': {'parent': 'networks', 'identified_by': 'network_id'},
+ 'subnets': {'parent': 'networks', 'identified_by': 'network_id'}
+}
exceptions.OverlappingAllocationPools: webob.exc.HTTPConflict,
exceptions.OutOfBoundsAllocationPool: webob.exc.HTTPBadRequest,
exceptions.InvalidAllocationPool: webob.exc.HTTPBadRequest,
+ exceptions.InvalidSharedSetting: webob.exc.HTTPConflict,
}
QUOTAS = quota.QUOTAS
self._resource = resource
self._attr_info = attr_info
self._policy_attrs = [name for (name, info) in self._attr_info.items()
- if 'required_by_policy' in info
- and info['required_by_policy']]
+ if info.get('required_by_policy')]
self._publisher_id = notifier_api.publisher_id('network')
def _is_visible(self, attr):
obj_list = obj_getter(request.context, **kwargs)
# Check authz
if do_authz:
+ # FIXME(salvatore-orlando): obj_getter might return references to
+ # other resources. Must check authZ on them too.
# Omit items from list that should not be visible
obj_list = [obj for obj in obj_list
if policy.check(request.context,
"get_%s" % self._resource,
- obj)]
+ obj,
+ plugin=self._plugin)]
return {self._collection: [self._view(obj,
fields_to_strip=fields_to_add)
for obj in obj_list]}
- def _item(self, request, id, do_authz=False):
+ def _item(self, request, id, do_authz=False, field_list=None):
"""Retrieves and formats a single element of the requested entity"""
- # NOTE(salvatore-orlando): The following ensures that fields which
- # are needed for authZ policy validation are not stripped away by the
- # plugin before returning.
- field_list, added_fields = self._do_field_list(fields(request))
kwargs = {'verbose': verbose(request),
'fields': field_list}
action = "get_%s" % self._resource
obj_getter = getattr(self._plugin, action)
obj = obj_getter(request.context, id, **kwargs)
-
# Check authz
+ # 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)
-
- return {self._resource: self._view(obj, fields_to_strip=added_fields)}
+ policy.enforce(request.context, action, obj, plugin=self._plugin)
+ return obj
def index(self, request):
"""Returns a list of the requested entity"""
def show(self, request, id):
"""Returns detailed information about the requested entity"""
try:
- return self._item(request, id, True)
+ # NOTE(salvatore-orlando): The following ensures that fields
+ # which are needed for authZ policy validation are not stripped
+ # away by the plugin before returning.
+ field_list, added_fields = self._do_field_list(fields(request))
+ return {self._resource:
+ self._view(self._item(request,
+ id,
+ do_authz=True,
+ field_list=field_list),
+ fields_to_strip=added_fields)}
except exceptions.PolicyNotAuthorized:
# To avoid giving away information, pretend that it
# doesn't exist
request,
item[self._resource],
)
- policy.enforce(
- request.context,
- action,
- item[self._resource],
- )
+ policy.enforce(request.context,
+ action,
+ item[self._resource],
+ plugin=self._plugin)
count = QUOTAS.count(request.context, self._resource,
self._plugin, self._collection,
item[self._resource]['tenant_id'])
request,
body[self._resource]
)
- policy.enforce(request.context, action, body[self._resource])
+ policy.enforce(request.context,
+ action,
+ body[self._resource],
+ plugin=self._plugin)
count = QUOTAS.count(request.context, self._resource,
self._plugin, self._collection,
body[self._resource]['tenant_id'])
kwargs = {self._resource: count + 1}
QUOTAS.limit_check(request.context, **kwargs)
except exceptions.PolicyNotAuthorized:
+ LOG.exception("Create operation not authorized")
raise webob.exc.HTTPForbidden()
obj_creator = getattr(self._plugin, action)
action = "delete_%s" % self._resource
# Check authz
- obj = self._item(request, id)[self._resource]
+ obj = self._item(request, id)
try:
- policy.enforce(request.context, action, obj)
+ policy.enforce(request.context,
+ action,
+ obj,
+ plugin=self._plugin)
except exceptions.PolicyNotAuthorized:
# To avoid giving away information, pretend that it
# doesn't exist
payload)
body = self._prepare_request_body(request.context, body, False)
action = "update_%s" % self._resource
+ # Load object to check authz
+ # but pass only attributes in the original body and required
+ # by the policy engine to the policy 'brain'
+ field_list = [name for (name, value) in self._attr_info.iteritems()
+ if ('required_by_policy' in value and
+ value['required_by_policy'] or
+ not 'default' in value)]
+ orig_obj = self._item(request, id, field_list=field_list)
+ orig_obj.update(body)
- # Check authz
- orig_obj = self._item(request, id)[self._resource]
try:
- policy.enforce(request.context, action, orig_obj)
+ policy.enforce(request.context,
+ action,
+ orig_obj,
+ plugin=self._plugin)
except exceptions.PolicyNotAuthorized:
# To avoid giving away information, pretend that it
# doesn't exist
if (('tenant_id' in res_dict and
res_dict['tenant_id'] != context.tenant_id and
not context.is_admin)):
- msg = _("Specifying 'tenant_id' other than authenticated"
+ msg = _("Specifying 'tenant_id' other than authenticated "
"tenant in request requires admin privileges")
raise webob.exc.HTTPBadRequest(msg)
raise webob.exc.HTTPBadRequest(_("Resource body required"))
body = body or {self._resource: {}}
-
if self._collection in body and allow_bulk:
bulk_body = [self._prepare_request_body(context,
{self._resource: b},
msg = _("Invalid input for %(attr)s. "
"Reason: %(reason)s.") % msg_dict
raise webob.exc.HTTPUnprocessableEntity(msg)
-
return body
def _validate_network_tenant_ownership(self, request, resource_item):
+ # TODO(salvatore-orlando): consider whether this check can be folded
+ # in the policy engine
if self._resource not in ('port', 'subnet'):
return
-
- network_owner = self._plugin.get_network(
+ network = self._plugin.get_network(
request.context,
- resource_item['network_id'],
- )['tenant_id']
+ resource_item['network_id'])
+ # do not perform the check on shared networks
+ if network.get('shared'):
+ return
+
+ network_owner = network['tenant_id']
if network_owner != resource_item['tenant_id']:
msg = _("Tenant %(tenant_id)s not allowed to "
class InvalidQuotaValue(QuantumException):
message = _("Change would make usage less than 0 for the following "
"resources: %(unders)s")
+
+
+class InvalidSharedSetting(QuantumException):
+ message = _("Unable to reconfigure sharing settings for network"
+ "%(network). Multiple tenants are using it")
query = context.session.query(model)
# NOTE(jkoelker) non-admin queries are scoped to their tenant_id
+ # NOTE(salvatore-orlando): unless the model allows for shared objects
if not context.is_admin and hasattr(model, 'tenant_id'):
- query = query.filter(model.tenant_id == context.tenant_id)
-
+ if hasattr(model, 'shared'):
+ query = query.filter((model.tenant_id == context.tenant_id) |
+ (model.shared))
+ else:
+ query = query.filter(model.tenant_id == context.tenant_id)
return query
def _get_by_id(self, context, model, id, joins=(), verbose=None):
subnet['cidr'])
return pools
+ def _validate_shared_update(self, context, id, original, updated):
+ # The only case that needs to be validated is when 'shared'
+ # goes from True to False
+ if updated['shared'] == original.shared or updated['shared']:
+ return
+ ports = self._model_query(
+ context, models_v2.Port).filter(
+ models_v2.Port.network_id == id).all()
+ subnets = self._model_query(
+ context, models_v2.Subnet).filter(
+ models_v2.Subnet.network_id == id).all()
+ tenant_ids = set([port['tenant_id'] for port in ports] +
+ [subnet['tenant_id'] for subnet in subnets])
+ # raise if multiple tenants found or if the only tenant found
+ # is not the owner of the network
+ if (len(tenant_ids) > 1 or len(tenant_ids) == 1 and
+ tenant_ids.pop() != original.tenant_id):
+ raise q_exc.InvalidSharedSetting(network=original.name)
+
def _make_network_dict(self, network, fields=None):
res = {'id': network['id'],
'name': network['name'],
'tenant_id': network['tenant_id'],
'admin_state_up': network['admin_state_up'],
'status': network['status'],
+ 'shared': network['shared'],
'subnets': [subnet['id']
for subnet in network['subnets']]}
id=n.get('id') or utils.str_uuid(),
name=n['name'],
admin_state_up=n['admin_state_up'],
+ shared=n['shared'],
status="ACTIVE")
context.session.add(network)
return self._make_network_dict(network)
n = network['network']
with context.session.begin():
network = self._get_network(context, id)
+ # validate 'shared' parameter
+ if 'shared' in n:
+ self._validate_shared_update(context, id, network, n)
network.update(n)
return self._make_network_dict(network)
subnets = orm.relationship(Subnet, backref='networks')
status = sa.Column(sa.String(16))
admin_state_up = sa.Column(sa.Boolean)
+ shared = sa.Column(sa.Boolean)
"""
Policy engine for quantum. Largely copied from nova.
"""
-
-import os.path
-
+from quantum.api.v2 import attributes
from quantum.common import exceptions
from quantum.openstack.common import cfg
import quantum.common.utils as utils
policy.set_brain(policy.Brain.load_json(data, default_rule))
-def check(context, action, target):
+def _get_resource_and_action(action):
+ data = action.split(':', 1)[0].split('_', 1)
+ return ("%ss" % data[-1], data[0] != 'get')
+
+
+def _is_attribute_explicitly_set(attribute_name, resource, target):
+ """Verify that an attribute is present and has a non-default value"""
+ if ('default' in resource[attribute_name] and
+ target.get(attribute_name, attributes.ATTR_NOT_SPECIFIED) !=
+ attributes.ATTR_NOT_SPECIFIED):
+ if (target[attribute_name] != resource[attribute_name]['default']):
+ return True
+ return False
+
+
+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, _w = _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 _create_access_rule_match(resource, is_write, shared):
+ if shared == resource[attributes.SHARED]:
+ return ('rule:%s:%s:%s' % (resource,
+ shared and 'shared' or 'private',
+ is_write and 'write' or 'read'), )
+
+
+def _build_perm_match(action, target):
+ """Create the permission rule match.
+
+ Given the current access right on a network (shared/private), and
+ the type of the current operation (read/write), builds a match
+ rule of the type <resource>:<sharing_mode>:<operation_type>
+ """
+ resource, is_write = _get_resource_and_action(action)
+ res_map = attributes.RESOURCE_ATTRIBUTE_MAP
+ if (resource in res_map and
+ attributes.SHARED in res_map[resource] and
+ attributes.SHARED in target):
+ return ('rule:%s:%s:%s' % (resource,
+ target[attributes.SHARED]
+ and 'shared' or 'private',
+ is_write and 'write' or 'read'), )
+
+
+def _build_match_list(action, target):
+ """Create the list of rules to match for a given action.
+
+ The list of policy rules to be matched is built in the following way:
+ 1) add entries for matching permission on objects
+ 2) add an entry for the specific action (e.g.: create_network)
+ 3) add an entry for attributes of a resource for which the action
+ is being executed (e.g.: create_network:shared)
+
+ """
+
+ match_list = ('rule:%s' % action,)
+ resource, is_write = _get_resource_and_action(action)
+ # assigning to variable with short name for improving readability
+ res_map = attributes.RESOURCE_ATTRIBUTE_MAP
+ if resource in res_map:
+ for attribute_name in res_map[resource]:
+ if _is_attribute_explicitly_set(attribute_name,
+ res_map[resource],
+ target):
+ attribute = res_map[resource][attribute_name]
+ if 'enforce_policy' in attribute and is_write:
+ match_list += ('rule:%s:%s' % (action,
+ attribute_name),)
+ # add permission-based rule (for shared resources)
+ perm_match = _build_perm_match(action, target)
+ if perm_match:
+ match_list += perm_match
+ # the policy engine must AND between all the rules
+ return [match_list]
+
+
+def check(context, action, target, plugin=None):
"""Verifies that the action is valid on the target in this context.
:param context: quantum context
:param action: string representing the action to be checked
this should be colon separated for clarity.
- :param object: dictionary representing the object of the action
+ :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
:return: Returns True if access is permitted else False.
"""
init()
- match_list = ('rule:%s' % action,)
+ real_target = _build_target(action, target, plugin, context)
+ match_list = _build_match_list(action, real_target)
credentials = context.to_dict()
- return policy.enforce(match_list, target, credentials)
+ return policy.enforce(match_list, real_target, credentials)
-def enforce(context, action, target):
+def enforce(context, action, target, plugin=None):
"""Verifies that the action is valid on the target in this context.
:param context: quantum context
:param action: string representing the action to be checked
this should be colon separated for clarity.
- :param object: dictionary representing the object of the action
+ :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
:raises quantum.exceptions.PolicyNotAllowed: if verification fails.
"""
init()
- match_list = ('rule:%s' % action,)
+ real_target = _build_target(action, target, plugin, context)
+ match_list = _build_match_list(action, real_target)
credentials = context.to_dict()
-
- policy.enforce(match_list, target, credentials,
+ policy.enforce(match_list, real_target, credentials,
exceptions.PolicyNotAuthorized, action=action)
fields=mock.ANY,
verbose=True)
+ def _do_field_list(self, resource, base_fields):
+ attr_info = attributes.RESOURCE_ATTRIBUTE_MAP[resource]
+ policy_attrs = [name for (name, info) in attr_info.items()
+ if info.get('required_by_policy')]
+ fields = base_fields
+ fields.extend(policy_attrs)
+ return fields
+
def test_fields(self):
instance = self.plugin.return_value
instance.get_networks.return_value = []
self.api.get(_get_path('networks'), {'fields': 'foo'})
+ fields = self._do_field_list('networks', ['foo'])
instance.get_networks.assert_called_once_with(mock.ANY,
filters=mock.ANY,
- fields=['foo',
- 'tenant_id'],
+ fields=fields,
verbose=mock.ANY)
def test_fields_multiple(self):
instance = self.plugin.return_value
instance.get_networks.return_value = []
+ fields = self._do_field_list('networks', ['foo', 'bar'])
self.api.get(_get_path('networks'), {'fields': ['foo', 'bar']})
instance.get_networks.assert_called_once_with(mock.ANY,
filters=mock.ANY,
- fields=['foo', 'bar',
- 'tenant_id'],
+ fields=fields,
verbose=mock.ANY)
def test_fields_multiple_with_empty(self):
instance = self.plugin.return_value
instance.get_networks.return_value = []
+ fields = self._do_field_list('networks', ['foo'])
self.api.get(_get_path('networks'), {'fields': ['foo', '']})
instance.get_networks.assert_called_once_with(mock.ANY,
filters=mock.ANY,
- fields=['foo',
- 'tenant_id'],
+ fields=fields,
verbose=mock.ANY)
def test_fields_empty(self):
self.api.get(_get_path('networks'), {'foo': 'bar', 'fields': 'foo'})
filters = {'foo': ['bar']}
+ fields = self._do_field_list('networks', ['foo'])
instance.get_networks.assert_called_once_with(mock.ANY,
filters=filters,
- fields=['foo',
- 'tenant_id'],
+ fields=fields,
verbose=mock.ANY)
def test_filters_with_verbose(self):
'fields': 'foo',
'verbose': 'true'})
filters = {'foo': ['bar']}
+ fields = self._do_field_list('networks', ['foo'])
instance.get_networks.assert_called_once_with(mock.ANY,
filters=filters,
- fields=['foo',
- 'tenant_id'],
+ fields=fields,
verbose=True)
'admin_state_up': True,
'status': "ACTIVE",
'tenant_id': real_tenant_id,
+ 'shared': False,
'subnets': []}
return_value = [input_dict]
instance = self.plugin.return_value
# expect full list returned
self.assertEqual(len(res.json['networks']), 1)
output_dict = res.json['networks'][0]
+ input_dict['shared'] = False
self.assertEqual(len(input_dict), len(output_dict))
for k, v in input_dict.iteritems():
self.assertEqual(v, output_dict[k])
def test_create_use_defaults(self):
net_id = _uuid()
initial_input = {'network': {'name': 'net1', 'tenant_id': _uuid()}}
- full_input = {'network': {'admin_state_up': True, 'subnets': []}}
+ full_input = {'network': {'admin_state_up': True,
+ 'shared': False,
+ 'subnets': []}}
full_input['network'].update(initial_input['network'])
return_value = {'id': net_id, 'status': "ACTIVE"}
# tenant_id should be fetched from env
initial_input = {'network': {'name': 'net1'}}
full_input = {'network': {'admin_state_up': True, 'subnets': [],
- 'tenant_id': tenant_id}}
+ 'shared': False, 'tenant_id': tenant_id}}
full_input['network'].update(initial_input['network'])
return_value = {'id': net_id, 'status': "ACTIVE"}
if req_tenant_id:
env = {'quantum.context': context.Context('', req_tenant_id)}
instance = self.plugin.return_value
- instance.get_network.return_value = {'tenant_id': real_tenant_id}
+ instance.get_network.return_value = {'tenant_id': real_tenant_id,
+ 'shared': False}
instance.delete_network.return_value = None
res = self.api.delete(_get_path('networks', id=str(uuid.uuid4())),
def _test_get(self, req_tenant_id, real_tenant_id, expected_code,
expect_errors=False):
env = {}
+ shared = False
if req_tenant_id:
env = {'quantum.context': context.Context('', req_tenant_id)}
- data = {'tenant_id': real_tenant_id}
+ if req_tenant_id.endswith('another'):
+ shared = True
+ env['quantum.context'].roles = ['tenant_admin']
+
+ data = {'tenant_id': real_tenant_id, 'shared': shared}
instance = self.plugin.return_value
instance.get_network.return_value = data
self._test_get(tenant_id + "bad", tenant_id,
exc.HTTPNotFound.code, expect_errors=True)
+ def test_get_keystone_shared_network(self):
+ tenant_id = _uuid()
+ self._test_get(tenant_id + "another", tenant_id, 200)
+
def _test_update(self, req_tenant_id, real_tenant_id, expected_code,
expect_errors=False):
env = {}
return_value.update(data['network'].copy())
instance = self.plugin.return_value
- instance.get_network.return_value = {'tenant_id': real_tenant_id}
+ instance.get_network.return_value = {'tenant_id': real_tenant_id,
+ 'shared': False}
instance.update_network.return_value = return_value
res = self.api.put_json(_get_path('networks',
def test_extended_create(self):
net_id = _uuid()
- data = {'network': {'name': 'net1', 'admin_state_up': True,
- 'tenant_id': _uuid(), 'subnets': [],
- 'v2attrs:something_else': "abc"}}
+ initial_input = {'network': {'name': 'net1', 'tenant_id': _uuid(),
+ 'v2attrs:something_else': "abc"}}
+ data = {'network': {'admin_state_up': True, 'subnets': [],
+ 'shared': False}}
+ data['network'].update(initial_input['network'])
+
return_value = {'subnets': [], 'status': "ACTIVE",
'id': net_id,
'v2attrs:something': "123"}
instance = self.plugin.return_value
instance.create_network.return_value = return_value
- res = self.api.post_json(_get_path('networks'), data)
+ res = self.api.post_json(_get_path('networks'), initial_input)
instance.create_network.assert_called_with(mock.ANY,
network=data)
data = {'network': {'name': name,
'admin_state_up': admin_status_up,
'tenant_id': self._tenant_id}}
- for arg in ('admin_state_up', 'tenant_id', 'public'):
+ for arg in ('admin_state_up', 'tenant_id', 'shared'):
# Arg must be present and not empty
if arg in kwargs and kwargs[arg]:
data['network'][arg] = kwargs[arg]
network_req = self.new_create_request('networks', data, fmt)
- if ('set_context' in kwargs and
- kwargs['set_context'] is True and
- 'tenant_id' in kwargs):
+ if (kwargs.get('set_context') and 'tenant_id' in kwargs):
# create a specific auth context for this request
network_req.environ['quantum.context'] = context.Context(
'', kwargs['tenant_id'])
return network_req.get_response(self.api)
- def _create_subnet(self, fmt, tenant_id, net_id, gateway_ip, cidr,
- allocation_pools=None, ip_version=4, enable_dhcp=True):
- data = {'subnet': {'tenant_id': tenant_id,
- 'network_id': net_id,
+ def _create_subnet(self, fmt, net_id, cidr,
+ expected_res_status=None, **kwargs):
+ data = {'subnet': {'network_id': net_id,
'cidr': cidr,
- 'ip_version': ip_version,
- 'tenant_id': self._tenant_id,
- 'enable_dhcp': enable_dhcp}}
- if gateway_ip:
- data['subnet']['gateway_ip'] = gateway_ip
- if allocation_pools:
- data['subnet']['allocation_pools'] = allocation_pools
+ 'ip_version': 4,
+ 'tenant_id': self._tenant_id}}
+ for arg in ('gateway_ip', 'allocation_pools',
+ 'ip_version', 'tenant_id',
+ 'enable_dhcp'):
+ # Arg must be present and not null (but can be false)
+ if arg in kwargs and kwargs[arg] is not None:
+ data['subnet'][arg] = kwargs[arg]
subnet_req = self.new_create_request('subnets', data, fmt)
- return subnet_req.get_response(self.api)
+ if (kwargs.get('set_context') and 'tenant_id' in kwargs):
+ # create a specific auth context for this request
+ subnet_req.environ['quantum.context'] = context.Context(
+ '', kwargs['tenant_id'])
+
+ subnet_res = subnet_req.get_response(self.api)
+ if expected_res_status:
+ self.assertEqual(subnet_res.status_int, expected_res_status)
+ return subnet_res
- def _create_port(self, fmt, net_id, custom_req_body=None,
- expected_res_status=None, **kwargs):
+ def _create_port(self, fmt, net_id, expected_res_status=None, **kwargs):
content_type = 'application/' + fmt
data = {'port': {'network_id': net_id,
'tenant_id': self._tenant_id}}
- for arg in ('admin_state_up', 'device_id', 'mac_address',
- 'name', 'fixed_ips'):
+ for arg in ('admin_state_up', 'device_id',
+ 'mac_address', 'fixed_ips',
+ 'name', 'tenant_id'):
# Arg must be present and not empty
if arg in kwargs and kwargs[arg]:
data['port'][arg] = kwargs[arg]
-
port_req = self.new_create_request('ports', data, fmt)
- return port_req.get_response(self.api)
+ if (kwargs.get('set_context') and 'tenant_id' in kwargs):
+ # create a specific auth context for this request
+ port_req.environ['quantum.context'] = context.Context(
+ '', kwargs['tenant_id'])
+
+ port_res = port_req.get_response(self.api)
+ if expected_res_status:
+ self.assertEqual(port_res.status_int, expected_res_status)
+ return port_res
+
+ def _list_ports(self, fmt, expected_res_status=None,
+ net_id=None, **kwargs):
+ query_params = None
+ if net_id:
+ query_params = "network_id=%s" % net_id
+ port_req = self.new_list_request('ports', fmt, query_params)
+ if ('set_context' in kwargs and
+ kwargs['set_context'] is True and
+ 'tenant_id' in kwargs):
+ # create a specific auth context for this request
+ port_req.environ['quantum.context'] = context.Context(
+ '', kwargs['tenant_id'])
+
+ port_res = port_req.get_response(self.api)
+ if expected_res_status:
+ self.assertEqual(port_res.status_int, expected_res_status)
+ return port_res
def _make_subnet(self, fmt, network, gateway, cidr,
allocation_pools=None, ip_version=4, enable_dhcp=True):
res = self._create_subnet(fmt,
- network['network']['tenant_id'],
- network['network']['id'],
- gateway,
- cidr,
+ net_id=network['network']['id'],
+ cidr=cidr,
+ gateway_ip=gateway,
+ tenant_id=network['network']['tenant_id'],
allocation_pools=allocation_pools,
ip_version=ip_version,
enable_dhcp=enable_dhcp)
req.get_response(self.api)
@contextlib.contextmanager
- def network(self, name='net1', admin_status_up=True, fmt='json'):
- res = self._create_network(fmt, name, admin_status_up)
+ def network(self, name='net1',
+ admin_status_up=True,
+ fmt='json',
+ **kwargs):
+ res = self._create_network(fmt,
+ name,
+ admin_status_up,
+ **kwargs)
network = self.deserialize(fmt, res)
+ # TODO(salvatore-orlando): do exception handling in this test module
+ # in a uniform way (we do it differently for ports, subnets, and nets
+ # Things can go wrong - raise HTTP exc with res code only
+ # so it can be caught by unit tests
+ if res.status_int >= 400:
+ raise webob.exc.HTTPClientError(code=res.status_int)
yield network
self._delete('networks', network['network']['id'])
res = port_req.get_response(self.api)
self.assertEquals(res.status_int, 403)
+ def test_create_port_public_network(self):
+ keys = [('admin_state_up', True), ('status', 'ACTIVE')]
+ with self.network(shared=True) as network:
+ port_res = self._create_port('json',
+ network['network']['id'],
+ 201,
+ tenant_id='another_tenant',
+ set_context=True)
+ port = self.deserialize('json', port_res)
+ for k, v in keys:
+ self.assertEquals(port['port'][k], v)
+ self.assertTrue('mac_address' in port['port'])
+
def test_list_ports(self):
with contextlib.nested(self.port(), self.port()) as (port1, port2):
req = self.new_list_request('ports', 'json')
self.assertTrue(port1['port']['id'] in ids)
self.assertTrue(port2['port']['id'] in ids)
+ def test_list_ports_public_network(self):
+ with self.network(shared=True) as network:
+ portres_1 = self._create_port('json',
+ network['network']['id'],
+ 201,
+ tenant_id='tenant_1',
+ set_context=True)
+ portres_2 = self._create_port('json',
+ network['network']['id'],
+ 201,
+ tenant_id='tenant_2',
+ set_context=True)
+ port1 = self.deserialize('json', portres_1)
+ port2 = self.deserialize('json', portres_2)
+
+ def _list_and_test_ports(expected_len, ports, tenant_id=None):
+ set_context = tenant_id is not None
+ port_res = self._list_ports('json',
+ 200,
+ network['network']['id'],
+ tenant_id=tenant_id,
+ set_context=set_context)
+ port_list = self.deserialize('json', port_res)
+ self.assertEqual(len(port_list['ports']), expected_len)
+ ids = [p['id'] for p in port_list['ports']]
+ for port in ports:
+ self.assertIn(port['port']['id'], ids)
+
+ # Admin request - must return both ports
+ _list_and_test_ports(2, [port1, port2])
+ # Tenant_1 request - must return single port
+ _list_and_test_ports(1, [port1], tenant_id='tenant_1')
+ # Tenant_2 request - must return single port
+ _list_and_test_ports(1, [port2], tenant_id='tenant_2')
+
def test_show_port(self):
with self.port() as port:
req = self.new_show_request('ports', port['port']['id'], 'json')
res = req.get_response(self.api)
self.assertEquals(res.status_int, 404)
+ def test_delete_port_public_network(self):
+ with self.network(shared=True) as network:
+ port_res = self._create_port('json',
+ network['network']['id'],
+ 201,
+ tenant_id='another_tenant',
+ set_context=True)
+
+ port = self.deserialize('json', port_res)
+ port_id = port['port']['id']
+ # delete the port
+ self._delete('ports', port['port']['id'])
+ # Todo: verify!!!
+
def test_update_port(self):
with self.port() as port:
data = {'port': {'admin_state_up': False}}
# Get a IPv4 and IPv6 address
tenant_id = subnet['subnet']['tenant_id']
net_id = subnet['subnet']['network_id']
- res = self._create_subnet(fmt, tenant_id, net_id=net_id,
+ res = self._create_subnet(fmt,
+ tenant_id=tenant_id,
+ net_id=net_id,
cidr='2607:f0d0:1002:51::0/124',
- ip_version=6, gateway_ip=None)
+ ip_version=6,
+ gateway_ip=None)
subnet2 = self.deserialize(fmt, res)
kwargs = {"fixed_ips":
[{'subnet_id': subnet['subnet']['id']},
def test_create_network(self):
name = 'net1'
keys = [('subnets', []), ('name', name), ('admin_state_up', True),
- ('status', 'ACTIVE')]
+ ('status', 'ACTIVE'), ('shared', False)]
with self.network(name=name) as net:
for k, v in keys:
self.assertEquals(net['network'][k], v)
+ def test_create_public_network(self):
+ name = 'public_net'
+ keys = [('subnets', []), ('name', name), ('admin_state_up', True),
+ ('status', 'ACTIVE'), ('shared', True)]
+ with self.network(name=name, shared=True) as net:
+ for k, v in keys:
+ self.assertEquals(net['network'][k], v)
+
+ def test_create_public_network_no_admin_tenant(self):
+ name = 'public_net'
+ keys = [('subnets', []), ('name', name), ('admin_state_up', True),
+ ('status', 'ACTIVE'), ('shared', True)]
+ with self.assertRaises(webob.exc.HTTPClientError) as ctx_manager:
+ with self.network(name=name,
+ shared=True,
+ tenant_id="another_tenant",
+ set_context=True):
+ pass
+ self.assertEquals(ctx_manager.exception.code, 403)
+
+ def test_update_network(self):
+ with self.network() as network:
+ data = {'network': {'name': 'a_brand_new_name'}}
+ req = self.new_update_request('networks',
+ data,
+ network['network']['id'])
+ res = self.deserialize('json', req.get_response(self.api))
+ self.assertEqual(res['network']['name'],
+ data['network']['name'])
+
+ def test_update_shared_network_noadmin_returns_403(self):
+ with self.network(shared=True) as network:
+ data = {'network': {'name': 'a_brand_new_name'}}
+ req = self.new_update_request('networks',
+ data,
+ network['network']['id'])
+ req.environ['quantum.context'] = context.Context('', 'somebody')
+ res = req.get_response(self.api)
+ # The API layer always returns 404 on updates in place of 403
+ self.assertEqual(res.status_int, 404)
+
+ def test_update_network_set_shared(self):
+ with self.network(shared=False) as network:
+ data = {'network': {'shared': True}}
+ req = self.new_update_request('networks',
+ data,
+ network['network']['id'])
+ res = self.deserialize('json', req.get_response(self.api))
+ self.assertTrue(res['network']['shared'])
+
+ def test_update_network_set_not_shared_single_tenant(self):
+ with self.network(shared=True) as network:
+ self._create_port('json',
+ network['network']['id'],
+ 201,
+ tenant_id=network['network']['tenant_id'],
+ set_context=True)
+ data = {'network': {'shared': False}}
+ req = self.new_update_request('networks',
+ data,
+ network['network']['id'])
+ res = self.deserialize('json', req.get_response(self.api))
+ self.assertFalse(res['network']['shared'])
+
+ def test_update_network_set_not_shared_other_tenant_returns_409(self):
+ with self.network(shared=True) as network:
+ self._create_port('json',
+ network['network']['id'],
+ 201,
+ tenant_id='somebody_else',
+ set_context=True)
+ data = {'network': {'shared': False}}
+ req = self.new_update_request('networks',
+ data,
+ network['network']['id'])
+ self.assertEqual(req.get_response(self.api).status_int, 409)
+
+ def test_update_network_set_not_shared_multi_tenants_returns_409(self):
+ with self.network(shared=True) as network:
+ self._create_port('json',
+ network['network']['id'],
+ 201,
+ tenant_id='somebody_else',
+ set_context=True)
+ self._create_port('json',
+ network['network']['id'],
+ 201,
+ tenant_id=network['network']['tenant_id'],
+ set_context=True)
+ data = {'network': {'shared': False}}
+ req = self.new_update_request('networks',
+ data,
+ network['network']['id'])
+ self.assertEqual(req.get_response(self.api).status_int, 409)
+
+ def test_update_network_set_not_shared_multi_tenants2_returns_409(self):
+ with self.network(shared=True) as network:
+ self._create_port('json',
+ network['network']['id'],
+ 201,
+ tenant_id='somebody_else',
+ set_context=True)
+ self._create_subnet('json',
+ network['network']['id'],
+ '10.0.0.0/24',
+ 201,
+ tenant_id=network['network']['tenant_id'],
+ set_context=True)
+ data = {'network': {'shared': False}}
+ req = self.new_update_request('networks',
+ data,
+ network['network']['id'])
+ self.assertEqual(req.get_response(self.api).status_int, 409)
+
def test_list_networks(self):
with self.network(name='net1') as net1:
with self.network(name='net2') as net2:
from quantum.common import exceptions
from quantum.common import utils
from quantum import context
+from quantum.openstack.common import importutils
from quantum.openstack.common import policy as common_policy
from quantum import policy
self._set_brain("default_noexist")
self.assertRaises(exceptions.PolicyNotAuthorized, policy.enforce,
self.context, "example:noexist", {})
+
+
+class QuantumPolicyTestCase(unittest.TestCase):
+
+ def setUp(self):
+ super(QuantumPolicyTestCase, self).setUp()
+ policy.reset()
+ policy.init()
+ self.rules = {
+ "admin_or_network_owner": [["role:admin"],
+ ["tenant_id:%(network_tenant_id)s"]],
+ "admin_only": [["role:admin"]],
+ "regular_user": [["role:user"]],
+ "default": [],
+
+ "networks:private:read": [["rule:admin_only"]],
+ "networks:private:write": [["rule:admin_only"]],
+ "networks:shared:read": [["rule:regular_user"]],
+ "networks:shared:write": [["rule:admin_only"]],
+
+ "create_network": [],
+ "create_network:shared": [["rule:admin_only"]],
+ "update_network": [],
+ "update_network:shared": [["rule:admin_only"]],
+
+ "get_network": [],
+ "create_port:mac": [["rule:admin_or_network_owner"]],
+ }
+
+ def fakepolicyinit():
+ common_policy.set_brain(common_policy.Brain(self.rules))
+
+ self.patcher = mock.patch.object(quantum.policy,
+ 'init',
+ new=fakepolicyinit)
+ self.patcher.start()
+ self.context = context.Context('fake', 'fake', roles=['user'])
+ plugin_klass = importutils.import_class(
+ "quantum.db.db_base_plugin_v2.QuantumDbPluginV2")
+ self.plugin = plugin_klass()
+
+ def tearDown(self):
+ self.patcher.stop()
+
+ def test_nonadmin_write_on_private_returns_403(self):
+ action = "update_network"
+ user_context = context.Context('', "user", roles=['user'])
+ # 384 is the int value of the bitmask for rw------
+ target = {'tenant_id': 'the_owner', 'shared': False}
+ self.assertRaises(exceptions.PolicyNotAuthorized, policy.enforce,
+ user_context, action, target, None)
+
+ def test_nonadmin_read_on_private_returns_403(self):
+ action = "get_network"
+ user_context = context.Context('', "user", roles=['user'])
+ # 384 is the int value of the bitmask for rw------
+ target = {'tenant_id': 'the_owner', 'shared': False}
+ self.assertRaises(exceptions.PolicyNotAuthorized, policy.enforce,
+ user_context, action, target, None)
+
+ def test_nonadmin_write_on_shared_returns_403(self):
+ action = "update_network"
+ user_context = context.Context('', "user", roles=['user'])
+ # 384 is the int value of the bitmask for rw-r--r--
+ target = {'tenant_id': 'the_owner', 'shared': True}
+ self.assertRaises(exceptions.PolicyNotAuthorized, policy.enforce,
+ user_context, action, target, None)
+
+ def test_nonadmin_read_on_shared_returns_200(self):
+ action = "get_network"
+ user_context = context.Context('', "user", roles=['user'])
+ # 420 is the int value of the bitmask for rw-r--r--
+ target = {'tenant_id': 'the_owner', 'shared': True}
+ result = policy.enforce(user_context, action, target, None)
+ self.assertEqual(result, None)
+
+ def _test_enforce_adminonly_attribute(self, action):
+ admin_context = context.get_admin_context()
+ target = {'shared': True}
+ result = policy.enforce(admin_context, action, target, None)
+ self.assertEqual(result, None)
+
+ def test_enforce_adminonly_attribute_create(self):
+ self._test_enforce_adminonly_attribute('create_network')
+
+ def test_enforce_adminonly_attribute_update(self):
+ self._test_enforce_adminonly_attribute('update_network')
+
+ def test_enforce_adminoly_attribute_nonadminctx_returns_403(self):
+ action = "create_network"
+ target = {'shared': True}
+ self.assertRaises(exceptions.PolicyNotAuthorized, policy.enforce,
+ self.context, action, target, None)
+
+ 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)
+ self.assertIsNone(result)
+
+ def test_enforce_parentresource_owner(self):
+
+ def fakegetnetwork(*args, **kwargs):
+ return {'tenant_id': 'fake'}
+
+ action = "create_port:mac"
+ with mock.patch.object(self.plugin, 'get_network', new=fakegetnetwork):
+ target = {'network_id': 'whatever'}
+ result = policy.enforce(self.context, action, target, self.plugin)
+ self.assertIsNone(result)
from quantum.wsgi import Serializer
-def create_request(path, body, content_type, method='GET', query_string=None):
+def create_request(path, body, content_type, method='GET',
+ query_string=None, context=None):
if query_string:
url = "%s?%s" % (path, query_string)
else:
req.headers = {}
req.headers['Accept'] = content_type
req.body = body
+ if context:
+ req.environ['quantum.context'] = context
return req
Code Review / openstack-build / neutron-build.git / commitdiff