OpenStack Networking (Neutron)
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 

429 lines
18 KiB

# 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 collections
from neutron_lib.agent import topics
from neutron_lib.callbacks import events
from neutron_lib.callbacks import registry
from neutron_lib import constants
from neutron_lib.plugins import directory
from neutron_lib import rpc as n_rpc
from neutron_lib.utils import net
from oslo_log import log as logging
import oslo_messaging
from oslo_utils import versionutils
from neutron.api.rpc.callbacks import resources
from neutron.api.rpc.handlers import resources_rpc
from neutron.db import securitygroups_rpc_base as sg_rpc_base
LOG = logging.getLogger(__name__)
class SecurityGroupServerRpcApi(object):
"""RPC client for security group methods in the plugin.
This class implements the client side of an rpc interface. This interface
is used by agents to call security group related methods implemented on the
plugin side. The other side of this interface is defined in
SecurityGroupServerRpcCallback. For more information about changing rpc
interfaces, see doc/source/contributor/internals/rpc_api.rst.
"""
def __init__(self, topic):
target = oslo_messaging.Target(
topic=topic, version='1.0',
namespace=constants.RPC_NAMESPACE_SECGROUP)
self.client = n_rpc.get_client(target)
def security_group_rules_for_devices(self, context, devices):
LOG.debug("Get security group rules "
"for devices via rpc %r", devices)
cctxt = self.client.prepare(version='1.1')
return cctxt.call(context, 'security_group_rules_for_devices',
devices=devices)
def security_group_info_for_devices(self, context, devices):
LOG.debug("Get security group information for devices via rpc %r",
devices)
call_version = '1.3'
cctxt = self.client.prepare(version=call_version)
return cctxt.call(context, 'security_group_info_for_devices',
devices=devices,
call_version=call_version)
class SecurityGroupServerRpcCallback(object):
"""Callback for SecurityGroup agent RPC in plugin implementations.
This class implements the server side of an rpc interface. The client side
can be found in SecurityGroupServerRpcApi. For more information on changing
rpc interfaces, see doc/source/contributor/internals/rpc_api.rst.
"""
# API version history:
# 1.1 - Initial version
# 1.2 - security_group_info_for_devices introduced as an optimization
# 1.3 - security_group_info_for_devices returns member_ips with new
# structure.
# NOTE: target must not be overridden in subclasses
# to keep RPC API version consistent across plugins.
target = oslo_messaging.Target(version='1.3',
namespace=constants.RPC_NAMESPACE_SECGROUP)
@property
def plugin(self):
return directory.get_plugin()
def _get_devices_info(self, context, devices):
return dict(
(port['id'], port)
for port in self.plugin.get_ports_from_devices(context, devices)
if port and not net.is_port_trusted(port)
)
def security_group_rules_for_devices(self, context, **kwargs):
"""Callback method to return security group rules for each port.
also convert remote_group_id rule
to source_ip_prefix and dest_ip_prefix rule
:params devices: list of devices
:returns: port correspond to the devices with security group rules
"""
devices_info = kwargs.get('devices')
ports = self._get_devices_info(context, devices_info)
return self.plugin.security_group_rules_for_ports(context, ports)
def security_group_info_for_devices(self, context, **kwargs):
"""Return security group information for requested devices.
:params devices: list of devices
:returns:
sg_info{
'security_groups': {sg_id: [rule1, rule2]}
'sg_member_ips': {sg_id: {'IPv4': set(), 'IPv6': set()}}
'devices': {device_id: {device_info}}
}
Note that sets are serialized into lists by rpc code.
"""
# The original client RPC version was 1.2 before this change.
call_version = kwargs.pop("call_version", '1.2')
_target_version = versionutils.convert_version_to_tuple(call_version)
devices_info = kwargs.get('devices')
ports = self._get_devices_info(context, devices_info)
sg_info = self.plugin.security_group_info_for_ports(context, ports)
if _target_version < (1, 3):
LOG.warning("RPC security_group_info_for_devices call has "
"inconsistent version between server and agents. "
"The server supports RPC version is 1.3 while "
"the agent is %s.", call_version)
return self.make_compatible_sg_member_ips(sg_info)
return sg_info
def make_compatible_sg_member_ips(self, sg_info):
sg_member_ips = sg_info.get('sg_member_ips', {})
sg_ids = sg_member_ips.keys()
for sg_id in sg_ids:
member_ips = sg_member_ips.get(sg_id, {})
ipv4_ips = member_ips.get("IPv4", set())
comp_ipv4_ips = set([ip for ip, _mac in ipv4_ips])
ipv6_ips = member_ips.get("IPv6", set())
comp_ipv6_ips = set([ip for ip, _mac in ipv6_ips])
comp_ips = {"IPv4": comp_ipv4_ips,
"IPv6": comp_ipv6_ips}
sg_member_ips[sg_id] = comp_ips
sg_info['sg_member_ips'] = sg_member_ips
return sg_info
class SecurityGroupAgentRpcApiMixin(object):
"""RPC client for security group methods to the agent.
This class implements the client side of an rpc interface. This interface
is used by plugins to call security group methods implemented on the
agent side. The other side of this interface can be found in
SecurityGroupAgentRpcCallbackMixin. For more information about changing
rpc interfaces, see doc/source/contributor/internals/rpc_api.rst.
"""
# history
# 1.1 Support Security Group RPC
SG_RPC_VERSION = "1.1"
def _get_security_group_topic(self):
return topics.get_topic_name(self.topic,
topics.SECURITY_GROUP,
topics.UPDATE)
def security_groups_rule_updated(self, context, security_groups):
"""Notify rule updated security groups."""
if not security_groups:
return
cctxt = self.client.prepare(version=self.SG_RPC_VERSION,
topic=self._get_security_group_topic(),
fanout=True)
cctxt.cast(context, 'security_groups_rule_updated',
security_groups=security_groups)
def security_groups_member_updated(self, context, security_groups):
"""Notify member updated security groups."""
if not security_groups:
return
cctxt = self.client.prepare(version=self.SG_RPC_VERSION,
topic=self._get_security_group_topic(),
fanout=True)
cctxt.cast(context, 'security_groups_member_updated',
security_groups=security_groups)
class SecurityGroupAgentRpcCallbackMixin(object):
"""A mix-in that enable SecurityGroup support in agent implementations.
This class implements the server side of an rpc interface. The client side
can be found in SecurityGroupAgentRpcApiMixin. For more information on
changing rpc interfaces, see doc/source/contributor/internals/rpc_api.rst.
The sg_agent reference implementation is available in neutron/agent
"""
# mix-in object should be have sg_agent
sg_agent = None
def _security_groups_agent_not_set(self):
LOG.warning("Security group agent binding currently not set. "
"This should be set by the end of the init "
"process.")
def security_groups_rule_updated(self, context, **kwargs):
"""Callback for security group rule update.
:param security_groups: list of updated security_groups
"""
security_groups = kwargs.get('security_groups', [])
LOG.debug("Security group rule updated on remote: %s",
security_groups)
if not self.sg_agent:
return self._security_groups_agent_not_set()
self.sg_agent.security_groups_rule_updated(security_groups)
def security_groups_member_updated(self, context, **kwargs):
"""Callback for security group member update.
:param security_groups: list of updated security_groups
"""
security_groups = kwargs.get('security_groups', [])
LOG.debug("Security group member updated on remote: %s",
security_groups)
if not self.sg_agent:
return self._security_groups_agent_not_set()
self.sg_agent.security_groups_member_updated(security_groups)
class SecurityGroupServerAPIShim(sg_rpc_base.SecurityGroupInfoAPIMixin):
"""Agent-side replacement for SecurityGroupServerRpcApi using local data.
This provides the same methods as SecurityGroupServerRpcApi but it reads
from the updates delivered to the push notifications cache rather than
calling the server.
"""
def __init__(self, rcache):
self.rcache = rcache
registry.subscribe(self._clear_child_sg_rules, 'SecurityGroup',
events.AFTER_DELETE)
registry.subscribe(self._add_child_sg_rules, 'SecurityGroup',
events.AFTER_UPDATE)
# set this attr so agent can adjust the timeout of the client
self.client = resources_rpc.ResourcesPullRpcApi().client
def register_legacy_sg_notification_callbacks(self, sg_agent):
self._sg_agent = sg_agent
registry.subscribe(self._handle_sg_rule_delete,
'SecurityGroupRule', events.AFTER_DELETE)
registry.subscribe(self._handle_sg_rule_update,
'SecurityGroupRule', events.AFTER_UPDATE)
registry.subscribe(self._handle_sg_member_delete,
'Port', events.AFTER_DELETE)
registry.subscribe(self._handle_sg_member_update,
'Port', events.AFTER_UPDATE)
self._register_legacy_ag_notification_callbacks(sg_agent)
def _register_legacy_ag_notification_callbacks(self, sg_agent):
for event in (events.AFTER_UPDATE, events.AFTER_DELETE):
registry.subscribe(self._handle_address_group_event,
resources.ADDRESSGROUP, event)
def security_group_info_for_devices(self, context, devices):
ports = self._get_devices_info(context, devices)
result = self.security_group_info_for_ports(context, ports)
return result
def security_group_rules_for_devices(self, context, devices):
# this is the legacy method that should never be called since
# security_group_info_for_devices will never throw an unsupported
# error.
raise NotImplementedError()
def get_secgroup_ids_for_address_group(self, address_group_id):
filters = {'remote_address_group_id': (address_group_id, )}
return set([rule.security_group_id for rule in
self.rcache.get_resources('SecurityGroupRule', filters)])
def _add_child_sg_rules(self, rtype, event, trigger, payload):
# whenever we receive a full security group, add all child rules
# because the server won't emit events for the individual rules on
# creation.
context = payload.context
updated = payload.latest_state
for rule in updated.rules:
self.rcache.record_resource_update(context, 'SecurityGroupRule',
rule)
def _clear_child_sg_rules(self, rtype, event, trigger, payload):
context = payload.context
existing = payload.states[0]
if not existing:
return
# the server can delete an entire security group without notifying
# about the security group rules. so we need to emulate a rule deletion
# when a security group is removed.
filters = {'security_group_id': (existing.id, )}
for rule in self.rcache.get_resources('SecurityGroupRule', filters):
self.rcache.record_resource_delete(context, 'SecurityGroupRule',
rule.id)
def _handle_sg_rule_delete(self, rtype, event, trigger, payload):
existing = payload.states[0]
if not existing:
return
sg_id = existing.security_group_id
self._sg_agent.security_groups_rule_updated([sg_id])
def _handle_sg_rule_update(self, rtype, event, trigger, payload):
updated = payload.latest_state
sg_id = updated.security_group_id
self._sg_agent.security_groups_rule_updated([sg_id])
def _handle_sg_member_delete(self, rtype, event, trigger, payload):
# received on port delete
existing = payload.states[0]
sgs = set(existing.security_group_ids) if existing else set()
if sgs:
self._sg_agent.security_groups_member_updated(sgs)
def _handle_sg_member_update(self, rtype, event, trigger, payload):
# received on port update
existing = payload.states[0]
updated = payload.latest_state
changed_fields = payload.metadata['changed_fields']
sgs = set(existing.security_group_ids) if existing else set()
if not changed_fields.intersection({'security_group_ids', 'fixed_ips',
'allowed_address_pairs'}):
# none of the relevant fields to SG calculations changed
return
sgs.update(set(updated.security_group_ids))
if sgs:
self._sg_agent.security_groups_member_updated(sgs)
def _handle_address_group_event(self, rtype, event, trigger, payload):
resource_id = payload.resource_id
if event == events.AFTER_UPDATE:
self._sg_agent.address_group_updated(resource_id)
else:
self._sg_agent.address_group_deleted(resource_id)
def _get_devices_info(self, context, devices):
# NOTE(kevinbenton): this format is required by the sg code, it is
# defined in get_port_from_device and mimics
# make_port_dict_with_security_groups in ML2 db
result = {}
for device in devices:
ovo = self.rcache.get_resource_by_id('Port', device)
if not ovo:
continue
port = ovo.to_dict()
# the caller expects trusted ports to be excluded from the result
if net.is_port_trusted(port):
continue
port['security_groups'] = list(ovo.security_group_ids)
port['security_group_rules'] = []
port['security_group_source_groups'] = []
port['security_group_remote_address_groups'] = []
port['fixed_ips'] = [str(f['ip_address'])
for f in port['fixed_ips']]
# NOTE(kevinbenton): this id==device is only safe for OVS. a lookup
# will be required for linux bridge and others that don't have the
# full port UUID
port['device'] = port['id']
port['port_security_enabled'] = getattr(
ovo.security, 'port_security_enabled', True)
result[device] = port
return result
def _select_ips_for_remote_group(self, context, remote_group_ids):
if not remote_group_ids:
return {}
ips_by_group = {rg: set() for rg in remote_group_ids}
filters = {'security_group_ids': tuple(remote_group_ids)}
for p in self.rcache.get_resources('Port', filters):
allowed_ips = [(str(addr.ip_address), str(addr.mac_address))
for addr in p.allowed_address_pairs]
port_ips = [(str(addr.ip_address), str(p.mac_address))
for addr in p.fixed_ips] + allowed_ips
for sg_id in p.security_group_ids:
if sg_id in ips_by_group:
ips_by_group[sg_id].update(set(port_ips))
return ips_by_group
def _select_ips_for_remote_address_group(self, context,
remote_address_group_ids):
if not remote_address_group_ids:
return {}
ips_by_group = {ag: set() for ag in remote_address_group_ids}
for ag_id in remote_address_group_ids:
ag = self.rcache.get_resource_by_id('AddressGroup', ag_id)
# NOTE(hangyang) Expected data structure is (ip, mac) tuple
ips = [(str(addr_assoc.address), None)
for addr_assoc in ag.addresses]
ips_by_group[ag_id].update(set(ips))
return ips_by_group
def _select_rules_for_ports(self, context, ports):
if not ports:
return []
results = []
sg_ids = set((sg_id for p in ports.values()
for sg_id in p['security_group_ids']))
rules_by_sgid = collections.defaultdict(list)
for sg_id in sg_ids:
filters = {'security_group_id': (sg_id, )}
for r in self.rcache.get_resources('SecurityGroupRule', filters):
rules_by_sgid[r.security_group_id].append(r)
for p in ports.values():
for sg_id in p['security_group_ids']:
for rule in rules_by_sgid[sg_id]:
results.append((p['id'], rule.to_dict()))
return results
def _select_sg_ids_for_ports(self, context, ports):
sg_ids = set((sg_id for p in ports.values()
for sg_id in p['security_group_ids']))
return [(sg_id, ) for sg_id in sg_ids]
def _is_security_group_stateful(self, context, sg_id):
sg = self.rcache.get_resource_by_id(resources.SECURITYGROUP, sg_id)
return sg.stateful