x/vmware-nsx
Code Issues Proposed changes

NSX|V - initial support for NSX policy

Browse Source 
This code adds an extension for policy-id in a security group.
when this feature is enabled (new nsxv config: use_nsx_policies):
- Each security group will be linked to an nsx policy.
- No rules will be added to any of the security groups
- Only admin can edit security groups (depending on the policy.json)
- the default security group will be using the new nsx.ini config
  default_policy_id

Change-Id: Iad5e90245c2f70ed88f65f0c5e6ec46cb2eedbbc
This commit is contained in:
Adit Sarfaty 2016-10-05 16:08:01 +03:00
parent a6f9a1f8e2
commit 5c1f2f5b30
13 changed files with 489 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
devstack/lib/vmware_nsx_v
View File

@ -118,6 +118,8 @@ function neutron_plugin_configure_service {
_nsxv_ini_set edge_ha "$NSXV_EDGE_HA"
_nsxv_ini_set exclusive_router_appliance_size "$NSXV_EXCLUSIVE_ROUTER_APPLIANCE_SIZE"
_nsxv_ini_set use_dvs_features "$NSXV_USE_DVS_FEATURES"
_nsxv_ini_set use_nsx_policies "$NSXV_USE_NSX_POLICIES"
_nsxv_ini_set default_policy_id "$NSXV_DEFAULT_POLICY_ID"
}
function neutron_plugin_setup_interface_driver {

7
devstack/tools/nsxv_cleanup.py
View File

@ -215,8 +215,11 @@ class VSMClient(object):
# Get layer3 sections related to security group
if response.status_code is 200:
l3_sections = response.json()['layer3Sections']['layer3Sections']
firewall_sections = [s for s in l3_sections if s['name'] !=
"Default Section Layer3"]
# do not delete the default section, or sections created by the
# service composer
firewall_sections = [s for s in l3_sections if (s['name'] !=
"Default Section Layer3" and
"NSX Service Composer" not in s['name'])]
else:
print("ERROR: wrong response status code! Exiting...")
sys.exit()

2
etc/policy.json
View File

@ -146,6 +146,8 @@
"get_security_group:logging": "rule:admin_only",
"create_security_group:provider": "rule:admin_only",
"create_port:provider_security_groups": "rule:admin_only",
"create_security_group:policy": "rule:admin_only",
"update_security_group:policy": "rule:admin_only",
"create_flow_classifier": "rule:admin_only",
"update_flow_classifier": "rule:admin_only",

6
vmware_nsx/common/config.py
View File

@ -617,6 +617,12 @@ nsxv_opts = [
help=_("(Optional) If set to True, the plugin will create "
"a redirect rule to send all the traffic to the "
"security partner")),
cfg.BoolOpt('use_nsx_policies', default=False,
help=_("If set to True, the plugin will use NSX policies "
"in the neutron security groups.")),
cfg.StrOpt('default_policy_id',
help=_("(Optional) If use_nsx_policies is True, this policy "
"will be used as the default policy for new tenants.")),
]
# Register the configuration options

33
vmware_nsx/db/extended_security_group.py
View File

