diff --git a/tools/harness-simulation/README.md b/tools/harness-simulation/README.md index 1412e9c43..ae9780690 100644 --- a/tools/harness-simulation/README.md +++ b/tools/harness-simulation/README.md @@ -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 ``` diff --git a/tools/harness-simulation/harness/Thread_Harness/Sniffer/SimSniffer.py b/tools/harness-simulation/harness/Thread_Harness/Sniffer/SimSniffer.py index fffdc831b..05fe5195d 100644 --- a/tools/harness-simulation/harness/Thread_Harness/Sniffer/SimSniffer.py +++ b/tools/harness-simulation/harness/Thread_Harness/Sniffer/SimSniffer.py @@ -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): diff --git a/tools/harness-simulation/harness/Thread_Harness/simulation/Sniffer/__init__.py b/tools/harness-simulation/harness/Thread_Harness/simulation/Sniffer/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/harness-simulation/harness/Thread_Harness/simulation/Sniffer/proto/__init__.py b/tools/harness-simulation/harness/Thread_Harness/simulation/Sniffer/proto/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tools/harness-simulation/harness/Thread_Harness/simulation/config.py b/tools/harness-simulation/harness/Thread_Harness/simulation/config.py index 44e866e55..930f2cf15 100644 --- a/tools/harness-simulation/harness/Thread_Harness/simulation/config.py +++ b/tools/harness-simulation/harness/Thread_Harness/simulation/config.py @@ -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' diff --git a/tools/harness-simulation/harness/install.bat b/tools/harness-simulation/harness/install.bat index 0b4a898b1..7d6a6d175 100644 --- a/tools/harness-simulation/harness/install.bat +++ b/tools/harness-simulation/harness/install.bat @@ -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 diff --git a/tools/harness-simulation/harness/requirements.txt b/tools/harness-simulation/harness/requirements.txt new file mode 100644 index 000000000..0d1d827d9 --- /dev/null +++ b/tools/harness-simulation/harness/requirements.txt @@ -0,0 +1,2 @@ +grpcio==1.20.1 +grpcio-tools==1.20.1 diff --git a/tools/harness-simulation/posix/install.sh b/tools/harness-simulation/posix/install.sh new file mode 100755 index 000000000..f82fbbd8b --- /dev/null +++ b/tools/harness-simulation/posix/install.sh @@ -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 diff --git a/tools/harness-simulation/posix/harness_dev_discovery.py b/tools/harness-simulation/posix/launch_testbed.py similarity index 79% rename from tools/harness-simulation/posix/harness_dev_discovery.py rename to tools/harness-simulation/posix/launch_testbed.py index 1587c998e..ba73d796b 100644 --- a/tools/harness-simulation/posix/harness_dev_discovery.py +++ b/tools/harness-simulation/posix/launch_testbed.py @@ -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) diff --git a/tools/harness-simulation/posix/requirements.txt b/tools/harness-simulation/posix/requirements.txt new file mode 100644 index 000000000..a6646e7b9 --- /dev/null +++ b/tools/harness-simulation/posix/requirements.txt @@ -0,0 +1,2 @@ +grpcio +grpcio-tools diff --git a/tools/harness-simulation/posix/sniffer_sim/pcap_codec.py b/tools/harness-simulation/posix/sniffer_sim/pcap_codec.py index ff25356bb..47f58ac10 100644 --- a/tools/harness-simulation/posix/sniffer_sim/pcap_codec.py +++ b/tools/harness-simulation/posix/sniffer_sim/pcap_codec.py @@ -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( ' bytes: + return b''.join(self._pcap_contents) diff --git a/tools/harness-simulation/posix/sniffer_sim/proto/sniffer.proto b/tools/harness-simulation/posix/sniffer_sim/proto/sniffer.proto new file mode 100644 index 000000000..620d4910e --- /dev/null +++ b/tools/harness-simulation/posix/sniffer_sim/proto/sniffer.proto @@ -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; +} diff --git a/tools/harness-simulation/posix/sniffer_sim/sniffer.py b/tools/harness-simulation/posix/sniffer_sim/sniffer.py index 0ef6899ec..accc6284e 100644 --- a/tools/harness-simulation/posix/sniffer_sim/sniffer.py +++ b/tools/harness-simulation/posix/sniffer_sim/sniffer.py @@ -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__': diff --git a/tools/harness-simulation/posix/sniffer_sim/sniffer_transport.py b/tools/harness-simulation/posix/sniffer_sim/sniffer_transport.py index 62cf179d3..2a9bff80f 100644 --- a/tools/harness-simulation/posix/sniffer_sim/sniffer_transport.py +++ b/tools/harness-simulation/posix/sniffer_sim/sniffer_transport.py @@ -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):