Allow self-sharing RBAC rules to be deleted without usage check

This patch allows the update and deletion of network
'access_as_shared' rule whose target tenant-id is the network owner
without checking for network usage.

Conflicts:
    neutron/tests/tempest/api/admin/test_shared_network_extension.py

Change-Id: I34d627da869f68c02e1abc4a19246698766d66e4
Closes-bug: #1517702
(cherry picked from commit f96b5439de)
This commit is contained in:
Stephen Ma 2016-04-26 15:07:35 -07:00
parent 3134048102
commit ef65c9c92c
3 changed files with 154 additions and 0 deletions

View File

@ -174,6 +174,9 @@ class NeutronDbPluginV2(db_base_plugin_common.DbBasePluginCommon,
raise exc.InvalidInput(error_message=msg)
tenant_to_check = None
self_sharing = policy['target_tenant'] == net['tenant_id']
if self_sharing:
return
if event == events.BEFORE_UPDATE:
new_tenant = kwargs['policy_update']['target_tenant']
if policy['target_tenant'] != new_tenant:

View File

@ -279,6 +279,17 @@ class RBACSharedNetworksTest(base.BaseAdminNetworkTest):
# anchor is gone, delete should pass
self.admin_client.delete_rbac_policy(wild['id'])
@test.idempotent_id('34d627da-a732-68c0-2e1a-bc4a19246698')
def test_delete_self_share_rule(self):
net = self.create_network()
self_share = self.client.create_rbac_policy(
object_type='network', object_id=net['id'],
action='access_as_shared',
target_tenant=net['tenant_id'])['rbac_policy']
port = self.client.create_port(network_id=net['id'])['port']
self.client.delete_rbac_policy(self_share['id'])
self.client.delete_port(port['id'])
@test.idempotent_id('86c3529b-1231-40de-803c-beefbeefbeef')
def test_tenant_can_delete_port_on_own_network(self):
net = self.create_network() # owned by self.client
@ -399,6 +410,18 @@ class RBACSharedNetworksTest(base.BaseAdminNetworkTest):
self.client.update_rbac_policy(pol['rbac_policy']['id'],
target_tenant='*')
@test.idempotent_id('34d627da-869f-68c0-2e1a-bc4a19246698')
def test_update_self_share_rule(self):
net = self.create_network()
self_share = self.client.create_rbac_policy(
object_type='network', object_id=net['id'],
action='access_as_shared',
target_tenant=net['tenant_id'])['rbac_policy']
port = self.client.create_port(network_id=net['id'])['port']
self.client.update_rbac_policy(self_share['id'],
target_tenant=self.client2.tenant_id)
self.client.delete_port(port['id'])
@test.idempotent_id('86c3529b-1231-40de-803c-aeeeeeee7fff')
def test_filtering_works_with_rbac_records_present(self):
resp = self._make_admin_net_and_subnet_shared_to_tenant_id(

View File

@ -0,0 +1,128 @@
# Copyright (c) 2016 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 mock
from neutron.common import constants
from neutron import context
import six
from neutron.callbacks import events
from neutron.db.db_base_plugin_v2 import NeutronDbPluginV2 as db_plugin_v2
from neutron.extensions import rbac as ext_rbac
from neutron.tests.unit.db import test_db_base_plugin_v2 as test_plugin
class NetworkRbacTestcase(test_plugin.NeutronDbPluginV2TestCase):
def setUp(self):
self.context = context.get_admin_context()
super(NetworkRbacTestcase, self).setUp()
def _make_networkrbac(self, network, target):
policy = {'rbac_policy': {'tenant_id': network['network']['tenant_id'],
'object_id': network['network']['id'],
'object_type': 'network',
'action': 'access_as_shared',
'target_tenant': target}}
return policy
def _setup_networkrbac_and_port(self, network, target_tenant):
policy = self._make_networkrbac(network, target_tenant)
netrbac = self.plugin.create_rbac_policy(self.context, policy)
test_port = {'port': {'name': 'test-port',
'network_id': network['network']['id'],
'mac_address': constants.ATTR_NOT_SPECIFIED,
'fixed_ips': constants.ATTR_NOT_SPECIFIED,
'admin_state_up': True,
'device_id': 'device_id',
'device_owner': 'device_owner',
'tenant_id': target_tenant}}
port = self.plugin.create_port(self.context, test_port)
return netrbac, port
def test_update_networkrbac_valid(self):
orig_target = 'test-tenant-2'
new_target = 'test-tenant-3'
with self.network() as net:
policy = self._make_networkrbac(net, orig_target)
netrbac = self.plugin.create_rbac_policy(self.context, policy)
update_policy = {'rbac_policy': {'target_tenant': new_target}}
netrbac2 = self.plugin.update_rbac_policy(self.context,
netrbac['id'],
update_policy)
policy['rbac_policy']['target_tenant'] = new_target
for k, v in six.iteritems(policy['rbac_policy']):
self.assertEqual(netrbac2[k], v)
def test_delete_networkrbac_in_use_fail(self):
with self.network() as net:
netrbac, _ = self._setup_networkrbac_and_port(
network=net, target_tenant='test-tenant-2')
self.assertRaises(ext_rbac.RbacPolicyInUse,
self.plugin.delete_rbac_policy,
self.context, netrbac['id'])
def test_delete_networkrbac(self):
with self.network() as net:
netrbac, port = self._setup_networkrbac_and_port(
network=net, target_tenant='test-tenant-4')
self.assertRaises(ext_rbac.RbacPolicyInUse,
self.plugin.delete_rbac_policy,
self.context, netrbac['id'])
self.plugin.delete_port(self.context, port['id'])
self.plugin.delete_rbac_policy(self.context, netrbac['id'])
self.assertRaises(ext_rbac.RbacPolicyNotFound,
self.plugin.get_rbac_policy,
self.context, netrbac['id'])
def test_delete_networkrbac_self_share(self):
net_id = 'my-network'
net_owner = 'my-tenant-id'
net = {'network': {'id': net_id, 'tenant_id': net_owner}}
policy = self._make_networkrbac(net, net_owner)['rbac_policy']
kwargs = {}
with mock.patch.object(db_plugin_v2, '_get_network') as get_net,\
mock.patch.object(db_plugin_v2,
'ensure_no_tenant_ports_on_network') as ensure:
get_net.return_value = net['network']
self.plugin.validate_network_rbac_policy_change(
None, events.BEFORE_DELETE, None,
self.context, 'network', policy, **kwargs)
self.assertEqual(0, ensure.call_count)
def test_update_self_share_networkrbac(self):
net_id = 'my-network'
net_owner = 'my-tenant-id'
net = {'network': {'id': net_id, 'tenant_id': net_owner}}
policy = self._make_networkrbac(net, net_owner)['rbac_policy']
kwargs = {'policy_update': {'target_tenant': 'new-target-tenant'}}
with mock.patch.object(db_plugin_v2, '_get_network') as get_net,\
mock.patch.object(db_plugin_v2,
'ensure_no_tenant_ports_on_network') as ensure:
get_net.return_value = net['network']
self.plugin.validate_network_rbac_policy_change(
None, events.BEFORE_UPDATE, None,
self.context, 'network', policy, **kwargs)
self.assertEqual(0, ensure.call_count)