@ -33,6 +33,7 @@ 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
from vmware_nsx.extensions import securitygrouppolicy as sg_policy
class NsxExtendedSecurityGroupProperties(model_base.BASEV2):
@ -45,6 +46,7 @@ class NsxExtendedSecurityGroupProperties(model_base.BASEV2):
logging = sa.Column(sa.Boolean, default=False, nullable=False)
provider = sa.Column(sa.Boolean, default=False, server_default=sql.false(),
nullable=False)
policy = sa.Column(sa.String(36))
security_group = orm.relationship(
securitygroups_db.SecurityGroup,
backref=orm.backref('ext_properties', lazy='joined',
@ -91,10 +93,12 @@ class ExtendedSecurityGroupPropertiesMixin(object):
properties = NsxExtendedSecurityGroupProperties(
security_group_id=sg_res['id'],
logging=sg_req.get(sg_logging.LOGGING, False),
provider=sg_req.get(provider_sg.PROVIDER, False))
provider=sg_req.get(provider_sg.PROVIDER, False),
policy=sg_req.get(sg_policy.POLICY))
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)
sg_res[sg_policy.POLICY] = sg_req.get(sg_policy.POLICY)
def _get_security_group_properties(self, context, security_group_id):
with context.session.begin(subtransactions=True):
@ -108,13 +112,20 @@ class ExtendedSecurityGroupPropertiesMixin(object):
def _process_security_group_properties_update(self, context,
sg_res, sg_req):
if (sg_logging.LOGGING in sg_req
if ((sg_logging.LOGGING in sg_req
and (sg_req[sg_logging.LOGGING] !=
sg_res.get(sg_logging.LOGGING, False))):
sg_res.get(sg_logging.LOGGING, False))) or
(sg_policy.POLICY in sg_req
and (sg_req[sg_policy.POLICY] !=
sg_res.get(sg_policy.POLICY)))):
prop = self._get_security_group_properties(context, sg_res['id'])
with context.session.begin(subtransactions=True):
prop.update({sg_logging.LOGGING: sg_req[sg_logging.LOGGING]})
sg_res[sg_logging.LOGGING] = sg_req[sg_logging.LOGGING]
prop.update({
sg_logging.LOGGING: sg_req.get(sg_logging.LOGGING, False),
sg_policy.POLICY: sg_req.get(sg_policy.POLICY)})
sg_res[sg_logging.LOGGING] = sg_req.get(sg_logging.LOGGING, False)
sg_res[sg_policy.POLICY] = sg_req.get(sg_policy.POLICY)
def _is_security_group_logged(self, context, security_group_id):
prop = self._get_security_group_properties(context, security_group_id)
@ -125,6 +136,11 @@ class ExtendedSecurityGroupPropertiesMixin(object):
security_group_id)
return sg_prop.provider
def _is_policy_security_group(self, context, security_group_id):
sg_prop = self._get_security_group_properties(context,
security_group_id)
return True if sg_prop.policy else False
def _check_provider_security_group_exists(self, context,
security_group_id):
# NOTE(roeyc): We want to retrieve the security-group info by calling
@ -251,10 +267,17 @@ class ExtendedSecurityGroupPropertiesMixin(object):
sg_id):
raise provider_sg.ProviderSecurityGroupDeleteNotAdmin(id=sg_id)
def _prevent_non_admin_delete_policy_sg(self, context, sg_id):
# Only someone who is an admin is allowed to delete this.
if not context.is_admin and self._is_policy_security_group(context,
sg_id):
raise sg_policy.PolicySecurityGroupDeleteNotAdmin(id=sg_id)
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
sg_res[sg_policy.POLICY] = sg_db.ext_properties.policy
def _extend_port_dict_provider_security_group(self, port_res, port_db):
# NOTE(arosen): this method overrides the one in the base

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

@ -1 +1 @@
7b5ec3caa9a4
e816d4fe9d4f

33
vmware_nsx/db/migration/alembic_migrations/versions/ocata/expand/e816d4fe9d4f_nsx_add_policy_security_group.py Normal file
View File

