x
/
vmware-nsx
Code Issues Proposed changes

Provider Security groups 

This patch set introduces a new feature called provider-security-groups.
Provider security groups allow the provider to create a security group
that is automatically attached to a specific tenants ports. The one
important thing to note is that rules inside of a provider security
group are set to DENY where as a normal security group they are set
to ALLOW. Provider security groups allow the admin tenant to block specific
traffic for any tenant they like by creatng a provider group. To use this
feature the admin tenant must first create a provider security group
on behalf of the other tenant (i.e):

$ neutron security-group-create no-pokemon-go-access --provider=True \
	--tenant-id=<shall remain nameless>

Then, whenever the above tenant id creates a port they will see a an
additional field on the port "provider-security-groups" which will
contain the uuid of the provider security group. This user can then
query neutron to see which rules are in it that are blocking them.

NOTE: one needs to use the correct policy.json file from this repo
for neutron inorder to prevent the tenant from removing the group.

Co-Authored-By: Aaron Rosen <aaronorosen@gmail.com>

Change-Id: I57b130437327b0bbe5cc0068695f226b76b4e2ba
This commit is contained in:
Roey Chen 2016-07-11 09:47:26 -07:00
parent 7a9336f09f
commit 2cfc1231dc
6 changed files with 583 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
etc/policy.json
View File

@ -144,9 +144,11 @@
"create_security_group:logging": "rule:admin_only",
"update_security_group:logging": "rule:admin_only",
"get_security_group:logging": "rule:admin_only",
"create_security_group:provider": "rule:admin_only",
"create_port:provider_security_groups": "rule:admin_only",
"create_flow_classifier": "rule:admin_only",
"update_flow_classifier": "rule:admin_only",
"delete_flow_classifier": "rule:admin_only",
"get_flow_classifier": "rule:admin_only"
"get_flow_classifier": "rule:admin_only",
}

213
vmware_nsx/db/extended_security_group.py
View File

