diff --git a/tools/harness-simulation/harness/Thread_Harness/Sniffer/SimSniffer.py b/tools/harness-simulation/harness/Thread_Harness/Sniffer/SimSniffer.py index 90bc43f3d..c5c55dbed 100644 --- a/tools/harness-simulation/harness/Thread_Harness/Sniffer/SimSniffer.py +++ b/tools/harness-simulation/harness/Thread_Harness/Sniffer/SimSniffer.py @@ -35,6 +35,7 @@ import netifaces import select import socket import struct +import threading import time import win32api import winreg as wr @@ -259,8 +260,19 @@ class SimSniffer(ISniffer): if response.status != sniffer_pb2.OK: raise RuntimeError('startSniffer error: %s' % sniffer_pb2.Status.Name(response.status)) + self._thread = threading.Thread(target=self._file_sync_main_loop) + self._thread.setDaemon(True) + self._thread.start() + self.is_active = True + @watched + def _file_sync_main_loop(self): + with open(self._local_pcapng_location, 'wb') as f: + for response in self._stub.TransferPcapng(sniffer_pb2.TransferPcapngRequest()): + f.write(response.content) + f.flush() + @watched def stopSniffer(self): if not self.is_active: @@ -270,8 +282,7 @@ class SimSniffer(ISniffer): if response.status != sniffer_pb2.OK: raise RuntimeError('stopSniffer error: %s' % sniffer_pb2.Status.Name(response.status)) - with open(self._local_pcapng_location, 'wb') as f: - f.write(response.pcap_content) + self._thread.join() self.is_active = False diff --git a/tools/harness-simulation/posix/sniffer_sim/proto/sniffer.proto b/tools/harness-simulation/posix/sniffer_sim/proto/sniffer.proto index 436bd3bf0..9d843f025 100644 --- a/tools/harness-simulation/posix/sniffer_sim/proto/sniffer.proto +++ b/tools/harness-simulation/posix/sniffer_sim/proto/sniffer.proto @@ -7,6 +7,9 @@ service Sniffer { // Start the sniffer rpc Start(StartRequest) returns (StartResponse) {} + // Transfer the capture file + rpc TransferPcapng(TransferPcapngRequest) returns (stream TransferPcapngResponse) {} + // Let the sniffer sniff these nodes only rpc FilterNodes(FilterNodesRequest) returns (FilterNodesResponse) {} @@ -41,6 +44,13 @@ message StartResponse { Status status = 1; } +message TransferPcapngRequest { +} + +message TransferPcapngResponse { + bytes content = 1; +} + message FilterNodesRequest { repeated int32 nodeids = 1; } @@ -54,5 +64,4 @@ 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 b059cf3da..f35213127 100644 --- a/tools/harness-simulation/posix/sniffer_sim/sniffer.py +++ b/tools/harness-simulation/posix/sniffer_sim/sniffer.py @@ -30,6 +30,7 @@ import argparse from concurrent import futures import enum +import fcntl import grpc import logging import os @@ -38,6 +39,7 @@ import socket import subprocess import tempfile import threading +import time import pcap_codec from proto import sniffer_pb2 @@ -66,13 +68,14 @@ class SnifferServicer(sniffer_pb2_grpc.Sniffer): self._transport = None self._thread = None self._thread_alive.clear() - self._pcapng_filename = None + self._file_sync_done.clear() self._tshark_proc = None def __init__(self, max_nodes_num): self._max_nodes_num = max_nodes_num self._thread_alive = threading.Event() - self._mutex = threading.Lock() # for self._denied_nodeids + self._file_sync_done = threading.Event() + self._nodeids_mutex = threading.Lock() # for `self._denied_nodeids` self._reset() def Start(self, request, context): @@ -94,11 +97,11 @@ class SnifferServicer(sniffer_pb2_grpc.Sniffer): if request.includeEthernet: self._state |= CaptureState.ETHERNET cmd += ['-i', 'docker0'] - self._pcapng_filename = os.path.join(tempdir, 'sim.pcapng') - cmd += ['-w', self._pcapng_filename, '-q', 'not ip and not tcp and not arp and not ether proto 0x8899'] + cmd += ['-w', '-', '-q', 'not ip and not tcp and not arp and not ether proto 0x8899'] self.logger.debug('Running command: %s', ' '.join(cmd)) - self._tshark_proc = subprocess.Popen(cmd) + self._tshark_proc = subprocess.Popen(cmd, stdout=subprocess.PIPE) + self._set_nonblocking(self._tshark_proc.stdout.fileno()) # Construct pcap codec after initiating tshark to avoid blocking self._pcap = pcap_codec.PcapCodec(request.channel, fifo_name) @@ -112,7 +115,7 @@ class SnifferServicer(sniffer_pb2_grpc.Sniffer): # Start the sniffer main loop thread self._thread = threading.Thread(target=self._sniffer_main_loop) - self._thread.daemon = True + self._thread.setDaemon(True) self._transport.open() self._thread_alive.set() self._thread.start() @@ -128,13 +131,44 @@ class SnifferServicer(sniffer_pb2_grpc.Sniffer): except socket.timeout: continue - with self._mutex: + with self._nodeids_mutex: denied_nodeids = self._denied_nodeids # Equivalent to RF enclosure if nodeid not in denied_nodeids: self._pcap.append(data) + def TransferPcapng(self, request, context): + """ Transfer the capture file. """ + + # Validate the state + if self._state == CaptureState.NONE: + return sniffer_pb2.FilterNodesResponse(status=sniffer_pb2.OPERATION_ERROR) + + # Synchronize the capture file + while True: + content = self._tshark_proc.stdout.read() + if content is None: + # Currently no captured packets + time.sleep(self.TIMEOUT) + elif content == b'': + # Reach EOF when tshark terminates + break + else: + # Forward the captured packets + yield sniffer_pb2.TransferPcapngResponse(content=content) + + self._file_sync_done.set() + + def _set_nonblocking(self, fd): + flags = fcntl.fcntl(fd, fcntl.F_GETFL) + if flags < 0: + raise RuntimeError('fcntl(F_GETFL) failed') + + flags |= os.O_NONBLOCK + if fcntl.fcntl(fd, fcntl.F_SETFL, flags) < 0: + raise RuntimeError('fcntl(F_SETFL) failed') + def FilterNodes(self, request, context): """ Only sniffer the specified nodes. """ @@ -150,7 +184,7 @@ class SnifferServicer(sniffer_pb2_grpc.Sniffer): if not 1 <= nodeid <= self._max_nodes_num: return sniffer_pb2.FilterNodesResponse(status=sniffer_pb2.VALUE_ERROR) - with self._mutex: + with self._nodeids_mutex: self._denied_nodeids = denied_nodeids return sniffer_pb2.FilterNodesResponse(status=sniffer_pb2.OK) @@ -161,28 +195,29 @@ class SnifferServicer(sniffer_pb2_grpc.Sniffer): self.logger.debug('call Stop') # Validate and change the state - if self._state == CaptureState.NONE: - return sniffer_pb2.StopResponse(status=sniffer_pb2.OPERATION_ERROR, pcap_content=b'') + if not (self._state & CaptureState.THREAD): + return sniffer_pb2.StopResponse(status=sniffer_pb2.OPERATION_ERROR) self._state = CaptureState.NONE self._thread_alive.clear() - self._thread.join(timeout=1) + self._thread.join() self._transport.close() self._pcap.close() self._tshark_proc.terminate() + self._file_sync_done.wait() + # `self._tshark_proc` becomes None after the next statement self._tshark_proc.wait() - with open(self._pcapng_filename, 'rb') as f: - pcap_content = f.read() - self._reset() - return sniffer_pb2.StopResponse(status=sniffer_pb2.OK, pcap_content=pcap_content) + return sniffer_pb2.StopResponse(status=sniffer_pb2.OK) def serve(address_port, max_nodes_num): - server = grpc.server(futures.ThreadPoolExecutor(max_workers=1)) + # One worker is used for `Start`, `FilterNodes` and `Stop` + # The other worker is used for `TransferPcapng`, which will be kept running by the client in a background thread + server = grpc.server(futures.ThreadPoolExecutor(max_workers=2)) sniffer_pb2_grpc.add_SnifferServicer_to_server(SnifferServicer(max_nodes_num), server) # add_secure_port requires a web domain server.add_insecure_port(address_port)