bgp: support md5 authentication for re-active sessions

For now, only Linux is supported.

Signed-off-by: FUJITA Tomonori <fujita.tomonori@lab.ntt.co.jp>

ryu/services/protocols/bgp/base.py

@@ -330,22 +330,16 @@ class Activity(object):
         addr, port = sock.getsockname()[:2]
         return (self._canonicalize_ip(addr), str(port))
 
-    def _listen_tcp(self, loc_addr, conn_handle):
-        """Creates a TCP server socket which listens on `port` number.
+    def _create_listen_socket(self, family, loc_addr):
+        s = socket.socket(family)
+        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        s.bind(loc_addr)
+        s.listen(1)
+        return s
 
-        For each connection `server_factory` starts a new protocol.
-        """
-        if ':' in loc_addr[0]:
-            server = hub.listen(loc_addr, family=socket.AF_INET6)
-        else:
-            server = hub.listen(loc_addr)
-
-        server_name = self.name + '_server@' + str(loc_addr)
-        self._asso_socket_map[server_name] = server
-
-        # We now wait for connection requests from client.
+    def _listen_socket_loop(self, s, conn_handle):
         while True:
-            sock, client_address = server.accept()
+            sock, client_address = s.accept()
             client_address, port = self.get_remotename(sock)
             LOG.debug('Connect request received from client for port'
                       ' %s:%s' % (client_address, port))
@@ -353,6 +347,45 @@ class Activity(object):
             self._asso_socket_map[client_name] = sock
             self._spawn(client_name, conn_handle, sock)
 
+    def _listen_tcp(self, loc_addr, conn_handle):
+        """Creates a TCP server socket which listens on `port` number.
+
+        For each connection `server_factory` starts a new protocol.
+        """
+        info = socket.getaddrinfo(None, loc_addr[1], socket.AF_UNSPEC,
+                                  socket.SOCK_STREAM, 0, socket.AI_PASSIVE)
+        listen_sockets = {}
+        for res in info:
+            af, socktype, proto, cannonname, sa = res
+            sock = None
+            try:
+                sock = socket.socket(af, socktype, proto)
+                sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+                if af == socket.AF_INET6:
+                    sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1)
+
+                sock.bind(sa)
+                sock.listen(50)
+                listen_sockets[sa] = sock
+            except socket.error:
+                if sock:
+                    sock.close()
+
+        count = 0
+        server = None
+        for sa in listen_sockets.keys():
+            name = self.name + '_server@' + str(sa[0])
+            if count == 0:
+                import eventlet
+                server = eventlet.spawn(self._listen_socket_loop,
+                                        listen_sockets[sa], conn_handle)
+
+                count += 1
+            else:
+                self._spawn(name, self._listen_socket_loop,
+                            listen_sockets[sa], conn_handle)
+        return server, listen_sockets
+
     def _connect_tcp(self, peer_addr, conn_handler, time_out=None,
                      bind_address=None, password=None):
         """Creates a TCP connection to given peer address.

ryu/services/protocols/bgp/core.py

@@ -21,6 +21,7 @@
 """
 import logging
 import netaddr
+import socket
 
 from ryu.lib.packet.bgp import BGP_ERROR_CEASE
 from ryu.lib.packet.bgp import BGP_ERROR_SUB_CONNECTION_RESET
@@ -40,6 +41,7 @@ from ryu.services.protocols.bgp.signals.emit import BgpSignalBus
 from ryu.services.protocols.bgp.speaker import BgpProtocol
 from ryu.services.protocols.bgp.utils.rtfilter import RouteTargetManager
 from ryu.services.protocols.bgp.utils import stats
+from ryu.lib import sockopt
 
 
 LOG = logging.getLogger('bgpspeaker.core')
@@ -221,7 +223,9 @@ class CoreService(Factory, Activity):
         server_addr = (CORE_IP, self._common_config.bgp_server_port)
         waiter = kwargs.pop('waiter')
         waiter.set()
-        server_thread = self._listen_tcp(server_addr, self.start_protocol)
+        server_thread, sockets = self._listen_tcp(server_addr,
+                                                  self.start_protocol)
+        self.listen_sockets = sockets
 
         server_thread.wait()
         processor_thread.wait()
@@ -358,7 +362,21 @@ class CoreService(Factory, Activity):
             out_route = FlexinetOutgoingRoute(path, route_dist)
             sink.enque_outgoing_msg(out_route)
 
+    def _set_password(self, address, password):
+        if netaddr.valid_ipv4(address):
+            family = socket.AF_INET
+        else:
+            family = socket.AF_INET6
+
+        for sock in self.listen_sockets.values():
+            if sock.family == family:
+                sockopt.set_tcp_md5sig(sock, address, password)
+
     def on_peer_added(self, peer):
+        if peer._neigh_conf.password:
+            self._set_password(peer._neigh_conf.ip_address,
+                               peer._neigh_conf.password)
+
         if self.started:
             self._spawn_activity(
                 peer, self.start_protocol
@@ -372,6 +390,10 @@ class CoreService(Factory, Activity):
             )
 
     def on_peer_removed(self, peer):
+        if peer._neigh_conf.password:
+            # seting zero length key means deleting the key
+            self._set_password(peer._neigh_conf.ip_address, '')
+
         if peer.rtc_as != self.asn:
             self._spawn(
                 'OLD_RTC_AS_HANDLER %s' % peer.rtc_as,