@ -0,0 +1,33 @@
# 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 'policy' attribute to security-group
Revision ID: e816d4fe9d4f
Revises: 7b5ec3caa9a4
Create Date: 2016-10-06 11:30:31.263918
"""
# revision identifiers, used by Alembic.
revision = 'e816d4fe9d4f'
down_revision = '7b5ec3caa9a4'
from alembic import op
import sqlalchemy as sa
def upgrade():
op.add_column('nsx_extended_security_group_properties',
sa.Column('policy', sa.String(36)))

68
vmware_nsx/extensions/securitygrouppolicy.py Normal file
View File

@ -0,0 +1,68 @@
# 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_lib import exceptions as nexception
POLICY = 'policy'
RESOURCE_ATTRIBUTE_MAP = {
'security_groups': {
POLICY: {
'allow_post': True,
'allow_put': True,
'enforce_policy': True,
'is_visible': True,
'default': None}
}
}
class PolicySecurityGroupDeleteNotAdmin(nexception.NotAuthorized):
message = _("Security group %(id)s is a policy security group and "
"requires an admin to delete it.")
class Securitygrouppolicy(extensions.ExtensionDescriptor):
"""Security group policy extension."""
@classmethod
def get_name(cls):
return "Security group policy"
@classmethod
def get_alias(cls):
return "security-group-policy"
@classmethod
def get_description(cls):
return "Security group policy extension."
@classmethod
def get_updated(cls):
return "2016-10-06T10: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 RESOURCE_ATTRIBUTE_MAP
else:
return {}

162
vmware_nsx/plugins/nsx_v/plugin.py
View File

@ -101,6 +101,7 @@ from vmware_nsx.extensions import providersecuritygroup as provider_sg
from vmware_nsx.extensions import routersize
from vmware_nsx.extensions import secgroup_rule_local_ip_prefix
from vmware_nsx.extensions import securitygrouplogging as sg_logging
from vmware_nsx.extensions import securitygrouppolicy as sg_policy
from vmware_nsx.plugins.nsx_v import availability_zones as nsx_az
from vmware_nsx.plugins.nsx_v import managers
from vmware_nsx.plugins.nsx_v import md_proxy as nsx_v_md_proxy
@ -212,9 +213,24 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
self.nsx_v)
self._availability_zones_data = nsx_az.ConfiguredAvailabilityZones()
self._validate_config()
self._use_nsx_policies = False
if cfg.CONF.nsxv.use_nsx_policies:
if not c_utils.is_nsxv_version_6_2(self.nsx_v.vcns.get_version()):
error = (_("NSX policies are not supported for version "
"%(ver)s.") %
{'ver': self.nsx_v.vcns.get_version()})
raise nsx_exc.NsxPluginException(err_msg=error)
# Support NSX policies in default security groups
self._use_nsx_policies = True
# enable the extension
self.supported_extension_aliases.append("security-group-policy")
self.sg_container_id = self._create_security_group_container()
self.default_section = self._create_cluster_default_fw_section()
self._process_security_groups_rules_logging()
self._router_managers = managers.RouterTypeManager(self)
if cfg.CONF.nsxv.use_dvs_features:
@ -713,9 +729,6 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
return list(set(
dvs.strip() for dvs in physical_network.split(',') if dvs))
def _get_default_security_group(self, context, tenant_id):
return self._ensure_default_security_group(context, tenant_id)
def _add_member_to_security_group(self, sg_id, vnic_id):
with locking.LockManager.get_lock('neutron-security-ops' + str(sg_id)):
try:
@ -3021,6 +3034,21 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
context,
securitygroup):
nsx_sg_id = self._create_nsx_security_group(context, securitygroup)
if self._use_nsx_policies:
# When using policies - no rules should be created.
# just add the security group to the policy on the backend.
self._update_nsx_security_group_policies(
securitygroup[sg_policy.POLICY], None, nsx_sg_id)
# Delete the neutron default rules (do not exist on the backend)
if securitygroup.get(ext_sg.SECURITYGROUPRULES):
with context.session.begin(subtransactions=True):
for rule in securitygroup[ext_sg.SECURITYGROUPRULES]:
rule_db = self._get_security_group_rule(context,
rule['id'])
context.session.delete(rule_db)
securitygroup.pop(ext_sg.SECURITYGROUPRULES)
else:
try:
self._create_fw_section_for_security_group(
context, securitygroup, nsx_sg_id)
@ -3030,14 +3058,61 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
if not securitygroup[provider_sg.PROVIDER]:
# Add Security Group to the Security Groups container in order to
# apply the default block rule. provider security-groups should not
# have a default blocking rule.
self._add_member_to_security_group(self.sg_container_id, nsx_sg_id)
# apply the default block rule.
# This is relevant for policies security groups too.
# provider security-groups should not have a default blocking rule.
self._add_member_to_security_group(self.sg_container_id,
nsx_sg_id)
def _validate_security_group(self, security_group, default_sg,
from_create=True):
if self._use_nsx_policies:
new_policy = None
if from_create:
# called from create_security_group
# must have a policy:
if not security_group.get(sg_policy.POLICY):
if default_sg:
# For default sg the default policy will be used
security_group[sg_policy.POLICY] = (
cfg.CONF.nsxv.default_policy_id)
else:
msg = _('A security group must be assigned to a '
'policy')
raise n_exc.InvalidInput(error_message=msg)
#TODO(asarfaty): add support for tenant sg with rules
new_policy = security_group[sg_policy.POLICY]
else:
# called from update_security_group
if sg_policy.POLICY in security_group:
new_policy = security_group[sg_policy.POLICY]
if not new_policy:
msg = _('A security group must be assigned to a '
'policy')
raise n_exc.InvalidInput(error_message=msg)
#TODO(asarfaty): add support for tenant sg with rules
# validate that the new policy exists
if new_policy and not self.nsx_v.vcns.validate_inventory(
new_policy):
msg = _('Policy %s was not found on the NSX') % new_policy
raise n_exc.InvalidInput(error_message=msg)
# Do not support logging with policy
if security_group.get(sg_logging.LOGGING):
msg = _('Cannot support logging when using NSX policies')
raise n_exc.InvalidInput(error_message=msg)
else:
# must not have a policy:
if security_group.get(sg_policy.POLICY):
msg = _('The security group cannot be assigned to a policy')
raise n_exc.InvalidInput(error_message=msg)
def create_security_group(self, context, security_group, default_sg=False):
"""Create a security group."""
sg_data = security_group['security_group']
sg_id = sg_data["id"] = str(uuid.uuid4())
self._validate_security_group(sg_data, default_sg, from_create=True)
with context.session.begin(subtransactions=True):
if sg_data.get(provider_sg.PROVIDER):
@ -3061,10 +3136,48 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
context = context.elevated()
super(NsxVPluginV2, self).delete_security_group(context, sg_id)
LOG.exception(_LE('Failed to create security group'))
return new_sg
def _update_security_group_with_policy(self, updated_group,
sg_data, nsx_sg_id):
"""Handle security group update when using NSX policies
Remove the security group from the old policies, and apply on the new
policies
"""
# Verify that the policy was not removed from the security group
if (sg_policy.POLICY in updated_group and
not updated_group[sg_policy.POLICY]):
msg = _('It is not allowed to remove the policy from security '
'group %s') % nsx_sg_id
raise n_exc.InvalidInput(error_message=msg)
if (updated_group.get(sg_policy.POLICY) and
updated_group[sg_policy.POLICY] != sg_data[sg_policy.POLICY]):
new_policy = updated_group[sg_policy.POLICY]
old_policy = sg_data[sg_policy.POLICY]
self._update_nsx_security_group_policies(
new_policy, old_policy, nsx_sg_id)
def _update_nsx_security_group_policies(self, new_policy, old_policy,
nsx_sg_id):
# update the NSX security group to use this policy
if old_policy:
with locking.LockManager.get_lock(
'neutron-security-policy-' + str(old_policy)):
self.nsx_sg_utils.del_nsx_security_group_from_policy(
old_policy, nsx_sg_id)
with locking.LockManager.get_lock(
'neutron-security-policy-' + str(new_policy)):
self.nsx_sg_utils.add_nsx_security_group_to_policy(
new_policy, nsx_sg_id)
def update_security_group(self, context, id, security_group):
s = security_group['security_group']
self._validate_security_group(s, False, from_create=False)
nsx_sg_id = nsx_db.get_nsx_security_group_id(context.session, id)
section_uri = self._get_section_uri(context.session, id)
section_needs_update = False
@ -3073,30 +3186,51 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
context, id, security_group)
# Reflect security-group name or description changes in the backend,
# dfw section name needs to be updated as well.
h, c = self.nsx_v.vcns.get_section(section_uri)
section = self.nsx_sg_utils.parse_section(c)
if set(['name', 'description']) & set(s.keys()):
nsx_sg_name = self.nsx_sg_utils.get_nsx_sg_name(sg_data)
section_name = self.nsx_sg_utils.get_nsx_section_name(sg_data)
self.nsx_v.vcns.update_security_group(
nsx_sg_id, nsx_sg_name, sg_data['description'])
# security groups with NSX policy - update the backend policy attached
# to the security group
if self._use_nsx_policies and sg_policy.POLICY in sg_data:
self._update_security_group_with_policy(s, sg_data, nsx_sg_id)
if self._use_nsx_policies:
# The rest of the update are not relevant to policies security
# groups as there is no matching section
self._process_security_group_properties_update(
context, sg_data, s)
return sg_data
# Get the backend section matching this security group
h, c = self.nsx_v.vcns.get_section(section_uri)
section = self.nsx_sg_utils.parse_section(c)
# dfw section name needs to be updated if the sg name was modified
if 'name' in s.keys():
section.attrib['name'] = section_name
section_needs_update = True
# Update the dfw section if security-group logging option has changed.
log_all_rules = cfg.CONF.nsxv.log_security_groups_allowed_traffic
self._process_security_group_properties_update(context, sg_data, s)
if not log_all_rules and context.is_admin:
section_needs_update |= self.nsx_sg_utils.set_rules_logged_option(
section, sg_data[sg_logging.LOGGING])
if section_needs_update:
# update the section with all the modifications
self.nsx_v.vcns.update_section(
section_uri, self.nsx_sg_utils.to_xml_string(section), h)
return sg_data
def delete_security_group(self, context, id):
"""Delete a security group."""
self._prevent_non_admin_delete_provider_sg(context, id)
self._prevent_non_admin_delete_policy_sg(context, id)
try:
# Find nsx rule sections
section_uri = self._get_section_uri(context.session, id)
@ -3196,6 +3330,13 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
"""
sg_rules = security_group_rules['security_group_rules']
sg_id = sg_rules[0]['security_group_rule']['security_group_id']
if self._use_nsx_policies:
# If policies are enabled - creating rules is forbidden
msg = _('Cannot create rules cannot for security group %s with'
' a policy') % sg_id
raise n_exc.InvalidInput(error_message=msg)
self._prevent_non_admin_delete_provider_sg(context, sg_id)
ruleids = set()
@ -3380,6 +3521,9 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
for res in az_resources:
inventory.append((res, 'availability_zones'))
if cfg.CONF.nsxv.default_policy_id:
inventory.append((cfg.CONF.nsxv.default_policy_id, 'policy'))
for moref, field in inventory:
if moref and not self.nsx_v.vcns.validate_inventory(moref):
error = _("Configured %s not found") % field

