New --remote-gateway option for network redundancy with multiple ISP

demo/demo

@@ -193,7 +193,7 @@ if 1:
                     % (folder, VERBOSE, registry, args))
     re6stnet(registry, 'registry', '--ip ' + REGISTRY, registry='http://localhost/')
     re6stnet(machine1, 'm1', '-I%s' % m1_if_0.name)
-    re6stnet(machine2, 'm2', prefix_len=80)
+    re6stnet(machine2, 'm2', '--remote-gateway 10.1.1.1', prefix_len=80)
     re6stnet(machine3, 'm3', '-i%s' % m3_if_0.name)
     re6stnet(machine4, 'm4', '-i%s' % m4_if_0.name)
     re6stnet(machine5, 'm5', '-i%s' % m5_if_0.name)

re6st/plib.py

@@ -39,15 +39,11 @@ def server(iface, max_clients, dh_path, pipe_fd, port, proto, encrypt, *args, **
         *args, **kw)
 
 
-def client(iface, server_address, encrypt, *args, **kw):
+def client(iface, address_list, encrypt, *args, **kw):
     remote = ['--nobind', '--client']
-    try:
-        for ip, port, proto in utils.address_list(server_address):
+    for ip, port, proto in address_list:
         remote += '--remote', ip, port, \
             'tcp-client' if proto == 'tcp' else proto
-    except ValueError, e:
-        logging.warning("Failed to parse node address %r (%s)",
-                        server_address, e)
     remote += args
     return openvpn(iface, encrypt, *remote, **kw)
 

re6st/tunnel.py

@@ -8,19 +8,71 @@ RTF_CACHE = 0x01000000  # cache entry
 
 # Be careful the refresh interval should let the routes be established
 
-class Connection:
 
-    def __init__(self, address, write_pipe, timeout, iface, prefix, encrypt,
-            ovpn_args):
-        self.process = plib.client(iface, address, encrypt,
-            '--tls-remote', '%u/%u' % (int(prefix, 2), len(prefix)),
+class MultiGatewayManager(dict):
+
+    def __init__(self, gateway):
+        if gateway:
+            self._gw = gateway
+        else:
+            self.add = self.remove = lambda _: None
+
+    def _route(self, cmd, dest, gw):
+        cmd = 'ip', '-4', 'route', cmd, '%s/32' % dest, 'via', gw
+        logging.trace('%r', cmd)
+        subprocess.call(cmd)
+
+    def add(self, ip_list):
+        for dest in ip_list:
+            try:
+                self[dest][1] += 1
+            except KeyError:
+                gw = self._gw(dest)
+                self[dest] = [gw, 0]
+                self._route('add', dest, gw)
+
+    def remove(self, ip_list):
+        for dest in ip_list:
+            gw, count = self[dest]
+            if count:
+                self[dest][1] = count - 1
+            else:
+                del self[dest]
+                try:
+                    self._route('del', dest, gw)
+                except:
+                    pass
+
+
+class Connection(object):
+
+    def __init__(self, address, iface, prefix):
+        self.address_list = list(utils.parse_address(address))
+        self.iface = iface
+        self.routes = 0
+        self._prefix = prefix
+
+    def __iter__(self):
+        if not hasattr(self, '_remote_ip_set'):
+            self._remote_ip_set = set(info[4][0]
+                for ip, port, proto in self.address_list
+                for info in socket.getaddrinfo(ip, port, socket.AF_INET, 0,
+                    getattr(socket, 'IPPROTO_' + proto.upper())))
+        return iter(self._remote_ip_set)
+
+    def open(self, write_pipe, timeout, encrypt, ovpn_args):
+        self.process = plib.client(self.iface, self.address_list, encrypt,
+            '--tls-remote', '%u/%u' % (int(self._prefix, 2), len(self._prefix)),
             '--connect-retry-max', '3', '--tls-exit',
             '--ping-exit', str(timeout),
             '--route-up', '%s %u' % (plib.ovpn_client, write_pipe),
             *ovpn_args)
-        self.iface = iface
-        self.routes = 0
-        self._prefix = prefix
+
+    def close(self):
+        try:
+            self.process.stop()
+        except (AttributeError, OSError):
+            pass # we already polled an exited process
 
     def refresh(self):
         # Check that the connection is alive
@@ -35,7 +87,7 @@ class TunnelManager(object):
 
     def __init__(self, write_pipe, peer_db, openvpn_args, timeout,
                 refresh, client_count, iface_list, network, prefix,
-                address, ip_changed, encrypt):
+                address, ip_changed, encrypt, remote_gateway):
         self._write_pipe = write_pipe
         self._peer_db = peer_db
         self._connecting = set()
@@ -49,9 +101,10 @@ class TunnelManager(object):
         self._network = network
         self._iface_list = iface_list
         self._prefix = prefix
