d9de2233d5
The test was originally implemented in this form as a tempest api test
case, but there is a problem with it being part of tempest suite,
because the suite should be safe to execute on a running cloud and
shouldn't affect other tenants. But sharing a network with everyone (*)
briefly impacts other tenants because they see a new shared network
popping up for a second in their net-list. It is also an issue because
if we are unlucky enough, some other tenant may start an instance
without specifying a particular port or network to use, in which case
there is a chance that nova allocates a port in this shared network and
attach it to the instance. When the rbac test case then tries to delete
the policy and the network, it fails because it's still in use.
The solution is to stop testing this scenario in tempest suite, instead
moving it into unit test suite.
A follow-up patch in neutron-tempest-plugin will clean up the bad test
case.
Change-Id: I199f639c95e8ae884ede46404370d7b64da3b309
Needed-By: Iba89a53b2715cf3a9c7485f2089f27d547fea308
Related-Bug: #1753209
(cherry picked from commit 6c772de103)
228 lines
11 KiB
Python
228 lines
11 KiB
Python
# 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_lib.callbacks import events
|
|
from neutron_lib import constants
|
|
from neutron_lib import context
|
|
|
|
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(plugin='ml2')
|
|
|
|
def _make_networkrbac(self, network, target, action='access_as_shared'):
|
|
policy = {'rbac_policy': {'tenant_id': network['network']['tenant_id'],
|
|
'object_id': network['network']['id'],
|
|
'object_type': 'network',
|
|
'action': action,
|
|
'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 _assert_external_net_state(self, net_id, is_external):
|
|
req = self.new_show_request('networks', net_id)
|
|
res = self.deserialize(self.fmt, req.get_response(self.api))
|
|
self.assertEqual(is_external, res['network']['router:external'])
|
|
|
|
def test_create_network_rbac_external(self):
|
|
with self.network() as ext_net:
|
|
net_id = ext_net['network']['id']
|
|
self._assert_external_net_state(net_id, is_external=False)
|
|
policy = self._make_networkrbac(ext_net,
|
|
'*',
|
|
'access_as_external')
|
|
self.plugin.create_rbac_policy(self.context, policy)
|
|
self._assert_external_net_state(net_id, is_external=True)
|
|
|
|
def test_update_network_rbac_external_valid(self):
|
|
orig_target = 'test-tenant-2'
|
|
new_target = 'test-tenant-3'
|
|
|
|
with self.network() as ext_net:
|
|
policy = self._make_networkrbac(ext_net,
|
|
orig_target,
|
|
'access_as_external')
|
|
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 policy['rbac_policy'].items():
|
|
self.assertEqual(netrbac2[k], v)
|
|
|
|
def test_delete_network_rbac_external(self):
|
|
with self.network() as ext_net:
|
|
net_id = ext_net['network']['id']
|
|
self._assert_external_net_state(net_id, is_external=False)
|
|
policy = self._make_networkrbac(ext_net,
|
|
'*',
|
|
'access_as_external')
|
|
net_rbac = self.plugin.create_rbac_policy(self.context, policy)
|
|
self._assert_external_net_state(net_id, is_external=True)
|
|
self.plugin.delete_rbac_policy(self.context, net_rbac['id'])
|
|
self._assert_external_net_state(net_id, is_external=False)
|
|
|
|
def test_delete_network_rbac_external_with_multi_rbac_policy(self):
|
|
with self.network() as ext_net:
|
|
net_id = ext_net['network']['id']
|
|
self._assert_external_net_state(net_id, is_external=False)
|
|
policy1 = self._make_networkrbac(ext_net,
|
|
'test-tenant-1',
|
|
'access_as_external')
|
|
net_rbac1 = self.plugin.create_rbac_policy(self.context, policy1)
|
|
self._assert_external_net_state(net_id, is_external=True)
|
|
policy2 = self._make_networkrbac(ext_net,
|
|
'test-tenant-2',
|
|
'access_as_external')
|
|
self.plugin.create_rbac_policy(self.context, policy2)
|
|
self._assert_external_net_state(net_id, is_external=True)
|
|
self.plugin.delete_rbac_policy(self.context, net_rbac1['id'])
|
|
self._assert_external_net_state(net_id, is_external=True)
|
|
|
|
def test_delete_external_network_shared_rbac(self):
|
|
with self.network() as ext_net:
|
|
net_id = ext_net['network']['id']
|
|
self.plugin.update_network(
|
|
self.context, net_id,
|
|
{'network': {'router:external': True}})
|
|
self._assert_external_net_state(net_id, is_external=True)
|
|
policy = self._make_networkrbac(ext_net, 'test-tenant-2')
|
|
net_rbac = self.plugin.create_rbac_policy(self.context, policy)
|
|
self.plugin.delete_rbac_policy(self.context, net_rbac['id'])
|
|
# Make sure that external attribute not changed.
|
|
self._assert_external_net_state(net_id, is_external=True)
|
|
|
|
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 policy['rbac_policy'].items():
|
|
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_port_presence_prevents_network_rbac_policy_deletion(self):
|
|
with self.network() as net:
|
|
netrbac, port = self._setup_networkrbac_and_port(
|
|
network=net, target_tenant='alice')
|
|
self.assertRaises(ext_rbac.RbacPolicyInUse,
|
|
self.plugin.delete_rbac_policy,
|
|
self.context, netrbac['id'])
|
|
|
|
# a wildcard policy should allow the specific policy to be deleted
|
|
# since it allows the remaining port
|
|
wild_policy = self._make_networkrbac(net, '*')
|
|
wild_policy = self.plugin.create_rbac_policy(self.context,
|
|
wild_policy)
|
|
self.plugin.delete_rbac_policy(self.context, netrbac['id'])
|
|
|
|
# now that wildcard is the only remaining, it should be subjected
|
|
# to to the same restriction
|
|
self.assertRaises(ext_rbac.RbacPolicyInUse,
|
|
self.plugin.delete_rbac_policy,
|
|
self.context, wild_policy['id'])
|
|
|
|
# similarly, we can't update the policy to a different tenant
|
|
update_policy = {'rbac_policy': {'target_tenant': 'bob'}}
|
|
self.assertRaises(ext_rbac.RbacPolicyInUse,
|
|
self.plugin.update_rbac_policy,
|
|
self.context, wild_policy['id'],
|
|
update_policy)
|
|
|
|
# after port anchor is gone, update and delete should pass
|
|
self.plugin.delete_port(self.context, port['id'])
|
|
self.plugin.update_rbac_policy(
|
|
self.context, wild_policy['id'], update_policy)
|
|
self.plugin.delete_rbac_policy(self.context, wild_policy['id'])
|
|
|
|
# check that policy is indeed gone
|
|
self.assertRaises(ext_rbac.RbacPolicyNotFound,
|
|
self.plugin.get_rbac_policy,
|
|
self.context, wild_policy['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)
|