From 22b043bf62328e89773fa61c9bb80770f81a4bef Mon Sep 17 00:00:00 2001 From: Simon Lin Date: Thu, 31 Dec 2020 03:29:27 +0800 Subject: [PATCH] [otci] add OpenThread Controller Interface library (#5779) This commit introduces the Proof of Concept implementation of the OpenThread Controller Interface (OTCI) which can be used to connect and manage various kinds of OT devices. This commit also adds the tests to verify that OTCI works for both simulation and real OT devices. - Device Connectors - CLI Virtual Time Simulation - CLI Real Time Simulation - CLI SOC Device - NCP Virtual Time Simulation (not all APIs work) - Real device via SSH - OTCI tests of simulation devices are executed in Github Actions - OTCI tests of real devices can be executed like below: # Install otci python library cd tests/scripts/thread-cert/otci && python3 setup.py install --user # Test OT CLI SOC device at /dev/ttyACM0 REAL_DEVICE=1 OT_CLI_SERIAL=/dev/ttyACM0 python3 tests/test_otci.py # Test OTBR device via SSH REAL_DEVICE=1 OTBR_SSH=172.16.243.151 python3 tests/test_otci.py --- .github/workflows/otci.yml | 70 + tests/scripts/thread-cert/simulator.py | 2 + tools/otci/LICENSE | 25 + tools/otci/README.md | 53 + tools/otci/otci/__init__.py | 49 + tools/otci/otci/command_handlers.py | 211 +++ tools/otci/otci/connectors.py | 161 ++ tools/otci/otci/constants.py | 32 + tools/otci/otci/errors.py | 64 + tools/otci/otci/otci.py | 1901 ++++++++++++++++++++++++ tools/otci/otci/types.py | 131 ++ tools/otci/otci/utils.py | 63 + tools/otci/setup.py | 50 + tools/otci/tests/test_otci.py | 515 +++++++ 14 files changed, 3327 insertions(+) create mode 100644 .github/workflows/otci.yml create mode 100644 tools/otci/LICENSE create mode 100644 tools/otci/README.md create mode 100644 tools/otci/otci/__init__.py create mode 100644 tools/otci/otci/command_handlers.py create mode 100644 tools/otci/otci/connectors.py create mode 100644 tools/otci/otci/constants.py create mode 100644 tools/otci/otci/errors.py create mode 100644 tools/otci/otci/otci.py create mode 100644 tools/otci/otci/types.py create mode 100644 tools/otci/otci/utils.py create mode 100644 tools/otci/setup.py create mode 100644 tools/otci/tests/test_otci.py diff --git a/.github/workflows/otci.yml b/.github/workflows/otci.yml new file mode 100644 index 000000000..add77d6b9 --- /dev/null +++ b/.github/workflows/otci.yml @@ -0,0 +1,70 @@ +# +# Copyright (c) 2020, 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. +# + +name: OTCI + +on: [push, pull_request] + +jobs: + + cancel-previous-runs: + runs-on: ubuntu-20.04 + steps: + - uses: rokroskar/workflow-run-cleanup-action@master + env: + GITHUB_TOKEN: "${{ secrets.GITHUB_TOKEN }}" + if: "github.ref != 'refs/heads/master'" + + cli-sim: + runs-on: ubuntu-20.04 + strategy: + matrix: + virtual_time: [0, 1] + env: + REFERENCE_DEVICE: 1 + VIRTUAL_TIME: ${{ matrix.virtual_time }} + REAL_DEVICE: 0 + steps: + - uses: actions/checkout@v2 + - name: Bootstrap + run: | + sudo rm /etc/apt/sources.list.d/* && sudo apt-get update + sudo apt-get --no-install-recommends install -y g++-multilib python3-setuptools python3-wheel + python3 -m pip install -r tests/scripts/thread-cert/requirements.txt + - name: Build + run: | + ./bootstrap + make -f examples/Makefile-simulation THREAD_VERSION=1.2 DUA=1 MLR=1 BACKBONE_ROUTER=1 CSL_RECEIVER=1 + - name: Install OTCI Python Library + run: | + (cd tools/otci && python3 setup.py install --user) + - name: Run + run: | + export PYTHONPATH=./tests/scripts/thread-cert/ + export OT_CLI=./output/simulation/bin/ot-cli-ftd + python3 tools/otci/tests/test_otci.py diff --git a/tests/scripts/thread-cert/simulator.py b/tests/scripts/thread-cert/simulator.py index 4dbe37e21..da04dec12 100644 --- a/tests/scripts/thread-cert/simulator.py +++ b/tests/scripts/thread-cert/simulator.py @@ -147,6 +147,8 @@ class VirtualTime(BaseSimulator): RADIO_ONLY = os.getenv('RADIO_DEVICE') is not None NCP_SIM = os.getenv('NODE_TYPE', 'sim') == 'ncp-sim' + _message_factory = None + def __init__(self, use_message_factory=True): super(VirtualTime, self).__init__() self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) diff --git a/tools/otci/LICENSE b/tools/otci/LICENSE new file mode 100644 index 000000000..d63767e70 --- /dev/null +++ b/tools/otci/LICENSE @@ -0,0 +1,25 @@ +Copyright (c) 2020, 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. diff --git a/tools/otci/README.md b/tools/otci/README.md new file mode 100644 index 000000000..788f9376e --- /dev/null +++ b/tools/otci/README.md @@ -0,0 +1,53 @@ +# OpenThread Control Interface + +The OpenThread Control Interface (OTCI) is a library which provides uniform python interfaces to connect and control various kinds of devices running OpenThread. + +## Supported device types + +- OpenThread CLI + - SOC device via Serial +- OpenThread NCP (limited support via [pyspinel](https://pypi.org/project/pyspinel/)) + - SOC device via Serial +- [OpenThread Border Router](https://github.com/openthread/ot-br-posix) + - OTBR device via SSH + +## Example + +```python +import otci + +# Connect to an OTBR device via SSH +node1 = otci.connect_otbr_ssh("192.168.1.101") + +# Connect to an OpenThread CLI device via Serial +node2 = otci.connect_cli_serial("/dev/ttyACM0")) + +# Start node1 to become Leader +node1.dataset_init_buffer() +node1.dataset_set_buffer(network_name='test', master_key='00112233445566778899aabbccddeeff', panid=0xface, channel=11) +node1.dataset_commit_buffer('active') + +node1.ifconfig_up() +node1.thread_start() +node1.wait(5) +assert node1.get_state() == "leader" + +# Start Commissioner on node1 +node1.commissioner_start() +node1.wait(3) + +node1.commissioner_add_joiner("TEST123",eui64='*') + +# Start node2 +node2.ifconfig_up() +node2.set_router_selection_jitter(1) + +# Start Joiner on node2 to join the network +node2.joiner_start("TEST123") +node2.wait(10, expect_line="Join success") + +# Wait for node 2 to become Router +node2.thread_start() +node2.wait(5) +assert node2.get_state() == "router" +``` diff --git a/tools/otci/otci/__init__.py b/tools/otci/otci/__init__.py new file mode 100644 index 000000000..8c3a1e5e8 --- /dev/null +++ b/tools/otci/otci/__init__.py @@ -0,0 +1,49 @@ +#!/usr/bin/env python3 +# +# Copyright (c) 2020, 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. +# + +from . import errors +from .constants import THREAD_VERSION_1_1, THREAD_VERSION_1_2 +from .otci import OTCI +from .otci import \ + connect_cli_sim, \ + connect_cli_serial, \ + connect_ncp_sim, \ + connect_cmd_handler, \ + connect_otbr_ssh +from .types import Rloc16, ChildId + +_connectors = [ + 'connect_cli_sim', + 'connect_cli_serial', + 'connect_ncp_sim', + 'connect_otbr_ssh', + 'connect_cmd_handler', +] + +__all__ = ['OTCI', 'errors', 'Rloc16', 'ChildId', 'THREAD_VERSION_1_1', 'THREAD_VERSION_1_2'] + _connectors diff --git a/tools/otci/otci/command_handlers.py b/tools/otci/otci/command_handlers.py new file mode 100644 index 000000000..c3b648460 --- /dev/null +++ b/tools/otci/otci/command_handlers.py @@ -0,0 +1,211 @@ +#!/usr/bin/env python3 +# +# Copyright (c) 2020, 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. +# +import logging +import queue +import re +import threading +import time +from abc import abstractmethod +from typing import Union, List, Pattern + +from .connectors import OtCliHandler +from .errors import ExpectLineTimeoutError, CommandError +from .utils import match_line + + +class OTCommandHandler: + """This abstract class defines interfaces of a OT Command Handler.""" + + @abstractmethod + def execute_command(self, cmd: str, timeout: float) -> List[str]: + """Method execute_command should execute the OT CLI command within a timeout (in seconds) and return the + command output as a list of lines. + + Note: each line SHOULD NOT contain '\r\n' at the end. The last line of output should be 'Done' or + 'Error : ' following OT CLI conventions. + """ + pass + + @abstractmethod + def close(self): + """Method close should close the OT Command Handler.""" + pass + + @abstractmethod + def wait(self, duration: float) -> List[str]: + """Method wait should wait for a given duration and return the OT CLI output during this period. + + Normally, OT CLI does not output when it's not executing any command. But OT CLI can also output + asynchronously in some cases (e.g. `Join Success` when Joiner joins successfully). + """ + pass + + +class OtCliCommandRunner(OTCommandHandler): + __PATTERN_COMMAND_DONE_OR_ERROR = re.compile( + r'(Done|Error|Error \d+:.*|.*: command not found)$') # "Error" for spinel-cli.py + + __PATTERN_LOG_LINE = re.compile(r'((\[(NONE|CRIT|WARN|NOTE|INFO|DEBG)\])' + r'|(-.*-+: )' # e.g. -CLI-----: + r')') + """regex used to filter logs""" + + __ASYNC_COMMANDS = { + 'scan', + } + + def __init__(self, otcli: OtCliHandler, is_spinel_cli=False): + self.__otcli: OtCliHandler = otcli + self.__is_spinel_cli = is_spinel_cli + self.__expect_command_echoback = not self.__is_spinel_cli + + self.__pending_lines = queue.Queue() + self.__should_close = threading.Event() + self.__otcli_reader = threading.Thread(target=self.__otcli_read_routine) + self.__otcli_reader.setDaemon(True) + self.__otcli_reader.start() + + def __repr__(self): + return repr(self.__otcli) + + def execute_command(self, cmd, timeout=10) -> None: + self.__otcli.writeline(cmd) + + if cmd in {'reset', 'factoryreset'}: + return [] + + if self.__expect_command_echoback: + self.__expect_line(timeout, cmd) + + output = self.__expect_line(timeout, + OtCliCommandRunner.__PATTERN_COMMAND_DONE_OR_ERROR, + asynchronous=cmd.split()[0] in OtCliCommandRunner.__ASYNC_COMMANDS) + return output + + def wait(self, duration: float) -> List[str]: + self.__otcli.wait(duration) + + output = [] + try: + while True: + line = self.__pending_lines.get_nowait() + output.append(line) + + except queue.Empty: + pass + + return output + + def close(self): + self.__should_close.set() + self.__otcli.close() + + # + # Private methods + # + + def __expect_line(self, timeout: float, expect_line: Union[str, Pattern], asynchronous=False) -> List[str]: + output = [] + + if not asynchronous: + while True: + try: + line = self.__pending_lines.get(timeout=timeout) + except queue.Empty: + raise ExpectLineTimeoutError(expect_line) + + output.append(line) + + if match_line(line, expect_line): + break + else: + done = False + while not done and timeout > 0: + lines = self.wait(1) + timeout -= 1 + + for line in lines: + output.append(line) + + if match_line(line, expect_line): + done = True + break + + if not done: + raise ExpectLineTimeoutError(expect_line) + + return output + + def __otcli_read_routine(self): + while not self.__should_close.isSet(): + line = self.__otcli.readline() + if line.startswith('> '): + line = line[2:] + + logging.debug('%s: %s', self.__otcli, line) + + if not OtCliCommandRunner.__PATTERN_LOG_LINE.match(line): + self.__pending_lines.put(line) + + +class OtbrSshCommandRunner(OTCommandHandler): + + def __init__(self, host, port, username, password): + import paramiko + + self.__host = host + self.__port = port + self.__ssh = paramiko.SSHClient() + self.__ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + self.__ssh.connect(host, + port=port, + username=username, + password=password, + allow_agent=False, + look_for_keys=False) + + def __repr__(self): + return f'{self.__host}:{self.__port}' + + def execute_command(self, cmd: str, timeout: float) -> List[str]: + sh_cmd = f'sudo ot-ctl "{cmd}"' + cmd_in, cmd_out, cmd_err = self.__ssh.exec_command(sh_cmd, timeout=int(timeout), bufsize=1024) + err = cmd_err.read().decode('utf-8') + if err: + raise CommandError(cmd, [err]) + + output = [l.rstrip('\r\n') for l in cmd_out.readlines()] + return output + + def close(self): + self.__ssh.close() + + def wait(self, duration: float) -> List[str]: + time.sleep(duration) + return [] diff --git a/tools/otci/otci/connectors.py b/tools/otci/otci/connectors.py new file mode 100644 index 000000000..1785f111f --- /dev/null +++ b/tools/otci/otci/connectors.py @@ -0,0 +1,161 @@ +#!/usr/bin/env python3 +# +# Copyright (c) 2020, 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. +# +import logging +import subprocess +import time +from abc import abstractmethod + + +class OtCliHandler: + """This abstract class defines interfaces for a OT CLI Handler.""" + + @abstractmethod + def readline(self) -> str: + """Method readline should return the next line read from OT CLI.""" + pass + + @abstractmethod + def writeline(self, s: str) -> None: + """Method writeline should write a line to the OT CLI. + + It should block until all characters are written to OT CLI. + """ + pass + + @abstractmethod + def wait(self, duration: float) -> None: + """Method wait should wait for a given duration. + + A normal implementation should just call `time.sleep(duration)`. This is intended for proceeding Virtual Time + Simulation instances. + """ + pass + + @abstractmethod + def close(self) -> None: + """Method close should close the OT CLI Handler.""" + pass + + +class Simulator: + """This abstract class defines interfaces for a Virtual Time Simulator.""" + + @abstractmethod + def go(self, duration: float): + """Proceed the simulator for a given duration (in seconds).""" + pass + + +class OtCliPopen(OtCliHandler): + """Connector for OT CLI process (a Popen instance).""" + + def __init__(self, proc: subprocess.Popen, nodeid: int, simulator: Simulator): + self.__otcli_proc = proc + self.__nodeid = nodeid + self.__simulator = simulator + + def __repr__(self): + return 'OTCli<%d>' % self.__nodeid + + def readline(self) -> str: + return self.__otcli_proc.stdout.readline().rstrip('\r\n') + + def writeline(self, s: str): + self.__otcli_proc.stdin.write(s + '\n') + self.__otcli_proc.stdin.flush() + + def wait(self, duration: float): + if self.__simulator is not None: + # Virtual time simulation + self.__simulator.go(duration) + else: + # Real time simulation + time.sleep(duration) + + def close(self): + self.__otcli_proc.stdin.close() + self.__otcli_proc.stdout.close() + self.__otcli_proc.wait() + + +class OtCliSim(OtCliPopen): + """Connector for OT CLI Simulation instances.""" + + def __init__(self, executable: str, nodeid: int, simulator: Simulator): + logging.info('%s: executable=%s', self.__class__.__name__, executable) + + proc = subprocess.Popen(args=[executable, str(nodeid)], + executable=executable, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + encoding='utf-8', + bufsize=1024) + super().__init__(proc, nodeid, simulator) + + +class OtNcpSim(OtCliHandler): + """Connector for OT NCP Simulation instances.""" + + def __init__(self, executable: str, nodeid: int, simulator: Simulator): + logging.info('%s: executable=%s', self.__class__.__name__, executable) + + proc = subprocess.Popen(args=f'spinel-cli.py -p "{executable}" -n {nodeid} 2>&1', + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + encoding='utf-8', + bufsize=1024, + shell=True) + super().__init__(proc, nodeid, simulator) + + +class OtCliSerial(OtCliHandler): + """Connector for OT CLI SOC devices via Serial.""" + + def __init__(self, dev: str, baudrate: int): + self.__dev = dev + self.__baudrate = baudrate + + import serial + self.__serial = serial.Serial(self.__dev, self.__baudrate, timeout=None, exclusive=True) + + def __repr__(self): + return self.__dev + + def readline(self) -> str: + line = self.__serial.readline().decode('utf-8').rstrip('\r\n') + return line + + def writeline(self, s: str): + self.__serial.write((s + '\n').encode('utf-8')) + + def wait(self, duration: float): + time.sleep(duration) + + def close(self): + self.__serial.close() diff --git a/tools/otci/otci/constants.py b/tools/otci/otci/constants.py new file mode 100644 index 000000000..4bf13c4cb --- /dev/null +++ b/tools/otci/otci/constants.py @@ -0,0 +1,32 @@ +#!/usr/bin/env python3 +# +# Copyright (c) 2020, 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. +# + +# Thread versions +THREAD_VERSION_1_1 = 2 +THREAD_VERSION_1_2 = 3 diff --git a/tools/otci/otci/errors.py b/tools/otci/otci/errors.py new file mode 100644 index 000000000..dc8fd59a3 --- /dev/null +++ b/tools/otci/otci/errors.py @@ -0,0 +1,64 @@ +#!/usr/bin/env python3 +# +# Copyright (c) 2020, 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. +# +from typing import List + + +class OTCIError(Exception): + """Base class for OTCI Errors.""" + pass + + +class ExpectLineTimeoutError(OTCIError): + """OTCI failed to find an expected line before timeout.""" + + def __init__(self, line): + super(ExpectLineTimeoutError, self).__init__("Expected line %r, but timed out" % line) + + +class CommandError(OTCIError): + """OTCI failed to execute a command.""" + + def __init__(self, cmd: str, output: List[str]): + self.__output = output + super(CommandError, self).__init__("Command error while executing %r:\n%s\n" % (cmd, '\n'.join(output))) + + def error(self) -> str: + return self.__output[-1] + + +class UnexpectedCommandOutput(OTCIError): + """OTCI got unexpected command output.""" + + def __init__(self, output: List[str]): + super(UnexpectedCommandOutput, self).__init__("Unexpected command output:\n%s\n" % '\n'.join(output)) + + +class InvalidArgumentsError(OTCIError): + """Invalid arguments.""" + pass diff --git a/tools/otci/otci/otci.py b/tools/otci/otci/otci.py new file mode 100644 index 000000000..65686a3c6 --- /dev/null +++ b/tools/otci/otci/otci.py @@ -0,0 +1,1901 @@ +#!/usr/bin/env python3 +# +# Copyright (c) 2020, 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. +# +import ipaddress +import logging +import re +import time +from collections import Counter +from typing import List, Collection, Union, Tuple, Optional, Dict, Pattern, Any + +from . import connectors +from .command_handlers import OTCommandHandler, OtCliCommandRunner, OtbrSshCommandRunner +from .connectors import Simulator +from .errors import UnexpectedCommandOutput, ExpectLineTimeoutError, CommandError, InvalidArgumentsError +from .types import ChildId, Rloc16, Ip6Addr, ThreadState, PartitionId, DeviceMode, RouterId, SecurityPolicy, Ip6Prefix +from .utils import match_line, constant_property + + +class OTCI(object): + """ + This class represents an OpenThread Controller Interface instance that provides versatile interfaces to + manipulate an OpenThread device. + """ + + def __init__(self, otcmd: OTCommandHandler): + """ + This method initializes an OTCI instance. + + :param otcmd: An OpenThread Command Handler instance to execute OpenThread CLI commands. + """ + self.__otcmd: OTCommandHandler = otcmd + self.__logger = logging.getLogger(name=str(self)) + + def __repr__(self): + """Gets the string representation of the OTCI instance.""" + return repr(self.__otcmd) + + def wait(self, duration: float, expect_line: Union[str, Pattern, Collection[Any]] = None): + """Wait for a given duration. + + :param duration: The duration (in seconds) wait for. + :param expect_line: The line expected to output if given. + Raise ExpectLineTimeoutError if expect_line is not found within the given duration. + """ + self.__logger and self.__logger.info("wait for %.3f seconds", duration) + if expect_line is None: + self.__otcmd.wait(duration) + else: + success = False + + while duration > 0: + output = self.__otcmd.wait(1) + if match_line(expect_line, output): + success = True + break + + if not success: + raise ExpectLineTimeoutError(expect_line) + + def close(self): + """Close the OTCI instance.""" + self.__otcmd.close() + + def execute_command(self, cmd: str, timeout: float = 10, silent: bool = False) -> List[str]: + """Execute the OpenThread CLI command. + + :param cmd: The command to execute. + :param timeout: The command timeout. + :param silent: Whether to run the command silent without logging. + :returns: The command output as a list of lines. + """ + if not silent: + self.__logger and self.__logger.info('> %s', cmd) + + output = self.__otcmd.execute_command(cmd, timeout) + + if not silent: + for line in output: + self.__logger and self.__logger.info('%s', line) + + if cmd in ('reset', 'factoryreset'): + self.__wait_reset() + return output + + if output[-1] == 'Done': + output = output[:-1] + return output + else: + raise CommandError(cmd, output) + + def set_logger(self, logger: logging.Logger): + """Set the logger for the OTCI instance, or None to disable logging.""" + self.__logger = logger + + # + # Constant properties + # + @constant_property + def version(self): + """Returns the firmware version. (e.g. "OPENTHREAD/20191113-01411-gb2d66e424-dirty; SIMULATION; Nov 14 2020 14:24:38")""" + return self.__parse_str(self.execute_command('version')) + + @constant_property + def thread_version(self): + """Get the Thread Version number.""" + return self.__parse_int(self.execute_command('thread version')) + + @constant_property + def api_version(self): + """Get API version number.""" + try: + return self.__parse_int(self.execute_command('version api')) + except ValueError: + # If the device does not have `version api` command, it will print the firmware version, which would lead to ValueError. + return 0 + + # + # Basic device operations + # + def ifconfig_up(self): + """Bring up the IPv6 interface.""" + self.execute_command('ifconfig up') + + def ifconfig_down(self): + """Bring down the IPv6 interface.""" + self.execute_command('ifconfig down') + + def get_ifconfig_state(self) -> bool: + """Get the status of the IPv6 interface.""" + return self.__parse_values(self.execute_command('ifconfig'), up=True, down=False) + + def thread_start(self): + """Enable Thread protocol operation and attach to a Thread network.""" + self.execute_command('thread start') + + def thread_stop(self): + """Disable Thread protocol operation and detach from a Thread network.""" + self.execute_command('thread stop') + + def reset(self): + """Signal a platform reset.""" + self.execute_command('reset') + + def factory_reset(self): + """Delete all stored settings, and signal a platform reset.""" + self.execute_command('factoryreset') + + # + # Network Operations + # + + def ping(self, ip: str, size: int = None, count: int = None, interval: int = None, hoplimit: int = None): + """Send an ICMPv6 Echo Request. + + :param ip: The target IPv6 address to ping. + :param size: The number of data bytes in the payload. + :param count: The number of ICMPv6 Echo Requests to be sent. + :param interval: The interval between two consecutive ICMPv6 Echo Requests in seconds. The value may have fractional form, for example 0.5. + :param hoplimit: The hoplimit of ICMPv6 Echo Request to be sent. + """ + cmd = f'ping {ip}' + + if size is not None: + cmd += f' {size}' + + if count is not None: + cmd += f' {count}' + + if interval is not None: + cmd += f' {interval}' + + if hoplimit is not None: + cmd += f' {hoplimit}' + + self.execute_command(cmd) + + def ping_stop(self): + """Stop sending ICMPv6 Echo Requests.""" + self.execute_command('ping stop') + + def scan(self, channel: int = None) -> List[Dict[str, Any]]: + """Perform an IEEE 802.15.4 Active Scan.""" + cmd = 'scan' + if channel is not None: + cmd += f' {channel}' + + output = self.execute_command(cmd, timeout=10) + if len(output) < 2: + raise UnexpectedCommandOutput(output) + + networks = [] + for line in output[2:]: + fields = line.strip().split('|') + + _, J, netname, extpanid, panid, extaddr, ch, dbm, lqi, _ = fields + networks.append({ + 'joinable': bool(int(J)), + 'network_name': netname.strip(), + 'extpanid': extpanid, + 'panid': int(panid, 16), + 'extaddr': extaddr, + 'channel': int(ch), + 'dbm': int(dbm), + 'lqi': int(lqi), + }) + + return networks + + def scan_energy(self, duration: float = None, channel: int = None) -> Dict[int, int]: + """Perform an IEEE 802.15.4 Energy Scan.""" + cmd = 'scan energy' + if duration is not None: + cmd += f' {duration * 1000:d}' + + if channel is not None: + cmd += f' {channel}' + + output = self.execute_command(cmd, timeout=10) + if len(output) < 2: + raise UnexpectedCommandOutput(output) + + channels = {} + for line in output[2:]: + fields = line.strip().split('|') + + _, Ch, RSSI, _ = fields + channels[int(Ch)] = int(RSSI) + + return channels + + def mac_send_data_request(self): + """Instruct an Rx-Off-When-Idle device to send a Data Request mac frame to its parent.""" + self.execute_command('mac send datarequest') + + def mac_send_empty_data(self): + """Instruct an Rx-Off-When-Idle device to send a Empty Data mac frame to its parent.""" + self.execute_command('mac send emptydata') + + # TODO: discover + # TODO: dns resolve [DNS server IP] [DNS server port] + # TODO: fake /a/an + # TODO: sntp query + + # + # Set or get device/network parameters + # + + def get_mode(self) -> str: + """Get the Thread Device Mode value. + + -: no flags set (rx-off-when-idle, minimal Thread device, stable network data) + r: rx-on-when-idle + d: Full Thread Device + n: Full Network Data + """ + return self.__parse_str(self.execute_command('mode')) + + def set_mode(self, mode: str): + """Set the Thread Device Mode value. + + -: no flags set (rx-off-when-idle, minimal Thread device, stable network data) + r: rx-on-when-idle + d: Full Thread Device + n: Full Network Data + """ + self.execute_command(f'mode {DeviceMode(mode)}') + + def get_extaddr(self) -> str: + """Get the IEEE 802.15.4 Extended Address.""" + return self.__parse_extaddr(self.execute_command('extaddr')) + + def set_extaddr(self, extaddr: str): + """Set the IEEE 802.15.4 Extended Address.""" + self.__validate_hex64b(extaddr) + self.execute_command(f'extaddr {extaddr}') + + def get_eui64(self) -> str: + """Get the factory-assigned IEEE EUI-64.""" + return self.__parse_eui64(self.execute_command('eui64')) + + def set_extpanid(self, extpanid: str): + """Set the Thread Extended PAN ID value.""" + self.__validate_extpanid(extpanid) + self.execute_command(f'extpanid {extpanid}') + + def get_extpanid(self) -> str: + """Get the Thread Extended PAN ID value.""" + return self.__parse_extpanid(self.execute_command('extpanid')) + + def set_channel(self, ch): + """Set the IEEE 802.15.4 Channel value.""" + self.execute_command('channel %d' % ch) + + def get_channel(self): + """Get the IEEE 802.15.4 Channel value.""" + return self.__parse_int(self.execute_command('channel')) + + def get_preferred_channel_mask(self) -> int: + """Get preferred channel mask.""" + return self.__parse_int(self.execute_command('channel preferred')) + + def get_supported_channel_mask(self): + """Get supported channel mask.""" + return self.__parse_int(self.execute_command('channel supported')) + + def get_panid(self): + """Get the IEEE 802.15.4 PAN ID value.""" + return self.__parse_int(self.execute_command('panid'), 16) + + def set_panid(self, panid): + """Get the IEEE 802.15.4 PAN ID value.""" + self.execute_command('panid %d' % panid) + + def set_network_name(self, name): + """Set network name.""" + self.execute_command('networkname %s' % self.__escape_escapable(name)) + + def get_network_name(self): + """Get network name.""" + return self.__parse_str(self.execute_command('networkname')) + + def get_master_key(self) -> str: + """Get the network master key.""" + return self.__parse_master_key(self.execute_command('masterkey')) + + def set_master_key(self, masterkey: str): + """Set the network master key.""" + self.__validate_master_key(masterkey) + self.execute_command(f'masterkey {masterkey}') + + def get_key_sequence_counter(self) -> int: + """Get the Thread Key Sequence Counter.""" + return self.__parse_int(self.execute_command('keysequence counter')) + + def set_key_sequence_counter(self, counter: int): + """Set the Thread Key Sequence Counter.""" + self.execute_command(f'keysequence counter {counter}') + + def get_key_sequence_guard_time(self) -> int: + """Get Thread Key Switch Guard Time (in hours).""" + return self.__parse_int(self.execute_command('keysequence guardtime')) + + def set_key_sequence_guard_time(self, hours: int): + """Set Thread Key Switch Guard Time (in hours) 0 means Thread Key Switch immediately if key index match.""" + self.execute_command(f'keysequence guardtime {hours}') + + def get_cca_threshold(self) -> int: + """Get the CCA threshold in dBm measured at antenna connector per IEEE 802.15.4 - 2015 section 10.1.4.""" + output = self.execute_command(f'ccathreshold') + val = self.__parse_str(output) + if not val.endswith(' dBm'): + raise UnexpectedCommandOutput(output) + + return int(val[:-4]) + + def set_cca_threshold(self, val: int): + """Set the CCA threshold measured at antenna connector per IEEE 802.15.4 - 2015 section 10.1.4.""" + self.execute_command(f'ccathreshold {val}') + + def get_promiscuous(self) -> bool: + """Get radio promiscuous property.""" + return self.__parse_Enabled_or_Disabled(self.execute_command('promiscuous')) + + def enable_promiscuous(self): + """Enable radio promiscuous operation and print raw packet content.""" + self.execute_command('promiscuous enable') + + def disable_promiscuous(self): + """Disable radio promiscuous operation.""" + self.execute_command('promiscuous disable') + + def get_txpower(self) -> int: + """Get the transmit power in dBm.""" + line = self.__parse_str(self.execute_command('txpower')) + if not line.endswith(' dBm'): + raise UnexpectedCommandOutput([line]) + + return int(line.split()[0]) + + def set_txpower(self, val: int): + """Set the transmit power in dBm.""" + self.execute_command(f'txpower {val}') + + # TODO: fem + # TODO: fem lnagain + # TODO: fem lnagain + # TODO: mac retries direct + # TODO: mac retries direct + # TODO: mac retries indirect + # TODO: mac retries indirect + + # + # Basic Node states and properties + # + + def get_state(self) -> ThreadState: + """Get the current Thread state.""" + return ThreadState(self.__parse_str(self.execute_command('state'))) + + def set_state(self, state: str): + """Try to switch to state detached, child, router or leader.""" + self.execute_command(f'state {state}') + + def get_rloc16(self) -> int: + """Get the Thread RLOC16 value.""" + return self.__parse_int(self.execute_command('rloc16'), 16) + + def get_router_id(self) -> int: + """Get the Thread Router ID value.""" + return self.get_rloc16() >> 10 + + def prefer_router_id(self, routerid: int): + """Prefer a Router ID when solicit router id from Leader.""" + self.execute_command(f'preferrouterid {routerid}') + + def is_singleton(self) -> bool: + return self.__parse_values(self.execute_command('singleton'), true=True, false=False) + + # + # RCP related utilities + # + + def get_rcp_version(self): + return self.__parse_str(self.execute_command('rcp version')) + + # + # Unsecure port utilities + # + + def get_unsecure_ports(self) -> List[int]: + """all ports from the allowed unsecured port list.""" + return self.__parse_int_list(self.execute_command('unsecureport get')) + + def add_unsecure_port(self, port: int): + """Add a port to the allowed unsecured port list.""" + self.execute_command(f'unsecureport add {port}') + + def remove_unsecure_port(self, port: int): + """Remove a port from the allowed unsecured port list.""" + self.execute_command(f'unsecureport remove {port}') + + def clear_unsecure_ports(self): + """Remove all ports from the allowed unsecured port list.""" + self.execute_command('unsecureport remove all') + + # + # Leader configurations + # + + def get_preferred_partition_id(self) -> PartitionId: + """Get the preferred Thread Leader Partition ID.""" + return PartitionId(self.__parse_int(self.execute_command(self.__get_partition_preferred_cmd()))) + + def set_preferred_partition_id(self, parid: int): + """Set the preferred Thread Leader Partition ID.""" + self.execute_command(f'{self.__get_partition_preferred_cmd()} {parid}') + + def __get_partition_preferred_cmd(self) -> str: + """""" + return 'partitionid preferred' if self.api_version >= 51 else 'leaderpartitionid' + + def get_leader_weight(self) -> int: + """Get the Thread Leader Weight.""" + return self.__parse_int(self.execute_command('leaderweight')) + + def set_leader_weight(self, weight: int): + """Set the Thread Leader Weight.""" + self.execute_command(f'leaderweight {weight}') + + __LEADER_DATA_KEY_MAP = { + 'Partition ID': 'partition_id', + 'Weighting': 'weight', + 'Data Version': 'data_ver', + 'Stable Data Version': 'stable_data_ver', + 'Leader Router ID': 'leader_id', + } + + def get_leader_data(self) -> Dict[str, int]: + """Get the Thread Leader Data.""" + data = {} + output = self.execute_command('leaderdata') + + try: + for line in output: + k, v = line.split(': ') + data[OTCI.__LEADER_DATA_KEY_MAP[k]] = int(v) + except KeyError: + raise UnexpectedCommandOutput(output) + + return data + + # + # Router configurations + # + + def get_router_selection_jitter(self): + """Get the ROUTER_SELECTION_JITTER value.""" + return self.__parse_int(self.execute_command('routerselectionjitter')) + + def set_router_selection_jitter(self, jitter): + """Set the ROUTER_SELECTION_JITTER value.""" + self.execute_command(f'routerselectionjitter {jitter}') + + def get_network_id_timeout(self) -> int: + """Get the NETWORK_ID_TIMEOUT parameter used in the Router role.""" + return self.__parse_int(self.execute_command('networkidtimeout')) + + def set_network_id_timeout(self, timeout: int): + """Set the NETWORK_ID_TIMEOUT parameter used in the Router role.""" + self.execute_command(f'networkidtimeout {timeout}') + + def get_parent_priority(self) -> int: + """Get the assigned parent priority value, -2 means not assigned.""" + return self.__parse_int(self.execute_command('parentpriority')) + + def set_parent_priority(self, priority: int): + """Set the assigned parent priority value: 1, 0, -1 or -2.""" + self.execute_command(f'parentpriority {priority}') + + def get_router_upgrade_threshold(self) -> int: + """Get the ROUTER_UPGRADE_THRESHOLD value.""" + return self.__parse_int(self.execute_command('routerupgradethreshold')) + + def set_router_upgrade_threshold(self, threshold: int): + """Set the ROUTER_UPGRADE_THRESHOLD value.""" + self.execute_command(f'routerupgradethreshold {threshold}') + + def get_router_downgrade_threshold(self): + """Set the ROUTER_DOWNGRADE_THRESHOLD value.""" + return self.__parse_int(self.execute_command('routerdowngradethreshold')) + + def set_router_downgrade_threshold(self, threshold: int): + """Get the ROUTER_DOWNGRADE_THRESHOLD value.""" + self.execute_command(f'routerdowngradethreshold {threshold}') + + def get_router_eligible(self) -> bool: + """Indicates whether the router role is enabled or disabled.""" + return self.__parse_Enabled_or_Disabled(self.execute_command('routereligible')) + + def enable_router_eligible(self): + """Disable the router role.""" + self.execute_command('routereligible enable') + + def disable_router_eligible(self): + """Disable the router role.""" + self.execute_command('routereligible disable') + + def get_router_list(self) -> List[RouterId]: + """Get allocated Router IDs.""" + line = self.__parse_str(self.execute_command('router list')) + return list(map(RouterId, line.strip().split())) + + def get_router_table(self) -> Dict[RouterId, Dict[str, Any]]: + """table of routers.""" + output = self.execute_command('router table') + if len(output) < 2: + raise UnexpectedCommandOutput(output) + + # + # Example output: + # + # | ID | RLOC16 | Next Hop | Path Cost | LQ In | LQ Out | Age | Extended MAC | + # +----+--------+----------+-----------+-------+--------+-----+------------------+ + # | 21 | 0x5400 | 21 | 0 | 3 | 3 | 5 | d28d7f875888fccb | + # | 56 | 0xe000 | 56 | 0 | 0 | 0 | 182 | f2d92a82c8d8fe43 | + # Done + # + + headers = self.__split_table_row(output[0]) + + table = {} + for line in output[2:]: + line = line.strip() + if not line: + continue + + fields = self.__split_table_row(line) + if len(fields) != len(headers): + raise UnexpectedCommandOutput(output) + + col = lambda colname: self.__get_table_col(colname, headers, fields) + id = col('ID') + + table[RouterId(id)] = { + 'id': RouterId(id), + 'rloc16': Rloc16(col('RLOC16'), 16), + 'next_hop': int(col('Next Hop')), + 'path_cost': int(col('Path Cost')), + 'lq_in': int(col('LQ In')), + 'lq_out': int(col('LQ Out')), + 'age': int(col('Age')), + 'extaddr': col('Extended MAC'), + } + + return table + + # TODO: router + + # + # Router utilities: Child management + # + + def get_child_table(self) -> Dict[ChildId, Dict[str, Any]]: + """Get the table of attached children.""" + output = self.execute_command('child table') + if len(output) < 2: + raise UnexpectedCommandOutput(output) + + # + # Example output: + # | ID | RLOC16 | Timeout | Age | LQ In | C_VN |R|D|N|Ver|CSL|QMsgCnt| Extended MAC | + # +-----+--------+------------+------------+-------+------+-+-+-+---+---+-------+------------------+ + # | 1 | 0xc801 | 240 | 24 | 3 | 131 |1|0|0| 3| 0 | 0 | 4ecede68435358ac | + # | 2 | 0xc802 | 240 | 2 | 3 | 131 |0|0|0| 3| 1 | 0 | a672a601d2ce37d8 | + # Done + # + + headers = self.__split_table_row(output[0]) + + table = {} + for line in output[2:]: + line = line.strip() + if not line: + continue + + fields = self.__split_table_row(line) + col = lambda colname: self.__get_table_col(colname, headers, fields) + + id = int(col("ID")) + r, d, n = int(col("R")), int(col("D")), int(col("N")) + mode = DeviceMode(f'{"r" if r else ""}{"d" if d else ""}{"n" if n else ""}') + + child = { + 'id': ChildId(id), + 'rloc16': Rloc16(col('RLOC16'), 16), + 'timeout': int(col('Timeout')), + 'age': int(col('Age')), + 'lq_in': int(col('LQ In')), + 'c_vn': int(col('C_VN')), + 'mode': mode, + 'extaddr': col('Extended MAC') + } + + if 'Ver' in headers: + child['ver'] = int(col('Ver')) + + if 'CSL' in headers: + child['csl'] = bool(int(col('CSL'))) + + if 'QMsgCnt' in headers: + child['qmsgcnt'] = int(col('QMsgCnt')) + + table[ChildId(id)] = child + + return table + + def __split_table_row(self, row: str) -> List[str]: + if not (row.startswith('|') and row.endswith('|')): + raise ValueError(row) + + fields = row.split('|') + fields = [x.strip() for x in fields[1:-1]] + return fields + + def __get_table_col(self, colname: str, headers: List[str], fields: List[str]) -> str: + return fields[headers.index(colname)] + + def get_child_list(self) -> List[ChildId]: + """Get attached Child IDs.""" + line = self.__parse_str(self.execute_command(f'child list')) + return [ChildId(id) for id in line.strip().split()] + + def get_child_info(self, child: Union[ChildId, Rloc16]) -> Dict[str, Any]: + output = self.execute_command(f'child {child}') + + info = {} + + for line in output: + k, v = line.split(': ') + if k == 'Child ID': + info['id'] = int(v) + elif k == 'Rloc': + info['rloc16'] = int(v, 16) + elif k == 'Ext Addr': + info['extaddr'] = v + elif k == 'Mode': + info['mode'] = DeviceMode(v) + elif k == 'Net Data': + info['c_vn'] = int(v) + elif k == 'Timeout': + info['timeout'] = int(v) + elif k == 'Age': + info['age'] = int(v) + elif k == 'Link Quality In': + info['lq_in'] = int(v) + elif k == 'RSSI': + info['rssi'] = int(v) + else: + self.__logger and self.__logger.warning("Child info %s: %s ignored", k, v) + + return info + + def get_child_ipaddrs(self) -> Dict[Rloc16, List[Ip6Addr]]: + """Get the list of IP addresses stored for MTD children. + + Note: Each MTD child might has multiple IP addresses. + """ + output = self.execute_command('childip') + + ipaddrs = {} + + for line in output: + rloc16, ip = line.split(': ') + rloc16 = Rloc16(rloc16, 16) + ipaddrs.setdefault(rloc16, []).append(Ip6Addr(ip.strip())) + + return ipaddrs + + # + # Child configurations + # + + def get_max_children(self) -> int: + """Get the Thread maximum number of allowed children.""" + return self.__parse_int(self.execute_command('childmax')) + + def set_max_children(self, val: int): + """Set the Thread maximum number of allowed children.""" + self.execute_command(f'childmax {val}') + + def get_child_ip_max(self) -> int: + """Get the maximum number of IP addresses that each MTD child may register with this device as parent.""" + return self.__parse_int(self.execute_command('childip max')) + + def set_child_ip_max(self, val: int): + """Get the maximum number of IP addresses that each MTD child may register with this device as parent.""" + self.execute_command(f'childip max {val}') + + def get_child_timeout(self): + """Get the Thread Child Timeout value.""" + return self.__parse_int(self.execute_command('childtimeout')) + + def set_child_timeout(self, timeout): + """Set the Thread Child Timeout value.""" + self.execute_command('childtimeout %d' % timeout) + + def get_child_supervision_interval(self) -> int: + """Get the Child Supervision Check Timeout value.""" + return self.__parse_int(self.execute_command('childsupervision interval')) + + def set_child_supervision_interval(self, val: int): + """Set the Child Supervision Interval value. + This command can only be used with FTD devices. + """ + self.execute_command(f'childsupervision interval {val}') + + def get_child_supervision_check_timeout(self) -> int: + """Get the Child Supervision Check Timeout value.""" + return self.__parse_int(self.execute_command('childsupervision checktimeout')) + + def set_child_supervision_check_timeout(self, val: int): + """Set the Child Supervision Check Timeout value.""" + self.execute_command(f'childsupervision checktimeout {val}') + + # + # Neighbor management + # + + def get_neighbor_list(self) -> List[Rloc16]: + """Get a list of RLOC16 of neighbors""" + line = self.__parse_str(self.execute_command('neighbor list')).strip() + return [Rloc16(id, 16) for id in line.split()] + + def get_neighbor_table(self) -> Dict[Rloc16, Dict[str, Any]]: + output = self.execute_command('neighbor table') + if len(output) < 2: + raise UnexpectedCommandOutput(output) + + # + # Example output: + # + # | Role | RLOC16 | Age | Avg RSSI | Last RSSI |R|D|N| Extended MAC | + # +------+--------+-----+----------+-----------+-+-+-+------------------+ + # | C | 0xcc01 | 96 | -46 | -46 |1|1|1| 1eb9ba8a6522636b | + # | R | 0xc800 | 2 | -29 | -29 |1|1|1| 9a91556102c39ddb | + # | R | 0xf000 | 3 | -28 | -28 |1|1|1| 0ad7ed6beaa6016d | + # Done + # + + headers = self.__split_table_row(output[0]) + + table = {} + for line in output[2:]: + line = line.strip() + if not line: + continue + + fields = self.__split_table_row(line) + col = lambda colname: self.__get_table_col(colname, headers, fields) + + role = col('Role') + is_router = role == 'R' + r, d, n = int(col('R')), int(col('D')), int(col('N')) + mode = DeviceMode(f'{"r" if r else ""}{"d" if d else ""}{"n" if n else ""}') + + rloc16 = Rloc16(col('RLOC16'), 16) + + table[rloc16] = { + 'is_router': is_router, + 'rloc16': rloc16, + 'age': int(col('Age')), + 'avg_rssi': int(col('Avg RSSI')), + 'last_rssi': int(col('Last RSSI')), + 'mode': mode, + 'extaddr': col('Extended MAC'), + } + + return table + + # + # SED/SSED configuration + # + + def get_poll_period(self) -> int: + """Get the customized data poll period of sleepy end device (milliseconds). + Only for Reference Device.""" + return self.__parse_int(self.execute_command('pollperiod')) + + def set_poll_period(self, poll_period: int): + """Set the customized data poll period (in milliseconds) for sleepy end device. + + Only for Reference Device.""" + self.execute_command(f'pollperiod {poll_period}') + + # TODO: csl + # TODO: csl channel + # TODO: csl period + # TODO: csl timeout + + _CSL_PERIOD_PATTERN = re.compile(r'(\d+)\(in units of 10 symbols\), \d+ms') + _CSL_TIMEOUT_PATTERN = re.compile(r'(\d+)s') + + def get_csl_config(self) -> Dict[str, int]: + """Get the CSL configuration.""" + output = self.execute_command('csl') + + cfg = {} + for line in output: + k, v = line.split(': ') + if k == 'Channel': + cfg['channel'] = int(v) + elif k == 'Timeout': + cfg['timeout'] = int(OTCI._CSL_TIMEOUT_PATTERN.match(v).group(1)) + elif k == 'Period': + cfg['period'] = int(OTCI._CSL_PERIOD_PATTERN.match(v).group(1)) + else: + logging.warning("Ignore unknown CSL parameter: %s: %s", k, v) + + return cfg + + def config_csl(self, channel: int = None, period: int = None, timeout: int = None): + """Configure CSL parameters. + + :param channel: Set CSL channel. + :param period: Set CSL period in units of 10 symbols. Disable CSL by setting this parameter to 0. + :param timeout: Set the CSL timeout in seconds. + """ + + if channel is None and period is None and timeout is None: + raise InvalidArgumentsError("Please specify at least 1 parameter to configure.") + + if channel is not None: + self.execute_command(f'csl channel {channel}') + + if period is not None: + self.execute_command(f'csl period {period}') + + if timeout is not None: + self.execute_command(f'csl timeout {timeout}') + + # + # Leader utilities + # + + def get_context_id_reuse_delay(self) -> int: + """Get the CONTEXT_ID_REUSE_DELAY value.""" + return self.__parse_int(self.execute_command('contextreusedelay')) + + def set_context_id_reuse_delay(self, val: int): + """Set the CONTEXT_ID_REUSE_DELAY value.""" + self.execute_command(f'contextreusedelay {val}') + + def release_router_id(self, routerid: int): + """Release a Router ID that has been allocated by the device in the Leader role.""" + self.execute_command(f'releaserouterid {routerid}') + + # Time Sync utilities + # TODO: networktime + # TODO: networktime + # TODO: delaytimermin + # TODO: delaytimermin + + # + # Commissioniner operations + # + + def commissioner_start(self): + """Start the Commissioner role.""" + self.execute_command('commissioner start') + + def commissioner_stop(self): + """Stop the Commissioner role.""" + self.execute_command('commissioner stop') + + def get_commissioiner_state(self) -> str: + """Get current Commissioner state (active or petitioning or disabled).""" + return self.__parse_str(self.execute_command('commissioner state')) + + def get_commissioner_session_id(self) -> int: + """Get current commissioner session id.""" + return self.__parse_int(self.execute_command('commissioner sessionid')) + + def commissioner_add_joiner(self, pskd, eui64=None, discerner=None, timeout=None): + """Add a Joiner entry. + + :param pskd: Pre-Shared Key for the Joiner. + :param eui64: The IEEE EUI-64 of the Joiner or '*' to match any Joiner + :param discerner: The Joiner discerner in format number/length. + :param timeout: Joiner timeout in seconds. + """ + if (eui64 is not None) == (discerner is not None): + raise InvalidArgumentsError("Please specify eui64 or discerner, but not both.") + + if eui64 is not None and eui64 != '*': + self.__validate_extaddr(eui64) + + cmd = f'commissioner joiner add {eui64 or discerner} {pskd}' + + if timeout is not None: + cmd += f' {timeout}' + + self.execute_command(cmd) + + def commissioner_remove_jointer(self, eui64=None, discerner=None): + if (eui64 is not None) == (discerner is not None): + raise InvalidArgumentsError("Please specify eui64 or discerner, but not both.") + + if eui64 is not None and eui64 != '*': + self.__validate_extaddr(eui64) + + self.execute_command(f'commissioner joiner remove {eui64 or discerner}') + + def set_commissioner_provisioning_url(self, url: str): + self.execute_command(f'commissioner provisioningurl {url}') + + # TODO: commissioner announce + # TODO: commissioner energy + # TODO: commissioner mgmtget + # TODO: commissioner mgmtset + # TODO: commissioner panid + + # + # Joiner operations + # + def joiner_start(self, psk: str, provisioning_url: str = None): + """Start the Joiner.""" + cmd = f'joiner start {psk}' + if provisioning_url is not None: + cmd += f' {provisioning_url}' + + self.execute_command(cmd) + + def joiner_stop(self): + """Stop the Joiner role.""" + self.execute_command('joiner stop') + + def get_joiner_id(self) -> str: + """Get the Joiner ID.""" + return self.__parse_joiner_id(self.execute_command('joiner id')) + + def get_joiner_port(self) -> int: + """Get the Joiner port.""" + return self.__parse_int(self.execute_command(f'joinerport')) + + def set_joiner_port(self, port: int): + """Set the Joiner port.""" + self.execute_command(f'joinerport {port}') + + # TODO: joiner discerner + + # + # Network Data utilities + # + def get_prefixes(self) -> List[Tuple[Ip6Prefix, str, str, Rloc16]]: + """Get the prefix list in the local Network Data.""" + output = self.execute_command('prefix') + return self.__parse_prefixes(output) + + def __parse_prefixes(self, output: List[str]) -> List[Tuple[Ip6Prefix, str, str, Rloc16]]: + prefixes = [] + + for line in output: + if line.startswith('- '): + line = line[2:] + + prefix, flags, prf, rloc16 = line.split()[:4] + prefixes.append((Ip6Prefix(prefix), flags, prf, Rloc16(rloc16, 16))) + + return prefixes + + def add_prefix(self, prefix: str, flags='paosr', prf='med'): + """Add a valid prefix to the Network Data.""" + self.execute_command(f'prefix add {prefix} {flags} {prf}') + + def remove_prefix(self, prefix: str): + """Invalidate a prefix in the Network Data.""" + self.execute_command(f'prefix remove {prefix}') + + def register_network_data(self): + self.execute_command('netdata register') + + def get_network_data(self) -> Dict[str, List]: + output = self.execute_command('netdata show') + + netdata = {} + if output.pop(0) != 'Prefixes:': + raise UnexpectedCommandOutput(output) + + prefixes_output = [] + while True: + line = output.pop(0) + if line == 'Routes:': + break + else: + prefixes_output.append(line) + + netdata['prefixes'] = self.__parse_prefixes(prefixes_output) + + print(prefixes_output, output) + routes_output = [] + while True: + line = output.pop(0) + if line == 'Services:': + break + else: + routes_output.append(line) + + print('routes_output:', routes_output, output) + netdata['routes'] = self.__parse_routes(routes_output) + netdata['services'] = self.__parse_services(output) + + return netdata + + def __parse_services(self, output: List[str]) -> List[Tuple[int, bytes, bytes, bool, Rloc16]]: + services = [] + for line in output: + line = line.split() + + enterprise_number, service_data, server_data = line[:3] + if line[3] == 's': + stable, rloc16 = True, line[4] + else: + stable, rloc16 = False, line[3] + + enterprise_number = int(enterprise_number) + service_data = self.__hex_to_bytes(service_data) + server_data = self.__hex_to_bytes(server_data) + rloc16 = Rloc16(rloc16, 16) + + services.append((enterprise_number, service_data, server_data, stable, rloc16)) + + return services + + def get_network_data_bytes(self) -> bytes: + """Get the raw Network Data.""" + hexstr = self.__parse_str(self.execute_command('netdata show -x')) + return bytes(int(hexstr[i:i + 2], 16) for i in range(0, len(hexstr), 2)) + + def get_routes(self) -> List[Tuple[str, bool, str, Rloc16]]: + """Get the external route list in the local Network Data.""" + return self.__parse_routes(self.execute_command('route')) + + def __parse_routes(self, output: List[str]) -> List[Tuple[str, bool, str, Rloc16]]: + routes = [] + for line in output: + line = line.split() + print(line) + if line[1] == 's': + prefix, _, prf, rloc16 = line + stable = True + else: + prefix, prf, rloc16 = line + stable = False + + rloc16 = Rloc16(rloc16, 16) + routes.append((prefix, stable, prf, rloc16)) + + return routes + + def add_route(self, prefix: str, stable=True, prf='med'): + """Add a valid external route to the Network Data.""" + cmd = f'route add {prefix}' + if stable: + cmd += ' s' + + cmd += f' {prf}' + self.execute_command(cmd) + + def remove_route(self, prefix: str): + """Invalidate a external route in the Network Data.""" + self.execute_command(f'route remove {prefix}') + + def add_service(self, enterprise_number: int, service_data: Union[str, bytes], server_data: Union[str, bytes]): + """Add service to the Network Data. + + enterpriseNumber: IANA enterprise number + serviceData: hex-encoded binary service data + serverData: hex-encoded binary server data + """ + service_data = self.__validate_hex_or_bytes(service_data) + server_data = self.__validate_hex_or_bytes(server_data) + self.execute_command(f'service add {enterprise_number} {service_data} {server_data}') + + def remove_service(self, enterprise_number, service_data): + """Remove service from Network Data. + + enterpriseNumber: IANA enterprise number + serviceData: hext-encoded binary service data + """ + service_data = self.__validate_hex_or_bytes(service_data) + self.execute_command(f'service remove {enterprise_number} {service_data}') + + # + # Dataset management + # + + def dataset_init_buffer(self, get_active_dataset=False, get_pending_dataset=False): + """Initialize operational dataset buffer.""" + if get_active_dataset and get_pending_dataset: + raise InvalidArgumentsError("Can not specify both `get_active_dataset` and `get_pending_dataset`.") + + if get_active_dataset: + self.execute_command(f'dataset init active') + elif get_pending_dataset: + self.execute_command(f'dataset init pending') + else: + self.execute_command(f'dataset init new') + + def dataset_commit_buffer(self, dataset: str): + if dataset in ('active', 'pending'): + cmd = f'dataset commit {dataset}' + else: + raise InvalidArgumentsError(f'Unkonwn dataset: {dataset}') + + self.execute_command(cmd) + + def dataset_clear_buffer(self): + """Reset operational dataset buffer.""" + self.execute_command('dataset clear') + + def get_dataset(self, dataset: str = 'buffer'): + if dataset in ('active', 'pending'): + cmd = f'dataset {dataset}' + elif dataset == 'buffer': + cmd = 'dataset' + else: + raise InvalidArgumentsError(f'Unkonwn dataset: {dataset}') + + output = self.execute_command(cmd) + return self.__parse_dataset(output) + + def __parse_dataset(self, output: List[str]) -> Dict[str, Any]: + # Example output: + # + # Active Timestamp: 1 + # Channel: 22 + # Channel Mask: 0x07fff800 + # Ext PAN ID: 5c93ae980ff22d35 + # Mesh Local Prefix: fdc7:55fe:6363:bd01::/64 + # Master Key: d1a8348d59fb1fac1d6c4f95007d487a + # Network Name: OpenThread-7caa + # PAN ID: 0x7caa + # PSKc: 167d89fd169e439ca0b8266de248090f + # Security Policy: 0, onrcb + + dataset = {} + + for line in output: + line = line.split(': ') + key, val = line[0], ': '.join(line[1:]) + + if key == 'Active Timestamp': + dataset['active_timestamp'] = int(val) + elif key == 'Channel': + dataset['channel'] = int(val) + elif key == 'Channel Mask': + dataset['channel_mask'] = int(val, 16) + elif key == 'Ext PAN ID': + dataset['extpanid'] = val + elif key == 'Mesh Local Prefix': + dataset['mesh_local_prefix'] = val + elif key == 'Master Key': + dataset['masterkey'] = val + elif key == 'Network Name': + dataset['network_name'] = val + elif key == 'PAN ID': + dataset['panid'] = int(val, 16) + elif key == 'PSKc': + dataset['pskc'] = val + elif key == 'Security Policy': + rotation_time, flags = val.split(', ') + rotation_time = int(rotation_time) + dataset['security_policy'] = SecurityPolicy(rotation_time, flags) + else: + raise UnexpectedCommandOutput(output) + + return dataset + + def get_dataset_bytes(self, dataset: str) -> bytes: + if dataset in ('active', 'pending'): + cmd = f'dataset {dataset} -x' + else: + raise InvalidArgumentsError(f'Unkonwn dataset: {dataset}') + + hexstr = self.__parse_str(self.execute_command(cmd)) + return self.__hex_to_bytes(hexstr) + + def dataset_set_buffer(self, + active_timestamp: int = None, + channel: int = None, + channel_mask: int = None, + extpanid: str = None, + mesh_local_prefix: str = None, + master_key: str = None, + network_name: str = None, + panid: int = None, + pskc: str = None, + security_policy: tuple = None, + pending_timestamp: int = None): + if active_timestamp is not None: + self.execute_command(f'dataset activetimestamp {active_timestamp}') + + if channel is not None: + self.execute_command(f'dataset channel {channel}') + + if channel_mask is not None: + self.execute_command(f'dataset channelmask {channel_mask}') + + if extpanid is not None: + self.execute_command(f'dataset extpanid {extpanid}') + + if mesh_local_prefix is not None: + self.execute_command(f'dataset meshlocalprefix {mesh_local_prefix}') + + if master_key is not None: + self.execute_command(f'dataset masterkey {master_key}') + + if network_name is not None: + self.execute_command(f'dataset networkname {self.__escape_escapable(network_name)}') + + if panid is not None: + self.execute_command(f'dataset panid {panid}') + + if pskc is not None: + self.execute_command(f'dataset pskc {pskc}') + + if security_policy is not None: + rotation_time, flags = security_policy + self.execute_command(f'dataset securitypolicy {rotation_time} {flags}') + + if pending_timestamp is not None: + self.execute_command(f'dataset pendingtimestamp {pending_timestamp}') + + # TODO: dataset mgmtgetcommand + # TODO: dataset mgmtsetcommand + # TODO: dataset set + + # + # Allowlist management + # + + def enable_allowlist(self): + self.execute_command(f'macfilter addr {self.__detect_allowlist_cmd()}') + + def disable_allowlist(self): + self.execute_command('macfilter addr disable') + + def add_allowlist(self, addr: str, rssi: int = None): + cmd = f'macfilter addr add {addr}' + + if rssi is not None: + cmd += f' {rssi}' + + self.execute_command(cmd) + + def remove_allowlist(self, addr: str): + self.execute_command(f'macfilter addr remove {addr}') + + def clear_allowlist(self): + self.execute_command('macfilter addr clear') + + def set_allowlist(self, allowlist: Collection[Union[str, Tuple[str, int]]]): + self.clear_allowlist() + + if allowlist is None: + self.disable_allowlist() + else: + self.enable_allowlist() + for item in allowlist: + if isinstance(item, str): + self.add_allowlist(item) + else: + addr, rssi = item[0], item[1] + self.add_allowlist(addr, rssi) + + # TODO: denylist + # TODO: macfilter rss + # TODO: macfilter rss add + # TODO: macfilter rss add-lqi + # TODO: macfilter rss remove + # TODO: macfilter rss clear + + def __detect_allowlist_cmd(self): + if self.api_version >= 28: + return 'allowlist' + else: + return '\x77\x68\x69\x74\x65\x6c\x69\x73\x74' + + # + # Unicast Addresses management + # + def add_ipaddr(self, ip: Union[str, ipaddress.IPv6Address]): + """Add an IPv6 address to the Thread interface.""" + self.execute_command(f'ipaddr add {ip}') + + def del_ipaddr(self, ip: Union[str, ipaddress.IPv6Address]): + """Delete an IPv6 address from the Thread interface.""" + self.execute_command(f'ipaddr del {ip}') + + def get_ipaddrs(self) -> Tuple[Ip6Addr]: + """Get all IPv6 addresses assigned to the Thread interface.""" + return tuple(map(Ip6Addr, self.execute_command('ipaddr'))) + + def has_ipaddr(self, ip: Union[str, ipaddress.IPv6Address]): + """Check if a Ip6 address was added to the Thread interface.""" + return ip in self.get_ipaddrs() + + def get_ipaddr_mleid(self) -> Ip6Addr: + """Get Thread Mesh Local EID address.""" + return self.__parse_ip6addr(self.execute_command('ipaddr mleid')) + + def get_ipaddr_linklocal(self) -> Ip6Addr: + """Get Thread link-local IPv6 address.""" + return self.__parse_ip6addr(self.execute_command('ipaddr linklocal')) + + def get_ipaddr_rloc(self) -> Ip6Addr: + """Get Thread Routing Locator (RLOC) address.""" + return self.__parse_ip6addr(self.execute_command('ipaddr rloc')) + + # + # Multicast Addresses management + # + + def add_ipmaddr(self, ip: Union[str, ipaddress.IPv6Address]): + """Subscribe the Thread interface to the IPv6 multicast address.""" + self.execute_command(f'ipmaddr add {ip}') + + def del_ipmaddr(self, ip: Union[str, ipaddress.IPv6Address]): + """Unsubscribe the Thread interface to the IPv6 multicast address.""" + self.execute_command(f'ipmaddr del {ip}') + + def get_ipmaddrs(self) -> Tuple[Ip6Addr]: + """Get all IPv6 multicast addresses subscribed to the Thread interface.""" + return tuple(map(Ip6Addr, self.execute_command('ipmaddr'))) + + def has_ipmaddr(self, ip: Union[str, ipaddress.IPv6Address]): + """Check if a Ip6 multicast address was subscribed by the Thread interface.""" + return ip in self.get_ipmaddrs() + + def get_ipmaddr_promiscuous(self) -> bool: + """Get multicast promiscuous mode.""" + return self.__parse_Enabled_or_Disabled(self.execute_command("ipmaddr promiscuous")) + + def enable_ipmaddr_promiscuous(self): + """Enable multicast promiscuous mode.""" + self.execute_command('ipmaddr promiscuous enable') + + def disable_ipmaddr_promiscuous(self): + """Disable multicast promiscuous mode.""" + self.execute_command('ipmaddr promiscuous disable') + + # + # Backbone Router Utilities + # + + # TODO: bbr mgmt ... + + def enable_backbone_router(self): + """Enable Backbone Router Service for Thread 1.2 FTD. + + SRV_DATA.ntf would be triggerred for attached device if there is no Backbone Router Service in Thread Network Data. + """ + self.execute_command('bbr enable') + + def disable_backbone_router(self): + """Disable Backbone Router Service for Thread 1.2 FTD. + + SRV_DATA.ntf would be triggerred if Backbone Router is Primary state. + """ + self.execute_command('bbr disable') + + def get_backbone_router_state(self) -> str: + """Get local Backbone state (Disabled or Primary or Secondary) for Thread 1.2 FTD.""" + self.__parse_str(self.execute_command('bbr state')) + + def get_primary_backbone_router_info(self) -> Optional[dict]: + """Show current Primary Backbone Router information for Thread 1.2 device.""" + output = self.execute_command('bbr') + + if len(output) < 1: + raise UnexpectedCommandOutput(output) + + line = output[0] + if line == 'BBR Primary: None': + return None + + if line != 'BBR Primary:': + raise UnexpectedCommandOutput(output) + + # Example output: + # BBR Primary: + # server16: 0xE400 + # seqno: 10 + # delay: 120 secs + # timeout: 300 secs + + dataset = {} + + for line in output[1:]: + key, val = line.split(':') + key, val = key.strip(), val.strip() + if key == 'server16': + dataset[key] = int(val, 16) + elif key == 'seqno': + dataset[key] = int(val) + elif key == 'delay': + if not val.endswith(' secs'): + raise UnexpectedCommandOutput(output) + dataset[key] = int(val.split()[0]) + elif key == 'timeout': + if not val.endswith(' secs'): + raise UnexpectedCommandOutput(output) + dataset[key] = int(val.split()[0]) + else: + raise UnexpectedCommandOutput(output) + + return dataset + + def register_backbone_router_dataset(self): + """Register Backbone Router Service for Thread 1.2 FTD. + + SRV_DATA.ntf would be triggerred for attached device. + """ + self.execute_command('bbr register') + + def get_backbone_router_config(self) -> dict: + """Show local Backbone Router configuration for Thread 1.2 FTD.""" + output = self.execute_command('bbr config') + # Example output: + # seqno: 10 + # delay: 120 secs + # timeout: 300 secs + + config = {} + + for line in output: + key, val = line.split(':') + key, val = key.strip(), val.strip() + if key == 'seqno': + config[key] = int(val) + elif key in ('delay', 'timeout'): + if not line.endswith(' secs'): + raise UnexpectedCommandOutput(output) + config[key] = int(val.split()[0]) + else: + raise UnexpectedCommandOutput(output) + + return config + + def set_backbone_router_config(self, seqno: int = None, delay: int = None, timeout: int = None): + """Configure local Backbone Router configuration for Thread 1.2 FTD. + + Call register_backbone_router_dataset() to explicitly register Backbone Router service to Leader for Secondary Backbone Router. + """ + if seqno is None and delay is None and timeout is None: + raise InvalidArgumentsError("Please specify seqno or delay or timeout") + + cmd = 'bbr config' + if seqno is not None: + cmd += f' seqno {seqno}' + + if delay is not None: + cmd += f' delay {delay}' + + if timeout is not None: + cmd += f' timeout {timeout}' + + self.execute_command(cmd) + + def get_backbone_router_jitter(self) -> int: + """Get jitter (in seconds) for Backbone Router registration for Thread 1.2 FTD.""" + return self.__parse_int(self.execute_command('bbr jitter')) + + def set_backbone_router_jitter(self, val: int): + """Set jitter (in seconds) for Backbone Router registration for Thread 1.2 FTD.""" + self.execute_command(f'bbr jitter {val}') + + # + # Thread 1.2 and DUA/MLR utilities + # + + def get_domain_name(self) -> str: + """Get the Thread Domain Name for Thread 1.2 device.""" + return self.__parse_str(self.execute_command('domainname')) + + def set_domain_name(self, name: str): + """Set the Thread Domain Name for Thread 1.2 device.""" + self.execute_command('domainname %s' % self.__escape_escapable(name)) + + # TODO: dua iid + # TODO: dua iid + # TODO: dua iid clear + # TODO: mlr reg ... [timeout] + + # + # Link metrics management + # + # TODO: linkmetrics mgmt forward [ldraX][pqmr] + # TODO: linkmetrics probe + # TODO: linkmetrics query single [pqmr] + # TODO: linkmetrics query forward + # TODO: linkquality + # TODO: linkquality + # + + # + # Logging + # + + def get_log_level(self) -> int: + """Get the log level.""" + return self.__parse_int(self.execute_command('log level')) + + def set_log_level(self, level: int): + """Set the log level.""" + self.execute_command(f'log level {level}') + + # + # Device performance related information + # + + def get_message_buffer_info(self) -> dict: + """Get the current message buffer information.""" + output = self.execute_command('bufferinfo') + + info = {} + + def _parse_val(val): + vals = val.split() + return int(vals[0]) if len(vals) == 1 else tuple(map(int, vals)) + + for line in output: + key, val = line.split(':') + key, val = key.strip(), val.strip() + info[key.replace(' ', '_')] = _parse_val(val) + + return info + + @constant_property + def counter_names(self): + """Get the supported counter names.""" + return tuple(self.execute_command('counters')) + + def get_counter(self, name: str) -> Counter: + """Reset the counter value.""" + output = self.execute_command(f'counters {name}') + + counter = Counter() + for line in output: + k, v = line.strip().split(': ') + counter[k] = int(v) + + return counter + + def reset_counter(self, name: str): + """Reset the counter value.""" + self.execute_command(f'counters {name} reset') + + def get_eidcache(self) -> Dict[Ip6Addr, Rloc16]: + """Get the EID-to-RLOC cache entries.""" + output = self.execute_command('eidcache') + cache = {} + + for line in output: + ip, rloc16 = line.split() + + cache[Ip6Addr(ip)] = Rloc16(rloc16, 16) + + return cache + + # + # UDP utilities + # + + def udp_open(self): + """Opens the example socket.""" + self.execute_command('udp open') + + def udp_close(self): + """Opens the example socket.""" + self.execute_command('udp close') + + def udp_bind(self, ip: str, port: int): + """Assigns a name (i.e. IPv6 address and port) to the example socket. + + :param ip: the IPv6 address or the unspecified IPv6 address (::). + :param port: the UDP port + """ + self.execute_command(f'udp bind {ip} {port}') + + def udp_connect(self, ip: str, port: int): + """Specifies the peer with which the socket is to be associated. + + ip: the peer's IPv6 address. + port: the peer's UDP port. + """ + self.execute_command(f'udp connect {ip} {port}') + + def udp_send(self, ip: str = None, port: int = None, text: str = None, random_bytes: int = None, hex: str = None): + """Send a few bytes over UDP. + + ip: the IPv6 destination address. + port: the UDP destination port. + type: the type of the message: _ -t: text payload in the value, same as without specifying the type. _ -s: autogenerated payload with specified length indicated in the value. + * -x: binary data in hexadecimal representation in the value. + """ + if (ip is None) != (port is None): + raise InvalidArgumentsError("Please specify both `ip` and `port`.") + + if (text is not None) + (random_bytes is not None) + (hex is not None) != 1: + raise InvalidArgumentsError("Please specify `text` or `random_bytes` or `hex`.") + + cmd = 'udp send' + + if ip is not None: + cmd += f' {ip} {port}' + + if text is not None: + cmd += f' -t {text}' + elif random_bytes is not None: + cmd += f' -s {random_bytes}' + elif hex is not None: + self.__validate_hex(hex) + cmd += f' -x {hex}' + + self.execute_command(cmd) + + def udp_get_link_security(self) -> bool: + """Gets whether the link security is enabled or disabled.""" + return self.__parse_Enabled_or_Disabled(self.execute_command('udp linksecurity')) + + def udp_enable_link_security(self): + """Enable link security.""" + self.execute_command('udp linksecurity enable') + + def udp_disable_link_security(self): + """Disable link security.""" + self.execute_command('udp linksecurity disable') + + # + # CoAP CLI (test) utilities + # + def coap_start(self): + """Starts the application coap service.""" + self.execute_command('coap start') + + def coap_stop(self): + """Stops the application coap service.""" + self.execute_command('coap stop') + + def coap_get(self, addr: str, uri_path: str, type: str = "con"): + cmd = f'coap get {addr} {uri_path} {type}' + self.execute_command(cmd) + + def coap_put(self, addr: str, uri_path: str, type: str = "con", payload: str = None): + cmd = f'coap put {addr} {uri_path} {type}' + + if payload is not None: + cmd += f' {payload}' + + self.execute_command(cmd) + + def coap_post(self, addr: str, uri_path: str, type: str = "con", payload: str = None): + cmd = f'coap post {addr} {uri_path} {type}' + + if payload is not None: + cmd += f' {payload}' + + self.execute_command(cmd) + + def coap_delete(self, addr: str, uri_path: str, type: str = "con", payload: str = None): + cmd = f'coap delete {addr} {uri_path} {type}' + + if payload is not None: + cmd += f' {payload}' + + self.execute_command(cmd) + + def coap_get_test_resource_path(self) -> str: + """Gets the URI path for the test resource.""" + return self.__parse_str(self.execute_command('coap resource')) + + def coap_set_test_resource_path(self, path: str): + """Sets the URI path for the test resource.""" + self.execute_command(f'coap resource {path}') + + def coap_test_set_resource_content(self, content: str): + """Sets the content sent by the test resource. If a CoAP client is observing the resource, a notification is sent to that client.""" + self.execute_command(f'coap set {content}') + + # TODO: coap observe
[type] + # TODO: coap cancel + # TODO: coap parameters ["default"| ] + # TODO: CoAP Secure utilities + + # + # Other TODOs + # + # TODO: netstat + # TODO: networkdiagnostic get .. + # TODO: networkdiagnostic reset .. + # TODO: parent + # TODO: pskc [-p] | + # + + # + # Private methods + # + + def __parse_str(self, output: List[str]) -> str: + if len(output) != 1: + raise UnexpectedCommandOutput(output) + + return output[0] + + def __parse_int_list(self, output: List[str]) -> List[int]: + line = self.__parse_str(output) + return list(map(int, line.strip().split())) + + def __parse_ip6addr(self, output: List[str]) -> Ip6Addr: + return Ip6Addr(self.__parse_str(output)) + + def __parse_int(self, output: List[str], base=10) -> int: + if len(output) != 1: + raise UnexpectedCommandOutput(output) + + return int(output[0], base) + + def __parse_master_key(self, output: List[str]) -> str: + masterkey = self.__parse_str(output) + + try: + self.__validate_master_key(masterkey) + except ValueError: + raise UnexpectedCommandOutput(output) + + return masterkey + + def __validate_master_key(self, masterkey: str): + if len(masterkey) != 32: + raise ValueError(masterkey) + + int(masterkey, 16) + + def __parse_hex64b(self, output: List[str]) -> str: + extaddr = self.__parse_str(output) + + try: + self.__validate_hex64b(extaddr) + except ValueError: + raise UnexpectedCommandOutput(output) + + return extaddr + + __parse_extaddr = __parse_hex64b + __parse_extpanid = __parse_hex64b + __parse_eui64 = __parse_hex64b + __parse_joiner_id = __parse_hex64b + + def __validate_hex64b(self, extaddr: str): + if len(extaddr) != 16: + raise ValueError(extaddr) + + self.__validate_hex(extaddr) + + def __validate_hex(self, hexstr: str): + if len(hexstr) % 2 != 0: + raise ValueError(hexstr) + + for i in range(0, len(hexstr), 2): + int(hexstr[i:i + 2], 16) + + __validate_extaddr = __validate_hex64b + __validate_extpanid = __validate_hex64b + + def __parse_Enabled_or_Disabled(self, output: List[str]) -> bool: + return self.__parse_values(output, Enabled=True, Disabled=False) + + def __parse_values(self, output: List[str], **vals) -> Any: + val = self.__parse_str(output) + if val not in vals: + raise UnexpectedCommandOutput(output) + + return vals[val] + + def __validate_hex_or_bytes(self, data: Union[str, bytes]) -> str: + if isinstance(data, bytes): + return ''.join('%02x' % c for c in data) + else: + self.__validate_hex(data) + return data + + def __hex_to_bytes(self, hexstr: str) -> bytes: + self.__validate_hex(hexstr) + return bytes(int(hexstr[i:i + 2], 16) for i in range(0, len(hexstr), 2)) + + def __escape_escapable(self, s: str) -> str: + """Escape CLI escapable characters in the given string. + """ + escapable_chars = '\\ \t\r\n' + for char in escapable_chars: + s = s.replace(char, '\\%s' % char) + return s + + def __wait_reset(self): + # reset would restart the otbr-agent executable. It's risky to send commands after reset too quickly because + # it might cause ot-ctl to quit abnormally. + # So we sleep for a while after reset. + time.sleep(3) + + +def connect_cli_sim(executable: str, nodeid: int, simulator: Optional[Simulator] = None) -> OTCI: + cli_handler = connectors.OtCliSim(executable, nodeid, simulator=simulator) + cmd_handler = OtCliCommandRunner(cli_handler) + return OTCI(cmd_handler) + + +def connect_cli_serial(dev: str, baudrate=115200) -> OTCI: + cli_handler = connectors.OtCliSerial(dev, baudrate) + cmd_handler = OtCliCommandRunner(cli_handler) + return OTCI(cmd_handler) + + +def connect_ncp_sim(executable: str, nodeid: int, simulator: Optional[Simulator] = None) -> OTCI: + ncp_handler = connectors.OtNcpSim(executable, nodeid, simulator=simulator) + cmd_handler = OtCliCommandRunner(ncp_handler, is_spinel_cli=True) + return OTCI(cmd_handler) + + +def connect_otbr_ssh(host: str, port: int = 22, username='pi', password='raspberry'): + cmd_handler = OtbrSshCommandRunner(host, port, username, password) + return OTCI(cmd_handler) + + +def connect_cmd_handler(cmd_handler: OTCommandHandler) -> OTCI: + return OTCI(cmd_handler) diff --git a/tools/otci/otci/types.py b/tools/otci/otci/types.py new file mode 100644 index 000000000..0ed8a8110 --- /dev/null +++ b/tools/otci/otci/types.py @@ -0,0 +1,131 @@ +#!/usr/bin/env python3 +# +# Copyright (c) 2020, 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. +# +import ipaddress +from collections import namedtuple + + +class ChildId(int): + """Represents a Child ID.""" + pass + + +class RouterId(int): + """Represents a Router ID.""" + pass + + +class Rloc16(int): + """Represents a RLOC16.""" + + def __repr__(self): + return hex(self) + + +class PartitionId(int): + """Represents a Thread Network Partition ID.""" + pass + + +class DeviceMode(str): + """Represents a device mode.""" + + def __new__(cls, o: str): + ins = str.__new__(cls, o) + + if ins != '-': + for c in ins: + if c not in 'rdn': + raise ValueError(o) + + # check for empty mode (SED should use "-") + if not ins: + raise ValueError(o) + + # check for duplicate chars + if len(ins) != len(set(ins)): + raise ValueError(o) + + return ins + + +class ThreadState(str): + """Represents a Thread state.""" + _VALID_VALUES = {'disabled', 'detached', 'child', 'router', 'leader'} + + def __new__(cls, o: str): + ins = str.__new__(cls, o) + + if ins not in ThreadState._VALID_VALUES: + raise ValueError(o) + + return ins + + +class Ip6Addr(ipaddress.IPv6Address): + """Represents an IPv6 address.""" + + def __eq__(self, other): + if isinstance(other, str): + other = ipaddress.IPv6Address(other) + + return super().__eq__(other) + + def __repr__(self): + return self.compressed + + def __hash__(self): + return super().__hash__() + + +class Ip6Prefix(ipaddress.IPv6Network): + """Represents an IPv6 prefix.""" + + def __eq__(self, other): + if isinstance(other, str): + other = ipaddress.IPv6Network(other) + + return super().__eq__(other) + + def __repr__(self): + return self.compressed + + def __hash__(self): + return super().__hash__() + + +SecurityPolicy = namedtuple('SecurityPolicy', ['rotation_time', 'flags']) +"""Represents a Security Policy configuration.""" + +if __name__ == '__main__': + assert Ip6Addr('2001:0:0:0:0:0:0:1') == '2001::1' + assert repr(Ip6Addr('2001:0:0:0:0:0:0:1')) == '2001::1' + assert str(Ip6Addr('2001:0:0:0:0:0:0:1')) == '2001::1' + assert Ip6Prefix('2001:0:0:0::/64') == '2001::/64' + assert repr(Ip6Prefix('2001:0:0:0::/64')) == '2001::/64' + assert str(Ip6Prefix('2001:0:0:0::/64')) == '2001::/64' diff --git a/tools/otci/otci/utils.py b/tools/otci/otci/utils.py new file mode 100644 index 000000000..3a52a8271 --- /dev/null +++ b/tools/otci/otci/utils.py @@ -0,0 +1,63 @@ +#!/usr/bin/env python3 +# +# Copyright (c) 2020, 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. +# +import functools +from typing import Union, Collection, Any, Pattern + + +def match_line(line: str, expect_line: Union[str, Pattern, Collection[Any]]) -> bool: + """Checks if a line is expected (matched by one of the given patterns).""" + if isinstance(expect_line, Pattern): + match = expect_line.match(line) + elif isinstance(expect_line, str): + match = (line == expect_line) + else: + match = any(match_line(line, x) for x in expect_line) + + return match + + +def cached(func): + """Decorator cached makes the function to cache its result and return it in duplicate calls.""" + prop_name = '__cached_' + func.__name__ + + @functools.wraps(func) + def _cached_func(self): + try: + return getattr(self, prop_name) + except AttributeError: + val = func(self) + setattr(self, prop_name, val) + return val + + return _cached_func + + +def constant_property(func): + """A constant property is a property that only evaluated once.""" + return property(cached(func)) diff --git a/tools/otci/setup.py b/tools/otci/setup.py new file mode 100644 index 000000000..f3d68685a --- /dev/null +++ b/tools/otci/setup.py @@ -0,0 +1,50 @@ +#!/usr/bin/env python3 +# +# Copyright (c) 2020, 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. +# +import setuptools + +with open("README.md", "r") as fh: + long_description = fh.read() + +setuptools.setup( + name="otci-openthread", + version="0.0.1", + author="The OpenThread Authors", + description="OpenThread Controller Interface", + long_description=long_description, + long_description_content_type="text/markdown", + url="https://github.com/openthread/openthread", + packages=setuptools.find_packages(), + classifiers=[ + "Programming Language :: Python :: 3", + "License :: OSI Approved :: BSD 3-Clause License", + "Operating System :: OS Independent", + ], + python_requires='>=3.6', + install_requires=['pySerial', 'paramiko', 'pyspinel'], +) diff --git a/tools/otci/tests/test_otci.py b/tools/otci/tests/test_otci.py new file mode 100644 index 000000000..ee7d48544 --- /dev/null +++ b/tools/otci/tests/test_otci.py @@ -0,0 +1,515 @@ +#!/usr/bin/env python3 +# +# Copyright (c) 2020, 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. +# +import json +import logging +import os +import subprocess +import unittest + +import otci +from otci.errors import CommandError + +logging.basicConfig(level=logging.INFO) + +TEST_CHANNEL = 22 +TEST_NETWORK_NAME = 'OT CI' +TEST_PANID = 0xeeee +TEST_MASTERKEY = 'ffeeddccbbaa99887766554433221100' + +REAL_DEVICE = int(os.getenv('REAL_DEVICE', '0')) + + +class TestOTCI(unittest.TestCase): + + def testCliRealDevice(self): + if not REAL_DEVICE: + self.skipTest('not for virtual device') + + if os.getenv('OTBR_SSH'): + node = otci.connect_otbr_ssh(os.getenv('OTBR_SSH')) + elif os.getenv('OT_CLI_SERIAL'): + node = otci.connect_cli_serial(os.getenv('OT_CLI_SERIAL')) + else: + self.fail("Please set OT_CLI_SERIAL or OTBR_SSH to test the real device.") + + node.factory_reset() + + self._test_otci_single_node(node) + + def testCliSimRealTime(self): + if REAL_DEVICE: + self.skipTest('not for real device') + + subprocess.check_call('rm -rf tmp/', shell=True) + VIRTUAL_TIME = int(os.getenv('VIRTUAL_TIME', "1")) + + import simulator + + if VIRTUAL_TIME: + sim = simulator.VirtualTime(use_message_factory=False) + else: + sim = None + + if os.getenv('OT_CLI'): + executable = os.getenv('OT_CLI') + connector = otci.connect_cli_sim + elif os.getenv('OT_NCP'): + executable = os.getenv('OT_NCP') + connector = otci.connect_ncp_sim + else: + self.fail("Please set OT_CLI to test virtual device") + + node1 = connector(executable, 1, simulator=sim) + self._test_otci_single_node(node1) + + node1.factory_reset() + + node2 = connector(executable, 2, simulator=sim) + node3 = connector(executable, 3, simulator=sim) + node4 = connector(executable, 4, simulator=sim) + + self._test_otci_example(node1, node2) + + node1.factory_reset() + node2.factory_reset() + + self._test_otci_multi_nodes(node1, node2, node3, node4) + + def _test_otci_single_node(self, leader): + logging.info('leader version: %r', leader.version) + logging.info('leader thread version: %r', leader.thread_version) + logging.info('API version: %r', leader.api_version) + logging.info('log level: %r', leader.get_log_level()) + + leader.enable_promiscuous() + self.assertTrue(leader.get_promiscuous()) + leader.disable_promiscuous() + self.assertFalse(leader.get_promiscuous()) + try: + logging.info("RCP version: %r", leader.get_rcp_version()) + except CommandError: + pass + + self.assertTrue(leader.get_router_eligible()) + leader.disable_router_eligible() + self.assertFalse(leader.get_router_eligible()) + leader.enable_router_eligible() + + self.assertFalse(leader.get_ifconfig_state()) + # ifconfig up + leader.ifconfig_up() + self.assertTrue(leader.get_ifconfig_state()) + + logging.info('leader eui64 = %r', leader.get_eui64()) + logging.info('leader extpanid = %r', leader.get_extpanid()) + logging.info('leader masterkey = %r', leader.get_master_key()) + + extaddr = leader.get_extaddr() + self.assertEqual(16, len(extaddr)) + int(extaddr, 16) + new_extaddr = 'aabbccddeeff0011' + leader.set_extaddr(new_extaddr) + self.assertEqual(new_extaddr, leader.get_extaddr()) + + leader.set_network_name(TEST_NETWORK_NAME) + + leader.set_master_key(TEST_MASTERKEY) + self.assertEqual(TEST_MASTERKEY, leader.get_master_key()) + + leader.set_panid(TEST_PANID) + self.assertEqual(TEST_PANID, leader.get_panid()) + + leader.set_channel(TEST_CHANNEL) + self.assertEqual(TEST_CHANNEL, leader.get_channel()) + + leader.set_network_name(TEST_NETWORK_NAME) + self.assertEqual(TEST_NETWORK_NAME, leader.get_network_name()) + + self.assertEqual('rdn', leader.get_mode()) + leader.set_mode('-') + self.assertEqual('-', leader.get_mode()) + leader.set_mode('rdn') + self.assertEqual('rdn', leader.get_mode()) + + logging.info('leader weight: %d', leader.get_leader_weight()) + leader.set_leader_weight(72) + + logging.info('domain name: %r', leader.get_domain_name()) + leader.set_domain_name("DefaultDomain2") + self.assertEqual("DefaultDomain2", leader.get_domain_name()) + + self.assertEqual(leader.get_preferred_partition_id(), 0) + leader.set_preferred_partition_id(0xabcddead) + self.assertEqual(leader.get_preferred_partition_id(), 0xabcddead) + + leader.thread_start() + leader.wait(5) + self.assertEqual('leader', leader.get_state()) + self.assertEqual(0xabcddead, leader.get_leader_data()['partition_id']) + logging.info('leader key sequence counter = %d', leader.get_key_sequence_counter()) + + rloc16 = leader.get_rloc16() + leader_id = leader.get_router_id() + self.assertEqual(rloc16, leader_id << 10) + + self.assertFalse(leader.get_child_list()) + self.assertEqual({}, leader.get_child_table()) + + leader.enable_allowlist() + leader.add_allowlist(leader.get_extaddr()) + leader.remove_allowlist(leader.get_extaddr()) + leader.set_allowlist([leader.get_extaddr()]) + leader.disable_allowlist() + + leader.add_ipmaddr('ff04::1') + leader.del_ipmaddr('ff04::1') + leader.add_ipmaddr('ff04::2') + logging.info('leader ipmaddrs: %r', leader.get_ipmaddrs()) + self.assertFalse(leader.has_ipmaddr('ff04::1')) + self.assertTrue(leader.has_ipmaddr('ff04::2')) + self.assertTrue(leader.get_ipaddr_rloc()) + self.assertTrue(leader.get_ipaddr_linklocal()) + self.assertTrue(leader.get_ipaddr_mleid()) + + leader.add_ipaddr('2001::1') + leader.del_ipaddr('2001::1') + leader.add_ipaddr('2001::2') + logging.info('leader ipaddrs: %r', leader.get_ipaddrs()) + self.assertFalse(leader.has_ipaddr('2001::1')) + self.assertTrue(leader.has_ipaddr('2001::2')) + + logging.info('leader bbr state: %r', leader.get_backbone_router_state()) + bbr_config = leader.get_backbone_router_config() + logging.info('leader bbr config: %r', bbr_config) + logging.info('leader PBBR: %r', leader.get_primary_backbone_router_info()) + + leader.set_backbone_router_config(seqno=bbr_config['seqno'] + 1, delay=10, timeout=301) + self.assertEqual({ + 'seqno': bbr_config['seqno'] + 1, + 'delay': 10, + 'timeout': 301 + }, leader.get_backbone_router_config()) + + leader.enable_backbone_router() + leader.wait(3) + + logging.info('leader bbr state: %r', leader.get_backbone_router_state()) + logging.info('leader bbr config: %r', leader.get_backbone_router_config()) + logging.info('leader PBBR: %r', leader.get_primary_backbone_router_info()) + + logging.info('leader bufferinfo: %r', leader.get_message_buffer_info()) + + logging.info('child ipaddrs: %r', leader.get_child_ipaddrs()) + logging.info('child ipmax: %r', leader.get_child_ip_max()) + leader.set_child_ip_max(2) + self.assertEqual(2, leader.get_child_ip_max()) + logging.info('childmax: %r', leader.get_max_children()) + + logging.info('counter names: %r', leader.counter_names) + for counter_name in leader.counter_names: + logging.info('counter %s: %r', counter_name, leader.get_counter(counter_name)) + leader.reset_counter(counter_name) + self.assertTrue(all(x == 0 for x in leader.get_counter(counter_name).values())) + + logging.info("CSL config: %r", leader.get_csl_config()) + leader.config_csl(channel=13, period=100, timeout=200) + logging.info("CSL config: %r", leader.get_csl_config()) + + logging.info("EID-to-RLOC cache: %r", leader.get_eidcache()) + + logging.info("ipmaddr promiscuous: %r", leader.get_ipmaddr_promiscuous()) + leader.enable_ipmaddr_promiscuous() + self.assertTrue(leader.get_ipmaddr_promiscuous()) + leader.disable_ipmaddr_promiscuous() + self.assertFalse(leader.get_ipmaddr_promiscuous()) + + logging.info("leader data: %r", leader.get_leader_data()) + logging.info("leader neighbor list: %r", leader.get_neighbor_list()) + logging.info("leader neighbor table: %r", leader.get_neighbor_table()) + logging.info("Leader external routes: %r", leader.get_routes()) + leader.add_route('2002::/64') + leader.register_network_data() + logging.info("Leader external routes: %r", leader.get_routes()) + + self.assertEqual(1, len(leader.get_router_list())) + self.assertEqual(1, len(leader.get_router_table())) + logging.info("Leader router table: %r", leader.get_router_table()) + + logging.info('scan: %r', leader.scan()) + logging.info('scan energy: %r', leader.scan_energy()) + + leader.add_service(44970, '112233', 'aabbcc') + leader.register_network_data() + leader.add_service(44971, b'\x11\x22\x33', b'\xaa\xbb\xcc\xdd') + + leader.add_prefix("2001::/64") + + logging.info("network data: %r", leader.get_network_data()) + logging.info("network data raw: %r", leader.get_network_data_bytes()) + + logging.info('txpower %r', leader.get_txpower()) + leader.set_txpower(-10) + self.assertEqual(-10, leader.get_txpower()) + + self.assertTrue(leader.is_singleton()) + + leader.coap_start() + leader.coap_set_test_resource_path('test') + leader.coap_test_set_resource_content('helloworld') + leader.coap_get(leader.get_ipaddr_rloc(), 'test') + leader.coap_put(leader.get_ipaddr_rloc(), 'test', 'con', 'xxx') + leader.coap_post(leader.get_ipaddr_rloc(), 'test', 'con', 'xxx') + leader.coap_delete(leader.get_ipaddr_rloc(), 'test', 'con', 'xxx') + leader.wait(1) + leader.coap_stop() + + leader.udp_open() + leader.udp_bind("::", 1234) + leader.udp_send(leader.get_ipaddr_rloc(), 1234, text='hello') + leader.udp_send(leader.get_ipaddr_rloc(), 1234, random_bytes=3) + leader.udp_send(leader.get_ipaddr_rloc(), 1234, hex='112233') + leader.wait(1) + leader.udp_close() + + logging.info('dataset: %r', leader.get_dataset()) + logging.info('dataset active: %r', leader.get_dataset('active')) + + leader.dataset_init_buffer() + leader.dataset_commit_buffer('pending') + leader.dataset_init_buffer(get_active_dataset=True) + leader.dataset_init_buffer(get_pending_dataset=True) + + logging.info('dataset: %r', leader.get_dataset()) + logging.info('dataset active: %r', leader.get_dataset('active')) + logging.info('dataset pending: %r', leader.get_dataset('pending')) + + logging.info('dataset active -x: %r', leader.get_dataset_bytes('active')) + logging.info('dataset pending -x: %r', leader.get_dataset_bytes('pending')) + + def _test_otci_example(self, node1, node2): + node1.dataset_init_buffer() + node1.dataset_set_buffer(network_name='test', + master_key='00112233445566778899aabbccddeeff', + panid=0xface, + channel=11) + node1.dataset_commit_buffer('active') + + node1.ifconfig_up() + node1.thread_start() + node1.wait(5) + assert node1.get_state() == "leader" + + node1.commissioner_start() + node1.wait(3) + + node1.commissioner_add_joiner("TEST123", eui64='*') + + node2.ifconfig_up() + node2.set_router_selection_jitter(1) + + node2.joiner_start("TEST123") + node2.wait(10, expect_line="Join success") + node2.thread_start() + node2.wait(5) + assert node2.get_state() == "router" + + def _test_otci_multi_nodes(self, leader, commissioner, child1, child2): + self.assertFalse(leader.get_ifconfig_state()) + + # ifconfig up + leader.ifconfig_up() + self.assertTrue(leader.get_ifconfig_state()) + + logging.info('leader eui64 = %r', leader.get_eui64()) + logging.info('leader extpanid = %r', leader.get_extpanid()) + logging.info('leader masterkey = %r', leader.get_master_key()) + + extaddr = leader.get_extaddr() + self.assertEqual(16, len(extaddr)) + int(extaddr, 16) + new_extaddr = 'aabbccddeeff0011' + leader.set_extaddr(new_extaddr) + self.assertEqual(new_extaddr, leader.get_extaddr()) + + leader.set_network_name(TEST_NETWORK_NAME) + + leader.set_master_key(TEST_MASTERKEY) + self.assertEqual(TEST_MASTERKEY, leader.get_master_key()) + + leader.set_panid(TEST_PANID) + self.assertEqual(TEST_PANID, leader.get_panid()) + + leader.set_channel(TEST_CHANNEL) + self.assertEqual(TEST_CHANNEL, leader.get_channel()) + + leader.set_network_name(TEST_NETWORK_NAME) + self.assertEqual(TEST_NETWORK_NAME, leader.get_network_name()) + + self.assertEqual('rdn', leader.get_mode()) + + leader.thread_start() + leader.wait(5) + self.assertEqual('leader', leader.get_state()) + logging.info('leader key sequence counter = %d', leader.get_key_sequence_counter()) + + rloc16 = leader.get_rloc16() + leader_id = leader.get_router_id() + self.assertEqual(rloc16, leader_id << 10) + + commissioner.ifconfig_up() + commissioner.set_channel(TEST_CHANNEL) + commissioner.set_panid(TEST_PANID) + commissioner.set_network_name(TEST_NETWORK_NAME) + commissioner.set_router_selection_jitter(1) + commissioner.set_master_key(TEST_MASTERKEY) + commissioner.thread_start() + + commissioner.wait(5) + + self.assertEqual('router', commissioner.get_state()) + + for dst_ip in leader.get_ipaddrs(): + commissioner.ping(dst_ip, size=10, count=1, interval=2, hoplimit=3) + + self.assertEqual('disabled', commissioner.get_commissioiner_state()) + commissioner.commissioner_start() + commissioner.wait(5) + self.assertEqual('active', commissioner.get_commissioiner_state()) + + logging.info('commissioner.get_network_id_timeout() = %d', commissioner.get_network_id_timeout()) + commissioner.set_network_id_timeout(60) + self.assertEqual(60, commissioner.get_network_id_timeout()) + + commissioner.commissioner_add_joiner('TEST123', eui64='*') + commissioner.wait(3) + + child1.ifconfig_up() + + logging.info("child1 scan: %r", child1.scan()) + logging.info("child1 scan energy: %r", child1.scan_energy()) + + child1.set_mode('rn') + child1.set_router_selection_jitter(1) + + child1.joiner_start('TEST123') + logging.info('joiner id = %r', child1.get_joiner_id()) + child1.wait(10, expect_line="Join success") + + child1.enable_allowlist() + child1.disable_allowlist() + child1.add_allowlist(commissioner.get_extaddr()) + child1.remove_allowlist(commissioner.get_extaddr()) + child1.set_allowlist([commissioner.get_extaddr()]) + + child1.thread_start() + child1.wait(3) + self.assertEqual('child', child1.get_state()) + + child1.thread_stop() + + child1.set_mode('n') + child1.set_poll_period(1000) + self.assertEqual(1000, child1.get_poll_period()) + + child1.thread_start() + child1.wait(3) + self.assertEqual('child', child1.get_state()) + + child2.ifconfig_up() + child2.set_mode('rn') + child2.set_router_selection_jitter(1) + + child2.joiner_start('TEST123') + logging.info('joiner id = %r', child2.get_joiner_id()) + child2.wait(10, expect_line="Join success") + + child2.enable_allowlist() + child2.disable_allowlist() + child2.add_allowlist(commissioner.get_extaddr()) + child2.remove_allowlist(commissioner.get_extaddr()) + child2.set_allowlist([commissioner.get_extaddr()]) + + child2.thread_start() + child2.wait(3) + self.assertEqual('child', child2.get_state()) + + child_table = commissioner.get_child_table() + logging.info('commissioiner child table: \n%s\n', json.dumps(child_table, indent=True)) + child_list = commissioner.get_child_list() + logging.info('commissioiner child list: %r', child_list) + for child_id in child_list: + logging.info('child %s info: %r', child_id, commissioner.get_child_info(child_id)) + + logging.info('child1 info: %r', commissioner.get_child_info(child1.get_rloc16())) + logging.info('child2 info: %r', commissioner.get_child_info(child2.get_rloc16())) + + self.assertEqual(set(commissioner.get_child_list()), set(commissioner.get_child_table().keys())) + + child1.add_ipmaddr('ff04::1') + child1.del_ipmaddr('ff04::1') + child1.add_ipmaddr('ff04::2') + logging.info('child1 ipmaddrs: %r', child1.get_ipmaddrs()) + self.assertFalse(child1.has_ipmaddr('ff04::1')) + self.assertTrue(child1.has_ipmaddr('ff04::2')) + + child1.add_ipaddr('2001::1') + child1.del_ipaddr('2001::1') + child1.add_ipaddr('2001::2') + logging.info('child1 ipaddrs: %r', child1.get_ipaddrs()) + self.assertFalse(child1.has_ipaddr('2001::1')) + self.assertTrue(child1.has_ipaddr('2001::2')) + + logging.info('child ipaddrs: %r', commissioner.get_child_ipaddrs()) + + logging.info("EID-to-RLOC cache: %r", leader.get_eidcache()) + + logging.info("leader neighbor list: %r", leader.get_neighbor_list()) + logging.info("leader neighbor table: %r", leader.get_neighbor_table()) + logging.info("prefixes: %r", commissioner.get_prefixes()) + commissioner.add_prefix('2001::/64') + commissioner.register_network_data() + commissioner.wait(1) + logging.info("prefixes: %r", commissioner.get_prefixes()) + + self.assertEqual(2, len(leader.get_router_list())) + self.assertEqual(2, len(leader.get_router_table())) + + self.assertFalse(leader.is_singleton()) + + # Shutdown + leader.thread_stop() + logging.info("node state: %s", leader.get_state()) + leader.ifconfig_down() + self.assertFalse(leader.get_ifconfig_state()) + + leader.close() + + +if __name__ == '__main__': + unittest.main()