37
vmware_nsx/plugins/nsx_v/vshield/securitygroup_utils.py
View File

@ -17,6 +17,8 @@ import xml.etree.ElementTree as et
from oslo_log import log as logging
from vmware_nsx.common import utils
WAIT_INTERVAL = 2000
MAX_ATTEMPTS = 5
@ -165,3 +167,38 @@ class NsxSecurityGroupUtils(object):
rule.attrib['logged'] = value
updated = True
return updated
def del_nsx_security_group_from_policy(self, policy_id, sg_id):
if not policy_id:
return
policy = self.nsxv_manager.vcns.get_security_policy(policy_id)
policy = utils.normalize_xml(policy)
# check if the security group is already bounded to the policy
for binding in policy.iter('securityGroupBinding'):
if binding.find('objectId').text == sg_id:
# delete this entry
policy.remove(binding)
return self.nsxv_manager.vcns.update_security_policy(
policy_id, et.tostring(policy))
def add_nsx_security_group_to_policy(self, policy_id, sg_id):
if not policy_id:
return
# Get the policy configuration
policy = self.nsxv_manager.vcns.get_security_policy(policy_id)
policy = utils.normalize_xml(policy)
# check if the security group is already bounded to the policy
for binding in policy.iter('securityGroupBinding'):
if binding.find('objectId').text == sg_id:
# Already there
return
# Add a new binding entry
new_binding = et.SubElement(policy, 'securityGroupBinding')
et.SubElement(new_binding, 'objectId').text = sg_id
return self.nsxv_manager.vcns.update_security_policy(
policy_id, et.tostring(policy))

