espressif/openthread
1
0
Fork 0
mirror of https://github.com/espressif/openthread.git synced 2026-08-18 16:39:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

[nexus] add test 1.1.8.2.1 for on-mesh commissioner joining (#12645)

Browse Source 
This commit adds Nexus test 1.1.8.2.1, "On Mesh Commissioner Joining
with JR, any commissioner, single (correct)". This test verifies that
the Joiner Router (DUT) correctly relays DTLS traffic between a Joiner
and an on-mesh Commissioner via RLY_RX.ntf and RLY_TX.ntf messages. It
also verifies that the JOIN_ENT.ntf message is encrypted with the KEK.

Changes:
- Added tests/nexus/test_1_1_8_2_1.cpp to implement the test logic,
  including DTLS key exporting for traffic decryption.
- Added tests/nexus/verify_1_1_8_2_1.py to verify the captured traffic
  using pktverify, ensuring correct relaying and encryption.
- Updated tests/nexus/verify_utils.py to support parsing Joiner-related
  CoAP TLVs (DTLS Encap, UDP Port, IID, Locator, KEK) and to handle
  16-bit TLV lengths.
- Integrated the test into the build system and test runner via
  tests/nexus/CMakeLists.txt and tests/nexus/run_nexus_tests.sh.
This commit is contained in:
Jonathan Hui
2026-03-08 15:39:28 -05:00
committed by GitHub
parent 1b43fe6694
commit 86bc9435ba
5 changed files with 851 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/nexus/CMakeLists.txt
+1
View File
@@ -199,6 +199,7 @@ ot_nexus_test(1_1_7_1_8 "cert;nexus")
ot_nexus_test(1_1_8_1_1 "cert;nexus")
ot_nexus_test(1_1_8_1_2 "cert;nexus")
ot_nexus_test(1_1_8_1_6 "cert;nexus")
ot_nexus_test(1_1_8_2_1 "cert;nexus")
ot_nexus_test(1_1_9_2_1 "cert;nexus")
ot_nexus_test(1_1_9_2_2 "cert;nexus")
ot_nexus_test(1_1_9_2_3 "cert;nexus")
tests/nexus/run_nexus_tests.sh
+1
View File
@@ -134,6 +134,7 @@ DEFAULT_TESTS=(
"1_1_8_1_1"
"1_1_8_1_2"
"1_1_8_1_6"
"1_1_8_2_1"
"1_1_9_2_1"
"1_1_9_2_2"
"1_1_9_2_3"
tests/nexus/test_1_1_8_2_1.cpp
+394
View File
@@ -0,0 +1,394 @@
/*
* Copyright (c) 2026, The OpenThread Authors.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are met:
* 1. Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* 2. Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
* 3. Neither the name of the copyright holder nor the
* names of its contributors may be used to endorse or promote products
* derived from this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
* POSSIBILITY OF SUCH DAMAGE.
*/
#include <stdio.h>
#include "platform/nexus_core.hpp"
#include "platform/nexus_node.hpp"
#include "meshcop/commissioner.hpp"
#include "meshcop/joiner.hpp"
#include "meshcop/meshcop_tlvs.hpp"
#include "thread/tmf.hpp"
namespace ot {
namespace Nexus {
/**
* Time to advance for a node to form a network and become leader, in milliseconds.
*/
static constexpr uint32_t kFormNetworkTime = 13 * Time::kOneSecondInMsec;
/**
* Time to advance for a node to join as a child and upgrade to a router, in milliseconds.
*/
static constexpr uint32_t kAttachToRouterTime = 200 * Time::kOneSecondInMsec;
/**
* Time to advance for the joining process to complete, in milliseconds.
*/
static constexpr uint32_t kJoiningProcessTime = 30 * Time::kOneSecondInMsec;
/**
* Timeout for Joiner entry in Commissioner, in seconds.
*/
static constexpr uint32_t kJoinerTimeout = 120;
/**
* PSKd used for joining.
*/
static const char kPskd[] = "J01NME";
/**
* The channel used for the test.
*/
static constexpr uint8_t kTestChannel = 11;
/**
* The Joiner UDP port.
*/
static constexpr uint16_t kJoinerPort = 49153;
/**
* The default JSON file name.
*/
static const char kDefaultJsonFile[] = "test_1_1_8_2_1.json";
#if OPENTHREAD_CONFIG_MBEDTLS_PROVIDES_SSL_KEY_EXPORT
/**
* The filename for the DTLS key log.
*/
static const char kKeyLogFilename[] = "test_1_1_8_2_1.keys";
/**
* The random buffer length in bytes.
*/
static constexpr uint8_t kRandomBufferLength = MeshCoP::SecureTransport::kSecureTransportRandomBufferSize;
static void HandleKeylog(void *aContext,
mbedtls_ssl_key_export_type aType,
const unsigned char *aMasterSecret,
size_t aMasterSecretLen,
const unsigned char aClientRandom[kRandomBufferLength],
const unsigned char aServerRandom[kRandomBufferLength],
mbedtls_tls_prf_types aTlsPrfType)
{
OT_UNUSED_VARIABLE(aContext);
OT_UNUSED_VARIABLE(aServerRandom);
OT_UNUSED_VARIABLE(aTlsPrfType);
if (aType == MBEDTLS_SSL_KEY_EXPORT_TLS12_MASTER_SECRET)
{
FILE *f = fopen(kKeyLogFilename, "a");
if (f != nullptr)
{
fprintf(f, "CLIENT_RANDOM ");
for (size_t i = 0; i < kRandomBufferLength; i++)
{
fprintf(f, "%02x", aClientRandom[i]);
}
fprintf(f, " ");
for (size_t i = 0; i < aMasterSecretLen; i++)
{
fprintf(f, "%02x", aMasterSecret[i]);
}
fprintf(f, "\n");
fclose(f);
}
}
}
#endif
void Test1_1_8_2_1(const char *aJsonFileName)
{
/**
* 8.2.1 On Mesh Commissioner Joining with JR, any commissioner, single (correct)
*
* 8.2.1.1 Topology
* - Commissioner (Leader)
* - Joiner Router
* - Joiner_1
*
* 8.2.1.2 Purpose & Description
* The purpose of this test case is to verify that the DUT sends and receives relayed DTLS traffic correctly when
* the correct PSKd is used by the Joiner. It also verifies that the DUT (as a Joiner Router) sends JOIN_ENT.ntf
* encrypted with KEK.
*
* Spec Reference | V1.1 Section | V1.3.0 Section
* ---------------|---------------|---------------
* Relay traffic | 8.4.3 / 8.4.4 | 8.4.3 / 8.4.4
*/
Core nexus;
Node &commNode = nexus.CreateNode();
Node &jrNode = nexus.CreateNode();
Node &joinerNode = nexus.CreateNode();
commNode.SetName("LEADER");
jrNode.SetName("ROUTER");
joinerNode.SetName("JOINER");
nexus.AdvanceTime(0);
Instance::SetLogLevel(kLogLevelNote);
Log("---------------------------------------------------------------------------------------");
Log("Step 0: All");
/**
* Step 0: All
* - Description: Form topology excluding the Joiner and including the DUT. The DUT may be joined to the network
* using either Thread Commissioning or OOB (out-of-band) Commissioning.
* - Pass Criteria: N/A.
*/
commNode.Form();
commNode.Get<Mac::Filter>().SetMode(Mac::Filter::kModeRssInOnly);
jrNode.Get<Mac::Filter>().SetMode(Mac::Filter::kModeRssInOnly);
{
MeshCoP::Dataset::Info datasetInfo;
SuccessOrQuit(commNode.Get<MeshCoP::ActiveDatasetManager>().Read(datasetInfo));
datasetInfo.Set<MeshCoP::Dataset::kChannel>(kTestChannel);
commNode.Get<MeshCoP::ActiveDatasetManager>().SaveLocal(datasetInfo);
}
nexus.AdvanceTime(kFormNetworkTime);
VerifyOrQuit(commNode.Get<Mle::Mle>().IsLeader());
jrNode.Join(commNode);
nexus.AdvanceTime(kAttachToRouterTime);
VerifyOrQuit(jrNode.Get<Mle::Mle>().IsRouter());
nexus.AdvanceTime(10 * Time::kOneSecondInMsec);
Log("---------------------------------------------------------------------------------------");
Log("Step 1: Joiner_1");
/**
* Step 1: Joiner_1
* - Description: PSKd configured on Joiner_1 is the same as PSKd configured on the Commissioner.
* - Pass Criteria: N/A.
*/
Log("---------------------------------------------------------------------------------------");
Log("Step 2: Commissioner");
/**
* Step 2: Commissioner
* - Description: Begin wireless sniffer and power on the Commissioner and Joiner Router.
* - Pass Criteria: N/A.
*/
SuccessOrQuit(commNode.Get<MeshCoP::Commissioner>().Start(nullptr, nullptr, nullptr));
nexus.AdvanceTime(Time::kOneSecondInMsec);
VerifyOrQuit(commNode.Get<MeshCoP::Commissioner>().IsActive());
SuccessOrQuit(commNode.Get<MeshCoP::Commissioner>().AddJoinerAny(kPskd, kJoinerTimeout));
commNode.Get<MeshCoP::JoinerRouter>().SetJoinerUdpPort(kJoinerPort);
jrNode.Get<MeshCoP::JoinerRouter>().SetJoinerUdpPort(kJoinerPort);
nexus.AdvanceTime(Time::kOneSecondInMsec);
#if OPENTHREAD_CONFIG_MBEDTLS_PROVIDES_SSL_KEY_EXPORT
static_cast<MeshCoP::SecureTransport &>(commNode.Get<Tmf::SecureAgent>())
.SetKeylogCallback(HandleKeylog, &commNode.Get<Tmf::SecureAgent>());
static_cast<MeshCoP::SecureTransport &>(joinerNode.Get<Tmf::SecureAgent>())
.SetKeylogCallback(HandleKeylog, &joinerNode.Get<Tmf::SecureAgent>());
#endif
Log("---------------------------------------------------------------------------------------");
Log("Step 3: Joiner_1");
/**
* Step 3: Joiner_1
* - Description: Initiate the DTLS-Handshake by sending a DTLS-ClientHello handshake record to the Commissioner.
* - Pass Criteria: Verify that the following details occur in the exchange between the Joiner, the Joiner_Router,
* and the Commissioner:
* - 1. UDP port (Specified by Commissioner in Discovery Response) is used as destination port for UDP datagrams
* from Joiner_1 to Commissioner.
* - 2. Joiner_1 sends an initial DTLS-ClientHello handshake record to the Joiner_Router.
* - 3. The Joiner_Router receives the initial DTLS-ClientHello handshake record and sends a RLY_RX.ntf message
* to the Commissioner containing:
* - CoAP URI-Path: NON POST coap://<C>/c/rx
* - CoAP Payload:
* - Joiner_1 DTLS Encapsulation TLV: Type = 17 (0x11), Length = Variable (8/16 bit size), Value = initial
* DTLS-ClientHello handshake record received from Joiner_1.
* - Joiner_1 UDP Port TLV: Type = 18 (0x12), Length = 2, Value = UDP port of Joiner_1.
* - Joiner_1 IID TLV: Type = 19 (0x13), Length = 8, Value = The IID based on the EUI-64 of Joiner_1.
* - Joiner_Router Locator TLV: Type = 20 (0x14), Length = 2, Value = RLOC of Joiner_Router.
* - 4. The Commissioner receives the RLY_RX.ntf message and sends a RLY_TX.ntf message to the Joiner_Router
* containing:
* - CoAP URI-Path: NON POST coap://<JR>/c/tx
* - CoAP Payload:
* - Joiner_1 DTLS Encapsulation TLV: Type = 17 (0x011), Length = Variable (8 bit size), Value =
* DTLS-HelloVerifyRequest handshake record destined for Joiner_1.
* - Joiner_1 UDP Port TLV: Type = 18 (0x12), Length = 2, Value = UDP port of Joiner_1.
* - Joiner_1 IID TLV: Type = 19 (0x13), Length = 8, Value = The IID based on the EUI-64 of Joiner_1.
* - Joiner_Router Locator TLV: Type = 20 (0x14), Length = 2, Value = RLOC of Joiner_Router.
* - 5. The Joiner_Router receives the RLY_TX.ntf message and sends the encapsulated DTLS-HelloVerifyRequest
* handshake record in one UDP datagram to Joiner_1.
* - 6. Joiner_1 receives the DTLS-HelloVerifyRequest handshake record and sends a subsequent DTLS-ClientHello
* handshake record in one UDP datagram to the Commissioner.
* - Verify that both DTLS-HelloVerifyRequest and subsequent DTLS-ClientHello contain the same cookie.
* - 7. The Joiner_Router receives the subsequent DTLS-ClientHello handshake record and sends a RLY_RX.ntf
* message to the Commissioner containing:
* - CoAP URI-Path: NON POST coap://<C>/c/rx
* - CoAP Payload:
* - Joiner_1 DTLS Encapsulation TLV: Type = 17 (0x11), Length = Variable (8/16 bit size), Value = subsequent
* DTLS-ClientHello handshake record received from Joiner_1.
* - Joiner_1 UDP Port TLV: Type = 18 (0x12), Length = 2, Value = UDP port of Joiner_1.
* - Joiner_1 IID TLV: Type = 19 (0x13), Length = 8, Value = The IID based on the EUI-64 of Joiner_1.
* - Joiner_Router Locator TLV: Type = 20 (0x14), Length = 2, Value = RLOC of Joiner_Router.
* - 8. The Commissioner receives the RLY_RX.ntf message and sends a RLY_TX.ntf message to the Joiner_Router
* containing:
* - CoAP URI-Path: NON POST coap://<JR>/c/tx
* - CoAP Payload:
* - Joiner_1 DTLS Encapsulation TLV: Type = 17 (0x11), Length = Variable (8/16 bit size), Value =
* DTLS-ServerHello, DTLS-ServerKeyExchange and DTLS-ServerHelloDone handshake records, in that order,
* destined for Joiner_1.
* - Joiner_1 UDP Port TLV: Type = 18 (0x12), Length = 2, Value = UDP port of Joiner_1.
* - Joiner_1 IID TLV: Type = 19 (0x13), Length = 8, Value = The IID based on the EUI-64 of Joiner_1.
* - Joiner_Router Locator TLV: Type = 20 (0x14), Length = 2, Value = RLOC of Joiner_Router.
* - 9. Joiner_Router receives the RLY_TX.ntf message and sends the encapsulated DTLS-ServerHello,
* DTLS-ServerKeyExchange and DTLS-ServerHelloDone handshake records in one UDP datagram to Joiner_1.
* - 10. Joiner_1 receives the DTLS-ServerHello, DTLS-ServerKeyExchange and DTLS-ServerHelloDone handshake
* records and sends a DTLS-ClientKeyExchange handshake record, a DTLS-ChangeCipherSpec record and an
* encrypted DTLS-Finished handshake record, in that order, to Joiner_Router.
* - 11. Joiner_Router receives the DTLS-ClientKeyExchange handshake record, the DTLS-ChangeCipherSpec record and
* the encrypted DTLS-Finished handshake record; it then sends a RLY_RX.ntf message to the Commissioner
* containing:
* - CoAP URI-Path: NON POST coap://<C>/c/rx
* - CoAP Payload:
* - Joiner_1 DTLS Encapsulation TLV: Type = 17 (0x11), Length = Variable (8/16 bit size), Value =
* DTLS-ClientKeyExchange handshake record, DTLS-ChangeCipherSpec record and encrypted DTLS-Finished
* handshake record received from Joiner_1.
* - Joiner_1 UDP Port TLV: Type = 18 (0x12), Length = 2, Value = UDP port of Joiner_1.
* - Joiner_1 IID TLV: Type = 19 (0x13), Length = 8, Value = The IID based on the EUI-64 of Joiner_1.
* - Joiner_Router Locator TLV: Type = 20 (0x14), Length = 2, Value = RLOC of Joiner_Router.
* - 12. The Commissioner receives the RLY_RX.ntf message and sends a RLY_TX.ntf message to the Joiner_Router
* containing:
* - CoAP URI-Path: NON POST coap://<JR>/c/tx
* - CoAP Payload:
* - Joiner_1 DTLS Encapsulation TLV: Type = 17 (0x11), Length = Variable (8/16 bit size), Value =
* DTLS-ChangeCipherSpec record and encrypted DTLS-Finished handshake record, in that order, destined
* for Joiner_1.
* - Joiner_1 UDP Port TLV: Type = 18 (0x12), Length = 2, Value = UDP port of Joiner_1.
* - Joiner_1 IID TLV: Type = 19 (0x13), Length = 8, Value = The IID based on the EUI-64 of Joiner_1.
* - Joiner_Router Locator TLV: Type = 20 (0x14), Length = 2, Value = RLOC of Joiner_Router.
* - 13. The Joiner_Router receives the RLY_TX.ntf message and sends the encapsulated DTLS-ChangeCipherSpec
* record and encrypted DTLS-Finished handshake record in one UDP datagram to Joiner_1.
* - 14. Joiner_1 receives the DTLS-ChangeCipherSpec record and the encrypted DTLS-Finished handshake record; it
* then sends a JOIN_FIN.req message in an encrypted DTLS-ApplicationData record in a single UDP datagram to
* the Joiner_Router.
* - The JOIN_FIN.req message must not contain a Provisioning URL TLV.
* - 15. The Joiner_Router receives the encrypted DTLS-ApplicationData record and sends a RLY_RX.ntf message to
* the Commissioner containing:
* - CoAP URI-Path: NON POST coap://<C>/c/rx
* - CoAP Payload:
* - Joiner_1 DTLS Encapsulation TLV: Type = 17 (0x11), Length = Variable (8/16 bit size), Value =
* JOIN_FIN.req message in an encrypted DTLS-ApplicationData record received from Joiner_1.
* - Joiner_1 UDP Port TLV: Type = 18 (0x12), Length = 2, Value = UDP port of Joiner_1.
* - Joiner_1 IID TLV: Type = 19 (0x13), Length = 8, Value = The IID based on the EUI-64 of Joiner_1.
* - Joiner_Router Locator TLV: Type = 20 (0x14), Length = 2, Value = RLOC of Joiner_Router.
* - 16. The Commissioner receives the RLY_RX.ntf message and sends a RLY_TX.ntf message to the Joiner_Router
* containing:
* - CoAP URI-Path: NON POST coap://<JR>/c/tx
* - CoAP Payload:
* - Joiner_1 DTLS Encapsulation TLV: Type = 17 (0x11), Length = Variable (8/16 bit size), Value =
* JOIN_FIN.rsp message in an encrypted DTLS-ApplicationData record destined for Joiner_1.
* - Joiner_1 UDP Port TLV: Type = 18 (0x12), Length = 2, Value = UDP port of Joiner_1.
* - Joiner_1 IID TLV: Type = 19 (0x13), Length = 8, Value = The IID based on the EUI-64 of Joiner_1.
* - Joiner_Router Locator TLV: Type = 20 (0x14), Length = 8, Value = RLOC of Joiner_Router.
* - Joiner_Router KEK TLV: Type = 21 (0x15), Length = 16, Value = KEK used for JOIN_ENT messages.
* - 17. The Joiner_Router receives the RLY_TX.ntf message and sends a JOIN_FIN.rsp and the encapsulated
* DTLS-ApplicationData to Joiner_1.
* - 18. The Joiner_Router sends an encrypted JOIN_ENT.ntf message to Joiner_1.
* - 19. Joiner_1 receives the encrypted JOIN_ENT.ntf message and sends an encrypted JOIN_ENT.ntf dummy
* response to the Joiner_Router.
* - 20. Joiner_1 sends an encrypted DTLS-Alert record with a code of 0 (close_notify) to the Joiner_Router.
* - 21. The Joiner_Router receives the encrypted DTLS-Alert record and sends a RLY_RX.ntf message to the
* Commissioner on behalf of Joiner_1. The RLY_RX.ntf contains:
* - CoAP URI-Path: NON POST coap://<C>/c/rx
* - CoAP Payload:
* - Joiner_1 DTLS Encapsulation TLV: Type = 17 (0x11), Length = Variable (8/16 bit size), Value = encrypted
* DTLS-Alert record (close_notify = 0) received from Joiner_1.
* - Joiner_1 UDP Port TLV: Type = 18 (0x12), Length = 2, Value = UDP port of Joiner_1.
* - Joiner_1 IID TLV: Type = 19 (0x13), Length = 8, Value = The IID based on the EUI-64 of Joiner_1.
* - Joiner_Router Locator TLV: Type = 20 (0x14), Length = 2, Value = RLOC of Joiner_Router.
* - 22. The Commissioner receives the RLY_RX.ntf messages and sends RLY_TX.ntf messages to the Joiner_Router
* for delivery to Joiner_1. The RLY_TX.ntf for Joiner_1 contains:
* - CoAP URI-Path: NON POST coap://<JR>/c/tx
* - CoAP Payload:
* - Joiner_1 DTLS Encapsulation TLV: Type = 17 (0x011), Length = Variable (8/16 bit size), Value = encrypted
* DTLS-Alert record (close_notify = 0) destined for Joiner_1.
* - Joiner_1 UDP Port TLV: Type = 18 (0x12), Length = 2, Value = UDP port of Joiner_1.
* - Joiner_1 IID TLV: Type = 19 (0x13), Length = 8, Value = The IID based on the EUI-64 of Joiner_1.
* - Joiner_Router Locator TLV: Type = 20 (0x14), Length = 2, Value = RLOC of Joiner_Router.
* - 23. The Joiner_Router receives the RLY_TX.ntf message and sends the encapsulated, encrypted DTLS-Alert
* record with a code of 0 (close_notify) to Joiner_1.
*/
joinerNode.Get<ThreadNetif>().Up();
SuccessOrQuit(joinerNode.Get<Mac::Mac>().SetPanChannel(kTestChannel));
joinerNode.Get<Mac::Mac>().SetSupportedChannelMask(Mac::ChannelMask(1 << kTestChannel));
joinerNode.AllowList(jrNode);
SuccessOrQuit(joinerNode.Get<MeshCoP::Seeker>().SetUdpPort(kJoinerPort));
SuccessOrQuit(
joinerNode.Get<MeshCoP::Joiner>().Start(kPskd, nullptr, nullptr, nullptr, nullptr, nullptr, nullptr, nullptr));
nexus.AdvanceTime(kJoiningProcessTime);
Log(" Joiner state: %s", MeshCoP::Joiner::StateToString(joinerNode.Get<MeshCoP::Joiner>().GetState()));
VerifyOrQuit(joinerNode.Get<MeshCoP::Joiner>().GetState() == MeshCoP::Joiner::kStateJoined ||
joinerNode.Get<MeshCoP::Joiner>().GetState() == MeshCoP::Joiner::kStateIdle);
{
NetworkKey networkKey;
NetworkKey joinerKey;
commNode.Get<KeyManager>().GetNetworkKey(networkKey);
joinerNode.Get<KeyManager>().GetNetworkKey(joinerKey);
VerifyOrQuit(joinerKey == networkKey);
}
nexus.SaveTestInfo(aJsonFileName);
}
} // namespace Nexus
} // namespace ot
int main(int argc, char *argv[])
{
const char *jsonFile = (argc > 2) ? argv[2] : ot::Nexus::kDefaultJsonFile;
ot::Nexus::Test1_1_8_2_1(jsonFile);
printf("All tests passed\n");
return 0;
}
tests/nexus/verify_1_1_8_2_1.py
+405
View File
@@ -0,0 +1,405 @@
#!/usr/bin/env python3
#
# Copyright (c) 2026, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import sys
import os
import json
# Add the current directory to sys.path to find verify_utils
CUR_DIR = os.path.dirname(os.path.abspath(__file__))
sys.path.append(CUR_DIR)
import verify_utils
from pktverify import consts
from pktverify.packet_verifier import PacketVerifier
JOIN_FIN_URI = '/c/jf'
JOIN_ENT_URI = '/c/je'
# Constants for ports to avoid magic numbers
DEFAULT_JOINER_PORT = 49153
TMF_PORT = 61631
NM_PROVISIONING_URL_TLV = 32
def verify_rly_rx(p, router_rloc16, leader_rloc16, joiner_port=None, joiner_iid=None):
pkt = p.must_next()
assert pkt.coap.opt.uri_path_recon == consts.RLY_RX_URI, f"Expected URI {consts.RLY_RX_URI}, got {pkt.coap.opt.uri_path_recon}"
assert pkt.coap.tlv.joiner_dtls_encap, "joiner_dtls_encap missing"
if joiner_port is not None:
assert pkt.coap.tlv.joiner_udp_port == joiner_port, f"Expected port {joiner_port}, got {pkt.coap.tlv.joiner_udp_port}"
if joiner_iid is not None:
assert pkt.coap.tlv.joiner_iid == joiner_iid, f"Expected IID {joiner_iid}, got {pkt.coap.tlv.joiner_iid}"
assert pkt.coap.tlv.joiner_router_locator == router_rloc16, f"Expected locator {router_rloc16}, got {pkt.coap.tlv.joiner_router_locator}"
return pkt
def verify_rly_tx(p, leader_rloc16, router_rloc16, joiner_port=None, joiner_iid=None, joiner_router_kek=False):
pkt = p.must_next()
assert pkt.coap.opt.uri_path_recon == consts.RLY_TX_URI, f"Expected URI {consts.RLY_TX_URI}, got {pkt.coap.opt.uri_path_recon}"
assert pkt.coap.tlv.joiner_dtls_encap, "joiner_dtls_encap missing"
if joiner_port is not None:
assert pkt.coap.tlv.joiner_udp_port == joiner_port, f"Expected port {joiner_port}, got {pkt.coap.tlv.joiner_udp_port}"
if joiner_iid is not None:
assert pkt.coap.tlv.joiner_iid == joiner_iid, f"Expected IID {joiner_iid}, got {pkt.coap.tlv.joiner_iid}"
assert pkt.coap.tlv.joiner_router_locator == router_rloc16, f"Expected locator {router_rloc16}, got {pkt.coap.tlv.joiner_router_locator}"
if joiner_router_kek:
assert pkt.coap.tlv.joiner_router_kek, "joiner_router_kek missing"
return pkt
def field_contains(field, value):
if hasattr(field, 'all_fields'):
return value in [int(f.get_default_value()) for f in field.all_fields]
elif isinstance(field, list):
return value in field
else:
try:
return int(field) == value
except (TypeError, ValueError):
return False
def verify(pv):
# 8.2.1 On Mesh Commissioner Joining with JR, any commissioner, single (correct)
#
# 8.2.1.1 Topology
# - Commissioner (Leader)
# - Joiner Router
# - Joiner_1
#
# 8.2.1.2 Purpose & Description
# The purpose of this test case is to verify that the DUT sends and receives relayed DTLS traffic correctly when
# the correct PSKd is used by the Joiner. It also verifies that the DUT (as a Joiner Router) sends JOIN_ENT.ntf
# encrypted with KEK.
#
# Spec Reference | V1.1 Section | V1.3.0 Section
# ---------------|---------------|---------------
# Relay traffic | 8.4.3 / 8.4.4 | 8.4.3 / 8.4.4
pkts = pv.pkts
pv.summary.show()
# Fail Condition 3: A DTLS-Alert record with a Fatal alert level is sent by either Joiner_1 or the Commissioner.
for p in pkts.filter(lambda p: 'dtls' in p and\
hasattr(p.dtls, 'record_content_type') and\
field_contains(p.dtls.record_content_type, 21)):
if hasattr(p.dtls, 'alert_message_level'):
levels = p.dtls.alert_message_level
if hasattr(levels, 'all_fields'):
levels = [f.get_default_value() for f in levels.all_fields]
else:
levels = [str(levels)]
for level in levels:
if level == '2':
raise ValueError(f"Fatal DTLS Alert found in packet {p.number}")
# Fail Condition 4: The Content-Format option in CoAP messages is present and is not application/octet-stream (42).
for p in pkts.filter(lambda p: 'coap' in p):
if hasattr(p.coap, 'opt_content_format'):
fmt = str(p.coap.opt_content_format)
if fmt != 'nullField' and fmt != '42':
raise ValueError(f"Invalid CoAP Content-Format {fmt} in packet {p.number}")
LEADER = pv.vars['LEADER']
ROUTER = pv.vars['ROUTER']
LEADER_RLOC16 = pv.vars['LEADER_RLOC16']
ROUTER_RLOC16 = pv.vars['ROUTER_RLOC16']
# Step 3.1: Discover Joiner Port from MLE Discovery Response
print("Step 3.1: Discover Joiner Port from MLE Discovery Response")
discovery_rsp = pkts.filter_wpan_src64(LEADER).\
filter_mle_cmd(consts.MLE_DISCOVERY_RESPONSE).\
must_next()
JOINER_PORTS = discovery_rsp.thread_meshcop.tlv.udp_port
if not isinstance(JOINER_PORTS, list):
JOINER_PORTS = [JOINER_PORTS]
JOINER_PORTS = [int(p) for p in JOINER_PORTS]
# We expect 49153 to be one of the ports
assert DEFAULT_JOINER_PORT in JOINER_PORTS, f"Expected {DEFAULT_JOINER_PORT} in {JOINER_PORTS}"
JOINER_PORT = DEFAULT_JOINER_PORT
pv.add_vars(JOINER_PORT=JOINER_PORT)
# Find the Joiner by its initial ClientHello
print("Finding Joiner from initial DTLS ClientHello")
initial_hello_pkt = pkts.filter(lambda p: p.udp.dstport == JOINER_PORT or p.udp.srcport == JOINER_PORT).\
must_next()
JOINER = initial_hello_pkt.wpan.src64
pv.add_vars(JOINER=JOINER)
# Step 3.2: Joiner_1 sends an initial DTLS-ClientHello
print("Step 3.2: Joiner_1 sends an initial DTLS-ClientHello")
pkts.index = (0, 0)
pkts.filter_wpan_src64(JOINER).\
filter_wpan_dst64(ROUTER).\
filter(lambda p: p.udp.dstport == JOINER_PORT).\
filter(lambda p: field_contains(p.dtls.handshake.type, 1)).\
must_next()
# Step 3.3: Joiner_Router receives the initial DTLS-ClientHello record and sends a RLY_RX.ntf message
print("Step 3.3: Joiner_Router sends RLY_RX.ntf to Commissioner")
rly_rx_filter = pkts.filter('wpan.src16 == {ROUTER_RLOC16}', ROUTER_RLOC16=ROUTER_RLOC16).\
filter('wpan.dst16 == {LEADER_RLOC16}', LEADER_RLOC16=LEADER_RLOC16).\
filter(lambda p: p.udp.dstport == TMF_PORT).\
filter(lambda p: p.coap.opt.uri_path_recon == consts.RLY_RX_URI)
rly_rx_pkt = verify_rly_rx(rly_rx_filter, ROUTER_RLOC16, LEADER_RLOC16, joiner_port=JOINER_PORT)
JOINER_IID = rly_rx_pkt.coap.tlv.joiner_iid
pv.add_vars(JOINER_IID=JOINER_IID)
# Step 3.4: The Commissioner receives the RLY_RX.ntf message and sends a RLY_TX.ntf message
print("Step 3.4: Commissioner sends RLY_TX.ntf to Joiner_Router")
rly_tx = pkts.filter('wpan.src16 == {LEADER_RLOC16}', LEADER_RLOC16=LEADER_RLOC16).\
filter('wpan.dst16 == {ROUTER_RLOC16}', ROUTER_RLOC16=ROUTER_RLOC16).\
filter(lambda p: p.udp.dstport == TMF_PORT).\
filter(lambda p: p.coap.opt.uri_path_recon == consts.RLY_TX_URI)
verify_rly_tx(rly_tx, LEADER_RLOC16, ROUTER_RLOC16, joiner_port=JOINER_PORT, joiner_iid=JOINER_IID)
# Step 3.5: The Joiner_Router receives the RLY_TX.ntf message and sends the DTLS-HelloVerifyRequest
print("Step 3.5: Joiner_Router sends DTLS-HelloVerifyRequest to Joiner_1")
pkts.index = (0, 0)
hvr_pkt = pkts.filter_wpan_src64(ROUTER).\
filter_wpan_dst64(JOINER).\
filter(lambda p: p.udp.srcport == JOINER_PORT).\
filter(lambda p: field_contains(p.dtls.handshake.type, 3)).\
must_next()
COOKIE = hvr_pkt.dtls.handshake.cookie
pv.add_vars(COOKIE=COOKIE)
# Step 3.6: Joiner_1 receives the DTLS-HelloVerifyRequest and sends subsequent DTLS-ClientHello
print("Step 3.6: Joiner_1 sends subsequent DTLS-ClientHello")
pkts.filter_wpan_src64(JOINER).\
filter_wpan_dst64(ROUTER).\
filter(lambda p: p.udp.dstport == JOINER_PORT).\
filter(lambda p: field_contains(p.dtls.handshake.type, 1) and\
p.dtls.handshake.cookie == COOKIE).\
must_next()
# Step 3.7: The Joiner_Router receives the subsequent DTLS-ClientHello and sends a RLY_RX.ntf
print("Step 3.7: Joiner_Router sends subsequent RLY_RX.ntf")
rly_rx = pkts.filter('wpan.src16 == {ROUTER_RLOC16}', ROUTER_RLOC16=ROUTER_RLOC16).\
filter('wpan.dst16 == {LEADER_RLOC16}', LEADER_RLOC16=LEADER_RLOC16).\
filter(lambda p: p.udp.dstport == TMF_PORT).\
filter(lambda p: p.coap.opt.uri_path_recon == consts.RLY_RX_URI)
verify_rly_rx(rly_rx, ROUTER_RLOC16, LEADER_RLOC16, joiner_port=JOINER_PORT, joiner_iid=JOINER_IID)
# Step 3.8: The Commissioner receives the RLY_RX.ntf and sends a RLY_TX.ntf (Server Hello etc.)
print("Step 3.8: Commissioner sends subsequent RLY_TX.ntf")
rly_tx = pkts.filter('wpan.src16 == {LEADER_RLOC16}', LEADER_RLOC16=LEADER_RLOC16).\
filter('wpan.dst16 == {ROUTER_RLOC16}', ROUTER_RLOC16=ROUTER_RLOC16).\
filter(lambda p: p.udp.dstport == TMF_PORT).\
filter(lambda p: p.coap.opt.uri_path_recon == consts.RLY_TX_URI)
verify_rly_tx(rly_tx, LEADER_RLOC16, ROUTER_RLOC16, joiner_port=JOINER_PORT, joiner_iid=JOINER_IID)
# Step 3.9: Joiner_Router receives the RLY_TX.ntf and sends DTLS-ServerHello etc. to Joiner_1
print("Step 3.9: Joiner_Router sends DTLS-ServerHello etc. to Joiner_1")
pkts.index = (0, 0)
pkts.filter_wpan_src64(ROUTER).\
filter_wpan_dst64(JOINER).\
filter(lambda p: p.udp.srcport == JOINER_PORT).\
filter(lambda p: field_contains(p.dtls.handshake.type, 2)).\
must_next()
# Step 3.10: Joiner_1 receives the DTLS-ServerHello etc. and sends DTLS-ClientKeyExchange etc.
print("Step 3.10: Joiner_1 sends DTLS-ClientKeyExchange etc.")
pkts.filter_wpan_src64(JOINER).\
filter_wpan_dst64(ROUTER).\
filter(lambda p: p.udp.dstport == JOINER_PORT).\
filter(lambda p: field_contains(p.dtls.record.content_type, 20) or\
field_contains(p.dtls.record.content_type, 22)).\
must_next()
# Step 3.11: Joiner_Router receives the DTLS records and sends a RLY_RX.ntf message
print("Step 3.11: Joiner_Router sends RLY_RX.ntf (handshake completion)")
rly_rx = pkts.filter('wpan.src16 == {ROUTER_RLOC16}', ROUTER_RLOC16=ROUTER_RLOC16).\
filter('wpan.dst16 == {LEADER_RLOC16}', LEADER_RLOC16=LEADER_RLOC16).\
filter(lambda p: p.udp.dstport == TMF_PORT).\
filter(lambda p: p.coap.opt.uri_path_recon == consts.RLY_RX_URI)
verify_rly_rx(rly_rx, ROUTER_RLOC16, LEADER_RLOC16, joiner_port=JOINER_PORT, joiner_iid=JOINER_IID)
# Step 3.12: The Commissioner receives the RLY_RX.ntf and sends a RLY_TX.ntf
print("Step 3.12: Commissioner sends RLY_TX.ntf (handshake completion)")
rly_tx = pkts.filter('wpan.src16 == {LEADER_RLOC16}', LEADER_RLOC16=LEADER_RLOC16).\
filter('wpan.dst16 == {ROUTER_RLOC16}', ROUTER_RLOC16=ROUTER_RLOC16).\
filter(lambda p: p.udp.dstport == TMF_PORT).\
filter(lambda p: p.coap.opt.uri_path_recon == consts.RLY_TX_URI)
verify_rly_tx(rly_tx, LEADER_RLOC16, ROUTER_RLOC16, joiner_port=JOINER_PORT, joiner_iid=JOINER_IID)
# Step 3.13: Joiner_Router receives the RLY_TX.ntf and sends DTLS-ChangeCipherSpec etc. to Joiner_1
print("Step 3.13: Joiner_Router sends DTLS-ChangeCipherSpec etc. to Joiner_1")
pkts.index = (0, 0)
pkts.filter_wpan_src64(ROUTER).\
filter_wpan_dst64(JOINER).\
filter(lambda p: p.udp.srcport == JOINER_PORT).\
filter(lambda p: field_contains(p.dtls.record.content_type, 20) or\
field_contains(p.dtls.record.content_type, 22)).\
must_next()
# Step 3.14: Joiner_1 receives the records and sends JOIN_FIN.req in DTLS-ApplicationData
print("Step 3.14: Joiner_1 sends JOIN_FIN.req")
pkt = pkts.filter_wpan_src64(JOINER).\
filter_wpan_dst64(ROUTER).\
filter(lambda p: p.udp.dstport == JOINER_PORT).\
filter(lambda p: field_contains(p.dtls.record.content_type, 23)).\
must_next()
# Verify JOIN_FIN.req does not contain Provisioning URL TLV
assert pkt.coap.opt.uri_path_recon == JOIN_FIN_URI, f"Expected URI {JOIN_FIN_URI}, got {pkt.coap.opt.uri_path_recon}"
assert NM_PROVISIONING_URL_TLV not in pkt.coap.tlv.type, "JOIN_FIN.req contains Provisioning URL TLV"
# Step 3.15: Joiner_Router receives the record and sends a RLY_RX.ntf message
print("Step 3.15: Joiner_Router sends RLY_RX.ntf (JOIN_FIN.req)")
rly_rx = pkts.filter('wpan.src16 == {ROUTER_RLOC16}', ROUTER_RLOC16=ROUTER_RLOC16).\
filter('wpan.dst16 == {LEADER_RLOC16}', LEADER_RLOC16=LEADER_RLOC16).\
filter(lambda p: p.udp.dstport == TMF_PORT).\
filter(lambda p: p.coap.opt.uri_path_recon == consts.RLY_RX_URI)
verify_rly_rx(rly_rx, ROUTER_RLOC16, LEADER_RLOC16, joiner_port=JOINER_PORT, joiner_iid=JOINER_IID)
# Step 3.16: The Commissioner receives the RLY_RX.ntf and sends a RLY_TX.ntf (JOIN_FIN.rsp)
print("Step 3.16: Commissioner sends RLY_TX.ntf (JOIN_FIN.rsp)")
rly_tx = pkts.filter('wpan.src16 == {LEADER_RLOC16}', LEADER_RLOC16=LEADER_RLOC16).\
filter('wpan.dst16 == {ROUTER_RLOC16}', ROUTER_RLOC16=ROUTER_RLOC16).\
filter(lambda p: p.udp.dstport == TMF_PORT).\
filter(lambda p: p.coap.opt.uri_path_recon == consts.RLY_TX_URI)
verify_rly_tx(rly_tx,
LEADER_RLOC16,
ROUTER_RLOC16,
joiner_port=JOINER_PORT,
joiner_iid=JOINER_IID,
joiner_router_kek=True)
# Step 3.17: The Joiner_Router receives the RLY_TX.ntf message and sends a JOIN_FIN.rsp
print("Step 3.17: Joiner_Router sends JOIN_FIN.rsp to Joiner_1")
pkts.filter_wpan_src64(ROUTER).\
filter_wpan_dst64(JOINER).\
filter(lambda p: p.udp.srcport == JOINER_PORT).\
filter(lambda p: field_contains(p.dtls.record.content_type, 23)).\
must_next()
# Step 3.18: The Joiner_Router sends an encrypted JOIN_ENT.ntf message to Joiner_1.
print("Step 3.18: Joiner_Router sends encrypted JOIN_ENT.ntf")
pkts.index = (0, 0)
pkts.filter_wpan_src64(ROUTER).\
filter_wpan_dst64(JOINER).\
filter(lambda p: p.wpan.security == True).\
must_next()
# Step 3.19: Joiner_1 receives the JOIN_ENT.ntf and sends response
print("Step 3.19: Joiner_1 sends JOIN_ENT.ntf response")
pkts.filter_wpan_src64(JOINER).\
filter_wpan_dst64(ROUTER).\
filter(lambda p: p.wpan.security == True).\
must_next()
# Step 3.20: Joiner_1 sends an encrypted DTLS-Alert record
print("Step 3.20: Joiner_1 sends DTLS-Alert (close_notify)")
pkts.index = (0, 0)
pkts.filter_wpan_src64(JOINER).\
filter_wpan_dst64(ROUTER).\
filter(lambda p: p.udp.dstport == JOINER_PORT).\
filter(lambda p: field_contains(p.dtls.record.content_type, 21)).\
filter(lambda p: field_contains(p.dtls.alert_message.desc, 0)).\
must_next()
# Step 3.21: The Joiner_Router receives the record and sends a RLY_RX.ntf message
print("Step 3.21: Joiner_Router sends RLY_RX.ntf (DTLS-Alert)")
rly_rx = pkts.filter('wpan.src16 == {ROUTER_RLOC16}', ROUTER_RLOC16=ROUTER_RLOC16).\
filter('wpan.dst16 == {LEADER_RLOC16}', LEADER_RLOC16=LEADER_RLOC16).\
filter(lambda p: p.udp.dstport == TMF_PORT).\
filter(lambda p: p.coap.opt.uri_path_recon == consts.RLY_RX_URI)
verify_rly_rx(rly_rx, ROUTER_RLOC16, LEADER_RLOC16, joiner_port=JOINER_PORT, joiner_iid=JOINER_IID)
# Step 3.22: The Commissioner receives the RLY_RX.ntf and sends a RLY_TX.ntf
print("Step 3.22: Commissioner sends RLY_TX.ntf (DTLS-Alert)")
rly_tx = pkts.filter('wpan.src16 == {LEADER_RLOC16}', LEADER_RLOC16=LEADER_RLOC16).\
filter('wpan.dst16 == {ROUTER_RLOC16}', ROUTER_RLOC16=ROUTER_RLOC16).\
filter(lambda p: p.udp.dstport == TMF_PORT).\
filter(lambda p: p.coap.opt.uri_path_recon == consts.RLY_TX_URI)
verify_rly_tx(rly_tx, LEADER_RLOC16, ROUTER_RLOC16, joiner_port=JOINER_PORT, joiner_iid=JOINER_IID)
# Step 3.23: The Joiner_Router receives the RLY_TX.ntf and sends the DTLS-Alert to Joiner_1
print("Step 3.23: Joiner_Router sends DTLS-Alert to Joiner_1")
pkts.filter_wpan_src64(ROUTER).\
filter_wpan_dst64(JOINER).\
filter(lambda p: p.udp.srcport == JOINER_PORT).\
filter(lambda p: field_contains(p.dtls.record.content_type, 21)).\
filter(lambda p: field_contains(p.dtls.alert_message.desc, 0)).\
must_next()
if __name__ == '__main__':
verify_utils.apply_patches()
json_file = os.path.abspath(sys.argv[1])
keys_file = json_file.replace('.json', '.keys')
with open(json_file, 'rt') as f:
data = json.load(f)
# Add Joiner port to DECODE_AS entries
consts.WIRESHARK_DECODE_AS_ENTRIES[f'udp.port=={DEFAULT_JOINER_PORT}'] = 'dtls'
consts.WIRESHARK_DECODE_AS_ENTRIES[f'dtls.port=={DEFAULT_JOINER_PORT}'] = 'coap'
wireshark_prefs = consts.WIRESHARK_OVERRIDE_PREFS.copy()
if os.path.exists(keys_file):
wireshark_prefs['tls.keylog_file'] = keys_file
existing_keys_str = wireshark_prefs.get('uat:ieee802154_keys', '')
keys = []
for line in existing_keys_str.split('\n'):
line = line.strip()
if line:
keys.append(line)
network_key = data.get('network_key')
if network_key:
new_key = f'"{network_key}","1","Thread hash"'
if not any(network_key in k for k in keys):
keys.append(new_key)
wireshark_prefs['uat:ieee802154_keys'] = '\n'.join(keys)
mesh_local_prefix = data.get('extra_vars', {}).get('mesh_local_prefix')
if mesh_local_prefix:
prefix_addr = mesh_local_prefix.split('/')[0]
wireshark_prefs['6lowpan.context0'] = f'{prefix_addr}/64'
try:
pv = PacketVerifier(json_file, wireshark_prefs=wireshark_prefs)
pv.add_common_vars()
for node_id, rloc16 in data.get('rloc16s', {}).items():
name = pv.test_info.get_node_name(int(node_id))
pv.add_vars(**{f'{name}_RLOC16': int(rloc16, 16)})
verify(pv)
print("Verification PASSED")
except Exception as e:
print(f"Verification FAILED: {e}")
import traceback
traceback.print_exc()
sys.exit(1)
tests/nexus/verify_utils.py
+50 -4
View File
@@ -45,11 +45,14 @@ if THREAD_CERT_DIR not in sys.path:
from pktverify import consts
from pktverify.packet_verifier import PacketVerifier
from pktverify import utils as pvutils
from pktverify import coap
from pktverify.coap import CoapTlvParser
from pktverify.addrs import Ipv6Addr
from pktverify.bytes import Bytes
# CSL period constants (in units of 10 symbols)
# Constants
NM_PROVISIONING_URL_TLV = 32
CSL_PERIOD_500MS = 500000 // consts.US_PER_TEN_SYMBOLS
CSL_PERIOD_3300MS = 3300000 // consts.US_PER_TEN_SYMBOLS
CSL_PERIOD_400MS = 400000 // consts.US_PER_TEN_SYMBOLS
@@ -133,7 +136,17 @@ def thread_coap_tlv_parse(t, v, layer=None):
kvs.append(('pan_id', struct.unpack('>H', v)[0]))
elif t == consts.NM_NETWORK_MESH_LOCAL_PREFIX_TLV and len(v) == 8 and not is_diag:
kvs.append(('mesh_local_prefix', v))
elif t == consts.NM_PROVISIONING_URL_TLV and not is_diag:
elif t == consts.NM_JOINER_DTLS_ENCAPSULATION_TLV and not is_diag:
kvs.append(('joiner_dtls_encap', v))
elif t == consts.NM_JOINER_UDP_PORT_TLV and len(v) == 2 and not is_diag:
kvs.append(('joiner_udp_port', struct.unpack('>H', v)[0]))
elif t == consts.NM_JOINER_IID_TLV and len(v) == 8 and not is_diag:
kvs.append(('joiner_iid', v))
elif t == consts.NM_JOINER_ROUTER_LOCATOR_TLV and len(v) == 2 and not is_diag:
kvs.append(('joiner_router_locator', struct.unpack('>H', v)[0]))
elif t == consts.NM_JOINER_ROUTER_KEK_TLV and len(v) == 16 and not is_diag:
kvs.append(('joiner_router_kek', v))
elif t == NM_PROVISIONING_URL_TLV and not is_diag:
kvs.append(('provisioning_url', v.decode('utf-8', errors='replace')))
elif t == consts.NM_FUTURE_TLV:
kvs.append(('future_tlv', v))
@@ -295,6 +308,11 @@ def apply_patches():
layer_fields._LAYER_FIELDS['coap.tlv.ext_pan_id'] = layer_fields._bytes
layer_fields._LAYER_FIELDS['coap.tlv.network_name'] = layer_fields._str
layer_fields._LAYER_FIELDS['coap.tlv.pskc'] = layer_fields._bytes
layer_fields._LAYER_FIELDS['coap.tlv.joiner_dtls_encap'] = layer_fields._bytes
layer_fields._LAYER_FIELDS['coap.tlv.joiner_udp_port'] = layer_fields._auto
layer_fields._LAYER_FIELDS['coap.tlv.joiner_iid'] = layer_fields._bytes
layer_fields._LAYER_FIELDS['coap.tlv.joiner_router_locator'] = layer_fields._auto
layer_fields._LAYER_FIELDS['coap.tlv.joiner_router_kek'] = layer_fields._bytes
layer_fields._LAYER_FIELDS['coap.tlv.security_policy'] = layer_fields._bytes
layer_fields._LAYER_FIELDS['coap.tlv.sec_policy_o'] = layer_fields._auto
layer_fields._LAYER_FIELDS['coap.tlv.sec_policy_n'] = layer_fields._auto
@@ -307,11 +325,39 @@ def apply_patches():
layer_fields._LAYER_FIELDS['coap.tlv.network_key'] = layer_fields._bytes
layer_fields._LAYER_FIELDS['coap.tlv.pan_id'] = layer_fields._auto
layer_fields._LAYER_FIELDS['coap.tlv.mesh_local_prefix'] = layer_fields._bytes
layer_fields._LAYER_FIELDS['coap.opt_content_format'] = layer_fields._auto
layer_fields._LAYER_FIELDS['dtls.alert_message_level'] = layer_fields._auto
layer_fields._LAYER_FIELDS['dtls.alert_message_desc'] = layer_fields._auto
layer_fields._LAYER_FIELDS['dtls.handshake.cookie'] = layer_fields._bytes
layer_fields._LAYER_FIELDS['thread_meshcop.tlv.discovery_version'] = layer_fields._dec
layer_fields._LAYER_FIELDS['thread_meshcop.tlv.discovery_native_commissioner'] = layer_fields._dec
layer_fields._LAYER_FIELDS['thread_meshcop.tlv.active_tstamp'] = layer_fields._list(layer_fields._thread_timestamp)
layer_fields._LAYER_FIELDS['thread_meshcop.tlv.pending_tstamp'] = layer_fields._list(
layer_fields._thread_timestamp)
def _parse_next_tlv_patched(payload, read_pos, layer=None) -> tuple:
assert read_pos <= len(payload)
if read_pos == len(payload):
return None, None, read_pos
t = payload[read_pos]
if read_pos + 1 >= len(payload):
return None, None, len(payload)
len_ = payload[read_pos + 1]
val_pos = read_pos + 2
if len_ == 255:
if read_pos + 3 >= len(payload):
return None, None, len(payload)
len_ = (payload[read_pos + 2] << 8) | payload[read_pos + 3]
val_pos = read_pos + 4
if len(payload) - val_pos < len_:
return None, None, len(payload)
kvs = coap.CoapTlvParser.parse(t, payload[val_pos:val_pos + len_], layer=layer)
return t, kvs, val_pos + len_
coap.CoapLayer._parse_next_tlv = staticmethod(_parse_next_tlv_patched)
layer_fields._LAYER_FIELDS['thread_meshcop.tlv.delay_timer'] = layer_fields._auto
layer_fields._LAYER_FIELDS['mle.tlv.link_query_options'] = layer_fields._bytes