Use new control socket of babeld to get routes

demo/demo

@@ -161,7 +161,8 @@ if 1:
         " -set_serial 0x120010db80042 -days %u" % CA_DAYS, shell=True)
     db_path = 'registry/registry.db'
     registry.screen('../re6st-registry @registry/re6st-registry.conf --db %s'
-        ' --mailhost %s -v%u' % (db_path, os.path.abspath('mbox'), VERBOSE))
+        ' --mailhost %s -v%u --control-socket registry/babeld.socket'
+        % (db_path, os.path.abspath('mbox'), VERBOSE))
     registry_url = 'http://%s/' % REGISTRY
     registry.Popen(('python', '-c', """if 1:
         import socket, time
@@ -194,8 +195,9 @@ if 1:
             p.communicate(str(token[0]))
             os.remove(dh_path)
             os.remove(folder + '/ca.crt')
-        node.screen('../re6stnet @%s/re6stnet.conf -v%u --registry %s %s'
-                    % (folder, VERBOSE, registry, args))
+        node.screen('../re6stnet @%s/re6stnet.conf -v%u --registry %s'
+                    ' --control-socket %s/babeld.socket'
+                    ' %s' % (folder, VERBOSE, registry, folder, args))
     re6stnet(registry, 'registry', '--ip ' + REGISTRY, registry='http://localhost/')
     re6stnet(machine1, 'm1', '-I%s' % m1_if_0.name)
     re6stnet(machine2, 'm2', '--remote-gateway 10.1.1.1', prefix_len=80)

re6st-registry

@@ -3,7 +3,7 @@ import httplib, logging, socket
 from BaseHTTPServer import BaseHTTPRequestHandler
 from SocketServer import ThreadingTCPServer
 from urlparse import parse_qsl
-from re6st import registry, utils
+from re6st import ctl, registry, utils
 
 # To generate server ca and key with serial for 2001:db8:42::/48
 #  openssl req -nodes -new -x509 -key ca.key -set_serial 0x120010db80042 -days 3650 -out ca.crt
@@ -80,6 +80,9 @@ def main():
     _('--anonymous-prefix-length', type=int,
             help="Length of allocated anonymous prefixes."
                  " If 0 or unset, registration by email is required")
+    _('--control-socket', metavar='CTL_SOCK', default=ctl.SOCK_PATH,
+        help="Socket path to use for communication between re6stnet and babeld"
+             " (option -R of Babel).")
     _('-l', '--logfile', default='/var/log/re6stnet/registry.log',
             help="Path to logging file.")
     _('-v', '--verbose', default=1, type=int,

re6st/ctl.py

+import logging, socket, struct
+from collections import namedtuple
+from . import utils
+
+SOCK_PATH = '/var/run/re6st-babeld.sock'
+
+uint16 = struct.Struct("!H")
+header = struct.Struct("!HI")
+
+class Struct(object):
+
+    def __init__(self, format, *args):
+        if args:
+            t = namedtuple(*args)
+        if isinstance(format, str):
+            s = struct.Struct("!" + format)
+            def encode(buffer, value):
+                buffer += s.pack(*value)
+            def decode(buffer, offset=0):
+                return offset + s.size, t(*s.unpack_from(buffer, offset))
+        else:
+            def encode(buffer, value):
+                for f, value in zip(format, value):
+                    f.encode(buffer, value)
+            def decode(buffer, offset=0):
+                r = []
+                for f in format:
+                    offset, x = f.decode(buffer, offset)
+                    r.append(x)
+                return offset, t(*r)
+        self.encode = encode
+        self.decode = decode
+
+class Array(object):
+
+    def __init__(self, item):
+        self._item = item
+
+    def encode(self, buffer, value):
+        buffer += uint16.pack(len(value))
+        encode = self._item.encode
+        for value in value:
+            encode(buffer, value)
+
+    def decode(self, buffer, offset=0):
+        r = []
+        o = offset + 2
+        decode = self._item.decode
+        for i in xrange(*uint16.unpack_from(buffer, offset)):
+            o, x = decode(buffer, o)
+            r.append(x)
+        return o, r
+
+class String(object):
+
+    @staticmethod
+    def encode(buffer, value):
+        buffer += value + "\0"
+
+    @staticmethod
+    def decode(buffer, offset=0):
+        i = buffer.index("\0", offset)
+        return i + 1, buffer[offset:i]
+
+
+class Buffer(object):
+
+    def __init__(self):
+        self._buf = bytearray()
+        self._r = self._w = 0
+
+
+    def __iadd__(self, value):
+        self._buf += value
+        return self
+
+    def __len__(self):
+        return len(self._buf)
+
+    def _seek(self, r):
+        n = len(self._buf)
+        if r < n:
+            self._r = r
+        else:
+            self._w -= n
+            del self._buf[:]
+            self._r = 0
+
+    # reading
+
+    @property
+    def ready(self):
+        return self._w <= len(self._buf)
+
+    def want(self, n):
+        self._w = self._r + n
+
+    def unpack_from(self, struct):
+        r = self._r
+        value = struct.unpack_from(self._buf, r)
+        self._seek(r + struct.size)
+        return value
+
+    def decode(self, decode):
+        r, value = decode(self._buf, self._r)
+        self._seek(r)
+        return value
+
+    try: # BBB: Python < 2.7.4 (http://bugs.python.org/issue10212)
+        uint16.unpack_from(bytearray(uint16.size))
+    except TypeError:
+        def unpack_from(self, struct):
+            r = self._r
+            x = r + struct.size
+            value = struct.unpack(buffer(self._buf)[r:x])
+            self._seek(x)
+            return value
+        def decode(self, decode):
+            r = self._r
+            size, value = decode(buffer(self._buf)[r:])
+            self._seek(r + size)
+            return value
+
+    # writing
+
+    def send(self, socket, *args):
+        r = self._r
+        self._seek(r + socket.send(self._buf[r:], *args))
+
+    def pack_into(self, struct, offset, *args):
+        struct.pack_into(self._buf, offset, *args)
+
+
+class Packet(object):
+
+    response_dict = {}
+
+    def __new__(cls, id, request, response=None):
+        if response:
+            cls.response_dict[id] = response.decode
+        if request:
+            def packet(*args):
+                self = object.__new__(cls)
+                self.id = id
+                self.args = args
+                self.request = request
+                return self
+            return packet
+
+    def write(self, buffer):
+        logging.trace('send %s%r', self.__class__.__name__,
+                                   (self.id,) + self.args)
+        offset = len(buffer)
+        buffer += '\0' * header.size
+        r = self.request
+        if isinstance(r, Struct):
+            r.encode(buffer, self.args)
+        else:
+            r.encode(buffer, *self.args)
+        buffer.pack_into(header, offset, self.id,
+                         len(buffer) - header.size - offset)
+
+
+Dump = Packet(1,
+  Struct("B"),
+  Struct((
+    Array(Struct((Struct("I", "index", "index"), String), "interface", "index name")),
+    Array(Struct("16sIHHHHHiHH", "neighbour", "address ifindex reach rxcost txcost rtt rttcost channel if_up")),
+    Array(Struct("16sBH", "xroute", "prefix plen metric")),
+    Array(Struct("16sBHHH8siiI16s16sB", "route", "prefix plen metric smoothed_metric refmetric id seqno age ifindex neigh_address nexthop flags")),
+    ), "dump", "interfaces neighbours xroutes routes"))
+
+
+class Babel(object):
+
+    _decode = None
+
+    def __init__(self, socket_path, handler, network):
+        self.network = network
+        self.write_buffer = Buffer()
+        self.read_buffer = Buffer()
+        self.read_buffer.want(header.size)
+        self.handler = handler
+        s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        def select(*args):
+            try:
+                s.connect(socket_path)
+            except socket.error:
+                return
+            s.send("\1")
+            s.setblocking(0)
+            del self.request_dump, self.select
+            self.socket = s
+            return self.select(*args)
+        self.select = select
+        self.request_dump = lambda: self.handle_dump((), (), (), ())
+
+    def send(self, packet):
+        packet.write(self.write_buffer)
+
+    def select(self, r, w, t):
+        s = self.socket
+        r[s] = self._read
+        if self.write_buffer:
+            w[s] = self._write
+
+    def _read(self):
+        d = self.socket.recv(65536)
+        if not d:
+            raise RuntimeError("connection to babeld closed")
+        b = self.read_buffer
+        b += d
+        while b.ready:
+            if self._decode:
+                packet = b.decode(self._decode)
+                self._decode = None
+                b.want(header.size)
+                name = packet.__class__.__name__
+                logging.trace('recv %r', packet)
+                try:
+                    h = getattr(self, "handle_" + name)
+                except AttributeError:
+                    h = getattr(self.handler, "babel_" + name)
+                h(*packet)
+            else:
+                packet_type, size = b.unpack_from(header)
+                self._decode = Packet.response_dict[packet_type]
+                b.want(size)
+
+    def _write(self):
+        self.write_buffer.send(self.socket)
+
+    def request_dump(self):
+        self.send(Dump(11)) # interfaces + neighbours + installed routes
+
+    def handle_dump(self, interfaces, neighbours, xroutes, routes):
+        # neighbours = {neigh_prefix: (neighbour, {dst_prefix: route})}
+        n = dict(((n.address, n.ifindex), (n, {})) for n in neighbours)
+        unidentified = set(n)
+        self.neighbours = neighbours = {}
+        a = len(self.network)
+        for route in routes:
+            assert route.flags & 1, route # installed
+            assert route.neigh_address == route.nexthop, route
+            address = route.neigh_address, route.ifindex
+            neigh_routes = n[address]
+            ip = utils.binFromRawIp(route.prefix)
+            if ip[:a] == self.network:
+                prefix = ip[a:route.plen]
+                if prefix and not route.refmetric:
+                    neighbours[prefix] = neigh_routes
+                    unidentified.remove(address)
+            else:
+                prefix = None
+            neigh_routes[1][prefix] = route
+        if unidentified:
+            routes = {}
+            for address in unidentified:
+                routes.update(n[address][1])
+            if routes:
+                neighbours[None] = None, routes
+                logging.trace("Routes via unidentified neighbours. %r",
+                              neighbours)
+        self.interfaces = dict((i.index, name) for i, name in interfaces)
+        self.handler.babel_dump()

re6st/plib.py

@@ -62,7 +62,7 @@ def client(iface, address_list, encrypt, *args, **kw):
 
 
 def router(subnet, hello_interval, table, log_path, state_path, pidfile,
-           tunnel_interfaces, *args, **kw):
+           tunnel_interfaces, control_socket, *args, **kw):
     s = utils.ipFromBin(subnet)
     n = len(subnet)
     cmd = ['babeld',
@@ -80,6 +80,8 @@ def router(subnet, hello_interval, table, log_path, state_path, pidfile,
         cmd += '-t%u' % table, '-T%u' % table
     else:
         cmd[-2:-2] = '-C', 'redistribute ip ::/0 eq 0'
+    if control_socket:
+        cmd += '-R', '%s' % control_socket
     for iface in tunnel_interfaces:
         cmd += '-C', 'interface %s legacy-rxcost 5120' % iface
     cmd += args

re6st/registry.py

@@ -18,15 +18,16 @@ Authenticated communication:
   - the last one that was really used by the client (!hello)
   - the one of the last handshake (hello)
 """
-import base64, hmac, hashlib, httplib, inspect, logging, mailbox, os, random
-import select, smtplib, socket, sqlite3, string, struct, sys, threading, time
+import base64, hmac, hashlib, httplib, inspect, logging
+import mailbox, os, random, select, smtplib, socket, sqlite3
+import string, struct, sys, threading, time, weakref
 from collections import deque
 from datetime import datetime
 from BaseHTTPServer import HTTPServer, BaseHTTPRequestHandler
 from email.mime.text import MIMEText
 from OpenSSL import crypto
 from urllib import splittype, splithost, splitport, urlencode
-from . import tunnel, utils
+from . import ctl, tunnel, utils
 
 HMAC_HEADER = "Re6stHMAC"
 RENEW_PERIOD = 30 * 86400
@@ -88,11 +89,21 @@ class RegistryServer(object):
         logging.info("Network: %s/%u", utils.ipFromBin(self.network),
                                        len(self.network))
         self.email = self.ca.get_subject().emailAddress
+
+        self.peers_lock = threading.Lock()
+        l = threading.Lock()
+        l.acquire()
+        self.wait_dump = l.acquire
+        self.babel_dump = l.release
+        self.ctl = ctl.Babel(config.control_socket,
+            weakref.proxy(self), self.network)
+
         self.onTimeout()
 
     def select(self, r, w, t):
         if self.timeout:
             t.append((self.timeout, self.onTimeout))
+        self.ctl.select(r, w, t)
 
     def onTimeout(self):
         # XXX: Because we use threads to process requests, the statements
@@ -306,10 +317,16 @@ class RegistryServer(object):
 
     @rpc
     def getBootstrapPeer(self, cn):
-        with self.lock:
+        with self.peers_lock:
             age, peers = self.peers
             if age < time.time() or not peers:
-                peers = [x[1] for x in utils.iterRoutes(self.network)]
+                self.ctl.request_dump()
+                self.wait_dump()
+                peers = [prefix
+                    for neigh_routes in self.ctl.neighbours.itervalues()
+                    for prefix in neigh_routes[1]
+                    if prefix]
+                peers.append(self.prefix)
                 random.shuffle(peers)
                 self.peers = time.time() + 60, peers
             peer = peers.pop()
@@ -318,6 +335,7 @@ class RegistryServer(object):
                 # so don't bother looping over above code
                 # (in case 'peers' is empty).
                 peer = self.prefix
+        with self.lock:
             address = utils.ipFromBin(self.network + peer), tunnel.PORT
             self.sock.sendto('\2', address)
             start = time.time()

re6st/utils.py

@@ -190,7 +190,10 @@ def makedirs(path):
             raise
 
 def binFromIp(ip):
-    ip1, ip2 = struct.unpack('>QQ', socket.inet_pton(socket.AF_INET6, ip))
+    return binFromRawIp(socket.inet_pton(socket.AF_INET6, ip))
+
+def binFromRawIp(ip):
+    ip1, ip2 = struct.unpack('>QQ', ip)
     return bin(ip1)[2:].rjust(64, '0') + bin(ip2)[2:].rjust(64, '0')
 
 
@@ -229,33 +232,6 @@ def binFromSubnet(subnet):
     p, l = subnet.split('/')
     return bin(int(p))[2:].rjust(int(l), '0')
 
-if 1:
-    def _iterRoutes():
-        with open('/proc/net/ipv6_route') as f:
-            routing_table = f.read()
-        for line in routing_table.splitlines():
-            line = line.split()
-            iface = line[-1]
-            if 0 < int(line[5], 16) < 1 << 31: # positive metric
-                yield (iface, bin(int(line[0], 16))[2:].rjust(128, '0'),
-                              int(line[1], 16))
-
-_iterRoutes.__doc__ = """Iterates over all routes
-
-    Amongst all returned routes starting with re6st prefix:
-    - one is the local one with our prefix
-    - any route with null prefix will be ignored
-    - other are reachable routes installed by babeld
-    """
-
-def iterRoutes(network, exclude_prefix=None):
-    a = len(network)
-    for iface, ip, prefix_len in _iterRoutes():
-        if ip[:a] == network:
-            prefix = ip[a:prefix_len]
-            if prefix and prefix != exclude_prefix:
-                yield iface, prefix
-
 def decrypt(key_path, data):
     p = Popen(('openssl', 'rsautl', '-decrypt', '-inkey', key_path),
               stdin=subprocess.PIPE, stdout=subprocess.PIPE)

re6stnet

@@ -3,7 +3,7 @@ import atexit, errno, logging, os, signal, socket
 import sqlite3, subprocess, sys, time, threading
 from collections import deque
 from OpenSSL import crypto
-from re6st import db, plib, tunnel, utils, version
+from re6st import ctl, db, plib, tunnel, utils, version
 from re6st.registry import RegistryClient, RENEW_PERIOD
 from re6st.utils import exit
 
@@ -60,6 +60,9 @@ def getConfig():
     _('--babel-pidfile', metavar='PID', default='/var/run/re6st-babeld.pid',
         help="Specify a file to write our process id to"
              " (option -I of Babel).")
+    _('--control-socket', metavar='CTL_SOCK', default=ctl.SOCK_PATH,
+        help="Socket path to use for communication between re6stnet and babeld"
+             " (option -R of Babel).")
     _('--hello', type=int, default=15,
         help="Hello interval in seconds, for both wired and wireless"
              " connections. OpenVPN ping-exit option is set to 4 times the"
@@ -303,8 +306,8 @@ def main():
             required('registry')
             peer_db = db.PeerDB(db_path, registry, config.key, network, prefix)
             cleanup.append(lambda: peer_db.cacheMinimize(config.client_count))
-            tunnel_manager = tunnel.TunnelManager(peer_db,
-                config.openvpn_args, timeout, config.tunnel_refresh,
+            tunnel_manager = tunnel.TunnelManager(config.control_socket,
+                peer_db, config.openvpn_args, timeout, config.tunnel_refresh,
                 config.client_count, config.iface_list, network, prefix,
                 address, ip_changed, config.encrypt, remote_gateway,
                 config.disable_proto, config.neighbour)
@@ -405,6 +408,7 @@ def main():
                 os.path.join(config.log, 'babeld.log'),
                 os.path.join(config.state, 'babeld.state'),
                 config.babel_pidfile, tunnel_interfaces,
+                config.control_socket,
                 *config.babel_args).stop)
             if config.up:
                 exit.release()