neutron/agent/metadata/agent.py

   1 # Copyright 2012 New Dream Network, LLC (DreamHost)
   2 #
   3 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   4 #    not use this file except in compliance with the License. You may obtain
   5 #    a copy of the License at
   6 #
   7 #         http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 #    Unless required by applicable law or agreed to in writing, software
  10 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  11 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  12 #    License for the specific language governing permissions and limitations
  13 #    under the License.
  14
  15 import hashlib
  16 import hmac
  17
  18 import httplib2
  19 from oslo_config import cfg
  20 from oslo_log import log as logging
  21 import oslo_messaging
  22 from oslo_service import loopingcall
  23 import six
  24 import six.moves.urllib.parse as urlparse
  25 import webob
  26
  27 from neutron._i18n import _, _LE, _LW
  28 from neutron.agent.linux import utils as agent_utils
  29 from neutron.agent.metadata import config
  30 from neutron.agent import rpc as agent_rpc
  31 from neutron.common import constants as n_const
  32 from neutron.common import rpc as n_rpc
  33 from neutron.common import topics
  34 from neutron.common import utils
  35 from neutron import context
  36 from neutron.openstack.common.cache import cache
  37
  38 LOG = logging.getLogger(__name__)
  39
  40 MODE_MAP = {
  41     config.USER_MODE: 0o644,
  42     config.GROUP_MODE: 0o664,
  43     config.ALL_MODE: 0o666,
  44 }
  45
  46
  47 class MetadataPluginAPI(object):
  48     """Agent-side RPC for metadata agent-to-plugin interaction.
  49
  50     This class implements the client side of an rpc interface used by the
  51     metadata service to make calls back into the Neutron plugin.  The server
  52     side is defined in
  53     neutron.api.rpc.handlers.metadata_rpc.MetadataRpcCallback.  For more
  54     information about changing rpc interfaces, see
  55     doc/source/devref/rpc_api.rst.
  56
  57     API version history:
  58         1.0 - Initial version.
  59     """
  60
  61     def __init__(self, topic):
  62         target = oslo_messaging.Target(
  63             topic=topic,
  64             namespace=n_const.RPC_NAMESPACE_METADATA,
  65             version='1.0')
  66         self.client = n_rpc.get_client(target)
  67
  68     def get_ports(self, context, filters):
  69         cctxt = self.client.prepare()
  70         return cctxt.call(context, 'get_ports', filters=filters)
  71
  72
  73 class MetadataProxyHandler(object):
  74
  75     def __init__(self, conf):
  76         self.conf = conf
  77         if self.conf.cache_url:
  78             self._cache = cache.get_cache(self.conf.cache_url)
  79         else:
  80             self._cache = False
  81
  82         self.plugin_rpc = MetadataPluginAPI(topics.PLUGIN)
  83         self.context = context.get_admin_context_without_session()
  84
  85     @webob.dec.wsgify(RequestClass=webob.Request)
  86     def __call__(self, req):
  87         try:
  88             LOG.debug("Request: %s", req)
  89
  90             instance_id, tenant_id = self._get_instance_and_tenant_id(req)
  91             if instance_id:
  92                 return self._proxy_request(instance_id, tenant_id, req)
  93             else:
  94                 return webob.exc.HTTPNotFound()
  95
  96         except Exception:
  97             LOG.exception(_LE("Unexpected error."))
  98             msg = _('An unknown error has occurred. '
  99                     'Please try your request again.')
 100             explanation = six.text_type(msg)
 101             return webob.exc.HTTPInternalServerError(explanation=explanation)
 102
 103     def _get_ports_from_server(self, router_id=None, ip_address=None,
 104                                networks=None):
 105         """Get ports from server."""
 106         filters = self._get_port_filters(router_id, ip_address, networks)
 107         return self.plugin_rpc.get_ports(self.context, filters)
 108
 109     def _get_port_filters(self, router_id=None, ip_address=None,
 110                           networks=None):
 111         filters = {}
 112         if router_id:
 113             filters['device_id'] = [router_id]
 114             filters['device_owner'] = n_const.ROUTER_INTERFACE_OWNERS
 115         if ip_address:
 116             filters['fixed_ips'] = {'ip_address': [ip_address]}
 117         if networks:
 118             filters['network_id'] = networks
 119
 120         return filters
 121
 122     @utils.cache_method_results
 123     def _get_router_networks(self, router_id):
 124         """Find all networks connected to given router."""
 125         internal_ports = self._get_ports_from_server(router_id=router_id)
 126         return tuple(p['network_id'] for p in internal_ports)
 127
 128     @utils.cache_method_results
 129     def _get_ports_for_remote_address(self, remote_address, networks):
 130         """Get list of ports that has given ip address and are part of
 131         given networks.
 132
 133         :param networks: list of networks in which the ip address will be
 134                          searched for
 135
 136         """
 137         return self._get_ports_from_server(networks=networks,
 138                                            ip_address=remote_address)
 139
 140     def _get_ports(self, remote_address, network_id=None, router_id=None):
 141         """Search for all ports that contain passed ip address and belongs to
 142         given network.
 143
 144         If no network is passed ports are searched on all networks connected to
 145         given router. Either one of network_id or router_id must be passed.
 146
 147         """
 148         if network_id:
 149             networks = (network_id,)
 150         elif router_id:
 151             networks = self._get_router_networks(router_id)
 152         else:
 153             raise TypeError(_("Either one of parameter network_id or router_id"
 154                               " must be passed to _get_ports method."))
 155
 156         return self._get_ports_for_remote_address(remote_address, networks)
 157
 158     def _get_instance_and_tenant_id(self, req):
 159         remote_address = req.headers.get('X-Forwarded-For')
 160         network_id = req.headers.get('X-Neutron-Network-ID')
 161         router_id = req.headers.get('X-Neutron-Router-ID')
 162
 163         ports = self._get_ports(remote_address, network_id, router_id)
 164
 165         if len(ports) == 1:
 166             return ports[0]['device_id'], ports[0]['tenant_id']
 167         return None, None
 168
 169     def _proxy_request(self, instance_id, tenant_id, req):
 170         headers = {
 171             'X-Forwarded-For': req.headers.get('X-Forwarded-For'),
 172             'X-Instance-ID': instance_id,
 173             'X-Tenant-ID': tenant_id,
 174             'X-Instance-ID-Signature': self._sign_instance_id(instance_id)
 175         }
 176
 177         nova_ip_port = '%s:%s' % (self.conf.nova_metadata_ip,
 178                                   self.conf.nova_metadata_port)
 179         url = urlparse.urlunsplit((
 180             self.conf.nova_metadata_protocol,
 181             nova_ip_port,
 182             req.path_info,
 183             req.query_string,
 184             ''))
 185
 186         h = httplib2.Http(
 187             ca_certs=self.conf.auth_ca_cert,
 188             disable_ssl_certificate_validation=self.conf.nova_metadata_insecure
 189         )
 190         if self.conf.nova_client_cert and self.conf.nova_client_priv_key:
 191             h.add_certificate(self.conf.nova_client_priv_key,
 192                               self.conf.nova_client_cert,
 193                               nova_ip_port)
 194         resp, content = h.request(url, method=req.method, headers=headers,
 195                                   body=req.body)
 196
 197         if resp.status == 200:
 198             LOG.debug(str(resp))
 199             req.response.content_type = resp['content-type']
 200             req.response.body = content
 201             return req.response
 202         elif resp.status == 403:
 203             LOG.warn(_LW(
 204                 'The remote metadata server responded with Forbidden. This '
 205                 'response usually occurs when shared secrets do not match.'
 206             ))
 207             return webob.exc.HTTPForbidden()
 208         elif resp.status == 400:
 209             return webob.exc.HTTPBadRequest()
 210         elif resp.status == 404:
 211             return webob.exc.HTTPNotFound()
 212         elif resp.status == 409:
 213             return webob.exc.HTTPConflict()
 214         elif resp.status == 500:
 215             msg = _(
 216                 'Remote metadata server experienced an internal server error.'
 217             )
 218             LOG.warn(msg)
 219             explanation = six.text_type(msg)
 220             return webob.exc.HTTPInternalServerError(explanation=explanation)
 221         else:
 222             raise Exception(_('Unexpected response code: %s') % resp.status)
 223
 224     def _sign_instance_id(self, instance_id):
 225         secret = self.conf.metadata_proxy_shared_secret
 226         if isinstance(secret, six.text_type):
 227             secret = secret.encode('utf-8')
 228         if isinstance(instance_id, six.text_type):
 229             instance_id = instance_id.encode('utf-8')
 230         return hmac.new(secret, instance_id, hashlib.sha256).hexdigest()
 231
 232
 233 class UnixDomainMetadataProxy(object):
 234
 235     def __init__(self, conf):
 236         self.conf = conf
 237         agent_utils.ensure_directory_exists_without_file(
 238             cfg.CONF.metadata_proxy_socket)
 239         self._init_state_reporting()
 240
 241     def _init_state_reporting(self):
 242         self.context = context.get_admin_context_without_session()
 243         self.state_rpc = agent_rpc.PluginReportStateAPI(topics.REPORTS)
 244         self.agent_state = {
 245             'binary': 'neutron-metadata-agent',
 246             'host': cfg.CONF.host,
 247             'topic': 'N/A',
 248             'configurations': {
 249                 'metadata_proxy_socket': cfg.CONF.metadata_proxy_socket,
 250                 'nova_metadata_ip': cfg.CONF.nova_metadata_ip,
 251                 'nova_metadata_port': cfg.CONF.nova_metadata_port,
 252                 'log_agent_heartbeats': cfg.CONF.AGENT.log_agent_heartbeats,
 253             },
 254             'start_flag': True,
 255             'agent_type': n_const.AGENT_TYPE_METADATA}
 256         report_interval = cfg.CONF.AGENT.report_interval
 257         if report_interval:
 258             self.heartbeat = loopingcall.FixedIntervalLoopingCall(
 259                 self._report_state)
 260             self.heartbeat.start(interval=report_interval)
 261
 262     def _report_state(self):
 263         try:
 264             self.state_rpc.report_state(
 265                 self.context,
 266                 self.agent_state,
 267                 use_call=self.agent_state.get('start_flag'))
 268         except AttributeError:
 269             # This means the server does not support report_state
 270             LOG.warn(_LW('Neutron server does not support state report.'
 271                          ' State report for this agent will be disabled.'))
 272             self.heartbeat.stop()
 273             return
 274         except Exception:
 275             LOG.exception(_LE("Failed reporting state!"))
 276             return
 277         self.agent_state.pop('start_flag', None)
 278
 279     def _get_socket_mode(self):
 280         mode = self.conf.metadata_proxy_socket_mode
 281         if mode == config.DEDUCE_MODE:
 282             user = self.conf.metadata_proxy_user
 283             if (not user or user == '0' or user == 'root'
 284                     or agent_utils.is_effective_user(user)):
 285                 # user is agent effective user or root => USER_MODE
 286                 mode = config.USER_MODE
 287             else:
 288                 group = self.conf.metadata_proxy_group
 289                 if not group or agent_utils.is_effective_group(group):
 290                     # group is agent effective group => GROUP_MODE
 291                     mode = config.GROUP_MODE
 292                 else:
 293                     # otherwise => ALL_MODE
 294                     mode = config.ALL_MODE
 295         return MODE_MAP[mode]
 296
 297     def run(self):
 298         server = agent_utils.UnixDomainWSGIServer('neutron-metadata-agent')
 299         server.start(MetadataProxyHandler(self.conf),
 300                      self.conf.metadata_proxy_socket,
 301                      workers=self.conf.metadata_workers,
 302                      backlog=self.conf.metadata_backlog,
 303                      mode=self._get_socket_mode())
 304         server.wait()