Merge branch 'develop' into 'master'

added first version ready for integration testing

See merge request !1

connectors/tun.py

@@ -24,9 +24,10 @@ class TunConsumer(BaseConsumer):
     def __init__(self, user, password, session, server, name, consumer_name):
         super(TunConsumer, self).__init__(user, password, session, server, name, consumer_name)
         self.dispatcher = {
-            "start": self.handle_start,
+            "tun.start": self.handle_start,
         }
 
+
     def handle_start(self, msg):
         """
         Function that will handle tun management messages
@@ -42,20 +43,25 @@ class TunConsumer(BaseConsumer):
 
         example:
         {
-            "_type": "start",
+            "_type": "tun.start",
             "ipv6_host": ":2",
             "ipv6_prefix": "cccc"
         }
 
         - stop_tun
         """
-
-        ipv6_host = msg.get("ipv6_host", None)
-        ipv6_prefix = msg.get("ipv6_prefix", None)
-        ipv4_host = msg.get("ipv4_host", None)
-        ipv4_network = msg.get("ipv4_network", None)
-        ipv4_netmask = msg.get("ipv4_netmask", None)
-
+        logging.info('starting tun interface')
+        try:
+            ipv6_host = msg.get("ipv6_host", None)
+            ipv6_prefix = msg.get("ipv6_prefix", None)
+            ipv4_host = msg.get("ipv4_host", None)
+            ipv4_network = msg.get("ipv4_network", None)
+            ipv4_netmask = msg.get("ipv4_netmask", None)
+        except AttributeError as ae:
+            logging.error(
+                    'Wrong message format: {0}'.format(msg.payload)
+            )
+            return
 
 
         params = {
@@ -73,25 +79,44 @@ class TunConsumer(BaseConsumer):
             sys.exit(1)
 
         elif sys.platform.startswith('linux'):
+            logging.info('Starting open tun [linux]')
             self.tun = OpenTunLinux(**params)
 
         elif sys.platform.startswith('darwin'):
+            logging.info('Starting open tun [darwin]')
             self.tun = OpenTunMACOS(**params)
         else:
             log.error('Agent TunTap not yet supported for: {0}'.format(sys.platform))
             sys.exit(1)
 
 
+    def handle_data(self, body, message):
+        """
+
+        Args:
+            msg:
+
+        Returns:
+
+        """
+        self.log.debug("HANDLE DATA from tun")
+        self.log.debug(("Payload", message.payload))
+        self.log.debug(("Properties", message.properties))
+        self.log.debug(("Headers", message.headers))
+        self.log.debug(("body", message.body))
 
     def handle_control(self, body, message):
         msg = None
         try:
+            # body is a json
             msg = json.loads(body)
             log.debug(message)
-        except ValueError as e:
+            if msg["_type"]:
+                log.debug('HANDLE CONTROL from tun processing event type: {0}'.format(msg["_type"]))
+        except (ValueError,KeyError) as e:
             message.ack()
             log.error(e)
-            log.error("Incorrect message: {0}".format(body))
+            log.error("Incorrect message: {0}".format(message.payload))
             return
         if msg["_type"] in self.dispatcher.keys():
             self.dispatcher[msg["_type"]](msg)

tests/myTest.py

+import json
+import logging
+import click
+from kombu import Connection, Exchange, Queue
+from kombu.mixins import ConsumerMixin
+
+from finterop import DEFAULT_IPV6_PREFIX
+from finterop.utils.tun import OpenTunLinux
+
+DEFAULT_PLATFORM = "f-interop.paris.inria.fr"
+LOG_FORMAT = ('%(levelname) -10s %(asctime)s %(name) -30s %(funcName) '
+              '-35s %(lineno) -5d: %(message)s')
+LOGGER = logging.getLogger("agent")
+
+logging.basicConfig(level=logging.DEBUG)
+log = logging.getLogger(__name__)
+
+__version__ = (0, 0, 1)
+
+
+class MyTest(ConsumerMixin):
+    def __init__(self, connection, name="myTest"):
+        self.name = name
+        self.connection = connection
+        self.producer = self.connection.Producer(serializer='json')
+
+        self.exchange = Exchange('default', type="topic", durable=True)
+        self.control_queue = Queue("control_{name}".format(name=name),
+                                   exchange=self.exchange,
+                                   durable=True,
+                                   routing_key="control.fromAgent.#")
+        self.data_queue = Queue("data_{name}".format(name=name),
+                                exchange=self.exchange,
+                                durable=True,
+                                routing_key="data.fromAgent.#")
+        self.tun = OpenTunLinux(
+            name=self.name,
+            rmq_connection=self.connection,
+            ipv6_host=":2",
+            ipv6_prefix=DEFAULT_IPV6_PREFIX
+        )
+
+        log.info("Let's bootstrap this.")
+        log.info("First let's start the tun device on this side")
+
+    def get_consumers(self, Consumer, channel):
+        return [
+            Consumer(queues=[self.control_queue],
+                     callbacks=[self.handle_control],
+                     no_ack=True,
+                     accept=['json']),
+            Consumer(queues=[self.data_queue],
+                     callbacks=[self.handle_data],
+                     no_ack=True,
+                     accept=["json"])
+        ]
+
+    def handle_control(self, body, message):
+        """
+        """
+        log.info("Let's handle control messages")
+        msg = None
+        log.debug("Here is the type of body: %s" % type(body))
+        log.debug("Here is the body")
+        log.debug(body)
+        log.debug("Here is the message")
+        log.debug(message)
+
+        # if type(body) == "dict":
+        #     msg = body
+        # else:
+        #     try:
+        #         msg = json.loads(body)
+        #         log.debug(message)
+        #     except ValueError as e:
+        #         message.ack()
+        #         log.error(e)
+        #         log.error("Incorrect message: {0}".format(body))
+
+        #     except TypeError as e:
+        #         message.ack()
+        #         log.error(e)
+        #         log.error("A problem with string / buffer happened?")
+
+        if body is not None:
+
+            log.debug("Just received that packet")
+            log.debug(body)
+            if "order" in body.keys():
+                if body["order"] == "bootstrap":
+                    self.handle_bootstrap()
+
+    def handle_bootstrap(self):
+        log.debug("Let's start the bootstrap")
+        self.producer.publish(json.dumps({"order": "tun.start",
+                                          "ipv6_host": ":1",
+                                          "ipv6_prefix": "bbbb"}),
+                              exchange=self.exchange,
+                              routing_key="control.toAgent.client")
+
+    def handle_data(self, body, message):
+        """
+        """
+        log.info("Let's handle data messages")
+        log.debug("Here is the type of body")
+        log.debug(type(body))
+        log.debug("Here is the body")
+        log.debug(body)
+        log.debug("Here is the message")
+        log.debug(message)
+        msg = json.loads(body)
+
+        # We are only with two nodes. Any packet from the client is for us.
+        # These lines do the routing between the two
+        if msg["routing_key"] == "data.fromAgent.client":
+            log.debug("Message was routed, therefore we can inject it on our tun")
+            self.tun._v6ToInternet_notif(sender="test",
+                                         signal="tun",
+                                         data=msg["data"])
+        else:
+            self.producer.publish(msg,
+                                  exchange=self.exchange,
+                                  routing_key="data.toAgent.client")
+
+    def handle_routing(self):
+        """
+        In charge of routing packets back and forth between client and server
+        Returns:
+
+        """
+        log.info("Should implement that")
+
+
+@click.group()
+@click.version_option(str(__version__))
+def cli():
+    """
+    myTest package
+    """
+    pass
+
+
+@cli.command("my_test")
+@click.argument("session")
+@click.option("--user",
+              default="finterop",
+              help="finterop AMQP username")
+@click.option("--password",
+              default="finterop",
+              help="finterop AMQP password")
+@click.option("--server",
+              default=DEFAULT_PLATFORM,
+              help="f-interop platform (default: %s)" % DEFAULT_PLATFORM)
+def my_test(session, user, password, server):
+    c = Connection(
+        'amqp://{user}:{password}@{server}/{session}'.format(user=user,
+                                                             password=password,
+                                                             session=session,
+                                                             server=server),
+        transport_options={'confirm_publish': True})
+
+    test = MyTest(c)
+    test.run()
+
+
+if __name__ == '__main__':
+    cli()

tests/readme.md

+Tests
+=====
+
+Tests performs tasks such as:
+
+- Send command to start features of agent such as tun, serial sniffing, ...
+- Packet forwarding between two remote agents.
+- Packet drop in test that target lossy links.
+- Packet injection in test that target fuzzing.

tests/snippet_base_test.py

+class MyTest(ConsumerMixin):
+
+    def __init__(self, connection, name="myTest"):
+        self.name = name
+        self.connection = connection
+        self.producer = self.connection.Producer(serializer='json')
+
+        self.exchange = Exchange('default', type="topic", durable=True)
+        self.control_queue = Queue("control_{name}".format(name=name),
+                                   exchange=self.exchange,
+                                   durable=True,
+                                   routing_key="control.fromAgent.#")
+        self.data_queue = Queue("data_{name}".format(name=name),
+                                exchange=self.exchange,
+                                durable=True,
+                                routing_key="data.fromAgent.#")
+        self.tun = OpenTunLinux(
+            name=self.name,
+            rmq_connection=self.connection,
+            ipv6_host=":2",
+            ipv6_prefix=DEFAULT_IPV6_PREFIX
+        )
+
+    def get_consumers(self, Consumer, channel):
+        return [
+            Consumer(queues=[self.control_queue],
+                     callbacks=[self.handle_control],
+                     no_ack=True,
+                     accept=['json']),
+            Consumer(queues=[self.data_queue],
+                     callbacks=[self.handle_data],
+                     no_ack=True,
+                     accept=["json"])
+        ]
\ No newline at end of file

tests/test_agent.py

+import json
+import logging
+
+import pika
+import pytest
+
+log = logging.getLogger(__name__)
+
+
+"""
+primer from 6p openwsn-fw/build/OpenMote-CC2538_armgcc/openstack/02b-MAChigh/sixtop.h
+
+//=========================== define ==========================================
+// 6P version
+#define IANA_6TOP_6P_VERSION   0x01
+// 6P command Id
+#define IANA_6TOP_CMD_NONE     0x00
+#define IANA_6TOP_CMD_ADD      0x01 // CMD_ADD      | add one or more cells
+#define IANA_6TOP_CMD_DELETE   0x02 // CMD_DELETE   | delete one or more cells
+#define IANA_6TOP_CMD_COUNT    0x03 // CMD_COUNT    | count scheduled cells
+#define IANA_6TOP_CMD_LIST     0x04 // CMD_LIST     | list the scheduled cells
+#define IANA_6TOP_CMD_CLEAR    0x05 // CMD_CLEAR    | clear all cells
+// 6P return code
+#define IANA_6TOP_RC_SUCCESS   0x06 // RC_SUCCESS  | operation succeeded
+#define IANA_6TOP_RC_VER_ERR   0x07 // RC_VER_ERR  | unsupported 6P version
+#define IANA_6TOP_RC_SFID_ERR  0x08 // RC_SFID_ERR | unsupported SFID
+#define IANA_6TOP_RC_BUSY      0x09 // RC_BUSY     | handling previous request
+#define IANA_6TOP_RC_RESET     0x0a // RC_RESET    | abort 6P transaction
+#define IANA_6TOP_RC_ERR       0x0b // RC_ERR      | operation failed
+
+// states of the sixtop-to-sixtop state machine
+typedef enum {
+    // ready for next event
+    SIX_IDLE                            = 0x00,
+    // sending
+    SIX_SENDING_REQUEST                 = 0x01,
+    // waiting for SendDone confirmation
+    SIX_WAIT_ADDREQUEST_SENDDONE        = 0x02,
+    SIX_WAIT_DELETEREQUEST_SENDDONE     = 0x03,
+    SIX_WAIT_COUNTREQUEST_SENDDONE      = 0x04,
+    SIX_WAIT_LISTREQUEST_SENDDONE       = 0x05,
+    SIX_WAIT_CLEARREQUEST_SENDDONE      = 0x06,
+    // waiting for response from the neighbor
+    SIX_WAIT_ADDRESPONSE                = 0x07,
+    SIX_WAIT_DELETERESPONSE             = 0x08,
+    SIX_WAIT_COUNTRESPONSE              = 0x09,
+    SIX_WAIT_LISTRESPONSE               = 0x0a,
+    SIX_WAIT_CLEARRESPONSE              = 0x0b,
+
+    // response senddone
+    SIX_REQUEST_RECEIVED                = 0x0c,
+    SIX_WAIT_RESPONSE_SENDDONE          = 0x0d
+} six2six_state_t;
+
+"""
+
+@pytest.fixture
+def producer():
+    connection = pika.BlockingConnection(
+        pika.connection.URLParameters('amqp://paul:iamthewalrus@f-interop.rennes.inria.fr/session01'))
+    channel = connection.channel()
+    channel.confirm_delivery()
+    return channel
+
+
+def test_ping(producer):
+    """
+    Launch a ping towards localhost through the agent
+
+    Returns:
+
+    """
+    d = {"version": "IPv6",
+         "count": 1,
+         "host": "localhost",
+         "payload": "cafe",
+         "interface": "tun0"}
+    producer.basic_publish(exchange='default',
+                           routing_key='control.ping.coucou.client',
+                           mandatory=True,
+                           body=json.dumps(d))
+
+
+def test_http(producer):
+    """
+
+    Returns:
+
+    """
+    d = {"verb": "GET",
+         "url": "http://f-interop.paris.inria.fr",
+         "data": {}
+         }
+    producer.basic_publish(exchange='default',
+                           routing_key='control.http.coucou.client',
+                           mandatory=True,
+                           body=json.dumps(d))
+
+
+def test_zmq(producer):
+    """
+
+    Returns:
+
+    """
+    d = {"zmq_socket_type": "req",
+         "url": "tcp://localhost:60000",
+         "payload": {"signal": "cmdToMote",
+                     "sender": "f-interop agent",
+                     "data": {
+                         "action": ["imageCommand", "6pAdd", "[6,7]"],
+                         "serialPort": "/dev/ttyUSB0"}}}
+    producer.basic_publish(exchange='default',
+                           routing_key='control.zeromq.req.client',
+                           mandatory=True,
+                           body=json.dumps(d))
+
+
+def test_6p_add(producer):
+    """
+
+    Used in 6TiSCH testing.
+    TODO: Implemented by which nodes? DAG ? 6N ?
+
+    """
+    d = {"zmq_socket_type": "req",
+         "url": "tcp://localhost:60000",
+         "payload": {"signal": "cmdToMote",
+                     "data": {
+                         "action": ["imageCommand", "6pAdd", "[6,7]"],
+                         "serialPort": "/dev/ttyUSB0"
+                     },
+                     "sender": "f-interop agent"}}
+
+    producer.basic_publish(exchange='default',
+                           routing_key='control.zeromq.req.client',
+                           mandatory=True,
+                           body=json.dumps(d))
+
+
+def test_6p_count(producer):
+    """
+
+    TODO: Write docs
+    Used in 6TiSCH testing.
+    TODO: Implemented by which nodes? DAG ? 6N ?
+
+    """
+    d = {"zmq_socket_type": "req",
+         "url": "tcp://localhost:60000",
+         "payload": {"signal": "cmdToMote",
+                     "data": {
+                         "action": ["imageCommand", "6pCount", "0"],
+                         "serialPort": "/dev/ttyUSB0"
+                     },
+                     "sender": "f-interop agent"}}
+
+    producer.basic_publish(exchange='default',
+                           routing_key='control.zeromq.req.client',
+                           mandatory=True,
+                           body=json.dumps(d))
+
+
+def test_6p_list(producer):
+    """
+    TODO: Write docs
+    Used in 6TiSCH testing.
+
+    TODO: Implemented by which nodes? DAG ? 6N ?
+
+    """
+    d = {"zmq_socket_type": "req",
+         "url": "tcp://localhost:60000",
+         "payload": {"signal": "cmdToMote",
+                     "data": {
+                         "action": ["imageCommand", "6pList", "0"],
+                         "serialPort": "/dev/ttyUSB0"
+                     },
+                     "sender": "f-interop agent"}}
+
+    producer.basic_publish(exchange='default',
+                           routing_key='control.zeromq.req.client',
+                           mandatory=True,
+                           body=json.dumps(d))
+
+
+def test_6p_clear(producer):
+    """
+    TODO: Write docs
+    Used in 6TiSCH testing.
+    TODO: Implemented by which nodes? DAG ? 6N ?
+
+    """
+    d = {"zmq_socket_type": "req",
+         "url": "tcp://localhost:60000",
+         "payload": {"signal": "cmdToMote",
+                     "data": {
+                         "action": ["imageCommand", "6pClear", "0"],
+                         "serialPort": "/dev/ttyUSB0"
+                     },
+                     "sender": "f-interop agent"}}
+
+    producer.basic_publish(exchange='default',
+                           routing_key='control.zeromq.req.client',
+                           mandatory=True,
+                           body=json.dumps(d))
+
+
+def test_6p_delete(producer):
+    """
+
+    TODO: Write docs
+    Use 3 arguments
+    Used in 6TiSCH testing.
+    TODO: Implemented by which nodes? DAG ? 6N ?
+
+    """
+    d = {"zmq_socket_type": "req",
+         "url": "tcp://localhost:60000",
+         "payload": {"signal": "cmdToMote",
+                     "data": {
+                         "action": ["imageCommand", "6pDelete", "[6,7]"],
+                         "serialPort": "/dev/ttyUSB0"
+                     },
+                     "sender": "f-interop agent"}}
+
+    producer.basic_publish(exchange='default',
+                           routing_key='control.zeromq.req.client',
+                           mandatory=True,
+                           body=json.dumps(d))
+
+
+def test_6p_answers(producer):
+    """
+    enable/disable 6P response (set value to 1 to enable, 0 to disable)
+    Test if 6P answers can be enable/disabled to create timeouts.
+    Used in 6TiSCH testing.
+    TODO: Implemented by which nodes? DAG ? 6N ?
+    """
+    d = {"zmq_socket_type": "req",
+         "url": "tcp://localhost:60000",
+         "payload": {"signal": "cmdToMote",
+                     "data": {
+                         "action": ["imageCommand", "6pResponse", "1"],
+                         "serialPort": "/dev/ttyUSB0"
+                     },
+                     "sender": "f-interop agent"}}
+
+    producer.basic_publish(exchange='default',
+                           routing_key='control.zeromq.req.client',
+                           mandatory=True,
+                           body=json.dumps(d))
+
+
+def test_send_dio(producer):
+    """
+
+    Used in 6TiSCH testing.
+    set the period at the which the device sends DIOs (value in milli-seconds)
+    TODO: Implemented by which nodes? DAG ? 6N ?
+    Returns:
+
+    """
+    d = {"zmq_socket_type": "req",
+         "url": "tcp://localhost:60000",
+         "payload": {"signal": "cmdToMote",
+                     "data": {
+                         "action": ["imageCommand", "dioPeriod", "1000"],
+                         "serialPort": "/dev/ttyUSB0"
+                     },
+                     "sender": "f-interop agent"}}
+
+    producer.basic_publish(exchange='default',
+                           routing_key='control.zeromq.req.client',
+                           mandatory=True,
+                           body=json.dumps(d))
+
+
+def test_send_dao(producer):
+    """
+
+    Used in 6TiSCH testing.
+    TODO: Implemented by which nodes? DAG ? 6N ?
+    set the period at the which the device sends DAOs (value in milli-seconds)
+    Returns:
+
+    """
+    d = {"zmq_socket_type": "req",
+         "url": "tcp://localhost:60000",
+         "payload": {"signal": "cmdToMote",
+                     "data": {
+                         "action": ["imageCommand", "daoPeriod", "1000"],
+                         "serialPort": "/dev/ttyUSB0"
+                     },
+                     "sender": "f-interop agent"}}
+
+    producer.basic_publish(exchange='default',
+                           routing_key='control.zeromq.req.client',
+                           body=json.dumps(d))
+
+
+def test_send_keep_alive(producer):
+    """
+    TODO: Implemented by which nodes? DAG ? 6N ?
+    Used in 6TiSCH testing.
+    set the timeout which the device send a keepAlive(value in slots)
+    """
+    d = {"zmq_socket_type": "req",
+         "url": "tcp://localhost:60000",
+         "payload": {"signal": "cmdToMote",
+                     "data": {
+                         "action": ["imageCommand", "kaPeriod", "1000"],
+                         "serialPort": "/dev/ttyUSB0"
+                     },
+                     "sender": "f-interop agent"}}
+