diff --git a/tools/cp-caps/README.md b/tools/cp-caps/README.md index 75924ef47..a6e2f90bf 100644 --- a/tools/cp-caps/README.md +++ b/tools/cp-caps/README.md @@ -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 git@github.com: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= Connect to the reference device via adb usb REF_CLI_SERIAL= Connect to the reference device via cli serial port REF_SSH= Connect to the reference device via ssh + SNIFFER_SERIAL= Use the RCP at the serial port as the sniffer (optional) + SNIFFER_RADIO_URL= Use the RCP at the radio URL as the sniffer (optional) + OT_EXTCAP= 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 diff --git a/tools/cp-caps/device_manager.py b/tools/cp-caps/device_manager.py index 77b42b7e1..dc8b83521 100644 --- a/tools/cp-caps/device_manager.py +++ b/tools/cp-caps/device_manager.py @@ -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()) diff --git a/tools/cp-caps/sniffer.py b/tools/cp-caps/sniffer.py new file mode 100644 index 000000000..abaaf8c1f --- /dev/null +++ b/tools/cp-caps/sniffer.py @@ -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 = ' 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 = ' 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())