x
/
vmware-nsx
Code Issues Proposed changes

Refactor provider SG validation 

- Prevent non-admin user from changing a provider SG (in addition to delete,
add rule & delete rule which were already prevented)
- rename the validation method and error
(please note - preventing SG creation is done with a policy.json rule)

Change-Id: Idcd1c6c7082b1bd26d0fbc19a399e01ecbf2fb0f
This commit is contained in:
Adit Sarfaty 2018-09-17 08:53:30 +03:00
parent bfcd1c9414
commit 37be04703a
5 changed files with 16 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
vmware_nsx/db/extended_security_group.py
View File

@ -337,11 +337,11 @@ class ExtendedSecurityGroupPropertiesMixin(object):
updated_port[provider_sg.PROVIDER_SECURITYGROUPS])
return provider_sg_changed
def _prevent_non_admin_delete_provider_sg(self, context, sg_id):
# Only someone who is an admin is allowed to delete this.
def _prevent_non_admin_edit_provider_sg(self, context, sg_id):
# Only someone who is an admin is allowed to modify a provider sg.
if not context.is_admin and self._is_provider_security_group(context,
sg_id):
raise provider_sg.ProviderSecurityGroupDeleteNotAdmin(id=sg_id)
raise provider_sg.ProviderSecurityGroupEditNotAdmin(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.

4
vmware_nsx/extensions/providersecuritygroup.py
View File

@ -61,9 +61,9 @@ class DefaultSecurityGroupIsNotProvider(nexception.InvalidInput):
"security-group.")
class ProviderSecurityGroupDeleteNotAdmin(nexception.NotAuthorized):
class ProviderSecurityGroupEditNotAdmin(nexception.NotAuthorized):
message = _("Security group %(id)s is a provider security group and "
"requires an admin to delete it.")
"requires an admin to modify it.")
class Providersecuritygroup(extensions.ExtensionDescriptor):

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

@ -4267,6 +4267,7 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
def update_security_group(self, context, id, security_group):
s = security_group['security_group']
self._validate_security_group(context, s, False, id=id)
self._prevent_non_admin_edit_provider_sg(context, id)
nsx_sg_id = nsx_db.get_nsx_security_group_id(context.session, id,
moref=True)
section_uri = self._get_section_uri(context.session, id)
@ -4323,7 +4324,7 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
def delete_security_group(self, context, id, delete_base=True):
"""Delete a security group."""
self._prevent_non_admin_delete_provider_sg(context, id)
self._prevent_non_admin_edit_provider_sg(context, id)
self._prevent_non_admin_delete_policy_sg(context, id)
policy = self._get_security_group_policy(context, id)
try:
@ -4441,7 +4442,7 @@ 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']
self._prevent_non_admin_delete_provider_sg(context, sg_id)
self._prevent_non_admin_edit_provider_sg(context, sg_id)
ruleids = set()
nsx_rules = []
@ -4517,7 +4518,7 @@ class NsxVPluginV2(addr_pair_db.AllowedAddressPairsMixin,
"""Delete a security group rule."""
rule_db = self._get_security_group_rule(context, id)
security_group_id = rule_db['security_group_id']
self._prevent_non_admin_delete_provider_sg(context, security_group_id)
self._prevent_non_admin_edit_provider_sg(context, security_group_id)
# Get the nsx rule from neutron DB and delete it
nsx_rule_id = nsxv_db.get_nsx_rule_id(context.session, id)

7
vmware_nsx/plugins/nsx_v3/plugin.py
View File

@ -4812,6 +4812,7 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
def update_security_group(self, context, id, security_group):
orig_secgroup = self.get_security_group(
context, id, fields=['id', 'name', 'description'])
self._prevent_non_admin_edit_provider_sg(context, id)
with db_api.context_manager.writer.using(context):
secgroup_res = (
super(NsxV3Plugin, self).update_security_group(context, id,
@ -4835,7 +4836,7 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
return secgroup_res
def delete_security_group(self, context, id):
self._prevent_non_admin_delete_provider_sg(context, id)
self._prevent_non_admin_edit_provider_sg(context, id)
nsgroup_id, section_id = nsx_db.get_sg_mappings(
context.session, id)
super(NsxV3Plugin, self).delete_security_group(context, id)
@ -4870,7 +4871,7 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
# group rules being added are part of the same security
# group. We should be validating that this is the case though...
sg_id = sg_rules[0]['security_group_rule']['security_group_id']
self._prevent_non_admin_delete_provider_sg(context, sg_id)
self._prevent_non_admin_edit_provider_sg(context, sg_id)
security_group = self.get_security_group(
context, sg_id)
@ -4900,7 +4901,7 @@ class NsxV3Plugin(agentschedulers_db.AZDhcpAgentSchedulerDbMixin,
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_delete_provider_sg(context, sg_id)
self._prevent_non_admin_edit_provider_sg(context, sg_id)
nsgroup_id, section_id = nsx_db.get_sg_mappings(context.session, sg_id)
fw_rule_id = nsx_db.get_sg_rule_mapping(context.session, id)
self.nsxlib.firewall_section.delete_rule(section_id, fw_rule_id)

6
vmware_nsx/tests/unit/extensions/test_provider_security_groups.py
View File

@ -106,20 +106,20 @@ class ProviderSecurityGroupTestPlugin(
return port_data
def delete_security_group(self, context, id):
self._prevent_non_admin_delete_provider_sg(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_delete_provider_sg(context, sg_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_delete_provider_sg(context, id)
self._prevent_non_admin_edit_provider_sg(context, id)
return super(ProviderSecurityGroupTestPlugin,
self).create_security_group_rule(context,
security_group_rule)