14
vmware_nsx/plugins/nsx_v/vshield/vcns.py
View File

@ -51,6 +51,7 @@ SPOOFGUARD_PREFIX = '/api/4.0/services/spoofguard'
TRUSTSTORE_PREFIX = '%s/%s' % (SERVICES_PREFIX, 'truststore')
EXCLUDELIST_PREFIX = '/api/2.1/app/excludelist'
SERVICE_INSERTION_PROFILE_PREFIX = '/api/2.0/si/serviceprofile'
SECURITY_POLICY_PREFIX = '/api/2.0/services/policy/securitypolicy'
#LbaaS Constants
LOADBALANCER_SERVICE = "loadbalancer/config"
@ -984,3 +985,16 @@ class Vcns(object):
uri = '%s/%s/%s/%s/%s' % (SERVICES_PREFIX, IPAM_POOL_SERVICE, pool_id,
'ipaddresses', ip_addr)
return self.do_request(HTTP_DELETE, uri)
def get_security_policy(self, policy_id):
# get the policy configuration as an xml string
uri = '%s/%s' % (SECURITY_POLICY_PREFIX, policy_id)
h, policy = self.do_request(HTTP_GET, uri, format='xml', decode=False)
return policy
def update_security_policy(self, policy_id, request):
# update the policy configuration. request should be an xml string
uri = '%s/%s' % (SECURITY_POLICY_PREFIX, policy_id)
return self.do_request(HTTP_PUT, uri, request,
format='xml',
decode=False, encode=True)

