411 lines
20 KiB
Python
411 lines
20 KiB
Python
# Copyright 2016 VMware, 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.
|
|
from unittest import mock
|
|
|
|
import webob.exc
|
|
|
|
from neutron.db import db_base_plugin_v2
|
|
from neutron.db import securitygroups_db
|
|
from neutron.extensions import securitygroup as ext_sg
|
|
from neutron.tests.unit.db import test_db_base_plugin_v2
|
|
from neutron.tests.unit.extensions import test_securitygroup
|
|
from neutron_lib import context as neutron_context
|
|
from neutron_lib.db import api as db_api
|
|
|
|
from vmware_nsx.db import extended_security_group
|
|
from vmware_nsx.extensions import providersecuritygroup as provider_sg
|
|
from vmware_nsx.tests.unit.nsx_p import test_plugin as test_nsxp_plugin
|
|
from vmware_nsx.tests.unit.nsx_v import test_plugin as test_nsxv_plugin
|
|
from vmware_nsx.tests.unit.nsx_v3 import test_plugin as test_nsxv3_plugin
|
|
|
|
|
|
PLUGIN_NAME = ('vmware_nsx.tests.unit.extensions.'
|
|
'test_provider_security_groups.ProviderSecurityGroupTestPlugin')
|
|
|
|
|
|
# FIXME(arosen): make common mixin for extended_security_group_properties and
|
|
# security_group_db_minxin.
|
|
class ProviderSecurityGroupTestPlugin(
|
|
db_base_plugin_v2.NeutronDbPluginV2,
|
|
extended_security_group.ExtendedSecurityGroupPropertiesMixin,
|
|
securitygroups_db.SecurityGroupDbMixin):
|
|
|
|
supported_extension_aliases = ["security-group",
|
|
provider_sg.ALIAS]
|
|
|
|
def create_security_group(self, context, security_group, default_sg=False):
|
|
secgroup = security_group['security_group']
|
|
with db_api.CONTEXT_WRITER.using(context):
|
|
# NOTE(arosen): a neutron security group by default adds rules
|
|
# that allow egress traffic. We do not want this behavior for
|
|
# provider security_groups
|
|
if secgroup.get(provider_sg.PROVIDER) is True:
|
|
secgroup_db = self.create_provider_security_group(
|
|
context, security_group)
|
|
else:
|
|
secgroup_db = (
|
|
super(ProviderSecurityGroupTestPlugin, self
|
|
).create_security_group(context, security_group,
|
|
default_sg))
|
|
|
|
self._process_security_group_properties_create(context,
|
|
secgroup_db,
|
|
secgroup,
|
|
default_sg)
|
|
return secgroup_db
|
|
|
|
def create_port(self, context, port, l2gw_port_check=False):
|
|
port_data = port['port']
|
|
|
|
with db_api.CONTEXT_WRITER.using(context):
|
|
self._ensure_default_security_group_on_port(context, port)
|
|
(sgids, provider_groups) = self._get_port_security_groups_lists(
|
|
context, port)
|
|
|
|
port_db = super(ProviderSecurityGroupTestPlugin, self).create_port(
|
|
context, port)
|
|
port_data.update(port_db)
|
|
|
|
# handle adding security groups to port
|
|
self._process_port_create_security_group(
|
|
context, port_db, sgids)
|
|
|
|
# handling adding provider security group to port if there are any
|
|
self._process_port_create_provider_security_group(
|
|
context, port_data, provider_groups)
|
|
return port_data
|
|
|
|
def update_port(self, context, id, port):
|
|
with db_api.CONTEXT_WRITER.using(context):
|
|
original_port = super(ProviderSecurityGroupTestPlugin,
|
|
self).get_port(context, id)
|
|
updated_port = super(ProviderSecurityGroupTestPlugin,
|
|
self).update_port(context, id, port)
|
|
|
|
self.update_security_group_on_port(context, id, port,
|
|
original_port, updated_port)
|
|
self._process_port_update_provider_security_group(
|
|
context, port, original_port, updated_port)
|
|
return self.get_port(context, id)
|
|
|
|
def _make_port_dict(self, port, fields=None, process_extensions=True):
|
|
port_data = super(
|
|
ProviderSecurityGroupTestPlugin, self)._make_port_dict(
|
|
port, fields=fields,
|
|
process_extensions=process_extensions)
|
|
self._remove_provider_security_groups_from_list(port_data)
|
|
return port_data
|
|
|
|
def delete_security_group(self, context, id):
|
|
self._prevent_non_admin_edit_provider_sg(context, id)
|
|
super(ProviderSecurityGroupTestPlugin,
|
|
self).delete_security_group(context, id)
|
|
|
|
def delete_security_group_rule(self, context, id):
|
|
rule_db = self._get_security_group_rule(context, id)
|
|
sg_id = rule_db['security_group_id']
|
|
self._prevent_non_admin_edit_provider_sg(context, sg_id)
|
|
return super(ProviderSecurityGroupTestPlugin,
|
|
self).delete_security_group_rule(context, id)
|
|
|
|
def create_security_group_rule(self, context, security_group_rule):
|
|
id = security_group_rule['security_group_rule']['security_group_id']
|
|
self._prevent_non_admin_edit_provider_sg(context, id)
|
|
return super(ProviderSecurityGroupTestPlugin,
|
|
self).create_security_group_rule(context,
|
|
security_group_rule)
|
|
|
|
|
|
class ProviderSecurityGroupExtTestCase(
|
|
test_securitygroup.SecurityGroupDBTestCase):
|
|
def setUp(self, plugin=PLUGIN_NAME, ext_mgr=None):
|
|
super(ProviderSecurityGroupExtTestCase, self).setUp(
|
|
plugin=plugin, ext_mgr=ext_mgr)
|
|
self._tenant_id = test_db_base_plugin_v2.TEST_TENANT_ID
|
|
# add provider group attributes
|
|
ext_sg.Securitygroup().update_attributes_map(
|
|
provider_sg.EXTENDED_ATTRIBUTES_2_0)
|
|
|
|
def _create_provider_security_group(self):
|
|
body = {'security_group': {'name': 'provider-deny',
|
|
'tenant_id': self._tenant_id,
|
|
'description': 'foobarzzkk',
|
|
'provider': True}}
|
|
security_group_req = self.new_create_request('security-groups', body)
|
|
return self.deserialize(self.fmt,
|
|
security_group_req.get_response(self.ext_api))
|
|
|
|
def test_create_provider_security_group(self):
|
|
# confirm this attribute is true
|
|
provider_secgroup = self._create_provider_security_group()
|
|
self.assertTrue(provider_secgroup['security_group']['provider'])
|
|
|
|
# provider security groups have no rules by default which is different
|
|
# from normal neutron security groups which by default include a rule
|
|
# to allow egress traffic. We confirm this here.
|
|
self.assertEqual(
|
|
provider_secgroup['security_group']['security_group_rules'], [])
|
|
|
|
def test_create_provider_security_groups_same_tenant(self):
|
|
provider_secgroup = self._create_provider_security_group()
|
|
self.assertTrue(provider_secgroup['security_group']['provider'])
|
|
|
|
# Verify that another one can also be created for the same tenant
|
|
provider_secgroup2 = self._create_provider_security_group()
|
|
self.assertTrue(provider_secgroup2['security_group']['provider'])
|
|
|
|
def test_create_port_gets_provider_sg(self):
|
|
# need to create provider security group first.
|
|
provider_secgroup = self._create_provider_security_group()
|
|
with self.port(tenant_id=self._tenant_id) as p:
|
|
# check that the provider security group is on port resource.
|
|
self.assertEqual(1, len(p['port']['provider_security_groups']))
|
|
self.assertEqual(provider_secgroup['security_group']['id'],
|
|
p['port']['provider_security_groups'][0])
|
|
|
|
# confirm there is still a default security group.
|
|
self.assertEqual(len(p['port']['security_groups']), 1)
|
|
|
|
def test_create_port_gets_multi_provider_sg(self):
|
|
# need to create provider security groups first.
|
|
provider_secgroup1 = self._create_provider_security_group()
|
|
provider_secgroup2 = self._create_provider_security_group()
|
|
with self.port(tenant_id=self._tenant_id) as p:
|
|
# check that the provider security group is on port resource.
|
|
self.assertEqual(2, len(p['port']['provider_security_groups']))
|
|
self.assertIn(provider_secgroup1['security_group']['id'],
|
|
p['port']['provider_security_groups'])
|
|
self.assertIn(provider_secgroup2['security_group']['id'],
|
|
p['port']['provider_security_groups'])
|
|
|
|
# confirm there is still a default security group.
|
|
self.assertEqual(len(p['port']['security_groups']), 1)
|
|
|
|
def test_create_port_with_no_provider_sg(self):
|
|
self._create_provider_security_group()
|
|
with self.port(tenant_id=self._tenant_id,
|
|
arg_list=('provider_security_groups', ),
|
|
provider_security_groups=[]) as p1:
|
|
self.assertEqual([], p1['port']['provider_security_groups'])
|
|
with self.port(tenant_id=self._tenant_id,
|
|
arg_list=('provider_security_groups', ),
|
|
provider_security_groups=None) as p1:
|
|
self.assertEqual([], p1['port']['provider_security_groups'])
|
|
|
|
def test_update_port_remove_provider_sg_with_empty_list(self):
|
|
# need to create provider security group first.
|
|
self._create_provider_security_group()
|
|
with self.port(tenant_id=self._tenant_id) as p:
|
|
body = {'port': {'provider_security_groups': []}}
|
|
req = self.new_update_request('ports', body, p['port']['id'])
|
|
port = self.deserialize(self.fmt, req.get_response(self.api))
|
|
# confirm that the group has been removed.
|
|
self.assertEqual([], port['port']['provider_security_groups'])
|
|
|
|
def test_update_port_remove_provider_sg_with_none(self):
|
|
# need to create provider security group first.
|
|
self._create_provider_security_group()
|
|
with self.port(tenant_id=self._tenant_id) as p:
|
|
body = {'port': {'provider_security_groups': None}}
|
|
req = self.new_update_request('ports', body, p['port']['id'])
|
|
port = self.deserialize(self.fmt, req.get_response(self.api))
|
|
# confirm that the group has been removed.
|
|
self.assertEqual([], port['port']['provider_security_groups'])
|
|
|
|
def test_cannot_update_port_with_provider_group_as_sec_group(self):
|
|
with self.port(tenant_id=self._tenant_id) as p:
|
|
provider_secgroup = self._create_provider_security_group()
|
|
sg_id = provider_secgroup['security_group']['id']
|
|
body = {'port': {'security_groups': [sg_id]}}
|
|
req = self.new_update_request('ports', body, p['port']['id'])
|
|
res = req.get_response(self.api)
|
|
self.assertEqual(webob.exc.HTTPBadRequest.code, res.status_int)
|
|
|
|
def test_cannot_update_port_with_sec_group_as_provider(self):
|
|
with self.security_group() as sg1:
|
|
with self.port(tenant_id=self._tenant_id) as p:
|
|
sg_id = sg1['security_group']['id']
|
|
body = {'port': {'provider_security_groups': [sg_id]}}
|
|
req = self.new_update_request('ports', body, p['port']['id'])
|
|
res = req.get_response(self.api)
|
|
self.assertEqual(webob.exc.HTTPBadRequest.code, res.status_int)
|
|
|
|
def test_cannot_update_port_with_different_tenant_provider_secgroup(self):
|
|
with self.port(tenant_id=self._tenant_id) as p:
|
|
tmp_tenant_id = self._tenant_id
|
|
self._tenant_id += "-alt"
|
|
pvd_sg = self._create_provider_security_group()
|
|
self._tenant_id = tmp_tenant_id
|
|
body = {'port': {'provider_security_groups': [
|
|
pvd_sg['security_group']['id']]}}
|
|
|
|
ctx = neutron_context.Context('', self._tenant_id)
|
|
req = self.new_update_request('ports', body,
|
|
p['port']['id'], context=ctx)
|
|
res = req.get_response(self.api)
|
|
self.assertEqual(webob.exc.HTTPNotFound.code, res.status_int)
|
|
|
|
def test_update_port_security_groups_only(self):
|
|
# We want to make sure that modifying security-groups on the port
|
|
# doesn't impact the provider security-group on this port.
|
|
provider_secgroup = self._create_provider_security_group()
|
|
with self.security_group() as sg1:
|
|
with self.port(tenant_id=self._tenant_id) as p:
|
|
sg_id = sg1['security_group']['id']
|
|
body = {'port': {'security_groups': [sg_id]}}
|
|
req = self.new_update_request('ports', body, p['port']['id'])
|
|
port = self.deserialize(self.fmt, req.get_response(self.api))
|
|
self.assertEqual(
|
|
[provider_secgroup['security_group']['id']],
|
|
port['port']['provider_security_groups'])
|
|
|
|
def test_update_port_security_groups(self):
|
|
with self.security_group() as sg1:
|
|
with self.port(tenant_id=self._tenant_id) as p:
|
|
# Port created before provider secgroup is created, so the port
|
|
# would not be associated with the pvd secgroup at this point.
|
|
provider_secgroup = self._create_provider_security_group()
|
|
pvd_sg_id = provider_secgroup['security_group']['id']
|
|
sg_id = sg1['security_group']['id']
|
|
body = {'port': {
|
|
'security_groups': [sg_id],
|
|
'provider_security_groups': [pvd_sg_id]}
|
|
}
|
|
req = self.new_update_request('ports', body, p['port']['id'])
|
|
port = self.deserialize(self.fmt, req.get_response(self.api))
|
|
self.assertEqual([pvd_sg_id],
|
|
port['port']['provider_security_groups'])
|
|
self.assertEqual([sg_id], port['port']['security_groups'])
|
|
|
|
def test_non_admin_cannot_delete_provider_sg_and_admin_can(self):
|
|
provider_secgroup = self._create_provider_security_group()
|
|
pvd_sg_id = provider_secgroup['security_group']['id']
|
|
|
|
# Try deleting the request as the normal tenant returns forbidden
|
|
# as a tenant is not allowed to delete this.
|
|
ctx = neutron_context.Context('', self._tenant_id)
|
|
self._delete('security-groups', pvd_sg_id,
|
|
expected_code=webob.exc.HTTPForbidden.code,
|
|
neutron_context=ctx)
|
|
# can be deleted though as admin
|
|
self._delete('security-groups', pvd_sg_id,
|
|
expected_code=webob.exc.HTTPNoContent.code)
|
|
|
|
def test_non_admin_cannot_delete_provider_sg_rule(self):
|
|
provider_secgroup = self._create_provider_security_group()
|
|
pvd_sg_id = provider_secgroup['security_group']['id']
|
|
|
|
data = {'security_group_rule': {'security_group_id': pvd_sg_id,
|
|
'direction': 'ingress',
|
|
'protocol': 'tcp',
|
|
'ethertype': 'IPv4',
|
|
'tenant_id': self._tenant_id}}
|
|
|
|
req = self.new_create_request('security-group-rules', data)
|
|
res = self.deserialize(self.fmt, req.get_response(self.ext_api))
|
|
|
|
sg_rule_id = res['security_group_rule']['id']
|
|
|
|
# Try deleting the request as the normal tenant returns forbidden
|
|
# as a tenant is not allowed to delete this.
|
|
ctx = neutron_context.Context('', self._tenant_id)
|
|
self._delete('security-group-rules', sg_rule_id,
|
|
expected_code=webob.exc.HTTPForbidden.code,
|
|
neutron_context=ctx)
|
|
# can be deleted though as admin
|
|
self._delete('security-group-rules', sg_rule_id,
|
|
expected_code=webob.exc.HTTPNoContent.code)
|
|
|
|
def test_non_admin_cannot_add_provider_sg_rule(self):
|
|
provider_secgroup = self._create_provider_security_group()
|
|
pvd_sg_id = provider_secgroup['security_group']['id']
|
|
|
|
data = {'security_group_rule': {'security_group_id': pvd_sg_id,
|
|
'direction': 'ingress',
|
|
'protocol': 'tcp',
|
|
'ethertype': 'IPv4',
|
|
'tenant_id': self._tenant_id}}
|
|
|
|
req = self.new_create_request(
|
|
'security-group-rules', data)
|
|
req.environ['neutron.context'] = neutron_context.Context(
|
|
'', self._tenant_id)
|
|
res = req.get_response(self.ext_api)
|
|
self.assertEqual(webob.exc.HTTPForbidden.code, res.status_int)
|
|
|
|
|
|
class TestNSXv3ProviderSecurityGrp(test_nsxv3_plugin.NsxV3PluginTestCaseMixin,
|
|
ProviderSecurityGroupExtTestCase):
|
|
|
|
def test_update_port_remove_provider_sg(self):
|
|
# need to create provider security group first.
|
|
self._create_provider_security_group()
|
|
with self.port(tenant_id=self._tenant_id) as p:
|
|
body = {'port': {'provider_security_groups': []}}
|
|
req = self.new_update_request('ports', body, p['port']['id'])
|
|
port = self.deserialize(self.fmt, req.get_response(self.api))
|
|
# confirm that the group has been removed.
|
|
self.assertEqual([], port['port']['provider_security_groups'])
|
|
# make sure that the security groups did not contain the provider
|
|
# security group
|
|
self.assertEqual(p['port']['security_groups'],
|
|
port['port']['security_groups'])
|
|
|
|
|
|
class TestNSXvProviderSecurityGroup(test_nsxv_plugin.NsxVPluginV2TestCase,
|
|
ProviderSecurityGroupExtTestCase):
|
|
def test_create_provider_security_group(self):
|
|
_create_section_tmp = self.fc2.create_section
|
|
|
|
def _create_section(*args, **kwargs):
|
|
return _create_section_tmp(*args, **kwargs)
|
|
|
|
with mock.patch.object(self.fc2, 'create_section',
|
|
side_effect=_create_section) as create_sec_mock:
|
|
super(TestNSXvProviderSecurityGroup,
|
|
self).test_create_provider_security_group()
|
|
create_sec_mock.assert_called_with('ip', mock.ANY,
|
|
insert_top=True,
|
|
insert_before=mock.ANY)
|
|
|
|
def test_create_provider_security_group_rule(self):
|
|
provider_secgroup = self._create_provider_security_group()
|
|
sg_id = provider_secgroup['security_group']['id']
|
|
_create_nsx_rule_tmp = self.plugin._create_nsx_rule
|
|
|
|
def m_create_nsx_rule(*args, **kwargs):
|
|
return _create_nsx_rule_tmp(*args, **kwargs)
|
|
|
|
with mock.patch.object(self.plugin, '_create_nsx_rule',
|
|
side_effect=m_create_nsx_rule) as create_rule_m:
|
|
with self.security_group_rule(security_group_id=sg_id):
|
|
create_rule_m.assert_called_with(mock.ANY, mock.ANY,
|
|
logged=mock.ANY,
|
|
action='deny')
|
|
|
|
|
|
class TestNSXpProviderSecurityGrp(test_nsxp_plugin.NsxPPluginTestCaseMixin,
|
|
ProviderSecurityGroupExtTestCase):
|
|
|
|
def test_create_provider_security_group_rule(self):
|
|
provider_secgroup = self._create_provider_security_group()
|
|
sg_id = provider_secgroup['security_group']['id']
|
|
|
|
with mock.patch("vmware_nsxlib.v3.policy.core_resources."
|
|
"NsxPolicyCommunicationMapApi.update_with_entries"
|
|
) as entry_create,\
|
|
self.security_group_rule(security_group_id=sg_id):
|
|
entry_create.assert_called_once()
|