Added TCP-handshake for TunChannel and tun_connect.py
[nepi.git] / src / nepi / testbeds / planetlab / metadata.py
index aac4d7b..7754c0a 100644 (file)
@@ -16,6 +16,7 @@ from nepi.util.constants import ApplicationStatus as AS, \
 import functools
 import os
 import os.path
+import weakref
 
 NODE = "Node"
 NODEIFACE = "NodeInterface"
@@ -27,6 +28,13 @@ NEPIDEPENDENCY = "NepiDependency"
 NS3DEPENDENCY = "NS3Dependency"
 INTERNET = "Internet"
 NETPIPE = "NetPipe"
+TUNFILTER = "TunFilter"
+CLASSQUEUEFILTER = "ClassQueueFilter"
+TOSQUEUEFILTER = "TosQueueFilter"
+
+TUNFILTERS = (TUNFILTER, CLASSQUEUEFILTER, TOSQUEUEFILTER)
+TAPFILTERS = (TUNFILTER, )
+ALLFILTERS = (TUNFILTER, CLASSQUEUEFILTER, TOSQUEUEFILTER)
 
 PL_TESTBED_ID = "planetlab"
 
@@ -103,10 +111,47 @@ def connect_tun_iface_peer(proto, testbed_instance, iface_guid, peer_iface_guid)
     iface = testbed_instance._elements[iface_guid]
     peer_iface = testbed_instance._elements[peer_iface_guid]
     iface.peer_iface = peer_iface
+    peer_iface.peer_iface = iface
     iface.peer_proto = \
-    iface.tun_proto = proto
+    iface.tun_proto = \
+    peer_iface.peer_proto = \
+    peer_iface.tun_proto = proto
     iface.tun_key = peer_iface.tun_key
 
+def connect_tun_iface_filter(testbed_instance, iface_guid, filter_guid):
+    iface = testbed_instance._elements[iface_guid]
+    filt = testbed_instance._elements[filter_guid]
+    traces = testbed_instance._get_traces(filter_guid)
+    if 'dropped_stats' in traces: 
+        args = filt.args if filt.args else ""
+        filt.args = ','.join(filt.args.split(',') + ["logdropped=true",])
+    iface.filter_module = filt
+    filt.iface_guid = iface_guid
+    filt.iface = weakref.ref(iface)
+
+    if filt.peer_guid:
+        connect_tun_iface_peer(filt.peer_proto, testbed_instance, filt.iface_guid, filt.peer_guid)
+
+def connect_filter_peer(proto, testbed_instance, filter_guid, peer_guid):
+    peer = testbed_instance._elements[peer_guid]
+    filt = testbed_instance._elements[filter_guid]
+    filt.peer_proto = proto
+    filt.peer_guid = peer_guid
+    if filt.iface_guid:
+        connect_tun_iface_peer(filt.peer_proto, testbed_instance, filt.iface_guid, filt.peer_guid)
+
+def connect_filter_filter(proto, testbed_instance, filter_guid, peer_guid):
+    peer = testbed_instance._elements[peer_guid]
+    filt = testbed_instance._elements[filter_guid]
+    filt.peer_proto = proto
+    peer.peer_proto = proto
+    if filt.iface_guid:
+        peer.peer_guid = filt.iface_guid
+    if peer.iface_guid:
+        filt.peer_guid = peer.iface_guid
+    if filt.iface_guid and filt.peer_guid:
+        connect_tun_iface_peer(filt.peer_proto, testbed_instance, filt.iface_guid, filt.peer_guid)
+
 def crossconnect_tun_iface_peer_init(proto, testbed_instance, iface_guid, peer_iface_data):
     iface = testbed_instance._elements[iface_guid]
     iface.peer_iface = None