@ -13,14 +13,22 @@
# License for the specific language governing permissions and limitations
# under the License.
from oslo_utils import uuidutils
import sqlalchemy as sa
from sqlalchemy import orm
from neutron.api.v2 import attributes
from neutron.common import utils as n_utils
from neutron.db import api as db_api
from neutron.db import db_base_plugin_v2
from neutron.db import model_base
from neutron.db import securitygroups_db
from neutron.extensions import securitygroup as ext_sg
from neutron_lib.api import validators
from neutron_lib import constants as n_constants
from vmware_nsx._i18n import _
from vmware_nsx.extensions import providersecuritygroup as provider_sg
from vmware_nsx.extensions import securitygrouplogging as sg_logging
@ -32,6 +40,7 @@ class NsxExtendedSecurityGroupProperties(model_base.BASEV2):
ondelete="CASCADE"),
primary_key=True)
logging = sa.Column(sa.Boolean, default=False, nullable=False)
provider = sa.Column(sa.Boolean, default=False, nullable=False)
security_group = orm.relationship(
securitygroups_db.SecurityGroup,
backref=orm.backref('ext_properties', lazy='joined',
@ -40,19 +49,55 @@ class NsxExtendedSecurityGroupProperties(model_base.BASEV2):
class ExtendedSecurityGroupPropertiesMixin(object):
# NOTE(arosen): here we add a relationship so that from the ports model
# it provides us access to SecurityGroupPortBinding and
# NsxExtendedSecurityGroupProperties
securitygroups_db.SecurityGroupPortBinding.extended_grp = orm.relationship(
'NsxExtendedSecurityGroupProperties',
foreign_keys="SecurityGroupPortBinding.security_group_id",
primaryjoin=("NsxExtendedSecurityGroupProperties.security_group_id"
"==SecurityGroupPortBinding.security_group_id"))
def create_provider_security_group(self, context, security_group):
"""Create a provider security group.
This method creates a security group that does not by default
enable egress traffic which normal neutron security groups do.
"""
s = security_group['security_group']
tenant_id = s['tenant_id']
with db_api.autonested_transaction(context.session):
security_group_db = securitygroups_db.SecurityGroup(
id=s.get('id') or (uuidutils.generate_uuid()),
description=s.get('description', ''),
tenant_id=tenant_id,
name=s.get('name', ''))
context.session.add(security_group_db)
secgroup_dict = self._make_security_group_dict(security_group_db)
secgroup_dict[provider_sg.PROVIDER] = True
return secgroup_dict
def _process_security_group_properties_create(self, context,
sg_res, sg_req):
sg_res, sg_req,
default_sg=False):
self._validate_security_group_properties_create(
context, sg_req, default_sg)
with context.session.begin(subtransactions=True):
properties = NsxExtendedSecurityGroupProperties(
security_group_id=sg_res['id'],
logging=sg_req.get(sg_logging.LOGGING, False))
logging=sg_req.get(sg_logging.LOGGING, False),
provider=sg_req.get(provider_sg.PROVIDER, False))
context.session.add(properties)
sg_res[sg_logging.LOGGING] = sg_req.get(sg_logging.LOGGING, False)
sg_res[provider_sg.PROVIDER] = sg_req.get(provider_sg.PROVIDER, False)
def _get_security_group_properties(self, context, security_group_id):
return context.session.query(
NsxExtendedSecurityGroupProperties).filter_by(
security_group_id=security_group_id).one()
with context.session.begin(subtransactions=True):
prop = context.session.query(
NsxExtendedSecurityGroupProperties).filter_by(
security_group_id=security_group_id).one()
return prop
def _process_security_group_properties_update(self, context,
sg_res, sg_req):
@ -68,9 +113,163 @@ class ExtendedSecurityGroupPropertiesMixin(object):
prop = self._get_security_group_properties(context, security_group_id)
return prop.logging
db_base_plugin_v2.NeutronDbPluginV2.register_dict_extend_funcs(
ext_sg.SECURITYGROUPS, ['_extend_security_group_with_properties'])
def _is_provider_security_group(self, context, security_group_id):
sg_prop = self._get_security_group_properties(context,
security_group_id)
return sg_prop.provider
def _check_provider_security_group_exists(self, context,
security_group_id):
# NOTE(roeyc): We want to retrieve the security-group info by calling
# get_security_group, this will also validate that the provider
# security-group belongs to the same tenant this request is made for.
sg = self.get_security_group(context, security_group_id)
if not sg[provider_sg.PROVIDER]:
raise provider_sg.SecurityGroupNotProvider(id=sg)
def _check_invalid_security_groups_specified(self, context, port):
if validators.is_attr_set(port.get(ext_sg.SECURITYGROUPS)):
for sg in port.get(ext_sg.SECURITYGROUPS, []):
# makes sure user doesn't add non-provider secgrp as secgrp
if self._is_provider_security_group(context, sg):
raise provider_sg.SecurityGroupIsProvider(id=sg)
if validators.is_attr_set(
port.get(provider_sg.PROVIDER_SECURITYGROUPS)):
# also check all provider groups are provider.
for sg in port.get(provider_sg.PROVIDER_SECURITYGROUPS, []):
self._check_provider_security_group_exists(context, sg)
def _get_tenant_provider_security_groups(self, context, tenant_id):
res = context.session.query(
NsxExtendedSecurityGroupProperties.security_group_id
).join(securitygroups_db.SecurityGroup).filter(
securitygroups_db.SecurityGroup.tenant_id == tenant_id,
NsxExtendedSecurityGroupProperties.provider == sa.true()).scalar()
return [res] if res else []
def _validate_security_group_properties_create(self, context,
security_group, default_sg):
self._validate_provider_security_group_create(context, security_group,
default_sg)
def _validate_provider_security_group_create(self, context, security_group,
default_sg):
if not security_group.get(provider_sg.PROVIDER, False):
return
if default_sg:
raise provider_sg.DefaultSecurityGroupIsNotProvider()
tenant_id = security_group['tenant_id']
ssg = self._get_tenant_provider_security_groups(context, tenant_id)
if ssg:
# REVISIT(roeyc): At the moment we only allow on provider
# security-group per tenant, this might change in the future.
raise Exception(_("Provider Security-group already exists"
"(%(pvdsg)s) for tenant %(tenant_id)s.")
% {'pvdsg': ssg, 'tenant_id': tenant_id})
def _get_provider_security_groups_on_port(self, context, port):
p = port['port']
tenant_id = p['tenant_id']
provider_sgs = p.get(provider_sg.PROVIDER_SECURITYGROUPS,
n_constants.ATTR_NOT_SPECIFIED)
if p.get('device_owner') and n_utils.is_port_trusted(p):
return
self._check_invalid_security_groups_specified(context, p)
if not validators.is_attr_set(provider_sgs):
if provider_sgs is n_constants.ATTR_NOT_SPECIFIED:
provider_sgs = self._get_tenant_provider_security_groups(
context, tenant_id)
else:
# Accept None as indication that this port should not be
# associated with any provider security-group.
provider_sgs = []
return provider_sgs
def _process_port_create_provider_security_group(self, context, p,
security_group_ids):
if validators.is_attr_set(security_group_ids):
for security_group_id in security_group_ids:
self._create_port_security_group_binding(context, p['id'],
security_group_id)
p[provider_sg.PROVIDER_SECURITYGROUPS] = security_group_ids or []
def _process_port_update_provider_security_group(self, context, port,
original_port,
updated_port):
p = port['port']
provider_sg_specified = (provider_sg.PROVIDER_SECURITYGROUPS in p and
p[provider_sg.PROVIDER_SECURITYGROUPS] !=
n_constants.ATTR_NOT_SPECIFIED)
provider_sg_changed = (
provider_sg_specified and not n_utils.compare_elements(
original_port[provider_sg.PROVIDER_SECURITYGROUPS],
p[provider_sg.PROVIDER_SECURITYGROUPS]))
sg_changed = (
set(original_port[ext_sg.SECURITYGROUPS]) !=
set(updated_port[ext_sg.SECURITYGROUPS]))
if provider_sg_changed:
port['port']['tenant_id'] = original_port['id']
port['port']['id'] = original_port['id']
updated_port[provider_sg.PROVIDER_SECURITYGROUPS] = (
self._get_provider_security_groups_on_port(context, port))
else:
if sg_changed:
self._check_invalid_security_groups_specified(context, p)
updated_port[provider_sg.PROVIDER_SECURITYGROUPS] = (
original_port[provider_sg.PROVIDER_SECURITYGROUPS])
if provider_sg_changed or sg_changed:
if not sg_changed:
query = context.session.query(
securitygroups_db.SecurityGroupPortBinding)
for sg in original_port[provider_sg.PROVIDER_SECURITYGROUPS]:
binding = query.filter_by(
port_id=p['id'], security_group_id=sg).one()
context.session.delete(binding)
self._process_port_create_provider_security_group(
context, updated_port,
updated_port[provider_sg.PROVIDER_SECURITYGROUPS])
def _extend_security_group_with_properties(self, sg_res, sg_db):
if sg_db.ext_properties:
sg_res[sg_logging.LOGGING] = sg_db.ext_properties.logging
sg_res[provider_sg.PROVIDER] = sg_db.ext_properties.provider
def _extend_port_dict_provider_security_group(self, port_res, port_db):
# NOTE(arosen): this method overrides the one in the base
# security group db class. The reason this is needed is because
# we are storing provider security groups in the same security
# groups db model. We need to do this here to remove the provider
# security groups and put those on the port resource as their
# own attribute.
# Security group bindings will be retrieved from the SQLAlchemy
# model. As they're loaded eagerly with ports because of the
# joined load they will not cause an extra query.
provider_groups = []
not_provider_groups = []
for sec_group_mapping in port_db.security_groups:
if sec_group_mapping.extended_grp.provider is True:
provider_groups.append(sec_group_mapping['security_group_id'])
else:
not_provider_groups.append(
sec_group_mapping['security_group_id'])
port_res[ext_sg.SECURITYGROUPS] = not_provider_groups
port_res[provider_sg.PROVIDER_SECURITYGROUPS] = provider_groups
return port_res
db_base_plugin_v2.NeutronDbPluginV2.register_dict_extend_funcs(
attributes.PORTS, ['_extend_port_dict_provider_security_group'])
db_base_plugin_v2.NeutronDbPluginV2.register_dict_extend_funcs(
ext_sg.SECURITYGROUPS, ['_extend_security_group_with_properties'])

