541 lines
19 KiB
Python
541 lines
19 KiB
Python
# Copyright (c) 2016 OpenStack Foundation.
|
|
# 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 hashlib
|
|
import hmac
|
|
|
|
import httplib2
|
|
import netaddr
|
|
from oslo_log import log
|
|
from oslo_utils import encodeutils
|
|
from ryu.lib.packet import arp
|
|
from ryu.lib.packet import ethernet
|
|
from ryu.lib.packet import ipv4
|
|
from ryu.ofproto import nicira_ext
|
|
import six
|
|
import six.moves.urllib.parse as urlparse
|
|
import webob
|
|
|
|
from dragonflow._i18n import _
|
|
from dragonflow.common import exceptions
|
|
from dragonflow import conf as cfg
|
|
from dragonflow.controller.common import arp_responder
|
|
from dragonflow.controller.common import constants as const
|
|
from dragonflow.controller import df_base_app
|
|
from dragonflow.db.models import constants as model_const
|
|
from dragonflow.db.models import l2
|
|
from dragonflow.db.models import switch
|
|
|
|
|
|
LOG = log.getLogger(__name__)
|
|
|
|
FLOW_IDLE_TIMEOUT = 60
|
|
|
|
# TODO(oanson) The TCP_* flag constants have already made it into ryu
|
|
# master, but not to pip. Once that is done, they should be taken from
|
|
# there. (ryu.lib.packet.tcp.TCP_SYN and ryu.lib.packet.tcp.TCP_ACK)
|
|
TCP_SYN = 0x002
|
|
TCP_ACK = 0x010
|
|
|
|
|
|
class MetadataServiceApp(df_base_app.DFlowApp):
|
|
def __init__(self, *args, **kwargs):
|
|
super(MetadataServiceApp, self).__init__(*args, **kwargs)
|
|
self._arp_responder = None
|
|
self._port_num = None
|
|
self._interface_mac = ""
|
|
self._ip = cfg.CONF.df_metadata.ip
|
|
self._port = cfg.CONF.df_metadata.port
|
|
self._interface = cfg.CONF.df_metadata.metadata_interface
|
|
|
|
def switch_features_handler(self, ev):
|
|
if self._interface_mac and self._port_num and self._port_num > 0:
|
|
# For reconnection, if the mac and port_num is set, re-download
|
|
# the flows.
|
|
self._add_tap_metadata_port(self._port_num, self._interface_mac)
|
|
|
|
@df_base_app.register_event(switch.SwitchPort, model_const.EVENT_CREATED)
|
|
@df_base_app.register_event(switch.SwitchPort, model_const.EVENT_UPDATED)
|
|
def switch_port_updated(self, switch_port, orig_switch_port=None):
|
|
if switch_port.name != cfg.CONF.df_metadata.metadata_interface:
|
|
return
|
|
|
|
port_num = switch_port.port_num
|
|
mac = switch_port.mac_in_use
|
|
if not port_num or not mac:
|
|
return
|
|
|
|
if port_num <= 0:
|
|
return
|
|
|
|
if port_num == self._port_num and mac == self._interface_mac:
|
|
return
|
|
|
|
self._add_tap_metadata_port(port_num, mac)
|
|
self._port_num = port_num
|
|
self._interface_mac = mac
|
|
|
|
@df_base_app.register_event(switch.SwitchPort, model_const.EVENT_DELETED)
|
|
def switch_port_deleted(self, switch_port):
|
|
if switch_port.name != cfg.CONF.df_metadata.metadata_interface:
|
|
return
|
|
|
|
self._remove_metadata_interface_flows()
|
|
|
|
def _remove_metadata_interface_flows(self):
|
|
if not self._port_num:
|
|
return
|
|
|
|
parser = self.parser
|
|
ofproto = self.ofproto
|
|
|
|
self.mod_flow(
|
|
table_id=const.INGRESS_CLASSIFICATION_DISPATCH_TABLE,
|
|
command=ofproto.OFPFC_DELETE,
|
|
priority=const.PRIORITY_MEDIUM,
|
|
match=parser.OFPMatch(in_port=self._port_num))
|
|
|
|
self._port_num = None
|
|
self._interface_mac = ""
|
|
|
|
def _add_tap_metadata_port(self, port_num, mac):
|
|
"""
|
|
Add the flows that can be added with the current available information:
|
|
Regular Client->Server packets have IP rewritten, and sent to OVS port
|
|
TCP Syn packets are sent to controller, so that response flows can be
|
|
added.
|
|
Packets from the OVS port are detected and sent for classification.
|
|
"""
|
|
self._port_num = port_num
|
|
ofproto = self.ofproto
|
|
parser = self.parser
|
|
self._add_incoming_flows()
|
|
# Regular packet
|
|
match = parser.OFPMatch(eth_type=ethernet.ether.ETH_TYPE_IP)
|
|
actions = self._get_rewrite_ip_and_output_actions(ofproto, parser)
|
|
inst = [parser.OFPInstructionActions(
|
|
ofproto.OFPIT_APPLY_ACTIONS,
|
|
actions,
|
|
)]
|
|
self.mod_flow(
|
|
table_id=const.METADATA_SERVICE_TABLE,
|
|
command=ofproto.OFPFC_ADD,
|
|
priority=const.PRIORITY_MEDIUM,
|
|
match=match,
|
|
inst=inst,
|
|
)
|
|
# TCP SYN packet
|
|
match = parser.OFPMatch(
|
|
eth_type=ethernet.ether.ETH_TYPE_IP,
|
|
ip_proto=ipv4.inet.IPPROTO_TCP,
|
|
tcp_flags=(TCP_SYN, TCP_SYN | TCP_ACK),
|
|
)
|
|
learn_actions = self._get_learn_actions(ofproto, parser)
|
|
learn_actions.extend(actions)
|
|
inst = [parser.OFPInstructionActions(
|
|
ofproto.OFPIT_APPLY_ACTIONS,
|
|
learn_actions,
|
|
)]
|
|
self.mod_flow(
|
|
table_id=const.METADATA_SERVICE_TABLE,
|
|
command=ofproto.OFPFC_ADD,
|
|
priority=const.PRIORITY_HIGH,
|
|
match=match,
|
|
inst=inst,
|
|
)
|
|
|
|
# ARP responder
|
|
match = parser.OFPMatch(in_port=port_num,
|
|
eth_type=ethernet.ether.ETH_TYPE_ARP)
|
|
actions = [
|
|
parser.NXActionResubmitTable(
|
|
table_id=const.METADATA_SERVICE_REPLY_TABLE),
|
|
parser.OFPActionOutput(ofproto.OFPP_IN_PORT, 0)]
|
|
inst = [parser.OFPInstructionActions(ofproto.OFPIT_APPLY_ACTIONS,
|
|
actions)]
|
|
self.mod_flow(
|
|
table_id=const.INGRESS_CLASSIFICATION_DISPATCH_TABLE,
|
|
command=ofproto.OFPFC_ADD,
|
|
priority=const.PRIORITY_MEDIUM,
|
|
match=match,
|
|
inst=inst,
|
|
)
|
|
self._create_arp_responder(mac)
|
|
|
|
# Response packet
|
|
match = parser.OFPMatch(in_port=port_num,
|
|
eth_type=ethernet.ether.ETH_TYPE_IP)
|
|
actions = [
|
|
parser.NXActionResubmitTable(
|
|
table_id=const.METADATA_SERVICE_REPLY_TABLE),
|
|
parser.NXActionResubmitTable(
|
|
table_id=const.INGRESS_DISPATCH_TABLE)
|
|
]
|
|
inst = [parser.OFPInstructionActions(ofproto.OFPIT_APPLY_ACTIONS,
|
|
actions)]
|
|
self.mod_flow(
|
|
table_id=const.INGRESS_CLASSIFICATION_DISPATCH_TABLE,
|
|
command=ofproto.OFPFC_ADD,
|
|
priority=const.PRIORITY_MEDIUM,
|
|
match=match,
|
|
inst=inst,
|
|
)
|
|
|
|
def _add_incoming_flows(self):
|
|
ofproto = self.ofproto
|
|
parser = self.parser
|
|
|
|
match = parser.OFPMatch(
|
|
eth_type=ethernet.ether.ETH_TYPE_IP,
|
|
ipv4_dst=const.METADATA_SERVICE_IP,
|
|
ip_proto=ipv4.inet.IPPROTO_TCP,
|
|
tcp_dst=const.METADATA_HTTP_PORT,
|
|
)
|
|
actions = [
|
|
parser.NXActionResubmitTable(
|
|
table_id=self.dfdp.apps['portsec'].exitpoints.services),
|
|
]
|
|
# Bypass the security group check for metadata request.
|
|
self.mod_flow(
|
|
table_id=self.dfdp.apps['portsec'].states.main,
|
|
command=ofproto.OFPFC_ADD,
|
|
priority=const.PRIORITY_VERY_HIGH,
|
|
match=match,
|
|
actions=actions)
|
|
|
|
inst = self._get_incoming_flow_instructions(ofproto, parser)
|
|
self.mod_flow(
|
|
table_id=const.SERVICES_CLASSIFICATION_TABLE,
|
|
command=ofproto.OFPFC_ADD,
|
|
priority=const.PRIORITY_MEDIUM,
|
|
match=match,
|
|
inst=inst)
|
|
|
|
def _get_incoming_flow_instructions(self, ofproto, parser):
|
|
actions = self._get_incoming_flow_actions(ofproto, parser)
|
|
inst = []
|
|
if actions:
|
|
inst.append(
|
|
parser.OFPInstructionActions(
|
|
ofproto.OFPIT_APPLY_ACTIONS,
|
|
actions
|
|
),
|
|
)
|
|
inst.append(
|
|
parser.OFPInstructionGotoTable(const.METADATA_SERVICE_TABLE)
|
|
)
|
|
return inst
|
|
|
|
def _get_incoming_flow_actions(self, ofproto, parser):
|
|
actions = []
|
|
if self._ip != const.METADATA_SERVICE_IP:
|
|
actions.append(parser.OFPActionSetField(ipv4_dst=self._ip))
|
|
if self._port != const.METADATA_HTTP_PORT:
|
|
actions.append(parser.OFPActionSetField(tcp_dst=self._port))
|
|
return actions
|
|
|
|
def _get_rewrite_ip_and_output_actions(self, ofproto, parser):
|
|
"""
|
|
Retrieve the actions that rewrite the dst IP field with the reg6
|
|
(the tunnel key), set the first bit of that field, and output to the
|
|
metadata service OVS port.
|
|
The IP is set to <reg6> | 0x8000000, so that the transparent proxy
|
|
can extract the <reg6> from the source IP address, and be able to
|
|
identify the source VM. reg6 holds the local DF id identifying the VM.
|
|
"""
|
|
return [
|
|
parser.NXActionRegMove(
|
|
src_field='reg6',
|
|
dst_field='ipv4_src',
|
|
n_bits=32,
|
|
),
|
|
parser.NXActionRegLoad(
|
|
ofs_nbits=nicira_ext.ofs_nbits(31, 31),
|
|
dst="ipv4_src",
|
|
value=1,),
|
|
parser.OFPActionOutput(
|
|
self._port_num,
|
|
ofproto.OFPCML_NO_BUFFER,
|
|
)
|
|
]
|
|
|
|
def _get_learn_actions(self, ofproto, parser):
|
|
return [
|
|
# Return flow
|
|
parser.NXActionLearn(
|
|
table_id=const.METADATA_SERVICE_REPLY_TABLE,
|
|
specs=[
|
|
# Match
|
|
parser.NXFlowSpecMatch(
|
|
src=ethernet.ether.ETH_TYPE_IP,
|
|
dst=('eth_type', 0),
|
|
n_bits=16,
|
|
),
|
|
parser.NXFlowSpecMatch(
|
|
src=ipv4.inet.IPPROTO_TCP,
|
|
dst=('ip_proto', 0),
|
|
n_bits=8,
|
|
),
|
|
parser.NXFlowSpecMatch(
|
|
src=1,
|
|
dst=('ipv4_dst', 31),
|
|
n_bits=1,
|
|
),
|
|
parser.NXFlowSpecMatch(
|
|
src=('reg6', 0),
|
|
dst=('ipv4_dst', 0),
|
|
n_bits=31,
|
|
),
|
|
parser.NXFlowSpecMatch(
|
|
src=('tcp_src', 0),
|
|
dst=('tcp_dst', 0),
|
|
n_bits=16,
|
|
),
|
|
# Actions
|
|
parser.NXFlowSpecLoad(
|
|
src=('ipv4_src', 0),
|
|
dst=('ipv4_dst', 0),
|
|
n_bits=32,
|
|
),
|
|
parser.NXFlowSpecLoad(
|
|
src=int(netaddr.IPAddress(const.METADATA_SERVICE_IP)),
|
|
dst=('ipv4_src', 0),
|
|
n_bits=32,
|
|
),
|
|
parser.NXFlowSpecLoad(
|
|
src=const.METADATA_HTTP_PORT,
|
|
dst=('tcp_src', 0),
|
|
n_bits=16,
|
|
),
|
|
parser.NXFlowSpecLoad(
|
|
src=('reg6', 0),
|
|
dst=('reg7', 0),
|
|
n_bits=32,
|
|
),
|
|
],
|
|
fin_idle_timeout=1,
|
|
fin_hard_timeout=1,
|
|
),
|
|
# ARP responder
|
|
parser.NXActionLearn(
|
|
table_id=const.METADATA_SERVICE_REPLY_TABLE,
|
|
priority=const.PRIORITY_HIGH,
|
|
specs=[
|
|
# Match
|
|
parser.NXFlowSpecMatch(
|
|
src=ethernet.ether.ETH_TYPE_ARP,
|
|
dst=('eth_type', 0),
|
|
n_bits=16,
|
|
),
|
|
parser.NXFlowSpecMatch(
|
|
src=('reg6', 0),
|
|
dst=('arp_tpa', 0),
|
|
n_bits=31,
|
|
),
|
|
parser.NXFlowSpecMatch(
|
|
src=arp.ARP_REQUEST,
|
|
dst=('arp_op', 0),
|
|
n_bits=8,
|
|
),
|
|
# Actions
|
|
parser.NXFlowSpecLoad(
|
|
src=0,
|
|
dst=('reg6', 0),
|
|
n_bits=32,
|
|
),
|
|
parser.NXFlowSpecLoad(
|
|
src=arp.ARP_REPLY,
|
|
dst=('arp_op', 0),
|
|
n_bits=8,
|
|
),
|
|
parser.NXFlowSpecLoad(
|
|
src=('eth_dst', 0),
|
|
dst=('arp_tha', 0),
|
|
n_bits=48,
|
|
),
|
|
parser.NXFlowSpecLoad(
|
|
src=int(netaddr.IPAddress(self._ip)),
|
|
dst=('arp_tpa', 0),
|
|
n_bits=32,
|
|
),
|
|
parser.NXFlowSpecLoad(
|
|
src=('eth_src', 0),
|
|
dst=('eth_src', 0),
|
|
n_bits=48,
|
|
),
|
|
parser.NXFlowSpecLoad(
|
|
src=('eth_src', 0),
|
|
dst=('arp_sha', 0),
|
|
n_bits=48,
|
|
),
|
|
parser.NXFlowSpecLoad(
|
|
src=('reg6', 0),
|
|
dst=('arp_spa', 0),
|
|
n_bits=32,
|
|
),
|
|
parser.NXFlowSpecLoad(
|
|
src=1,
|
|
dst=('arp_spa', 31),
|
|
n_bits=1,
|
|
),
|
|
],
|
|
idle_timeout=30,
|
|
)
|
|
]
|
|
|
|
def _create_arp_responder(self, mac):
|
|
self._arp_responder = arp_responder.ArpResponder(
|
|
self,
|
|
None,
|
|
const.METADATA_SERVICE_IP,
|
|
mac
|
|
)
|
|
self._arp_responder.add()
|
|
|
|
|
|
class BaseMetadataProxyHandler(object):
|
|
@webob.dec.wsgify(RequestClass=webob.Request)
|
|
def __call__(self, req):
|
|
try:
|
|
LOG.debug("Request: %s", req)
|
|
return self.proxy_request(req)
|
|
except Exception:
|
|
LOG.exception("Unexpected error.")
|
|
msg = _('An unknown error has occurred. '
|
|
'Please try your request again.')
|
|
explanation = six.text_type(msg)
|
|
return webob.exc.HTTPInternalServerError(explanation=explanation)
|
|
|
|
def proxy_request(self, req):
|
|
headers = self.get_headers(req)
|
|
url = urlparse.urlunsplit((
|
|
self.get_scheme(req),
|
|
self.get_host(req),
|
|
self.get_path_info(req),
|
|
self.get_query_string(req),
|
|
''))
|
|
h = self.create_http_client(req)
|
|
resp, content = h.request(
|
|
url,
|
|
method=req.method,
|
|
headers=headers,
|
|
body=req.body
|
|
)
|
|
if resp.status == 200:
|
|
LOG.debug(str(resp))
|
|
return self.create_response(req, resp, content)
|
|
elif resp.status == 403:
|
|
LOG.warning(
|
|
'The remote metadata server responded with Forbidden. This '
|
|
'response usually occurs when shared secrets do not match.')
|
|
return webob.exc.HTTPForbidden()
|
|
elif resp.status == 400:
|
|
return webob.exc.HTTPBadRequest()
|
|
elif resp.status == 404:
|
|
return webob.exc.HTTPNotFound()
|
|
elif resp.status == 409:
|
|
return webob.exc.HTTPConflict()
|
|
elif resp.status == 500:
|
|
msg = (
|
|
'Remote metadata server experienced an internal server error.'
|
|
)
|
|
LOG.warning(msg)
|
|
explanation = six.text_type(msg)
|
|
return webob.exc.HTTPInternalServerError(explanation=explanation)
|
|
else:
|
|
raise Exception(_('Unexpected response code: %s') % resp.status)
|
|
|
|
def get_headers(self, req):
|
|
return req.headers
|
|
|
|
def create_response(self, req, resp, content):
|
|
req.response.content_type = resp['content-type']
|
|
req.response.body = content
|
|
return req.response
|
|
|
|
def get_scheme(self, req):
|
|
return req.scheme
|
|
|
|
def get_host(self, req):
|
|
return req.host
|
|
|
|
def get_path_info(self, req):
|
|
return req.path
|
|
|
|
def get_query_string(self, req):
|
|
return req.query_string
|
|
|
|
def create_http_client(self, req):
|
|
return httplib2.Http()
|
|
|
|
|
|
class DFMetadataProxyHandler(BaseMetadataProxyHandler):
|
|
def __init__(self, conf, nb_api):
|
|
super(DFMetadataProxyHandler, self).__init__()
|
|
self.conf = conf
|
|
self.nb_api = nb_api
|
|
|
|
def get_headers(self, req):
|
|
remote_addr = req.remote_addr
|
|
if not remote_addr:
|
|
raise exceptions.NoRemoteIPProxyException()
|
|
tunnel_key = int(netaddr.IPAddress(remote_addr) & ~0x80000000)
|
|
lport = self._get_logical_port_by_tunnel_key(tunnel_key)
|
|
headers = dict(req.headers)
|
|
tenant_id = lport.topic
|
|
instance_id = lport.device_id
|
|
ip = lport.ip
|
|
headers.update({
|
|
'X-Forwarded-For': str(ip),
|
|
'X-Tenant-ID': tenant_id,
|
|
'X-Instance-ID': instance_id,
|
|
'X-Instance-ID-Signature': self._sign_instance_id(instance_id),
|
|
})
|
|
return headers
|
|
|
|
def get_host(self, req):
|
|
return '{}:{}'.format(
|
|
self.conf.nova_metadata_host,
|
|
self.conf.nova_metadata_port,
|
|
)
|
|
|
|
def get_scheme(self, req):
|
|
return self.conf.nova_metadata_protocol
|
|
|
|
def create_http_client(self, req):
|
|
h = httplib2.Http(
|
|
ca_certs=self.conf.auth_ca_cert,
|
|
disable_ssl_certificate_validation=self.conf.nova_metadata_insecure
|
|
)
|
|
if self.conf.nova_client_cert and self.conf.nova_client_priv_key:
|
|
h.add_certificate(self.conf.nova_client_priv_key,
|
|
self.conf.nova_client_cert,
|
|
self.get_host(req))
|
|
return h
|
|
|
|
def _get_logical_port_by_tunnel_key(self, tunnel_key):
|
|
lports = self.nb_api.get_all(l2.LogicalPort)
|
|
for lport in lports:
|
|
if lport.unique_key == tunnel_key:
|
|
return lport
|
|
raise exceptions.LogicalPortNotFoundByTunnelKey(key=tunnel_key)
|
|
|
|
# Taken from Neurton: neutron/agent/metadata/agent.py
|
|
def _sign_instance_id(self, instance_id):
|
|
secret = self.conf.metadata_proxy_shared_secret
|
|
secret = encodeutils.to_utf8(secret)
|
|
instance_id = encodeutils.to_utf8(instance_id)
|
|
return hmac.new(secret, instance_id, hashlib.sha256).hexdigest()
|