sfa/openstack/security_group.py

   1 from sfa.util.sfalogging import logger
   2
   3
   4 class SecurityGroup:
   5
   6     def __init__(self, driver):
   7         self.client = driver.shell.nova_manager
   8
   9     def create_security_group(self, name):
  10         try:
  11             self.client.security_groups.create(name=name, description=name)
  12         except Exception as ex:
  13             logger.log_exc("Failed to add security group")
  14             raise
  15
  16     def delete_security_group(self, name):
  17         try:
  18             security_group = self.client.security_groups.find(name=name)
  19             self.client.security_groups.delete(security_group.id)
  20         except Exception as ex:
  21             logger.log_exc("Failed to delete security group")
  22
  23     def _validate_port_range(self, port_range):
  24         from_port = to_port = None
  25         if isinstance(port_range, str):
  26             ports = port_range.split(':')
  27             if len(ports) > 1:
  28                 from_port = int(ports[0])
  29                 to_port = int(ports[1])
  30             else:
  31                 from_port = to_port = int(ports[0])
  32         return (from_port, to_port)
  33
  34     def _validate_icmp_type_code(self, icmp_type_code):
  35         from_port = to_port = None
  36         if isinstance(icmp_type_code, str):
  37             code_parts = icmp_type_code.split(':')
  38             if len(code_parts) > 1:
  39                 try:
  40                     from_port = int(code_parts[0])
  41                     to_port = int(code_parts[1])
  42                 except ValueError:
  43                     logger.error('port must be an integer.')
  44         return (from_port, to_port)
  45
  46     def add_rule_to_group(self, group_name=None, protocol='tcp', cidr_ip='0.0.0.0/0',
  47                           port_range=None, icmp_type_code=None,
  48                           source_group_name=None, source_group_owner_id=None):
  49
  50         try:
  51             from_port, to_port = self._validate_port_range(port_range)
  52             icmp_type = self._validate_icmp_type_code(icmp_type_code)
  53             if icmp_type and icmp_type[0] and icmp_type[1]:
  54                 from_port, to_port = icmp_type[0], icmp_type[1]
  55
  56             group = self.client.security_groups.find(name=group_name)
  57             self.client.security_group_rules.create(group.id,
  58                                                     protocol, from_port, to_port, cidr_ip)
  59         except Exception as ex:
  60             logger.log_exc("Failed to add rule to group %s" % group_name)
  61
  62     def remove_rule_from_group(self, group_name=None, protocol='tcp', cidr_ip='0.0.0.0/0',
  63                                port_range=None, icmp_type_code=None,
  64                                source_group_name=None, source_group_owner_id=None):
  65         try:
  66             from_port, to_port = self._validate_port_range(port_range)
  67             icmp_type = self._validate_icmp_type_code(icmp_type_code)
  68             if icmp_type:
  69                 from_port, to_port = icmp_type[0], icmp_type[1]
  70             group = self.client.security_groups.find(name=group_name)
  71             filter = {
  72                 'id': group.id,
  73                 'from_port': from_port,
  74                 'to_port': to_port,
  75                 'cidr_ip': ip,
  76                 'ip_protocol': protocol,
  77             }
  78             rule = self.client.security_group_rules.find(**filter)
  79             if rule:
  80                 self.client.security_group_rules.delete(rule)
  81         except Exception as ex:
  82             logger.log_exc("Failed to remove rule from group %s" % group_name)