Merge TCP handshake stuff
[nepi.git] / src / nepi / testbeds / planetlab / scripts / tun_connect.py
index 3697dc5..f015bad 100644 (file)
@@ -20,11 +20,17 @@ import traceback
 
 import tunchannel
 
+try:
+    import iovec
+    HAS_IOVEC = True
+except:
+    HAS_IOVEC = False
+
 tun_name = 'tun0'
 tun_path = '/dev/net/tun'
 hostaddr = socket.gethostbyname(socket.gethostname())
 
-usage = "usage: %prog [options] <remote-endpoint>"
+usage = "usage: %prog [options]"
 
 parser = optparse.OptionParser(usage=usage)
 
@@ -37,9 +43,9 @@ parser.add_option(
     default = "/dev/net/tun",
     help = "TUN/TAP device file path or file descriptor number")
 parser.add_option(
-    "-p", "--port", dest="port", metavar="PORT", type="int",
+    "-p", "--peer-port", dest="peer_port", metavar="PEER_PORT", type="int",
     default = 15000,
-    help = "Peering TCP port to connect or listen to.")
+    help = "Remote TCP/UDP port to connect to.")
 parser.add_option(
     "--pass-fd", dest="pass_fd", metavar="UNIX_SOCKET",
     default = None,
@@ -47,7 +53,6 @@ parser.add_option(
            "If given, all other connectivity options are ignored, tun_connect will "
            "simply wait to be killed after passing the file descriptor, and it will be "
            "the receiver's responsability to handle the tunneling.")
-
 parser.add_option(
     "-m", "--mode", dest="mode", metavar="MODE",
     default = "none",
@@ -56,6 +61,11 @@ parser.add_option(
         "by using the proper interface (tunctl for tun/tap, /vsys/fd_tuntap.control for pl-tun/pl-tap), "
         "and it will be brought up (with ifconfig for tun/tap, with /vsys/vif_up for pl-tun/pl-tap). You have "
         "to specify an VIF_ADDRESS and VIF_MASK in any case (except for none).")
+parser.add_option(
+    "-t", "--protocol", dest="protocol", metavar="PROTOCOL",
+    default = None,
+    help = 
+        "Set protocol. One of tcp, udp, fd, gre. In any mode except none, a TUN/TAP will be created.")
 parser.add_option(
     "-A", "--vif-address", dest="vif_addr", metavar="VIF_ADDRESS",
     default = None,
@@ -68,13 +78,18 @@ parser.add_option(
     help = 
         "See mode. This specifies the VIF_MASK, "
         "a number indicating the network type (ie: 24 for a C-class network).")
+parser.add_option(
+    "-P", "--port", dest="port", type="int", metavar="PORT", 
+    default = None,
+    help = 
+        "This specifies the LOCAL_PORT. This will be the local bind port for UDP/TCP.")
 parser.add_option(
     "-S", "--vif-snat", dest="vif_snat", 
     action = "store_true",
     default = False,
     help = "See mode. This specifies whether SNAT will be enabled for the virtual interface. " )
 parser.add_option(
-    "-P", "--vif-pointopoint", dest="vif_pointopoint",  metavar="DST_ADDR",
+    "-Z", "--vif-pointopoint", dest="vif_pointopoint",  metavar="DST_ADDR",
     default = None,
     help = 
         "See mode. This specifies the remote endpoint's virtual address, "
@@ -86,11 +101,16 @@ parser.add_option(
     help = 
         "See mode. This specifies the interface's transmission queue length. " )
 parser.add_option(
-    "-u", "--udp", dest="udp", metavar="PORT", type="int",
+    "-b", "--bwlimit", dest="bwlimit", metavar="BYTESPERSECOND", type="int",
+    default = None,
+    help = 
+        "This specifies the interface's emulated bandwidth in bytes per second." )
+parser.add_option(
+    "-a", "--peer-address", dest="peer_addr", metavar="PEER_ADDRESS",
     default = None,
     help = 
-        "Bind to the specified UDP port locally, and send UDP datagrams to the "
-        "remote endpoint, creating a tunnel through UDP rather than TCP." )
+        "This specifies the PEER_ADDRESS, "
+        "the IP address of the remote interface.")
 parser.add_option(
     "-k", "--key", dest="cipher_key", metavar="KEY",
     default = None,
@@ -98,8 +118,8 @@ parser.add_option(
         "Specify a symmetric encryption key with which to protect packets across "
         "the tunnel. python-crypto must be installed on the system." )
 parser.add_option(
-    "-K", "--gre-key", dest="gre_key", metavar="KEY", type="int",
-    default = None,
+    "-K", "--gre-key", dest="gre_key", metavar="KEY", type="string",
+    default = "true",
     help = 
         "Specify a demultiplexing 32-bit numeric key for GRE." )
 parser.add_option(
@@ -117,6 +137,13 @@ parser.add_option(
     default = None,
     help = "If specified, packets won't be logged to standard output, "
            "but dumped to a pcap-formatted trace in the specified file. " )
+parser.add_option(
+    "--multicast-forwarder", dest="multicast_fwd", 
+    default = None,
+    help = "If specified, multicast packets will be forwarded to "
+           "the specified unix-domain socket. If the device uses ethernet "
+           "frames, ethernet headers will be stripped and IP packets "
+           "will be forwarded, prefixed with the interface's address." )
 parser.add_option(
     "--filter", dest="filter_module", metavar="PATH",
     default = None,
@@ -175,7 +202,7 @@ parser.add_option(
     help = "If specified, packets won't be logged to standard output, "
            "but dumped to a pcap-formatted trace in the specified file. " )
 
-(options, remaining_args) = parser.parse_args(sys.argv[1:])
+(options,args) = parser.parse_args(sys.argv[1:])
 
 options.cipher = {
     'aes' : 'AES',
@@ -434,8 +461,8 @@ def pl_vif_start(tun_path, tun_name):
     if options.vif_txqueuelen is not None:
         stdin.write("txqueuelen=%d\n" % (options.vif_txqueuelen,))
     if options.mode.startswith('pl-gre'):
-        stdin.write("gre=%d\n" % (options.gre_key,))
-        stdin.write("remote=%s\n" % (remaining_args[0],))
+        stdin.write("gre=%s\n" % (options.gre_key,))
+        stdin.write("remote=%s\n" % (options.peer_addr,))
     stdin.close()
     
     t.join()
@@ -484,7 +511,7 @@ def pl_vif_stop(tun_path, tun_name):
     del lock, lockfile
 
 
-def tun_fwd(tun, remote, reconnect = None, accept_local = None, accept_remote = None, slowlocal = True):
+def tun_fwd(tun, remote, reconnect = None, accept_local = None, accept_remote = None, slowlocal = True, bwlimit = None):
     global TERMINATE
     
     tunqueue = options.vif_txqueuelen or 1000
@@ -496,7 +523,7 @@ def tun_fwd(tun, remote, reconnect = None, accept_local = None, accept_remote =
         with_pi = options.mode.startswith('pl-'),
         ether_mode = tun_name.startswith('tap'),
         cipher_key = options.cipher_key,
-        udp = options.udp,
+        udp = options.protocol == 'udp',
         TERMINATE = TERMINATE,
         stderr = None,
         reconnect = reconnect,
@@ -505,7 +532,9 @@ def tun_fwd(tun, remote, reconnect = None, accept_local = None, accept_remote =
         cipher = options.cipher,
         accept_local = accept_local,
         accept_remote = accept_remote,
-        slowlocal = slowlocal
+        queueclass = queueclass,
+        slowlocal = slowlocal,
+        bwlimit = bwlimit
     )
 
 
@@ -607,6 +636,13 @@ else:
     filter_init = None
     filter_run = None
     filter_close = None
+    queueclass = None
+
+# install multicast forwarding hook
+if options.multicast_fwd:
+    print >>sys.stderr, "Connecting to mcast filter"
+    mcfwd_sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
+    tunchannel.nonblock(mcfwd_sock.fileno())
 
 # be careful to roll back stuff on exceptions
 tun_path, tun_name = modeinfo['alloc'](tun_path, tun_name)
@@ -632,8 +668,64 @@ signal.signal(signal.SIGTERM, _finalize)
 try:
     tcpdump = None
     reconnect = None
-    
-    if options.pass_fd:
+    mcastthread = None
+
+    # install multicast forwarding hook
+    if options.multicast_fwd:
+        print >>sys.stderr, "Installing mcast filter"
+        
+        if HAS_IOVEC:
+            writev = iovec.writev
+        else:
+            os_write = os.write
+            map_ = map
+            str_ = str
+            def writev(fileno, *stuff):
+                os_write(''.join(map_(str_,stuff)))
+        
+        def accept_packet(packet, direction, 
+                _up_accept=accept_packet, 
+                sock=mcfwd_sock, 
+                sockno=mcfwd_sock.fileno(),
+                etherProto=tunchannel.etherProto,
+                etherStrip=tunchannel.etherStrip,
+                etherMode=tun_name.startswith('tap'),
+                multicast_fwd = options.multicast_fwd,
+                vif_addr = socket.inet_aton(options.vif_addr),
+                connected = [], writev=writev,
+                len=len, ord=ord):
+            if _up_accept:
+                rv = _up_accept(packet, direction)
+                if not rv:
+                    return rv
+
+            if direction == 1:
+                # Incoming... what?
+                if etherMode:
+                    if etherProto(packet)=='\x08\x00':
+                        fwd = etherStrip(packet)
+                    else:
+                        fwd = None
+                else:
+                    fwd = packet
+                if fwd is not None and len(fwd) >= 20:
+                    if (ord(fwd[16]) & 0xf0) == 0xe0:
+                        # Forward it
+                        if not connected:
+                            try:
+                                sock.connect(multicast_fwd)
+                                connected.append(None)
+                            except:
+                                traceback.print_exc(file=sys.stderr)
+                        if connected:
+                            try:
+                                writev(sockno, vif_addr,fwd)
+                            except:
+                                traceback.print_exc(file=sys.stderr)
+            return 1
+
+    
+    if options.protocol == 'fd':
         if accept_packet or filter_init:
             raise NotImplementedError, "--pass-fd and --filter are not compatible"
         
@@ -671,7 +763,7 @@ try:
             while not TERM:
                 time.sleep(1)
         remote = None
-    elif options.mode.startswith('pl-gre'):
+    elif options.protocol == "gre":
         if accept_packet or filter_init:
             raise NotImplementedError, "--mode %s and --filter are not compatible" % (options.mode,)
         
@@ -681,102 +773,29 @@ try:
             TERM = TERMINATE
             while not TERM:
                 time.sleep(1)
-        remote = remaining_args[0]
-    elif options.udp:
+        remote = options.peer_addr
+    elif options.protocol == "udp":
         # connect to remote endpoint
-        if remaining_args and not remaining_args[0].startswith('-'):
-            print >>sys.stderr, "Listening at: %s:%d" % (hostaddr,options.udp)
-            print >>sys.stderr, "Connecting to: %s:%d" % (remaining_args[0],options.port)
-            rsock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, 0)
-            retrydelay = 1.0
-            for i in xrange(30):
-                if TERMINATE:
-                    raise OSError, "Killed"
-                try:
-                    rsock.bind((hostaddr,options.udp))
-                    break
-                except socket.error:
-                    # wait a while, retry
-                    print >>sys.stderr, "%s: Could not bind. Retrying in a sec..." % (time.strftime('%c'),)
-                    time.sleep(min(30.0,retrydelay))
-                    retrydelay *= 1.1
-            else:
-                rsock.bind((hostaddr,options.udp))
-            rsock.connect((remaining_args[0],options.port))
+        if options.peer_addr and options.peer_port:
+            rsock = tunchannel.udp_establish(TERMINATE, hostaddr, options.port, 
+                    options.peer_addr, options.peer_port)
+            remote = os.fdopen(rsock.fileno(), 'r+b', 0)
         else:
             print >>sys.stderr, "Error: need a remote endpoint in UDP mode"
             raise AssertionError, "Error: need a remote endpoint in UDP mode"
-        
-        # Wait for other peer
-        endme = False
-        def keepalive():
-            while not endme and not TERMINATE:
-                try:
-                    rsock.send('')
-                except:
-                    pass
-                time.sleep(1)
-            try:
-                rsock.send('')
-            except:
-                pass
-        keepalive_thread = threading.Thread(target=keepalive)
-        keepalive_thread.start()
-        retrydelay = 1.0
-        for i in xrange(30):
-            if TERMINATE:
-                raise OSError, "Killed"
-            try:
-                heartbeat = rsock.recv(10)
-                break
-            except:
-                time.sleep(min(30.0,retrydelay))
-                retrydelay *= 1.1
-        else:
-            heartbeat = rsock.recv(10)
-        endme = True
-        keepalive_thread.join()
-        
-        remote = os.fdopen(rsock.fileno(), 'r+b', 0)
-    else:
+    elif options.protocol == "tcp":
         # connect to remote endpoint
-        if remaining_args and not remaining_args[0].startswith('-'):
-            print >>sys.stderr, "Connecting to: %s:%d" % (remaining_args[0],options.port)
-            rsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
-            retrydelay = 1.0
-            for i in xrange(30):
-                if TERMINATE:
-                    raise OSError, "Killed"
-                try:
-                    rsock.connect((remaining_args[0],options.port))
-                    break
-                except socket.error:
-                    # wait a while, retry
-                    print >>sys.stderr, "%s: Could not connect. Retrying in a sec..." % (time.strftime('%c'),)
-                    time.sleep(min(30.0,retrydelay))
-                    retrydelay *= 1.1
-            else:
-                rsock.connect((remaining_args[0],options.port))
+        if options.peer_addr and options.peer_port:
+            rsock = tunchannel.tcp_establish(TERMINATE, hostaddr, options.port,
+                    options.peer_addr, options.peer_port)
+            remote = os.fdopen(rsock.fileno(), 'r+b', 0)
         else:
-            print >>sys.stderr, "Listening at: %s:%d" % (hostaddr,options.port)
-            lsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
-            retrydelay = 1.0
-            for i in xrange(30):
-                if TERMINATE:
-                    raise OSError, "Killed"
-                try:
-                    lsock.bind((hostaddr,options.port))
-                    break
-                except socket.error:
-                    # wait a while, retry
-                    print >>sys.stderr, "%s: Could not bind. Retrying in a sec..." % (time.strftime('%c'),)
-                    time.sleep(min(30.0,retrydelay))
-                    retrydelay *= 1.1
-            else:
-                lsock.bind((hostaddr,options.port))
-            lsock.listen(1)
-            rsock,raddr = lsock.accept()
-        remote = os.fdopen(rsock.fileno(), 'r+b', 0)
+            print >>sys.stderr, "Error: need a remote endpoint in TCP mode"
+            raise AssertionError, "Error: need a remote endpoint in TCP mode"
+    else:
+        msg = "Error: Invalid protocol %s" % options.protocol
+        print >>sys.stderr, msg 
+        raise AssertionError, msg
 
     if filter_init:
         filter_local, filter_remote = filter_init()
@@ -810,12 +829,13 @@ try:
         # Ignore errors, we might not have enough privileges,
         # or perhaps there is no os.nice support in the system
         pass
-
+    
     if not filter_init:
         tun_fwd(tun, remote,
             reconnect = reconnect,
             accept_local = accept_packet,
             accept_remote = accept_packet,
+            bwlimit = options.bwlimit,
             slowlocal = True)
     else:
         # Hm...
@@ -846,6 +866,7 @@ try:
             tun_fwd(filter_remote_fd, remote,
                 reconnect = reconnect,
                 accept_remote = accept_packet,
+                bwlimit = options.bwlimit,
                 slowlocal = False)
         
         localthread = threading.Thread(target=localside)
@@ -865,6 +886,12 @@ finally:
     # tidy shutdown in every case - swallow exceptions
     TERMINATE.append(None)
     
+    if mcastthread:
+        try:
+            mcastthread.stop()
+        except:
+            pass
+    
     if filter_thread:
         try:
             filter_thread.join()