mirror of
https://github.com/espressif/openthread.git
synced 2026-11-08 15:33:21 +00:00
[cp-caps] add sniffer support (#13722)
Some RCP capabilities can only be verified over the air, e.g. the timing of frames transmitted by the RCP. This commit adds an optional sniffer to the RCP capability test setup. The sniffer is an RCP driven by `ot-extcap` from ot-br-posix. The new `Sniffer` class runs `ot-extcap --capture` on a channel, writing to a temporary PCAPNG file, and parses the captured frames, each with its timestamp in microseconds taken by the sniffer radio. A failure of `ot-extcap` (e.g. to initialize the sniffer RCP) is reported, and `ot-extcap` is stopped at exit if the sniffer is still capturing. `DeviceManager.sniffer` provides the sniffer if `SNIFFER_SERIAL` or `SNIFFER_RADIO_URL` is set, with `OT_EXTCAP` set to the `ot-extcap` executable.
This commit is contained in:
Yakun Xu
+28
-5
@@ -8,17 +8,20 @@ This test is used for testing RCP capabilities.
|
||||
+-------+
|
||||
+---------------| PC |----------------+
|
||||
| +-------+ |
|
||||
| ADB/SSH | ADB/SSH/SERIAL
|
||||
| |
|
||||
+-------+ +------------------+
|
||||
| ADB/SSH | | ADB/SSH/SERIAL
|
||||
| | SERIAL |
|
||||
+-------+ | +------------------+
|
||||
| DUT |<-----------Thread-------->| Reference Device |
|
||||
+-------+ +------------------+
|
||||
|
||||
+-------+ | +------------------+
|
||||
+-----------+
|
||||
| Sniffer | (optional)
|
||||
+-----------+
|
||||
```
|
||||
|
||||
- PC : The computer to run the test script.
|
||||
- DUT : The device under test.
|
||||
- Reference Device : The device that supports all tested features.
|
||||
- Sniffer : An RCP that captures frames with timestamps (e.g. to check the timing of frames). It is optional and only used by some test cases.
|
||||
|
||||
### Reference Device
|
||||
|
||||
@@ -26,6 +29,23 @@ The [nRF52840DK][ot-nrf528xx-nrf52840] is set as the reference device by default
|
||||
|
||||
[ot-nrf528xx-nrf52840]: https://github.com/openthread/ot-nrf528xx/blob/main/src/nrf52840/README.md
|
||||
|
||||
### Sniffer
|
||||
|
||||
The sniffer is an RCP (e.g. an nRF52840 dongle running `ot-rcp`) driven by [ot-extcap][ot-extcap] from ot-br-posix. The test cases capture frames by running `ot-extcap` on the channel of interest, and read each frame with its timestamp in microseconds taken by the sniffer radio.
|
||||
|
||||
[ot-extcap]: https://github.com/openthread/ot-br-posix/tree/main/tools/extcap
|
||||
|
||||
Quick guide to building `ot-extcap`:
|
||||
|
||||
```bash
|
||||
$ git clone [email protected]:openthread/ot-br-posix.git
|
||||
$ cd ot-br-posix/
|
||||
$ git submodule update --init
|
||||
$ ./script/cmake-build extcap
|
||||
```
|
||||
|
||||
The executable is at `build/extcap/ot-extcap`. Set `OT_EXTCAP` to its full path (unless it is in the `PATH`), and `SNIFFER_SERIAL` (or `SNIFFER_RADIO_URL`) to the sniffer RCP.
|
||||
|
||||
Quick guide to setting up the nRF52840DK:
|
||||
|
||||
```bash
|
||||
@@ -97,6 +117,9 @@ Device configurations:
|
||||
REF_ADB_USB=<serial_number> Connect to the reference device via adb usb
|
||||
REF_CLI_SERIAL=<serial_device> Connect to the reference device via cli serial port
|
||||
REF_SSH=<device_ip> Connect to the reference device via ssh
|
||||
SNIFFER_SERIAL=<serial_device> Use the RCP at the serial port as the sniffer (optional)
|
||||
SNIFFER_RADIO_URL=<radio_url> Use the RCP at the radio URL as the sniffer (optional)
|
||||
OT_EXTCAP=<ot_extcap_path> Full path to the ot-extcap executable used by the sniffer (default: ot-extcap)
|
||||
|
||||
Test cases:
|
||||
test_csl test whether the RCP supports CSL transmitter
|
||||
|
||||
@@ -39,6 +39,7 @@ from typing import Optional
|
||||
import otci
|
||||
from otci import OTCI
|
||||
from otci.errors import ExpectLineTimeoutError, InvalidArgumentsError
|
||||
from sniffer import Sniffer
|
||||
|
||||
CP_CAPABILITY_VERSION = "0.1.1-dev"
|
||||
|
||||
@@ -79,6 +80,7 @@ class DeviceManager(object):
|
||||
def __init__(self):
|
||||
self.__dut = self.__connect_dut()
|
||||
self.__ref = self.__connect_reference_device()
|
||||
self.__sniffer = self.__create_sniffer()
|
||||
|
||||
if self.__get_debug_enabled():
|
||||
logger = logging.getLogger()
|
||||
@@ -92,6 +94,11 @@ class DeviceManager(object):
|
||||
def ref(self) -> OTCI:
|
||||
return self.__ref
|
||||
|
||||
@property
|
||||
def sniffer(self) -> Optional[Sniffer]:
|
||||
"""The sniffer, or None if no sniffer is configured (`SNIFFER_SERIAL` or `SNIFFER_RADIO_URL`)."""
|
||||
return self.__sniffer
|
||||
|
||||
def output_test_result_string(self, name: str, value: str, align_length: int = DEFAULT_FORMAT_ALIGN_LENGTH):
|
||||
prefix = (name + ' ').ljust(align_length, '-')
|
||||
print(f'{prefix} {value}')
|
||||
@@ -119,6 +126,16 @@ class DeviceManager(object):
|
||||
def __get_adb_key(self) -> Optional[str]:
|
||||
return os.getenv('ADB_KEY', None)
|
||||
|
||||
def __create_sniffer(self) -> Optional[Sniffer]:
|
||||
if os.getenv('SNIFFER_RADIO_URL'):
|
||||
radio_url = os.getenv('SNIFFER_RADIO_URL')
|
||||
elif os.getenv('SNIFFER_SERIAL'):
|
||||
radio_url = f'spinel+hdlc+uart://{os.getenv("SNIFFER_SERIAL")}?uart-baudrate=115200'
|
||||
else:
|
||||
return None
|
||||
|
||||
return Sniffer(os.getenv('OT_EXTCAP', 'ot-extcap'), radio_url)
|
||||
|
||||
def __connect_dut(self) -> OTCI:
|
||||
if os.getenv('DUT_ADB_TCP'):
|
||||
node = otci.connect_otbr_adb_tcp(os.getenv('DUT_ADB_TCP'), adb_key=self.__get_adb_key())
|
||||
|
||||
@@ -0,0 +1,244 @@
|
||||
#!/usr/bin/env python3
|
||||
#
|
||||
# Copyright (c) 2026, 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.
|
||||
#
|
||||
"""Captures IEEE 802.15.4 frames with an OpenThread RCP sniffer driven by `ot-extcap` (from ot-br-posix)."""
|
||||
|
||||
import atexit
|
||||
import logging
|
||||
import os
|
||||
import signal
|
||||
import struct
|
||||
import subprocess
|
||||
import tempfile
|
||||
import time
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import IO, List, Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class SniffedFrame:
|
||||
"""Represents a frame captured by the sniffer.
|
||||
|
||||
Attributes:
|
||||
timestamp_us: The receive timestamp in microseconds, taken by the sniffer radio (end of SFD).
|
||||
channel: The channel the frame was received on.
|
||||
psdu: The PSDU of the frame in hex, including the FCS.
|
||||
"""
|
||||
timestamp_us: int
|
||||
channel: int
|
||||
psdu: str
|
||||
|
||||
|
||||
class Sniffer(object):
|
||||
"""Captures frames on a channel by running `ot-extcap --capture` with an RCP as the sniffer radio."""
|
||||
|
||||
DEFAULT_START_DELAY = 2.0 # Time (in seconds) for `ot-extcap` to initialize the RCP and start capturing.
|
||||
DEFAULT_STOP_TIMEOUT = 5.0
|
||||
|
||||
def __init__(self, extcap: str, radio_url: str, start_delay: float = DEFAULT_START_DELAY):
|
||||
"""
|
||||
:param extcap: The path to the `ot-extcap` executable.
|
||||
:param radio_url: The radio URL of the sniffer RCP, e.g. `spinel+hdlc+uart:///dev/ttyACM2?uart-baudrate=115200`.
|
||||
:param start_delay: The time (in seconds) to wait for `ot-extcap` to start capturing.
|
||||
"""
|
||||
self.__extcap = extcap
|
||||
self.__radio_url = radio_url
|
||||
self.__start_delay = start_delay
|
||||
self.__proc: Optional[subprocess.Popen] = None
|
||||
self.__log: Optional[IO[bytes]] = None
|
||||
self.__pcapng_path: Optional[str] = None
|
||||
|
||||
def start(self, channel: int):
|
||||
"""Starts capturing frames on the channel."""
|
||||
if self.__proc is not None:
|
||||
raise RuntimeError('The sniffer is already capturing')
|
||||
|
||||
fd, self.__pcapng_path = tempfile.mkstemp(prefix='cp-caps-sniffer-', suffix='.pcapng')
|
||||
os.close(fd)
|
||||
|
||||
args = [
|
||||
self.__extcap, '--capture', '--extcap-interface', 'spinel_radio_url', '--radio-url', self.__radio_url,
|
||||
'--channel',
|
||||
str(channel), '--fifo', self.__pcapng_path
|
||||
]
|
||||
logging.debug('Sniffer: %s', ' '.join(args))
|
||||
|
||||
# The output of `ot-extcap` goes to a file rather than a pipe, which is not read while capturing and could
|
||||
# block `ot-extcap` once full. It is only read if `ot-extcap` stops unexpectedly.
|
||||
self.__log = tempfile.TemporaryFile(prefix='cp-caps-sniffer-', suffix='.log')
|
||||
try:
|
||||
self.__proc = subprocess.Popen(args, stdout=self.__log, stderr=subprocess.STDOUT)
|
||||
except Exception:
|
||||
self.__log.close()
|
||||
self.__log = None
|
||||
self.__remove_pcapng()
|
||||
raise
|
||||
|
||||
# Stops `ot-extcap` (which holds the sniffer RCP) even if the caller exits without stopping the sniffer.
|
||||
atexit.register(self.close)
|
||||
|
||||
time.sleep(self.__start_delay)
|
||||
|
||||
if self.__proc.poll() is not None:
|
||||
self.__raise_stopped_unexpectedly()
|
||||
|
||||
def stop(self) -> List[SniffedFrame]:
|
||||
"""Stops capturing frames and returns the captured frames.
|
||||
|
||||
Raises RuntimeError if `ot-extcap` stopped before (e.g. it failed to initialize the sniffer RCP).
|
||||
"""
|
||||
if self.__proc is None:
|
||||
raise RuntimeError('The sniffer is not capturing')
|
||||
|
||||
if self.__proc.poll() is not None:
|
||||
self.__raise_stopped_unexpectedly()
|
||||
|
||||
self.__proc.send_signal(signal.SIGTERM)
|
||||
|
||||
try:
|
||||
self.__proc.wait(self.DEFAULT_STOP_TIMEOUT)
|
||||
except subprocess.TimeoutExpired:
|
||||
self.__proc.kill()
|
||||
self.__proc.wait()
|
||||
|
||||
self.__release_proc()
|
||||
|
||||
try:
|
||||
with open(self.__pcapng_path, 'rb') as f:
|
||||
frames = self.parse_pcapng(f.read())
|
||||
finally:
|
||||
self.__remove_pcapng()
|
||||
|
||||
return frames
|
||||
|
||||
def close(self):
|
||||
"""Stops capturing frames if capturing, and discards the captured frames."""
|
||||
if self.__proc is not None:
|
||||
if self.__proc.poll() is None:
|
||||
self.__proc.kill()
|
||||
self.__proc.wait()
|
||||
|
||||
self.__release_proc()
|
||||
self.__remove_pcapng()
|
||||
|
||||
def __raise_stopped_unexpectedly(self):
|
||||
self.__log.seek(0)
|
||||
output = self.__log.read().decode('utf-8', errors='replace').strip()
|
||||
returncode = self.__proc.returncode
|
||||
|
||||
self.__release_proc()
|
||||
self.__remove_pcapng()
|
||||
|
||||
raise RuntimeError(
|
||||
f'The sniffer stopped unexpectedly (exit code {returncode}, see the ot-extcap logs in syslog):'
|
||||
f' {output}')
|
||||
|
||||
def __release_proc(self):
|
||||
self.__log.close()
|
||||
self.__log = None
|
||||
self.__proc = None
|
||||
atexit.unregister(self.close)
|
||||
|
||||
def __remove_pcapng(self):
|
||||
if self.__pcapng_path is not None:
|
||||
os.remove(self.__pcapng_path)
|
||||
self.__pcapng_path = None
|
||||
|
||||
@staticmethod
|
||||
def parse_pcapng(data: bytes) -> List[SniffedFrame]:
|
||||
"""Parses the frames of a little-endian PCAPNG capture written by `ot-extcap`.
|
||||
|
||||
The frames are Enhanced Packet Blocks of the `IEEE802_15_4_TAP` link type, with timestamps in microseconds.
|
||||
Malformed blocks are skipped, and the parsing stops at a truncated block (e.g. the capture was stopped while
|
||||
writing it).
|
||||
"""
|
||||
BLOCK_HEADER_FORMAT = '<II' # Type, total length
|
||||
BLOCK_TRAILER_LENGTH = 4 # Total length
|
||||
PCAPNG_EPB_TYPE = 6
|
||||
|
||||
frames = []
|
||||
offset = 0
|
||||
|
||||
while offset + struct.calcsize(BLOCK_HEADER_FORMAT) <= len(data):
|
||||
block_type, block_length = struct.unpack_from(BLOCK_HEADER_FORMAT, data, offset)
|
||||
|
||||
if block_length < struct.calcsize(BLOCK_HEADER_FORMAT) + BLOCK_TRAILER_LENGTH or \
|
||||
offset + block_length > len(data):
|
||||
break
|
||||
|
||||
if block_type == PCAPNG_EPB_TYPE:
|
||||
frame = Sniffer.__parse_epb(data[offset:offset + block_length - BLOCK_TRAILER_LENGTH])
|
||||
if frame is not None:
|
||||
frames.append(frame)
|
||||
|
||||
offset += block_length
|
||||
|
||||
return frames
|
||||
|
||||
@staticmethod
|
||||
def __parse_epb(block: bytes) -> Optional[SniffedFrame]:
|
||||
"""Parses an Enhanced Packet Block (without its trailer), or returns None if it is malformed."""
|
||||
EPB_HEADER_FORMAT = '<IIIIIII' # Type, total length, interface, timestamp high/low, captured/original length
|
||||
TAP_HEADER_FORMAT = '<HH' # Version, length (including the TLVs)
|
||||
TAP_TLV_HEADER_FORMAT = '<HH' # Type, length
|
||||
TAP_CHANNEL_TYPE = 3
|
||||
TAP_CHANNEL_FORMAT = '<H' # Channel (followed by the channel page)
|
||||
|
||||
if len(block) < struct.calcsize(EPB_HEADER_FORMAT):
|
||||
return None
|
||||
|
||||
_, _, _, ts_high, ts_low, captured_length, _ = struct.unpack_from(EPB_HEADER_FORMAT, block, 0)
|
||||
packet = block[struct.calcsize(EPB_HEADER_FORMAT):]
|
||||
|
||||
if captured_length > len(packet) or captured_length < struct.calcsize(TAP_HEADER_FORMAT):
|
||||
return None
|
||||
|
||||
packet = packet[:captured_length]
|
||||
_, tap_length = struct.unpack_from(TAP_HEADER_FORMAT, packet, 0)
|
||||
|
||||
if tap_length < struct.calcsize(TAP_HEADER_FORMAT) or tap_length > len(packet):
|
||||
return None
|
||||
|
||||
channel = 0
|
||||
tlv_offset = struct.calcsize(TAP_HEADER_FORMAT)
|
||||
|
||||
while tlv_offset + struct.calcsize(TAP_TLV_HEADER_FORMAT) <= tap_length:
|
||||
tlv_type, tlv_length = struct.unpack_from(TAP_TLV_HEADER_FORMAT, packet, tlv_offset)
|
||||
value_offset = tlv_offset + struct.calcsize(TAP_TLV_HEADER_FORMAT)
|
||||
|
||||
if value_offset + tlv_length > tap_length:
|
||||
return None
|
||||
|
||||
if tlv_type == TAP_CHANNEL_TYPE and tlv_length >= struct.calcsize(TAP_CHANNEL_FORMAT):
|
||||
channel = struct.unpack_from(TAP_CHANNEL_FORMAT, packet, value_offset)[0]
|
||||
|
||||
# TLV values are padded to 4-byte boundaries.
|
||||
tlv_offset = value_offset + ((tlv_length + 3) & ~3)
|
||||
|
||||
return SniffedFrame(timestamp_us=(ts_high << 32) | ts_low, channel=channel, psdu=packet[tap_length:].hex())
|
||||
Reference in New Issue
Block a user