-        self._address = utils.address_str(address)
+        self._address = utils.dump_address(address)
         self._ip_changed = ip_changed
         self._encrypt = encrypt
+        self._gateway_manager = MultiGatewayManager(remote_gateway)
         self._served = set()
 
         self.sock = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
@@ -131,10 +184,8 @@ class TunnelManager(object):
                      int(prefix, 2), len(prefix))
         connection = self._connection_dict.pop(prefix)
         self.freeInterface(connection.iface)
-        try:
-            connection.process.stop()
-        except OSError:
-            pass # we already polled an exited process
+        connection.close()
+        self._gateway_manager.remove(connection)
         logging.trace('Connection with %u/%u killed',
                       int(prefix, 2), len(prefix))
 
@@ -146,8 +197,9 @@ class TunnelManager(object):
         logging.info('Establishing a connection with %u/%u',
                      int(prefix, 2), len(prefix))
         iface = self.getFreeInterface(prefix)
-        self._connection_dict[prefix] = Connection(address, self._write_pipe,
-            self._timeout, iface, prefix, self._encrypt, self._ovpn_args)
+        self._connection_dict[prefix] = c = Connection(address, iface, prefix)
+        self._gateway_manager.add(c)
+        c.open(self._write_pipe, self._timeout, self._encrypt, self._ovpn_args)
         self._peer_db.connecting(prefix, 1)
         return True
 
@@ -301,7 +353,7 @@ class TunnelManager(object):
     def _ovpn_route_up(self, common_name, ip):
         self._peer_db.connecting(utils.binFromSubnet(common_name), 0)
         if self._ip_changed:
-            self._address = utils.address_str(self._ip_changed(ip))
+            self._address = utils.dump_address(self._ip_changed(ip))
 
     def handlePeerEvent(self):
         msg, address = self.sock.recvfrom(1<<16)

re6st/utils.py

@@ -139,14 +139,17 @@ def subnetFromCert(cert_path):
         cert = crypto.load_certificate(crypto.FILETYPE_PEM, f.read())
         return cert.get_subject().CN
 
-def address_str(address):
+def dump_address(address):
     return ';'.join(map(','.join, address))
 
-
-def address_list(address_list):
-    return list(tuple(address.split(','))
-        for address in address_list.split(';'))
-
+def parse_address(address_list):
+    for address in address_list.split(';'):
+        try:
+            ip, port, proto = address.split(',')
+            yield ip, str(port), proto
+        except ValueError, e:
+            logging.warning("Failed to parse node address %r (%s)",
+                            address, e)
 
 def binFromSubnet(subnet):
     p, l = subnet.split('/')

re6stnet

 #!/usr/bin/python
-import atexit, errno, logging, os, select, signal
+import atexit, errno, logging, os, random, select, signal
 import sqlite3, subprocess, sys, time, traceback
 from re6st import plib, utils, db, tunnel
 
@@ -97,6 +97,9 @@ def getConfig():
         help="Interval in seconds between two tunnel refresh: the worst"
              " tunnel is closed if the number of client tunnels has reached"
              " its maximum number (client-count).")
+    _('--remote-gateway', action='append', dest='gw_list',
+        help="Force each tunnel to be created through one the given gateways,"
+             " randomly.")
     _('--client', metavar='HOST,PORT,PROTO[;...]',
         help="Do not run any OpenVPN server, but only 1 OpenVPN client,"
              " with specified remotes. Any other option not required in this"
@@ -145,6 +148,8 @@ def main():
         else:
             pp = (1194, 'udp'), (1194, 'tcp')
         ip_changed = lambda ip: [(ip, str(port), proto) for port, proto in pp]
+        remote_gateway = config.gw_list and (lambda _:
+            random.choice(config.gw_list))
         forwarder = None
         if config.ip == 'upnp' or not config.ip:
             logging.info('Attempting automatic configuration via UPnP...')
@@ -207,7 +212,7 @@ def main():
             tunnel_manager = tunnel.TunnelManager(write_pipe, peer_db,
                 config.openvpn_args, timeout, config.tunnel_refresh,
                 config.client_count, config.iface_list, network, prefix,
-                address, ip_changed, config.encrypt)
+                address, ip_changed, config.encrypt, remote_gateway)
             tunnel_interfaces += tunnel_manager.new_iface_list
         else:
             tunnel_manager = write_pipe = None
@@ -230,7 +235,8 @@ def main():
             ip('addrlabel', 'prefix', my_network, 'label', '99')
             # prepare persistent interfaces
             if config.client:
-                cleanup.append(plib.client('re6stnet', config.client,
+                cleanup.append(plib.client('re6stnet',
+                    utils.parse_address(config.client),
                     config.encrypt, '--ping-restart', str(timeout),
                     *config.openvpn_args).stop)
             elif server_tunnels: