text: # Copyright 2012, Nachi Ueno, NTT MCL, 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.
  
  import netaddr
  from neutron_lib import constants as const
  from neutron_lib.utils import helpers
  from oslo_log import log as logging
  
  from neutron._i18n import _
  from neutron.callbacks import events
  from neutron.callbacks import registry
  from neutron.callbacks import resources
  from neutron.db import api as db_api
  from neutron.db.models import allowed_address_pair as aap_models
  from neutron.db.models import securitygroup as sg_models
  from neutron.db import models_v2
  from neutron.db import securitygroups_db as sg_db
  from neutron.extensions import securitygroup as ext_sg
  
  LOG = logging.getLogger(__name__)
  
  
  DIRECTION_IP_PREFIX = {'ingress': 'source_ip_prefix',
                         'egress': 'dest_ip_prefix'}
  
  DHCP_RULE_PORT = {4: (67, 68, const.IPv4), 6: (547, 546, const.IPv6)}
  
  
  @registry.has_registry_receivers
  class SecurityGroupServerNotifierRpcMixin(sg_db.SecurityGroupDbMixin):
      """Mixin class to add agent-based security group implementation."""
  
      @registry.receives(resources.PORT, [events.AFTER_CREATE,
                                          events.AFTER_UPDATE,
                                          events.AFTER_DELETE])
      def notify_sg_on_port_change(self, resource, event, trigger, context,
                                   port, *args, **kwargs):
          """Trigger notification to other SG members on port changes."""
          self.notify_security_groups_member_updated(context, port)
  
      def create_security_group_rule(self, context, security_group_rule):
          rule = super(SecurityGroupServerNotifierRpcMixin,
                       self).create_security_group_rule(context,
                                                        security_group_rule)
          sgids = [rule['security_group_id']]
          self.notifier.security_groups_rule_updated(context, sgids)
          return rule
  
      def create_security_group_rule_bulk(self, context, security_group_rules):
          rules = super(SecurityGroupServerNotifierRpcMixin,
                        self).create_security_group_rule_bulk_native(
                            context, security_group_rules)
          sgids = set([r['security_group_id'] for r in rules])
          self.notifier.security_groups_rule_updated(context, list(sgids))
          return rules
  
      def delete_security_group_rule(self, context, sgrid):
          rule = self.get_security_group_rule(context, sgrid)
          super(SecurityGroupServerNotifierRpcMixin,
                self).delete_security_group_rule(context, sgrid)
          self.notifier.security_groups_rule_updated(context,
                                                     [rule['security_group_id']])
  
      def check_and_notify_security_group_member_changed(
              self, context, original_port, updated_port):
          sg_change = not helpers.compare_elements(
              original_port.get(ext_sg.SECURITYGROUPS),
              updated_port.get(ext_sg.SECURITYGROUPS))
          if sg_change:
              self.notify_security_groups_member_updated_bulk(
                  context, [original_port, updated_port])
          elif original_port['fixed_ips'] != updated_port['fixed_ips']:
              self.notify_security_groups_member_updated(context, updated_port)
  
      def is_security_group_member_updated(self, context,
                                           original_port, updated_port):
          """Check security group member updated or not.
  
          This method returns a flag which indicates request notification
          is required and does not perform notification itself.
          It is because another changes for the port may require notification.
          """
          need_notify = False
          if (original_port['fixed_ips'] != updated_port['fixed_ips'] or
              original_port['mac_address'] != updated_port['mac_address'] or
              not helpers.compare_elements(
                  original_port.get(ext_sg.SECURITYGROUPS),
                  updated_port.get(ext_sg.SECURITYGROUPS))):
              need_notify = True
          return need_notify
  
      def notify_security_groups_member_updated_bulk(self, context, ports):
          """Notify update event of security group members for ports.
  
          The agent setups the iptables rule to allow
          ingress packet from the dhcp server (as a part of provider rules),
          so we need to notify an update of dhcp server ip
          address to the plugin agent.
          """
          sec_groups = set()
          for port in ports:
              # NOTE (Swami): ROUTER_INTERFACE_OWNERS check is required
              # since it includes the legacy router interface device owners
              # and DVR router interface device owners.
              if (port['device_owner'] not in
                  [const.DEVICE_OWNER_DHCP, const.ROUTER_INTERFACE_OWNERS]):
                  sec_groups |= set(port.get(ext_sg.SECURITYGROUPS))
  
          if sec_groups:
              self.notifier.security_groups_member_updated(
                  context, list(sec_groups))
  
      def notify_security_groups_member_updated(self, context, port):
          self.notify_security_groups_member_updated_bulk(context, [port])
  
  
  class SecurityGroupInfoAPIMixin(object):
      """API for retrieving security group info for SG agent code."""
  
      def get_port_from_device(self, context, device):
          """Get port dict from device name on an agent.
  
          Subclass must provide this method or get_ports_from_devices.
  
          :param device: device name which identifies a port on the agent side.
          What is specified in "device" depends on a plugin agent implementation.
          For example, it is a port ID in OVS agent and netdev name in Linux
          Bridge agent.
          :return: port dict returned by DB plugin get_port(). In addition,
          it must contain the following fields in the port dict returned.
          - device
          - security_groups
          - security_group_rules,
          - security_group_source_groups
          - fixed_ips
          """
          raise NotImplementedError(_("%s must implement get_port_from_device "
                                      "or get_ports_from_devices.")
                                    % self.__class__.__name__)
  
      def get_ports_from_devices(self, context, devices):
          """Bulk method of get_port_from_device.
  
          Subclasses may override this to provide better performance for DB
          queries, backend calls, etc.
          """
          return [self.get_port_from_device(context, device)
                  for device in devices]
  
      def security_group_info_for_ports(self, context, ports):
          sg_info = {'devices': ports,
                     'security_groups': {},
                     'sg_member_ips': {}}
          rules_in_db = self._select_rules_for_ports(context, ports)
          remote_security_group_info = {}
          for (port_id, rule_in_db) in rules_in_db:
              remote_gid = rule_in_db.get('remote_group_id')
              security_group_id = rule_in_db.get('security_group_id')
              ethertype = rule_in_db['ethertype']
              if ('security_group_source_groups'
                  not in sg_info['devices'][port_id]):
                  sg_info['devices'][port_id][
                      'security_group_source_groups'] = []
  
              if remote_gid:
                  if (remote_gid
                      not in sg_info['devices'][port_id][
                          'security_group_source_groups']):
                      sg_info['devices'][port_id][
                          'security_group_source_groups'].append(remote_gid)
                  if remote_gid not in remote_security_group_info:
                      remote_security_group_info[remote_gid] = {}
                  if ethertype not in remote_security_group_info[remote_gid]:
                      # this set will be serialized into a list by rpc code
                      remote_security_group_info[remote_gid][ethertype] = set()
  
              direction = rule_in_db['direction']
              rule_dict = {
                  'direction': direction,
                  'ethertype': ethertype}
  
              for key in ('protocol', 'port_range_min', 'port_range_max',
                          'remote_ip_prefix', 'remote_group_id'):
                  if rule_in_db.get(key) is not None:
                      if key == 'remote_ip_prefix':
                          direction_ip_prefix = DIRECTION_IP_PREFIX[direction]
                          rule_dict[direction_ip_prefix] = rule_in_db[key]
                          continue
                      rule_dict[key] = rule_in_db[key]
              if security_group_id not in sg_info['security_groups']:
                  sg_info['security_groups'][security_group_id] = []
              if rule_dict not in sg_info['security_groups'][security_group_id]:
                  sg_info['security_groups'][security_group_id].append(
                      rule_dict)
          # Update the security groups info if they don't have any rules
          sg_ids = self._select_sg_ids_for_ports(context, ports)
          for (sg_id, ) in sg_ids:
              if sg_id not in sg_info['security_groups']:
                  sg_info['security_groups'][sg_id] = []
  
          sg_info['sg_member_ips'] = remote_security_group_info
          # the provider rules do not belong to any security group, so these
          # rules still reside in sg_info['devices'] [port_id]
          self._apply_provider_rule(context, sg_info['devices'])
  
          return self._get_security_group_member_ips(context, sg_info)
  
      def _get_security_group_member_ips(self, context, sg_info):
          ips = self._select_ips_for_remote_group(
              context, sg_info['sg_member_ips'].keys())
          for sg_id, member_ips in ips.items():
              for ip in member_ips:
                  ethertype = 'IPv%d' % netaddr.IPNetwork(ip).version
                  if ethertype in sg_info['sg_member_ips'][sg_id]:
                      sg_info['sg_member_ips'][sg_id][ethertype].add(ip)
          return sg_info
  
      def _select_remote_group_ids(self, ports):
          remote_group_ids = []
          for port in ports.values():
              for rule in port.get('security_group_rules'):
                  remote_group_id = rule.get('remote_group_id')
                  if remote_group_id:
                      remote_group_ids.append(remote_group_id)
          return remote_group_ids
  
      def _convert_remote_group_id_to_ip_prefix(self, context, ports):
          remote_group_ids = self._select_remote_group_ids(ports)
          ips = self._select_ips_for_remote_group(context, remote_group_ids)
          for port in ports.values():
              updated_rule = []
              for rule in port.get('security_group_rules'):
                  remote_group_id = rule.get('remote_group_id')
                  direction = rule.get('direction')
                  direction_ip_prefix = DIRECTION_IP_PREFIX[direction]
                  if not remote_group_id:
                      updated_rule.append(rule)
                      continue
  
                  port['security_group_source_groups'].append(remote_group_id)
                  base_rule = rule
                  for ip in ips[remote_group_id]:
                      if ip in port.get('fixed_ips', []):
                          continue
                      ip_rule = base_rule.copy()
                      version = netaddr.IPNetwork(ip).version
                      ethertype = 'IPv%s' % version
                      if base_rule['ethertype'] != ethertype:
                          continue
                      ip_rule[direction_ip_prefix] = str(
                          netaddr.IPNetwork(ip).cidr)
                      updated_rule.append(ip_rule)
              port['security_group_rules'] = updated_rule
          return ports
  
      def _add_ingress_dhcp_rule(self, port):
          for ip_version in (4, 6):
              # only allow DHCP servers to talk to the appropriate IP address
              # to avoid getting leases that don't match the Neutron IPs
              prefix = '32' if ip_version == 4 else '128'
              dests = ['%s/%s' % (ip, prefix) for ip in port['fixed_ips']
                       if netaddr.IPNetwork(ip).version == ip_version]
              if ip_version == 4:
                  # v4 dhcp servers can also talk to broadcast
                  dests.append('255.255.255.255/32')
              elif ip_version == 6:
                  # v6 dhcp responses can target link-local addresses
                  dests.append('fe80::/64')
              source_port, dest_port, ethertype = DHCP_RULE_PORT[ip_version]
              for dest in dests:
                  dhcp_rule = {'direction': 'ingress',
                               'ethertype': ethertype,
                               'protocol': 'udp',
                               'port_range_min': dest_port,
                               'port_range_max': dest_port,
                               'source_port_range_min': source_port,
                               'source_port_range_max': source_port,
                               'dest_ip_prefix': dest}
                  port['security_group_rules'].append(dhcp_rule)
  
      def _add_ingress_ra_rule(self, port):
          has_v6 = [ip for ip in port['fixed_ips']
                    if netaddr.IPNetwork(ip).version == 6]
          if not has_v6:
              return
          ra_rule = {'direction': 'ingress',
                     'ethertype': const.IPv6,
                     'protocol': const.PROTO_NAME_IPV6_ICMP,
                     'source_port_range_min': const.ICMPV6_TYPE_RA}
          port['security_group_rules'].append(ra_rule)
  
      def _apply_provider_rule(self, context, ports):
          for port in ports.values():
              self._add_ingress_ra_rule(port)
              self._add_ingress_dhcp_rule(port)
  
      def security_group_rules_for_ports(self, context, ports):
          rules_in_db = self._select_rules_for_ports(context, ports)
          for (port_id, rule_in_db) in rules_in_db:
              port = ports[port_id]
              direction = rule_in_db['direction']
              rule_dict = {
                  'security_group_id': rule_in_db['security_group_id'],
                  'direction': direction,
                  'ethertype': rule_in_db['ethertype'],
              }
              for key in ('protocol', 'port_range_min', 'port_range_max',
                          'remote_ip_prefix', 'remote_group_id'):
                  if rule_in_db.get(key) is not None:
                      if key == 'remote_ip_prefix':
                          direction_ip_prefix = DIRECTION_IP_PREFIX[direction]
                          rule_dict[direction_ip_prefix] = rule_in_db[key]
                          continue
                      rule_dict[key] = rule_in_db[key]
              port['security_group_rules'].append(rule_dict)
          self._apply_provider_rule(context, ports)
          return self._convert_remote_group_id_to_ip_prefix(context, ports)
  
      def _select_ips_for_remote_group(self, context, remote_group_ids):
          """Get all IP addresses (including allowed addr pairs) for each sg.
  
          Return dict of lists of IPs keyed by group_id.
          """
          raise NotImplementedError()
  
      def _select_rules_for_ports(self, context, ports):
          """Get all security group rules associated with a list of ports.
  
          Return list of tuples of (port_id, sg_rule)
          """
          raise NotImplementedError()
  
      def _select_sg_ids_for_ports(self, context, ports):
          """Return security group IDs for a list of ports.
  
          Return list of tuples with a single element of sg_id.
          """
          raise NotImplementedError()
  
  
  class SecurityGroupServerRpcMixin(SecurityGroupInfoAPIMixin,
                                    SecurityGroupServerNotifierRpcMixin):
      """Server-side RPC mixin using DB for SG notifications and responses."""
  
      @db_api.retry_if_session_inactive()
      def _select_sg_ids_for_ports(self, context, ports):
          if not ports:
              return []
          sg_binding_port = sg_models.SecurityGroupPortBinding.port_id
          sg_binding_sgid = sg_models.SecurityGroupPortBinding.security_group_id
          query = context.session.query(sg_binding_sgid)
          query = query.filter(sg_binding_port.in_(ports.keys()))
          return query.all()
  
      @db_api.retry_if_session_inactive()
      def _select_rules_for_ports(self, context, ports):
          if not ports:
              return []
          sg_binding_port = sg_models.SecurityGroupPortBinding.port_id
          sg_binding_sgid = sg_models.SecurityGroupPortBinding.security_group_id
  
          sgr_sgid = sg_models.SecurityGroupRule.security_group_id
  
          query = context.session.query(sg_binding_port,
                                        sg_models.SecurityGroupRule)
          query = query.join(sg_models.SecurityGroupRule,
                             sgr_sgid == sg_binding_sgid)
          query = query.filter(sg_binding_port.in_(ports.keys()))
          return query.all()
  
      @db_api.retry_if_session_inactive()
      def _select_ips_for_remote_group(self, context, remote_group_ids):
          ips_by_group = {}
          if not remote_group_ids:
              return ips_by_group
          for remote_group_id in remote_group_ids:
              ips_by_group[remote_group_id] = set()
  
          ip_port = models_v2.IPAllocation.port_id
          sg_binding_port = sg_models.SecurityGroupPortBinding.port_id
          sg_binding_sgid = sg_models.SecurityGroupPortBinding.security_group_id
  
          # Join the security group binding table directly to the IP allocation
          # table instead of via the Port table skip an unnecessary intermediary
          query = context.session.query(sg_binding_sgid,
                                        models_v2.IPAllocation.ip_address,
                                        aap_models.AllowedAddressPair.ip_address)
          query = query.join(models_v2.IPAllocation,
                             ip_port == sg_binding_port)
          # Outerjoin because address pairs may be null and we still want the
          # IP for the port.
          query = query.outerjoin(
              aap_models.AllowedAddressPair,
              sg_binding_port == aap_models.AllowedAddressPair.port_id)
          query = query.filter(sg_binding_sgid.in_(remote_group_ids))
          # Each allowed address pair IP record for a port beyond the 1st
          # will have a duplicate regular IP in the query response since
          # the relationship is 1-to-many. Dedup with a set
          for security_group_id, ip_address, allowed_addr_ip in query:
              ips_by_group[security_group_id].add(ip_address)
              if allowed_addr_ip:
                  ips_by_group[security_group_id].add(allowed_addr_ip)
          return ips_by_group
  