@@ -133,6 +178,20 @@ def crossconnect_tun_iface_peer_both(proto, testbed_instance, iface_guid, peer_i
     crossconnect_tun_iface_peer_init(proto, testbed_instance, iface_guid, peer_iface_data)
     crossconnect_tun_iface_peer_compl(proto, testbed_instance, iface_guid, peer_iface_data)
 
+def crossconnect_filter_peer_init(proto, testbed_instance, filter_guid, peer_data):
+    filt = testbed_instance._elements[filter_guid]
+    filt.peer_proto = proto
+    crossconnect_tun_iface_peer_init(filt.peer_proto, testbed_instance, filt.iface_guid, peer_data)
+
+def crossconnect_filter_peer_compl(proto, testbed_instance, filter_guid, peer_data):
+    filt = testbed_instance._elements[filter_guid]
+    filt.peer_proto = proto
+    crossconnect_tun_iface_peer_compl(filt.peer_proto, testbed_instance, filt.iface_guid, peer_data)
+
+def crossconnect_filter_peer_both(proto, testbed_instance, filter_guid, peer_data):
+    crossconnect_filter_peer_init(proto, testbed_instance, iface_guid, peer_iface_data)
+    crossconnect_filter_peer_compl(proto, testbed_instance, iface_guid, peer_iface_data)
+
 def connect_dep(testbed_instance, node_guid, app_guid):
     node = testbed_instance._elements[node_guid]
     app = testbed_instance._elements[app_guid]
@@ -220,6 +279,21 @@ def create_tapiface(testbed_instance, guid):
     
     testbed_instance.elements[guid] = element
 
+def create_tunfilter(testbed_instance, guid):
+    parameters = testbed_instance._get_parameters(guid)
+    element = testbed_instance._make_tun_filter(parameters)
+    testbed_instance.elements[guid] = element
+
+def create_classqueuefilter(testbed_instance, guid):
+    parameters = testbed_instance._get_parameters(guid)
+    element = testbed_instance._make_class_queue_filter(parameters)
+    testbed_instance.elements[guid] = element
+
+def create_tosqueuefilter(testbed_instance, guid):
+    parameters = testbed_instance._get_parameters(guid)
+    element = testbed_instance._make_tos_queue_filter(parameters)
+    testbed_instance.elements[guid] = element
+
 def create_application(testbed_instance, guid):
     parameters = testbed_instance._get_parameters(guid)
     element = testbed_instance._make_application(parameters)
@@ -276,6 +350,7 @@ def start_application(testbed_instance, guid):
     app.stdout = "stdout" in traces
     app.stderr = "stderr" in traces
     app.buildlog = "buildlog" in traces
+    app.outout = "output" in traces
     
     app.start()
 
@@ -348,36 +423,19 @@ def preconfigure_tuniface(testbed_instance, guid):
     element.validate()
     
     # First-phase setup
-    if element.peer_proto:
-        if element.peer_iface and isinstance(element.peer_iface, testbed_instance._interfaces.TunIface):
-            # intra tun
-            listening = id(element) < id(element.peer_iface)
-        else:
-            # cross tun
-            if not element.tun_addr or not element.tun_port:
-                listening = True
-            elif not element.peer_addr or not element.peer_port:
-                listening = True
-            else:
-                # both have addresses...
-                # ...the one with the lesser address listens
-                listening = element.tun_addr < element.peer_addr
-        element.prepare( 
-            'tun-%s' % (guid,),
-             listening)
+    element.prepare('tun-%s' % (guid,))
 
 def postconfigure_tuniface(testbed_instance, guid):
     element = testbed_instance._elements[guid]
     
     # Second-phase setup
-    element.setup()
+    element.launch()
     
-def wait_tuniface(testbed_instance, guid):
+def prestart_tuniface(testbed_instance, guid):
     element = testbed_instance._elements[guid]
     
     # Second-phase setup
-    element.async_launch_wait()
-    
+    element.wait()
 
 def configure_node(testbed_instance, guid):
     node = testbed_instance._elements[guid]
@@ -509,6 +567,12 @@ connector_types = dict({
                 "max": 1, 
                 "min": 0
             }),
+    "->fd": dict({
+                "help": "TUN device file descriptor slot", 
+                "name": "->fd",
+                "max": 1, 
+                "min": 0
+            }),
    })
 
 connections = [
@@ -584,6 +648,24 @@ connections = [
         "init_code": functools.partial(connect_tun_iface_peer,"gre"),
         "can_cross": False
     }),
