import logging
import enum
-from neutronclient.common.exceptions import NotFound
+from neutronclient.common.exceptions import NotFound, Conflict
+
+from snaps.openstack.openstack_creator import OpenStackNetworkObject
from snaps.openstack.utils import keystone_utils
from snaps.openstack.utils import neutron_utils
logger = logging.getLogger('OpenStackSecurityGroup')
-class OpenStackSecurityGroup:
+class OpenStackSecurityGroup(OpenStackNetworkObject):
"""
- Class responsible for creating Security Groups
+ Class responsible for managing a Security Group in OpenStack
"""
def __init__(self, os_creds, sec_grp_settings):
:param os_creds: The credentials to connect with OpenStack
:param sec_grp_settings: The settings used to create a security group
"""
- self.__os_creds = os_creds
+ super(self.__class__, self).__init__(os_creds)
+
self.sec_grp_settings = sec_grp_settings
- self.__neutron = None
- self.__keystone = None
# Attributes instantiated on create()
self.__security_group = None
# dict where the rule settings object is the key
self.__rules = dict()
- def create(self, cleanup=False):
+ def initialize(self):
+ """
+ Loads existing security group.
+ :return: the security group domain object
+ """
+ super(self.__class__, self).initialize()
+
+ self.__security_group = neutron_utils.get_security_group(
+ self._neutron, sec_grp_settings=self.sec_grp_settings)
+ if self.__security_group:
+ # Populate rules
+ existing_rules = neutron_utils.get_rules_by_security_group(
+ self._neutron, self.__security_group)
+
+ for existing_rule in existing_rules:
+ # For Custom Rules
+ rule_setting = self.__get_setting_from_rule(existing_rule)
+ self.__rules[rule_setting] = existing_rule
+
+ return self.__security_group
+
+ def create(self):
"""
Responsible for creating the security group.
- :param cleanup: Denotes whether or not this is being called for cleanup
- :return: the OpenStack security group object
+ :return: the security group domain object
"""
- self.__neutron = neutron_utils.neutron_client(self.__os_creds)
- self.__keystone = keystone_utils.keystone_client(self.__os_creds)
+ self.initialize()
- logger.info(
- 'Creating security group %s...' % self.sec_grp_settings.name)
+ if not self.__security_group:
+ logger.info(
+ 'Creating security group %s...' % self.sec_grp_settings.name)
- self.__security_group = neutron_utils.get_security_group(
- self.__neutron, self.sec_grp_settings.name)
- if not self.__security_group and not cleanup:
- # Create the security group
+ keystone = keystone_utils.keystone_client(self._os_creds)
self.__security_group = neutron_utils.create_security_group(
- self.__neutron, self.__keystone,
+ self._neutron, keystone,
self.sec_grp_settings)
# Get the rules added for free
auto_rules = neutron_utils.get_rules_by_security_group(
- self.__neutron, self.__security_group)
+ self._neutron, self.__security_group)
ctr = 0
for auto_rule in auto_rules:
# Create the custom rules
for sec_grp_rule_setting in self.sec_grp_settings.rule_settings:
- custom_rule = neutron_utils.create_security_group_rule(
- self.__neutron, sec_grp_rule_setting)
- self.__rules[sec_grp_rule_setting] = custom_rule
+ try:
+ custom_rule = neutron_utils.create_security_group_rule(
+ self._neutron, sec_grp_rule_setting)
+ self.__rules[sec_grp_rule_setting] = custom_rule
+ except Conflict as e:
+ logger.warn('Unable to create rule due to conflict - %s',
+ e)
# Refresh security group object to reflect the new rules added
self.__security_group = neutron_utils.get_security_group(
- self.__neutron, self.sec_grp_settings.name)
- else:
- # Populate rules
- existing_rules = neutron_utils.get_rules_by_security_group(
- self.__neutron, self.__security_group)
-
- for existing_rule in existing_rules:
- # For Custom Rules
- rule_setting = self.__get_setting_from_rule(existing_rule)
- ctr = 0
- if not rule_setting:
- # For Free Rules
- rule_setting = self.__generate_rule_setting(existing_rule)
- ctr += 1
-
- self.__rules[rule_setting] = existing_rule
+ self._neutron, sec_grp_settings=self.sec_grp_settings)
return self.__security_group
:return: the newly instantiated SecurityGroupRuleSettings object
"""
sec_grp = neutron_utils.get_security_group_by_id(
- self.__neutron, rule.security_group_id)
+ self._neutron, rule.security_group_id)
setting = SecurityGroupRuleSettings(
description=rule.description,
"""
for setting, rule in self.__rules.items():
try:
- neutron_utils.delete_security_group_rule(self.__neutron, rule)
+ neutron_utils.delete_security_group_rule(self._neutron, rule)
except NotFound as e:
logger.warning('Rule not found, cannot delete - ' + str(e))
pass
if self.__security_group:
try:
- neutron_utils.delete_security_group(self.__neutron,
+ neutron_utils.delete_security_group(self._neutron,
self.__security_group)
except NotFound as e:
logger.warning(
:param rule_setting: the rule configuration
"""
rule_setting.sec_grp_name = self.sec_grp_settings.name
- new_rule = neutron_utils.create_security_group_rule(self.__neutron,
+ new_rule = neutron_utils.create_security_group_rule(self._neutron,
rule_setting)
self.__rules[rule_setting] = new_rule
self.sec_grp_settings.rule_settings.append(rule_setting)
if rule_id or rule_setting:
if rule_id:
rule_to_remove = neutron_utils.get_rule_by_id(
- self.__neutron, self.__security_group, rule_id)
+ self._neutron, self.__security_group, rule_id)
elif rule_setting:
rule_to_remove = self.__rules.get(rule_setting)
if rule_to_remove:
- neutron_utils.delete_security_group_rule(self.__neutron,
+ neutron_utils.delete_security_group_rule(self._neutron,
rule_to_remove)
rule_setting = self.__get_setting_from_rule(rule_to_remove)
if rule_setting:
if isinstance(rule_setting, SecurityGroupRuleSettings):
self.rule_settings.append(rule_setting)
else:
+ rule_setting['sec_grp_name'] = self.name
self.rule_settings.append(SecurityGroupRuleSettings(
**rule_setting))
out['protocol'] = self.protocol.name
if self.sec_grp_name:
sec_grp = neutron_utils.get_security_group(
- neutron, self.sec_grp_name)
+ neutron, sec_grp_name=self.sec_grp_name)
if sec_grp:
out['security_group_id'] = sec_grp.id
else:
Code Review / snaps.git / blobdiff