2
vmware_nsx/db/migration/alembic_migrations/versions/EXPAND_HEAD
View File

@ -1 +1 @@
633514d94b93
1b4eaffe4f31

34
vmware_nsx/db/migration/alembic_migrations/versions/newton/expand/1b4eaffe4f31_nsx_provider_security_group.py Normal file
View File

@ -0,0 +1,34 @@
# Copyright 2016 VMware, Inc.
#
# 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.
"""NSX Adds a 'provider' attribute to security-group
Revision ID: 1b4eaffe4f31
Revises: 633514d94b93
Create Date: 2016-07-17 11:30:31.263918
"""
# revision identifiers, used by Alembic.
revision = '1b4eaffe4f31'
down_revision = '633514d94b93'
from alembic import op
import sqlalchemy as sa
def upgrade():
op.add_column('nsx_extended_security_group_properties',
sa.Column('provider', sa.Boolean(), default=False,
server_default=sa.false(), nullable=False))

95
vmware_nsx/extensions/providersecuritygroup.py Normal file
View File

@ -0,0 +1,95 @@
# 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 neutron.api import extensions
from neutron.extensions import securitygroup
from neutron_lib.api import converters
from neutron_lib import constants
from neutron_lib import exceptions as nexception
from vmware_nsx._i18n import _
PROVIDER = 'provider'
PROVIDER_SECURITYGROUPS = 'provider_security_groups'
EXTENDED_ATTRIBUTES_2_0 = {
'security_groups': {
PROVIDER: {
'allow_post': True,
'allow_put': False,
'convert_to': converters.convert_to_boolean,
'default': False,
'enforce_policy': True,
'is_visible': True}
},
'ports': {PROVIDER_SECURITYGROUPS: {
'allow_post': True,
'allow_put': True,
'is_visible': True,
'convert_to': securitygroup.convert_to_uuid_list_or_none,
'default': constants.ATTR_NOT_SPECIFIED}
}
}
NUM_PROVIDER_SGS_ON_PORT = 1
class SecurityGroupNotProvider(nexception.InvalidInput):
message = _("Security group %(id)s is not a provider security group.")
class SecurityGroupIsProvider(nexception.InvalidInput):
message = _("Security group %(id)s is a provider security group and "
"cannot be specified via the security group field.")
class DefaultSecurityGroupIsNotProvider(nexception.InvalidInput):
message = _("Can't create default security-group as a provider "
"security-group.")
class Providersecuritygroup(extensions.ExtensionDescriptor):
"""Provider security-group extension."""
@classmethod
def get_name(cls):
return "Provider security group"
@classmethod
def get_alias(cls):
return "provider-security-group"
@classmethod
def get_description(cls):
return "Admin controlled security groups with blocking rules."
@classmethod
def get_updated(cls):
return "2016-07-13T10:00:00-00:00"
def get_required_extensions(self):
return ["security-group"]
@classmethod
def get_resources(cls):
"""Returns Ext Resources."""
return []
def get_extended_resources(self, version):
if version == "2.0":
return EXTENDED_ATTRIBUTES_2_0
else:
return {}