+    dict({
+        "from": (TESTBED_ID, TUNIFACE, "fd->"),
+        "to":   (TESTBED_ID, TUNFILTERS, "->fd"),
+        "init_code": connect_tun_iface_filter,
+        "can_cross": False
+    }),
+    dict({
+        "from": (TESTBED_ID, TUNFILTERS, "tcp"),
+        "to":   (TESTBED_ID, TUNIFACE, "tcp"),
+        "init_code": functools.partial(connect_filter_peer,"tcp"),
+        "can_cross": False
+    }),
+    dict({
+        "from": (TESTBED_ID, TUNFILTERS, "udp"),
+        "to":   (TESTBED_ID, TUNIFACE, "udp"),
+        "init_code": functools.partial(connect_filter_peer,"udp"),
+        "can_cross": False
+    }),
     dict({
         "from": (TESTBED_ID, TAPIFACE, "tcp"),
         "to":   (TESTBED_ID, TAPIFACE, "tcp"),
@@ -602,6 +684,48 @@ connections = [
         "init_code": functools.partial(connect_tun_iface_peer,"gre"),
         "can_cross": False
     }),
+    dict({
+        "from": (TESTBED_ID, TAPIFACE, "fd->"),
+        "to":   (TESTBED_ID, TAPFILTERS, "->fd"),
+        "init_code": connect_tun_iface_filter,
+        "can_cross": False
+    }),
+    dict({
+        "from": (TESTBED_ID, TAPFILTERS, "tcp"),
+        "to":   (TESTBED_ID, TAPIFACE, "tcp"),
+        "init_code": functools.partial(connect_filter_peer,"tcp"),
+        "can_cross": False
+    }),
+    dict({
+        "from": (TESTBED_ID, TAPFILTERS, "udp"),
+        "to":   (TESTBED_ID, TAPIFACE, "udp"),
+        "init_code": functools.partial(connect_filter_peer,"udp"),
+        "can_cross": False
+    }),
+    dict({
+        "from": (TESTBED_ID, TUNFILTERS, "tcp"),
+        "to":   (TESTBED_ID, TUNFILTERS, "tcp"),
+        "init_code": functools.partial(connect_filter_filter,"tcp"),
+        "can_cross": False
+    }),
+    dict({
+        "from": (TESTBED_ID, TUNFILTERS, "udp"),
+        "to":   (TESTBED_ID, TUNFILTERS, "udp"),
+        "init_code": functools.partial(connect_filter_filter,"udp"),
+        "can_cross": False
+    }),
+    dict({
+        "from": (TESTBED_ID, TAPFILTERS, "tcp"),
+        "to":   (TESTBED_ID, TAPFILTERS, "tcp"),
+        "init_code": functools.partial(connect_filter_filter,"tcp"),
+        "can_cross": False
+    }),
+    dict({
+        "from": (TESTBED_ID, TAPFILTERS, "udp"),
+        "to":   (TESTBED_ID, TAPFILTERS, "udp"),
+        "init_code": functools.partial(connect_filter_filter,"udp"),
+        "can_cross": False
+    }),
     dict({
         "from": (TESTBED_ID, TUNIFACE, "tcp"),
         "to":   (None, None, "tcp"),
@@ -656,6 +780,20 @@ connections = [
         "compl_code": functools.partial(crossconnect_tun_iface_peer_both,"gre"),
         "can_cross": True
     }),
+    dict({
+        "from": (TESTBED_ID, ALLFILTERS, "tcp"),
+        "to":   (None, None, "tcp"),
+        "init_code": functools.partial(crossconnect_filter_peer_init,"tcp"),
+        "compl_code": functools.partial(crossconnect_filter_peer_compl,"tcp"),
+        "can_cross": True
+    }),
+    dict({
+        "from": (TESTBED_ID, ALLFILTERS, "udp"),
+        "to":   (None, None, "udp"),
+        "init_code": functools.partial(crossconnect_filter_peer_init,"udp"),
+        "compl_code": functools.partial(crossconnect_filter_peer_compl,"udp"),
+        "can_cross": True
+    }),
 ]
 
 attributes = dict({
@@ -805,8 +943,8 @@ attributes = dict({
                 "value": True,
                 "validation_function": validation.is_bool
             }),
-    "device_name": dict({
-                "name": "name",
+    "if_name": dict({
+                "name": "if_name",
                 "help": "Device name",
                 "type": Attribute.STRING,
                 "flags": Attribute.ExecReadOnly | Attribute.ExecImmutable,
@@ -832,6 +970,15 @@ attributes = dict({
                 "value": False,
                 "validation_function": validation.is_bool
             }),
+    "multicast":  dict({
+                "name": "multicast", 
+                "help": "Enable multicast forwarding on this device. "
+                        "Note that you still need a multicast routing daemon "
+                        "in the node.",
+                "type": Attribute.BOOL,
+                "value": False,
+                "validation_function": validation.is_bool
+            }),
     "pointopoint":  dict({
                 "name": "pointopoint", 
                 "help": "If the interface is a P2P link, the remote endpoint's IP "
@@ -840,6 +987,14 @@ attributes = dict({
                 "flags": Attribute.ExecReadOnly | Attribute.ExecImmutable,
                 "validation_function": validation.is_string
             }),
+    "bwlimit":  dict({
+                "name": "bwlimit", 
+                "help": "Emulated transmission speed (in kbytes per second)",
+                "type": Attribute.INTEGER,
+                "range" : (1,10*2**20),
+                "flags": Attribute.ExecReadOnly | Attribute.ExecImmutable,
+                "validation_function": validation.is_integer
+            }),
     "txqueuelen":  dict({
                 "name": "txqueuelen", 
                 "help": "Transmission queue length (in packets)",
@@ -998,6 +1153,20 @@ attributes = dict({
                 "range": (0,60000),
                 "validation_function": validation.is_integer,
             }),
+    "module": dict({
+                "name": "module",
+                "help": "Path to a .c or .py source for a filter module, or a binary .so",
+                "type": Attribute.STRING,
+                "flags": Attribute.ExecReadOnly | Attribute.ExecImmutable,
+                "validation_function": validation.is_string
+            }),
+    "args": dict({
+                "name": "args",
+                "help": "Module arguments - comma-separated list of name=value pairs",
+                "type": Attribute.STRING,
+                "flags": Attribute.ExecReadOnly | Attribute.ExecImmutable,
+                "validation_function": validation.is_string
+            }),
     })
 
 traces = dict({
@@ -1027,9 +1196,17 @@ traces = dict({
                 "name": "pcap",
                 "help": "PCAP trace of all packets going through the interface",
               }),
+    "output": dict({
+                "name": "output",
+                "help": "Extra output trace for applications. When activated this trace can be referenced with wildcard a reference from an Application command line. Ex: command: 'tcpdump -w {#[elemet-label].trace[trace-id].[name|path]#}' ",
+              }),
+    "dropped_stats": dict({
+                "name": "dropped_stats",
+                "help": "Information on dropped packets on a filer or queue associated to a network interface",
+            }),
     })
 
-create_order = [ INTERNET, NODE, NODEIFACE, TAPIFACE, TUNIFACE, NETPIPE, NEPIDEPENDENCY, NS3DEPENDENCY, DEPENDENCY, APPLICATION ]
+create_order = [ INTERNET, NODE, NODEIFACE, CLASSQUEUEFILTER, TOSQUEUEFILTER, TUNFILTER, TAPIFACE, TUNIFACE, NETPIPE, NEPIDEPENDENCY, NS3DEPENDENCY, DEPENDENCY, APPLICATION ]
 
 configure_order = [ INTERNET, Parallel(NODE), NODEIFACE, Parallel(TAPIFACE), Parallel(TUNIFACE), NETPIPE, Parallel(NEPIDEPENDENCY), Parallel(NS3DEPENDENCY), Parallel(DEPENDENCY), Parallel(APPLICATION) ]
 
@@ -1078,9 +1255,9 @@ factories_info = dict({
             "create_function": create_tuniface,
             "preconfigure_function": preconfigure_tuniface,
             "configure_function": postconfigure_tuniface,
-            "prestart_function": wait_tuniface,
+            "prestart_function": prestart_tuniface,
             "box_attributes": [
-                "up", "device_name", "mtu", "snat", "pointopoint",
+                "up", "if_name", "mtu", "snat", "pointopoint", "multicast", "bwlimit",
                 "txqueuelen",
                 "tun_proto", "tun_addr", "tun_port", "tun_key", "tun_cipher",
             ],
@@ -1094,9 +1271,9 @@ factories_info = dict({
             "create_function": create_tapiface,
             "preconfigure_function": preconfigure_tuniface,
             "configure_function": postconfigure_tuniface,
-            "prestart_function": wait_tuniface,
+            "prestart_function": prestart_tuniface,
             "box_attributes": [
-                "up", "device_name", "mtu", "snat", "pointopoint",
+                "up", "if_name", "mtu", "snat", "pointopoint", "multicast", "bwlimit",
                 "txqueuelen",
                 "tun_proto", "tun_addr", "tun_port", "tun_key", "tun_cipher",
             ],
@@ -1104,6 +1281,110 @@ factories_info = dict({
             "connector_types": ["node","udp","tcp","fd->","gre"],
             "tags": [tags.INTERFACE, tags.ALLOW_ADDRESSES],
         }),
+    TUNFILTER: dict({
+            "help": "TUN/TAP stream filter\n\n"
+                    "If specified, it should be either a .py or .so module. "
+                    "It will be loaded, and all incoming and outgoing packets "
+                    "will be routed through it. The filter will not be responsible "
+                    "for buffering, packet queueing is performed in tun_connect "
+                    "already, so it should not concern itself with it. It should "
+                    "not, however, block in one direction if the other is congested.\n"
+                    "\n"
+                    "Modules are expected to have the following methods:\n"
+                    "\tinit(**args)\n"
+                    "\t\tIf arguments are given, this method will be called with the\n"
+                    "\t\tgiven arguments (as keyword args in python modules, or a single\n"
+                    "\taccept_packet(packet, direction):\n"
+                    "\t\tDecide whether to drop the packet. Direction is 0 for packets "
+                        "coming from the local side to the remote, and 1 is for packets "
+                        "coming from the remote side to the local. Return a boolean, "
+                        "true if the packet is not to be dropped.\n"
+                    "\tfilter_init():\n"
+                    "\t\tInitializes a filtering pipe (filter_run). It should "
+                        "return two file descriptors to use as a bidirectional "
+                        "pipe: local and remote. 'local' is where packets from the "
+                        "local side will be written to. After filtering, those packets "
+                        "should be written to 'remote', where tun_connect will read "
+                        "from, and it will forward them to the remote peer. "
+                        "Packets from the remote peer will be written to 'remote', "
+                        "where the filter is expected to read from, and eventually "
+                        "forward them to the local side. If the file descriptors are "
+                        "not nonblocking, they will be set to nonblocking. So it's "
+                        "better to set them from the start like that.\n"
+                    "\tfilter_run(local, remote):\n"
+                    "\t\tIf filter_init is provided, it will be called repeatedly, "
+                        "in a separate thread until the process is killed. It should "
+                        "sleep at most for a second.\n"
+                    "\tfilter_close(local, remote):\n"
+                    "\t\tCalled then the process is killed, if filter_init was provided. "
+                        "It should, among other things, close the file descriptors.\n"
+                    "\n"
+                    "Python modules are expected to return a tuple in filter_init, "
+                    "either of file descriptors or file objects, while native ones "
+                    "will receive two int*.\n"
+                    "\n"
+                    "Python modules can additionally contain a custom queue class "
+                    "that will replace the FIFO used by default. The class should "
+                    "be named 'queueclass' and contain an interface compatible with "
+                    "collections.deque. That is, indexing (especiall for q[0]), "
+                    "bool(q), popleft, appendleft, pop (right), append (right), "
+                    "len(q) and clear.",
+            "category": FC.CATEGORY_CHANNELS,
+            "create_function": create_tunfilter,
+            "box_attributes": [
+                "module", "args",
+                "tun_proto", "tun_addr", "tun_port", "tun_key", "tun_cipher",
+            ],
+            "connector_types": ["->fd","udp","tcp"],
+        }),
+    CLASSQUEUEFILTER : dict({
+            "help": "TUN classfull queue, uses a separate queue for each user-definable class.\n\n"
+                    "It takes two arguments, both of which have sensible defaults:\n"
+                    "\tsize: the base size of each class' queue\n"
+                    "\tclasses: the class definitions, which follow the following syntax:\n"
+                    '\t   <CLASSLIST> ::= <CLASS> ":" CLASSLIST\n'
+                    '\t                |  <CLASS>\n'
+                    '\t   <CLASS>     ::= <PROTOLIST> "*" <PRIORITYSPEC>\n'
+                    '\t                |  <DFLTCLASS>\n'
+                    '\t   <DFLTCLASS> ::= "*" <PRIORITYSPEC>\n'
+                    '\t   <PROTOLIST> ::= <PROTO> "." <PROTOLIST>\n'
+                    '\t                |  <PROTO>\n'
+                    '\t   <PROTO>     ::= <NAME> | <NUMBER>\n'
+                    '\t   <NAME>      ::= --see http://en.wikipedia.org/wiki/List_of_IP_protocol_numbers --\n'
+                    '\t                   --only in lowercase, with special characters removed--\n'
+                    '\t                   --or see below--\n'
+                    '\t   <NUMBER>    ::= [0-9]+\n'
+                    '\t   <PRIORITYSPEC> ::= <THOUGHPUT> [ "#" <SIZE> ] [ "p" <PRIORITY> ]\n'
+                    '\t   <THOUGHPUT> ::= NUMBER -- default 1\n'
+                    '\t   <PRIORITY>  ::= NUMBER -- default 0\n'
+                    '\t   <SIZE>      ::= NUMBER -- default 1\n'
+                    "\n"
+                    "Size, thoughput and priority are all relative terms. "
+                    "Sizes are multipliers for the size argument, thoughput "
+                    "is applied relative to other classes and the same with "
+                    "priority.",
+            "category": FC.CATEGORY_CHANNELS,
+            "create_function": create_classqueuefilter,
+            "box_attributes": [
+                "args",
+                "tun_proto", "tun_addr", "tun_port", "tun_key", "tun_cipher",
+            ],
+            "connector_types": ["->fd","udp","tcp"],
+            "traces": ["dropped_stats"],
+        }),
+    TOSQUEUEFILTER : dict({
+            "help": "TUN classfull queue that classifies according to the TOS (RFC 791) IP field.\n\n"
+                    "It takes a size argument that specifies the size of each class. As TOS is a "
+                    "subset of DiffServ, this queue half-implements DiffServ.",
+            "category": FC.CATEGORY_CHANNELS,
+            "create_function": create_tosqueuefilter,
+            "box_attributes": [
+                "args",
+                "tun_proto", "tun_addr", "tun_port", "tun_key", "tun_cipher",
+            ],
+            "connector_types": ["->fd","udp","tcp"],
+        }),
+
     APPLICATION: dict({
             "help": "Generic executable command line application",
             "category": FC.CATEGORY_APPLICATIONS,
@@ -1116,7 +1397,7 @@ factories_info = dict({
                                "depends", "build-depends", "build", "install",
                                "sources", "rpm-fusion" ],
             "connector_types": ["node"],
-            "traces": ["stdout", "stderr", "buildlog"],
+            "traces": ["stdout", "stderr", "buildlog", "output"],
             "tags": [tags.APPLICATION],
         }),
     DEPENDENCY: dict({