123
vmware_nsx/tests/unit/extensions/test_security_group_policy.py Normal file
View File

@ -0,0 +1,123 @@
# 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.
import mock
from oslo_config import cfg
import webob.exc
from neutron.api.v2 import attributes as attr
from neutron import context
from neutron.tests.unit.extensions import test_securitygroup
from vmware_nsx.extensions import securitygrouppolicy as ext_policy
from vmware_nsx.tests.unit.nsx_v import test_plugin
from vmware_nsx.tests.unit.nsx_v.vshield import fake_vcns
PLUGIN_NAME = 'vmware_nsx.plugin.NsxVPlugin'
class SecGroupPolicyExtensionTestCase(
test_plugin.NsxVPluginV2TestCase,
test_securitygroup.SecurityGroupDBTestCase):
def setUp(self, plugin=PLUGIN_NAME, ext_mgr=None):
cfg.CONF.set_override('use_nsx_policies', True, group='nsxv')
cfg.CONF.set_override('default_policy_id', 'policy-1', group='nsxv')
# This feature is enabled only since 6.2
with mock.patch.object(fake_vcns.FakeVcns,
'get_version',
return_value="6.2.3"):
super(SecGroupPolicyExtensionTestCase, self).setUp(
plugin=plugin, ext_mgr=ext_mgr)
self._tenant_id = 'foobar'
# add policy security group attribute
attr.RESOURCE_ATTRIBUTE_MAP['security_groups'].update(
ext_policy.RESOURCE_ATTRIBUTE_MAP['security_groups'])
def tearDown(self):
# remove policy security group attribute
del attr.RESOURCE_ATTRIBUTE_MAP['security_groups']['policy']
super(SecGroupPolicyExtensionTestCase, self).tearDown()
def _create_secgroup_with_policy(self, policy_id):
body = {'security_group': {'name': 'sg-policy',
'tenant_id': self._tenant_id,
'policy': policy_id}}
security_group_req = self.new_create_request('security-groups', body)
return security_group_req.get_response(self.ext_api)
def test_secgroup_create_with_policy(self):
policy_id = 'policy-5'
res = self._create_secgroup_with_policy(policy_id)
sg = self.deserialize(self.fmt, res)
self.assertEqual(policy_id, sg['security_group']['policy'])
def test_secgroup_create_without_policy(self):
res = self._create_secgroup_with_policy(None)
self.assertEqual(400, res.status_int)
def test_secgroup_create_with_illegal_policy(self):
with mock.patch.object(fake_vcns.FakeVcns,
'validate_inventory',
return_value=False):
policy_id = 'bad-policy'
res = self._create_secgroup_with_policy(policy_id)
self.assertEqual(400, res.status_int)
def test_secgroup_update_with_policy(self):
old_policy = 'policy-5'
new_policy = 'policy-6'
res = self._create_secgroup_with_policy(old_policy)
sg = self.deserialize(self.fmt, res)
data = {'security_group': {'policy': new_policy}}
req = self.new_update_request('security-groups', data,
sg['security_group']['id'])
updated_sg = self.deserialize(self.fmt, req.get_response(self.ext_api))
self.assertEqual(new_policy, updated_sg['security_group']['policy'])
def test_secgroup_update_no_policy_change(self):
old_policy = 'policy-5'
res = self._create_secgroup_with_policy(old_policy)
sg = self.deserialize(self.fmt, res)
data = {'security_group': {'description': 'abc'}}
req = self.new_update_request('security-groups', data,
sg['security_group']['id'])
updated_sg = self.deserialize(self.fmt, req.get_response(self.ext_api))
self.assertEqual(old_policy, updated_sg['security_group']['policy'])
def test_secgroup_update_remove_policy(self):
old_policy = 'policy-5'
new_policy = None
res = self._create_secgroup_with_policy(old_policy)
sg = self.deserialize(self.fmt, res)
data = {'security_group': {'policy': new_policy}}
req = self.new_update_request('security-groups', data,
sg['security_group']['id'])
res = req.get_response(self.ext_api)
self.assertEqual(400, res.status_int)
def test_non_admin_cannot_delete_policy_sg_and_admin_can(self):
policy_id = 'policy-5'
res = self._create_secgroup_with_policy(policy_id)
sg = self.deserialize(self.fmt, res)
sg_id = sg['security_group']['id']
# Try deleting the request as a normal user returns forbidden
# as a tenant is not allowed to delete this.
ctx = context.Context('', self._tenant_id)
self._delete('security-groups', sg_id,
expected_code=webob.exc.HTTPForbidden.code,
neutron_context=ctx)
# can be deleted though as admin
self._delete('security-groups', sg_id,
expected_code=webob.exc.HTTPNoContent.code)

10
vmware_nsx/tests/unit/nsx_v/vshield/fake_vcns.py
View File

@ -1341,3 +1341,13 @@ class FakeVcns(object):
response = self._get_bad_req_response(
msg, 120054, 'core-services')
return self.return_helper(header, response)
def get_security_policy(self, policy_id):
response_text = (
"<?xml version=\"1.0\" encoding=\"UTF-8\"?>"
"<securityPolicy><objectId>%s</objectId>"
"</securityPolicy>") % policy_id
return response_text
def update_security_policy(self, policy_id, request):
pass