[thci] change implementation of sniffer simulation from SSH to gRPC (#7983)

The sniffer simulation now supports node filter functionality (equivalent to RF enclosure in the real world).
2026-08-08 03:37:46 +00:00 · 2022-08-08 09:19:07 -07:00
parent 24fd146e10
commit 62fcf85106
14 changed files with 343 additions and 135 deletions
@@ -11,19 +11,25 @@ Platform developers should modify the THCI implementation and/or the SI implemen
 ## POSIX Environment Setup

 1. Build OpenThread to generate standalone OpenThread simulation `ot-cli-ftd`. For example, run the following command in the top directory of OpenThread.
+
   ```bash
   $ script/cmake-build simulation
   ```
+
   Then `ot-cli-ftd` is built in the directory `build/simulation/examples/apps/cli/`.

+2. Run the installation script.
+
+   ```bash
+   $ tools/harness-simulation/posix/install.sh
+   ```
+
 ## Test Harness Environment Setup

 1. Double click the file `harness\install.bat` on the machine which installed Harness.

 2. Check the configuration file `C:\GRL\Thread1.2\Thread_Harness\simulation\config.py`

-   - Edit the value of `REMOTE_USERNAME` to the username expected to connect to on the remote POSIX environment.
-   - Edit the value of `REMOTE_PASSWORD` to the password corresponding to the username above.
   - Edit the value of `REMOTE_OT_PATH` to the absolute path where the top directory of the OpenThread repository is located.

 3. Add the additional simulation device information in `harness\Web\data\deviceInputFields.xml` to `C:\GRL\Thread1.2\Web\data\deviceInputFields.xml`.
@@ -34,9 +40,9 @@ Platform developers should modify the THCI implementation and/or the SI implemen

   ```bash
   $ cd tools/harness-simulation/posix
-   $ python harness_dev_discovery.py \
-         --interface=eth0            \
-         --ot1.1=24                  \
+   $ python launch_testbed.py \
+         --interface=eth0     \
+         --ot1.1=24           \
         --sniffer=2
   ```

@@ -27,27 +27,23 @@
 # POSSIBILITY OF SUCH DAMAGE.
 #

+import grpc
 import ipaddress
+import json
 import netifaces
-import os
-import paramiko
 import select
 import socket
 import struct
 import subprocess
 import time
+import win32api
 import winreg as wr

 from ISniffer import ISniffer
 from THCI.OpenThread import watched
-from simulation.config import (
-    REMOTE_PORT,
-    REMOTE_USERNAME,
-    REMOTE_PASSWORD,
-    REMOTE_OT_PATH,
-    REMOTE_SNIFFER_OUTPUT_PREFIX,
-    EDITCAP_PATH,
-)
+from simulation.config import EDITCAP_PATH
+from simulation.Sniffer.proto import sniffer_pb2
+from simulation.Sniffer.proto import sniffer_pb2_grpc

 DISCOVERY_ADDR = ('ff02::114', 12345)

@@ -68,12 +64,15 @@ class SimSniffer(ISniffer):
    @watched
    def __init__(self, **kwargs):
        self.channel = kwargs.get('channel')
-        self.ipaddr = kwargs.get('addressofDevice')
+        self.addr_port = kwargs.get('addressofDevice')
        self.is_active = False
        self._local_pcapng_location = None
-        self._ssh = None
-        self._remote_pcap_location = None
-        self._remote_pid = None
+        if self.addr_port is not None:
+            self._sniffer = grpc.insecure_channel(self.addr_port)
+            self._stub = sniffer_pb2_grpc.SnifferStub(self._sniffer)
+
+        # Close the sniffer only when Harness exits
+        win32api.SetConsoleCtrlHandler(self.__disconnect, True)

    def __repr__(self):
        return '%r' % self.__dict__
@@ -130,13 +129,14 @@ class SimSniffer(ISniffer):
        start = time.time()
        while time.time() - start < SCAN_TIME:
            if select.select([sock], [], [], 1)[0]:
-                addr, _ = sock.recvfrom(1024)
-                devs.add(addr)
+                data, _ = sock.recvfrom(1024)
+                data = json.loads(data)
+                devs.add((data['add'], data['por']))
            else:
                # Re-send the request, due to unreliability of UDP especially on WLAN
                sock.sendto(('Sniffer').encode(), DISCOVERY_ADDR)

-        devs = [SimSniffer(addressofDevice=addr, channel=None) for addr in devs]
+        devs = [SimSniffer(addressofDevice=self._encode_address_port(addr, port), channel=None) for addr, port in devs]
        self.log('List of SimSniffers: %r', devs)

        return devs
@@ -145,22 +145,10 @@ class SimSniffer(ISniffer):
    def startSniffer(self, channelToCapture, captureFileLocation, includeEthernet=False):
        self.channel = channelToCapture
        self._local_pcapng_location = captureFileLocation
-        self._remote_pcap_location = os.path.join(REMOTE_SNIFFER_OUTPUT_PREFIX, self.ipaddr.split('@')[0] + '.pcap')

-        self._ssh = paramiko.SSHClient()
-        self._ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
-        remote_ip = self.ipaddr.split('@')[1]
-        self._ssh.connect(remote_ip, port=REMOTE_PORT, username=REMOTE_USERNAME, password=REMOTE_PASSWORD)
-
-        _, stdout, _ = self._ssh.exec_command(
-            'echo $$ && exec python3 %s -o %s -c %d' %
-            (os.path.join(REMOTE_OT_PATH, 'tools/harness-simulation/posix/sniffer_sim/sniffer.py'),
-             self._remote_pcap_location, self.channel))
-        self._remote_pid = int(stdout.readline())
-
-        self.log('local pcapng location = %s', self._local_pcapng_location)
-        self.log('remote pcap location = %s', self._remote_pcap_location)
-        self.log('remote pid = %d', self._remote_pid)
+        response = self._stub.Start(sniffer_pb2.StartRequest(channel=self.channel))
+        if response.status != sniffer_pb2.OK:
+            raise RuntimeError(f'startSniffer error: {sniffer_pb2.Status.Name(response.status)}')

        self.is_active = True

@@ -168,30 +156,27 @@ class SimSniffer(ISniffer):
    def stopSniffer(self):
        if not self.is_active:
            return
-        self.is_active = False

-        assert self._ssh is not None
-        self._ssh.exec_command('kill -s TERM %d' % self._remote_pid)
-        # Wait to make sure the file is closed
-        time.sleep(3)
+        response = self._stub.Stop(sniffer_pb2.StopRequest())
+        if response.status != sniffer_pb2.OK:
+            raise RuntimeError(f'stopSniffer error: {sniffer_pb2.Status.Name(response.status)}')

        # Truncate suffix from .pcapng to .pcap
        local_pcap_location = self._local_pcapng_location[:-2]

-        with self._ssh.open_sftp() as sftp:
-            sftp.get(self._remote_pcap_location, local_pcap_location)
-
-        self._ssh.close()
+        with open(local_pcap_location, 'wb') as f:
+            f.write(response.pcap_content)

        cmd = [EDITCAP_PATH, '-F', 'pcapng', local_pcap_location, self._local_pcapng_location]
        self.log('running editcap: %r', cmd)
        subprocess.Popen(cmd).wait()
        self.log('editcap done')

-        self._local_pcapng_location = None
-        self._ssh = None
-        self._remote_pcap_location = None
-        self._remote_pid = None
+        self.is_active = False
+
+    def __disconnect(self, dwCtrlType):
+        if self._sniffer is not None:
+            self._sniffer.close()

    @watched
    def setChannel(self, channelToCapture):
@@ -211,7 +196,7 @@ class SimSniffer(ISniffer):

    @watched
    def getSnifferAddress(self):
-        return self.ipaddr
+        return self.addr_port

    @watched
    def globalReset(self):
@@ -27,12 +27,6 @@
 # POSSIBILITY OF SUCH DAMAGE.
 #

-REMOTE_PORT = 22
-REMOTE_USERNAME = 'pi'
-REMOTE_PASSWORD = 'raspberry'
-
-REMOTE_SNIFFER_OUTPUT_PREFIX = '/tmp/'
-
-REMOTE_OT_PATH = '/home/pi/work/src/openthread-pr/'
+REMOTE_OT_PATH = '/home/pi/openthread/'

 EDITCAP_PATH = r'C:\Program Files (x86)\Wireshark_Thread\editcap.exe'
@@ -1,2 +1,39 @@
+:: Copyright (c) 2022, 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.
+::
+
 xcopy /E /Y Thread_Harness %systemdrive%\GRL\Thread1.2\Thread_Harness
 copy /Y ..\..\harness-thci\OpenThread.py %systemdrive%\GRL\Thread1.2\Thread_Harness\THCI
+xcopy /E /Y ..\posix\sniffer_sim\proto %systemdrive%\GRL\Thread1.2\Thread_Harness\simulation\Sniffer\proto
+
+%systemdrive%\GRL\Thread1.2\Python27\python.exe -m pip install --upgrade pip
+%systemdrive%\GRL\Thread1.2\Python27\python.exe -m pip install -r requirements.txt
+
+set BASEDIR=%systemdrive%\GRL\Thread1.2\Thread_Harness
+
+%systemdrive%\GRL\Thread1.2\Python27\python.exe -m grpc_tools.protoc -I%BASEDIR% --python_out=%BASEDIR% --grpc_python_out=%BASEDIR% simulation/Sniffer/proto/sniffer.proto
+
+pause
@@ -0,0 +1,2 @@
+grpcio==1.20.1
+grpcio-tools==1.20.1
@@ -0,0 +1,37 @@
+#!/bin/bash
+#
+# Copyright (c) 2022, 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.
+#
+
+set -euxo pipefail
+
+BASE_DIR=$(dirname "$0")
+SNIFFER_DIR="${BASE_DIR}/sniffer_sim"
+
+pip3 install -r "${BASE_DIR}/requirements.txt"
+
+python3 -m grpc_tools.protoc -I"${SNIFFER_DIR}" --python_out="${SNIFFER_DIR}" --grpc_python_out="${SNIFFER_DIR}" proto/sniffer.proto
@@ -30,16 +30,21 @@
 import argparse
 import ctypes
 import ctypes.util
+import ipaddress
 import json
 import logging
 import os
+import signal
 import socket
 import struct
+import subprocess
+import sys

 GROUP = 'ff02::114'
 PORT = 12345
 MAX_OT11_NUM = 33
 MAX_SNIFFER_NUM = 4
+SNIFFER_SERVER_PORT_BASE = 50051


 def if_nametoindex(ifname: str) -> int:
@@ -78,6 +83,11 @@ def init_socket(ifname: str, group: str, port: int) -> socket.socket:
    return s


+def _advertise(s: socket.socket, dst, info):
+    logging.info('Advertise: %r', info)
+    s.sendto(json.dumps(info).encode('utf-8'), dst)
+
+
 def advertise_ftd(s: socket.socket, dst, ven: str, ver: str, add: str, por: int, number: int):
    # Node ID of ot-cli-ftd is 1-indexed
    for i in range(1, number + 1):
@@ -88,15 +98,27 @@ def advertise_ftd(s: socket.socket, dst, ven: str, ver: str, add: str, por: int,
            'add': f'{i}@{add}',
            'por': por,
        }
-        logging.info('Advertise: %r', info)
-        s.sendto(json.dumps(info).encode('utf-8'), dst)
+        _advertise(s, dst, info)


 def advertise_sniffer(s: socket.socket, dst, add: str, number: int):
    for i in range(number):
-        info = 'Sniffer_%d@%s' % (i, add)
-        logging.info('Advertise: %r', info)
-        s.sendto(info.encode('utf-8'), dst)
+        info = {
+            'add': add,
+            'por': i + SNIFFER_SERVER_PORT_BASE,
+        }
+        _advertise(s, dst, info)
+
+
+def start_sniffer(addr: str, port: int) -> subprocess.Popen:
+    if isinstance(ipaddress.ip_address(addr), ipaddress.IPv6Address):
+        server = f'[{addr}]:{port}'
+    else:
+        server = f'{addr}:{port}'
+
+    cmd = ['python3', 'sniffer_sim/sniffer.py', '--grpc-server', server]
+    logging.info('Executing command:  %s', ' '.join(cmd))
+    return subprocess.Popen(cmd)


 def main():
@@ -113,7 +135,7 @@ def main():
                        required=True,
                        help='the interface used for discovery')