244
vmware_nsx/tests/unit/extensions/test_provider_security_groups.py Normal file
View File

@ -0,0 +1,244 @@
# 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 neutron.api.v2 import attributes as attr
from neutron import context
from neutron.db import db_base_plugin_v2
from neutron.db import securitygroups_db
from neutron.tests.unit.extensions import test_securitygroup
import webob.exc
from vmware_nsx.db import extended_security_group
from vmware_nsx.extensions import providersecuritygroup as provider_sg
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-security-group"]
def create_security_group(self, context, security_group, default_sg=False):
secgroup = security_group['security_group']
with context.session.begin(subtransactions=True):
# 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 context.session.begin(subtransactions=True):
self._ensure_default_security_group_on_port(context, port)
sgids = self._get_security_groups_on_port(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
provider_groups = self._get_provider_security_groups_on_port(
context, port)
self._process_port_create_provider_security_group(
context, port_data, provider_groups)
return port_data
def update_port(self, context, id, port):
with context.session.begin(subtransactions=True):
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)
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 = 'foobar'
# add provider group attributes
attr.RESOURCE_ATTRIBUTE_MAP['security_groups'].update(
provider_sg.EXTENDED_ATTRIBUTES_2_0['security_groups'])
attr.RESOURCE_ATTRIBUTE_MAP['ports'].update(
provider_sg.EXTENDED_ATTRIBUTES_2_0['ports'])
def tearDown(self):
# remove provider security group attributes
del attr.RESOURCE_ATTRIBUTE_MAP['security_groups']['provider']
del attr.RESOURCE_ATTRIBUTE_MAP['ports']['provider_security_groups']
super(ProviderSecurityGroupExtTestCase, self).tearDown()
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_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(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_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 = 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'])
# TODO(arosen): add nsxv3 test case mixin when ready
# TODO(roeyc): add nsxv test case mixin when ready