diff --git a/tests/scripts/Makefile.am b/tests/scripts/Makefile.am index 17ed5fd43..c120be3d2 100644 --- a/tests/scripts/Makefile.am +++ b/tests/scripts/Makefile.am @@ -135,6 +135,7 @@ EXTRA_DIST = \ thread-cert/node.py \ thread-cert/node_cli.py \ thread-cert/sniffer.py \ + thread-cert/sniffer_transport.py \ thread-cert/test_coap.py \ thread-cert/test_common.py \ thread-cert/test_crypto.py \ diff --git a/tests/scripts/thread-cert/message.py b/tests/scripts/thread-cert/message.py index 287317981..ac4e39efa 100644 --- a/tests/scripts/thread-cert/message.py +++ b/tests/scripts/thread-cert/message.py @@ -250,7 +250,7 @@ class Message(object): assert self.ipv6_packet.ipv6_header.hop_limit == hop_limit def __repr__(self): - return "Message(type={})".format(MessageType.name[self.type]) + return "Message(type={})".format(self.type) class MessagesSet(object): diff --git a/tests/scripts/thread-cert/sniffer.py b/tests/scripts/thread-cert/sniffer.py index 5b51b6ff5..1e371c641 100644 --- a/tests/scripts/thread-cert/sniffer.py +++ b/tests/scripts/thread-cert/sniffer.py @@ -30,9 +30,6 @@ import collections import io import logging -import os -import select -import socket import threading try: @@ -41,6 +38,7 @@ except ImportError: import queue as Queue import message +import sniffer_transport class Sniffer: @@ -49,16 +47,8 @@ class Sniffer: logger = logging.getLogger("sniffer.Sniffer") - POLL_TIMEOUT = 0.10 - RECV_BUFFER_SIZE = 4096 - BASE_PORT = 9000 - - WELLKNOWN_NODE_ID = 34 - - PORT_OFFSET = int(os.getenv('PORT_OFFSET', "0")) - def __init__(self, nodeid, message_factory): """ Args: @@ -69,101 +59,65 @@ class Sniffer: self.nodeid = nodeid self._message_factory = message_factory - self._socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - if self._socket is None: - raise RuntimeError("Could not create socket.") - - self._socket.bind(self.address) + # Create transport + transport_factory = sniffer_transport.SnifferTransportFactory() + self._transport = transport_factory.create_transport(nodeid) self._thread = None - self._thread_alive_event = threading.Event() - - self._poll = select.poll() + self._thread_alive = threading.Event() + self._thread_alive.clear() self._buckets = collections.defaultdict(Queue.Queue) def __del__(self): - if self._socket is None: - return + del self._transport - self._socket.close() - del self._socket + def _sniffer_main_loop(self): + """ Sniffer main loop. """ - def _nodeid_to_address(self, nodeid, ip_address="localhost"): - return "", self.BASE_PORT + (self.PORT_OFFSET * self.WELLKNOWN_NODE_ID) + nodeid + self.logger.debug("Sniffer started.") - def _address_to_nodeid(self, address): - ip_address, port = address - return (port - self.BASE_PORT - (self.PORT_OFFSET * self.WELLKNOWN_NODE_ID)) + while self._thread_alive.is_set(): + data, nodeid = self._transport.recv(self.RECV_BUFFER_SIZE) - def _recv(self, fd): - """ Receive data from socket with passed file descriptor. """ + msg = self._message_factory.create(io.BytesIO(data)) - data, address = self._socket.recvfrom(self.RECV_BUFFER_SIZE) + if msg is not None: + self.logger.debug("Received message: {}".format(msg)) + self._buckets[nodeid].put(msg) - msg = self._message_factory.create(io.BytesIO(data)) - - if msg is None: - self.logger.debug("Received 6LowPAN fragment.") - return - - nodeid = self._address_to_nodeid(address) - - self._buckets[nodeid].put(msg) - - @property - def address(self): - """ Sniffer address. """ - - return self._nodeid_to_address(self.nodeid) - - def _run(self): - """ Receive thread main loop. """ - - while self._thread_alive_event.is_set(): - reported_events = self._poll.poll(self.POLL_TIMEOUT) - - for fd_event_pair in reported_events: - fd, event = fd_event_pair - - if event & select.POLLIN or event & select.POLLPRI: - self._recv(fd) - - elif event & select.POLLERR: - self.logger.error("Error condition of some sort") - self._thread_alive_event.clear() - break - - elif event & select.POLLNVAL: - self.logger.error("Invalid request: descriptor not open") - self._thread_alive_event.clear() - break + self.logger.debug("Sniffer stopped.") def start(self): """ Start sniffing. """ - self._poll.register(self._socket, select.POLLIN | select.POLLPRI | select.POLLERR | select.POLLNVAL) - - self._thread = threading.Thread(target=self._run) + self._thread = threading.Thread(target=self._sniffer_main_loop) self._thread.daemon = True - self._thread_alive_event.set() + self._transport.open() + + self._thread_alive.set() self._thread.start() def stop(self): """ Stop sniffing. """ - self._poll.unregister(self._socket) - - self._thread_alive_event.clear() + self._thread_alive.clear() self._thread.join() self._thread = None + self._transport.close() + def get_messages_sent_by(self, nodeid): """ Get sniffed messages. Note! This method flushes the message queue so calling this method again will return only the newly logged messages. + Args: + nodeid (int): node id + + Returns: + MessagesSet: a set with received messages. """ bucket = self._buckets[nodeid] messages = [] diff --git a/tests/scripts/thread-cert/sniffer_transport.py b/tests/scripts/thread-cert/sniffer_transport.py new file mode 100644 index 000000000..7115ea0f7 --- /dev/null +++ b/tests/scripts/thread-cert/sniffer_transport.py @@ -0,0 +1,158 @@ +#!/usr/bin/python +# +# Copyright (c) 2016, The OpenThread Authors. +# All rights reserved. +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# 1. Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# 2. Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# 3. Neither the name of the copyright holder nor the +# names of its contributors may be used to endorse or promote products +# derived from this software without specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. +# + +import os +import socket +import sys + + +class SnifferTransport(object): + """ Interface for transport that allows eavesdrop other nodes. """ + + def open(self): + """ Open transport. + + Raises: + RuntimeError: when transport is already opened or when transport opening failed. + """ + raise NotImplementedError + + def close(self): + """ Close transport. + + Raises: + RuntimeError: when transport is already closed. + """ + raise NotImplementedError + + @property + def is_opened(self): + """ Check if transport is opened. + + Returns: + bool: True if the transport is opened, False in otherwise + """ + raise NotImplementedError + + def send(self, data, nodeid): + """ Send data to the node with nodeid. + + Args: + data (bytearray): outcoming data. + nodeid (int): node id + + Returns: + int: number of sent bytes + """ + raise NotImplementedError + + def recv(self, bufsize): + """ Receive data sent by other node. + + Args: + bufsize (int): size of buffer for incoming data. + + Returns: + A tuple contains data and node id. + + For example: + (bytearray([0x00, 0x01...], 1) + """ + raise NotImplementedError + + +class SnifferSocketTransport(SnifferTransport): + """ Socket based implementation of sniffer transport. """ + + BASE_PORT = 9000 + + WELLKNOWN_NODE_ID = 34 + + PORT_OFFSET = int(os.getenv('PORT_OFFSET', "0")) + + def __init__(self, nodeid): + self._nodeid = nodeid + self._socket = None + + def __del__(self): + if not self.is_opened: + return + + self.close() + + def _nodeid_to_address(self, nodeid, ip_address=""): + return ip_address, self.BASE_PORT + (self.PORT_OFFSET * self.WELLKNOWN_NODE_ID) + nodeid + + def _address_to_nodeid(self, address): + _, port = address + return (port - self.BASE_PORT - (self.PORT_OFFSET * self.WELLKNOWN_NODE_ID)) + + def open(self): + if self.is_opened: + raise RuntimeError("Transport is already opened.") + + self._socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + + if not self.is_opened: + raise RuntimeError("Transport opening failed.") + + self._socket.bind(self._nodeid_to_address(self._nodeid)) + + def close(self): + if not self.is_opened: + raise RuntimeError("Transport is closed.") + + self._socket.close() + self._socket = None + + @property + def is_opened(self): + return bool(self._socket is not None) + + def send(self, data, nodeid): + address = self._nodeid_to_address(nodeid) + + return self._socket.sendto(data, address) + + def recv(self, bufsize): + data, address = self._socket.recvfrom(bufsize) + + nodeid = self._address_to_nodeid(address) + + return bytearray(data), nodeid + + +class SnifferTransportFactory(object): + + def create_transport(self, nodeid): + if sys.platform != "win32": + return SnifferSocketTransport(nodeid) + + else: + raise NotImplementedError