-    # Determine the number of OpenThread 1.1 FTD simulations to be "detected" and then initiated
+    # Determine the number of OpenThread 1.1 FTD simulations to be "detected" and then started
    parser.add_argument('--ot1.1',
                        dest='ot11_num',
                        type=int,
@@ -121,7 +143,7 @@ def main():
                        default=0,
                        help=f'the number of OpenThread FTD simulations, no more than {MAX_OT11_NUM}')

-    # Determine the number of sniffer simulations to be initiated and then detected
+    # Determine the number of sniffer simulations to be started and then detected
    parser.add_argument('-s',
                        '--sniffer',
                        dest='sniffer_num',
@@ -145,10 +167,27 @@ def main():
    # Get the local IP address on the specified interface
    addr = get_ipaddr(args.ifname)

+    # Start the sniffer
+    sniffer_procs = []
+    for i in range(args.sniffer_num):
+        sniffer_procs.append(start_sniffer(addr, i + SNIFFER_SERVER_PORT_BASE))
+
    s = init_socket(args.ifname, GROUP, PORT)

    logging.info('Advertising on interface %s group %s ...', args.ifname, GROUP)

+    # Terminate all sniffer simulation server processes and then exit
+    def exit_handler(signum, context):
+        # Return code is non-zero if any return code of the processes is non-zero
+        ret = 0
+        for sniffer_proc in sniffer_procs:
+            sniffer_proc.terminate()
+            ret = max(ret, sniffer_proc.wait())
+        sys.exit(ret)
+
+    signal.signal(signal.SIGINT, exit_handler)
+    signal.signal(signal.SIGTERM, exit_handler)
+
    # Loop, printing any data we receive
    while True:
        data, src = s.recvfrom(64)
@@ -0,0 +1,2 @@
+grpcio
+grpcio-tools
@@ -42,16 +42,14 @@ PCAP_VERSION_MINOR = 4
 class PcapCodec(object):
    """ Utility class for .pcap formatters. """

-    def __init__(self, filename, channel):
+    def __init__(self, channel):
        self._dlt = DLT_IEEE802_15_4_WITHFCS
-        if not filename.endswith('.pcap'):
-            raise ValueError('Filename should end with .pcap')
-        self._pcap_file = open(filename, 'wb')
-        self._pcap_file.write(self.encode_header())
+        self._pcap_contents = [self._encode_header()]
        self._channel = channel

-    def encode_header(self):
+    def _encode_header(self):
        """ Return a pcap file header. """
+
        return struct.pack(
            '<LHHLLLL',
            PCAP_MAGIC_NUMBER,
@@ -63,7 +61,7 @@ class PcapCodec(object):
            self._dlt,
        )

-    def encode_frame(self, frame, sec, usec):
+    def _encode_frame(self, frame, sec, usec):
        """ Return a pcap encapsulation of the given frame. """

        # Ignore the first byte storing channel.
@@ -76,6 +74,7 @@ class PcapCodec(object):

    def _get_timestamp(self):
        """ Return the internal timestamp. """
+
        timestamp = time.time()
        timestamp_sec = int(timestamp)
        timestamp_usec = int((timestamp - timestamp_sec) * 1000000)
@@ -89,11 +88,7 @@ class PcapCodec(object):
            return

        timestamp = self._get_timestamp()
-        pkt = self.encode_frame(frame, *timestamp)
-        self._pcap_file.write(pkt)
-        self._pcap_file.flush()
+        self._pcap_contents.append(self._encode_frame(frame, *timestamp))

-    def close(self):
-        """ Close the pcap file. """
-
-        self._pcap_file.close()
+    def pop_all(self) -> bytes:
+        return b''.join(self._pcap_contents)
@@ -0,0 +1,55 @@
+syntax = "proto3";
+
+package sniffer;
+
+// Sniffer simulation
+service Sniffer {
+    // Start the sniffer
+    rpc Start(StartRequest) returns (StartResponse) {}
+
+    // Let the sniffer sniff these nodes only
+    rpc FilterNodes(FilterNodesRequest) returns (FilterNodesResponse) {}
+
+    // Stop the sniffer
+    rpc Stop(StopRequest) returns (StopResponse) {}
+}
+
+// Possible Status which the RPCs may return
+enum Status {
+    // Default value which is unused
+    STATUS_UNSPECIFIED = 0;
+
+    // Everything goes well
+    OK = 1;
+
+    // Unable to run the specified RPC currently
+    OPERATION_ERROR = 2;
+
+    // The parameters passed to the RPC is erroneous
+    VALUE_ERROR = 3;
+}
+
+message StartRequest {
+    // Specify the channel that the sniffer is going to sniff
+    int32 channel = 1;
+}
+
+message StartResponse {
+    Status status = 1;
+}
+
+message FilterNodesRequest {
+    repeated int32 nodeids = 1;
+}
+
+message FilterNodesResponse {
+    Status status = 1;
+}
+
+message StopRequest {
+}
+
+message StopResponse {
+    Status status = 1;
+    bytes pcap_content = 2;
+}
@@ -28,104 +28,162 @@
 #

 import argparse
+from concurrent import futures
+import enum
+import grpc
 import logging
 import signal
-import time
 import pcap_codec
-import sys
 import threading

+from proto import sniffer_pb2
+from proto import sniffer_pb2_grpc
 import sniffer_transport


-class Sniffer:
-    """ Class representing the Sniffing node, whose main task is listening.
-    """
+class SnifferServicer(sniffer_pb2_grpc.Sniffer):
+    """ Class representing the Sniffing node, whose main task is listening. """

-    logger = logging.getLogger('sniffer.Sniffer')
+    logger = logging.getLogger('sniffer.SnifferServicer')

    RECV_BUFFER_SIZE = 4096
+    MAX_NODES_NUM = 33

-    def __init__(self, filename, channel):
-        self._pcap = pcap_codec.PcapCodec(filename, channel)
-
-        # Create transport
-        transport_factory = sniffer_transport.SnifferTransportFactory()
-        self._transport = transport_factory.create_transport()
+    class State(enum.Enum):
+        STOPPED = 0
+        RUNNING = 1

+    def _reset(self):
+        self._state = SnifferServicer.State.STOPPED
+        self._pcap = None
+        self._allowed_nodeids = None
+        self._transport = None
        self._thread = None
-        self._thread_alive = threading.Event()
        self._thread_alive.clear()

+    def __init__(self):
+        self._thread_alive = threading.Event()
+        self._mutex = threading.Lock()  # for self._allowed_nodeids
+        self._reset()
+
    def _sniffer_main_loop(self):
        """ Sniffer main loop. """

        self.logger.debug('Sniffer started.')

        while self._thread_alive.is_set():
+            # Avoid being blocked endlessly when there is no data
+            if not self._transport.ready(0.1):
+                continue
            data, nodeid = self._transport.recv(self.RECV_BUFFER_SIZE)
-            self._pcap.append(data)
+
+            with self._mutex:
+                allowed_nodeids = self._allowed_nodeids
+
+            # Equivalent to RF enclosure
+            if allowed_nodeids is None or nodeid in allowed_nodeids:
+                self._pcap.append(data)

        self.logger.debug('Sniffer stopped.')

-    def start(self):
+    def Start(self, request, context):
        """ Start sniffing. """

+        self.logger.debug('call Start')
+
+        # Validate and change the state
+        if self._state != SnifferServicer.State.STOPPED:
+            return sniffer_pb2.StartResponse(status=sniffer_pb2.OPERATION_ERROR)
+        self._state = SnifferServicer.State.RUNNING
+
+        self._pcap = pcap_codec.PcapCodec(request.channel)
+
+        # Sniffer all nodes in default, i.e. there is no RF enclosure
+        # In this case, self._allowed_nodeids is set to None
+        self._allowed_nodeids = None
+
+        # Create transport
+        transport_factory = sniffer_transport.SnifferTransportFactory()
+        self._transport = transport_factory.create_transport()
+
+        # Start the sniffer main loop thread
        self._thread = threading.Thread(target=self._sniffer_main_loop)
        self._thread.daemon = True
-
        self._transport.open()
-
        self._thread_alive.set()
        self._thread.start()

-    def stop(self):
-        """ Stop sniffing. """
+        return sniffer_pb2.StartResponse(status=sniffer_pb2.OK)
+
+    def FilterNodes(self, request, context):
+        """ Only sniffer the specified nodes. """
+
+        self.logger.debug('call FilterNodes')
+
+        # Validate the state
+        if self._state != SnifferServicer.State.RUNNING:
+            return sniffer_pb2.FilterNodesResponse(status=sniffer_pb2.OPERATION_ERROR)
+
+        allowed_nodeids = set(request.nodeids)
+        # Validate the node IDs
+        for nodeid in allowed_nodeids:
+            if not 1 <= nodeid <= self.MAX_NODES_NUM:
+                return sniffer_pb2.FilterNodesResponse(status=sniffer_pb2.VALUE_ERROR)
+
+        with self._mutex:
+            self._allowed_nodeids = allowed_nodeids
+
+        return sniffer_pb2.FilterNodesResponse(status=sniffer_pb2.OK)
+
+    def Stop(self, request, context):
+        """ Stop sniffing, and return the pcap bytes. """
+
+        self.logger.debug('call Stop')
+
+        # Validate and change the state
+        if self._state != SnifferServicer.State.RUNNING:
+            return sniffer_pb2.StopResponse(status=sniffer_pb2.OPERATION_ERROR, pcap_content=b'')
+        self._state = SnifferServicer.State.STOPPED

        self._thread_alive.clear()
-
+        self._thread.join(timeout=1)
        self._transport.close()

-        self._thread.join(timeout=1)
-        self._thread = None
+        pcap_content = self._pcap.pop_all()
+        self._reset()

-    def close(self):
-        """ Close the pcap file. """
+        return sniffer_pb2.StopResponse(status=sniffer_pb2.OK, pcap_content=pcap_content)

-        self._pcap.close()
+
+def serve(address_port):
+    server = grpc.server(futures.ThreadPoolExecutor(max_workers=1))
+    sniffer_pb2_grpc.add_SnifferServicer_to_server(SnifferServicer(), server)
+    # add_secure_port requires a web domain
+    server.add_insecure_port(address_port)
+    logging.info('server starts on %s', address_port)
+    server.start()
+
+    def exit_handler(signum, context):
+        server.stop(1)
+
+    signal.signal(signal.SIGINT, exit_handler)
+    signal.signal(signal.SIGTERM, exit_handler)
+
+    server.wait_for_termination()


 def run_sniffer():
+    logging.basicConfig(level=logging.INFO)
+
    parser = argparse.ArgumentParser()
-    parser.add_argument('-o',
-                        '--output',
-                        dest='output',
+    parser.add_argument('--grpc-server',
+                        dest='grpc_server',
                        type=str,
                        required=True,
-                        help='the path of the output .pcap file')
-    parser.add_argument('-c',
-                        '--channel',
-                        dest='channel',
-                        type=int,
-                        required=True,
-                        help='the channel which is sniffered')
+                        help='the address of the sniffer server')
    args = parser.parse_args()

-    sniffer = Sniffer(args.output, args.channel)
-    sniffer.start()
-
-    def atexit(signum, frame):
-        sniffer.stop()
-        sniffer.close()
-        sys.exit(0)
-
-    signal.signal(signal.SIGTERM, atexit)
-
-    while sniffer._thread_alive.is_set():
-        time.sleep(0.5)
-
-    sniffer.stop()
-    sniffer.close()
+    serve(args.grpc_server)


 if __name__ == '__main__':
@@ -28,6 +28,7 @@
 #

 import os
+import select
 import socket


@@ -100,12 +101,6 @@ class SnifferSocketTransport(SnifferTransport):
    def __init__(self):
        self._socket = None

-    def __del__(self):
-        if not self.is_opened:
-            return
-
-        self.close()
-
    def _nodeid_to_port(self, nodeid: int):
        return self.BASE_PORT + (self.PORT_OFFSET * (self.MAX_NETWORK_SIZE + 1)) + nodeid

@@ -152,6 +147,9 @@ class SnifferSocketTransport(SnifferTransport):

        return bytearray(data), nodeid

+    def ready(self, timeout):
+        return select.select([self._socket], [], [], timeout)[0]
+

 class SnifferTransportFactory(object):