espressif/openthread
1
0
Fork 0
mirror of https://github.com/espressif/openthread.git synced 2026-08-03 01:17:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

[thread-cert] refactor case 8.1.1 using pktverify (#5883)

Browse Source
This commit is contained in:
Jing Ma
2020-12-09 07:34:07 -08:00
committed by GitHub
parent 73126f7099
commit fdc8a97594
4 changed files with 196 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/scripts/thread-cert/Cert_8_1_01_Commissioning.py
+185 -55
View File
@@ -31,25 +31,45 @@ import unittest
import command
import dtls
import mle
import thread_cert
from command import CheckType
from pktverify.consts import MLE_DISCOVERY_REQUEST, MLE_DISCOVERY_RESPONSE, HANDSHAKE_CLIENT_HELLO, HANDSHAKE_SERVER_HELLO, HANDSHAKE_SERVER_KEY_EXCHANGE, HANDSHAKE_SERVER_HELLO_DONE, HANDSHAKE_CLIENT_KEY_EXCHANGE, HANDSHAKE_HELLO_VERIFY_REQUEST, CONTENT_APPLICATION_DATA, NM_EXTENDED_PAN_ID_TLV, NM_NETWORK_NAME_TLV, NM_STEERING_DATA_TLV, NM_COMMISSIONER_UDP_PORT_TLV, NM_JOINER_UDP_PORT_TLV, NM_DISCOVERY_REQUEST_TLV, NM_DISCOVERY_RESPONSE_TLV, THREAD_DISCOVERY_TLV, CONTENT_CHANGE_CIPHER_SPEC, CONTENT_HANDSHAKE, CONTENT_ALERT
from pktverify.packet_verifier import PacketVerifier
COMMISSIONER = 1
JOINER = 2
# Test Purpose and Description:
# -----------------------------
# The purpose of this test case is to verify the DTLS sessions between the
# on-mesh Commissioner and a Joiner when the correct PSKd is used
#
# Note that many of the messages/records exchanged are encrypted
# and cannot be observe
#
# Test Topology:
# -------------
# Commissioner
# |
# Joiner
#
# DUT Types:
# ----------
# Commissioner
# Joiner
class Cert_8_1_01_Commissioning(thread_cert.TestCase):
SUPPORT_NCP = False
TOPOLOGY = {
COMMISSIONER: {
'name': 'COMMISSIONER',
'masterkey': '00112233445566778899aabbccddeeff',
'mode': 'rdn',
'panid': 0xface
},
JOINER: {
'name': 'JOINER',
'masterkey': 'deadbeefdeadbeefdeadbeefdeadbeef',
'mode': 'rdn',
'router_selection_jitter': 1
@@ -73,71 +93,181 @@ class Cert_8_1_01_Commissioning(thread_cert.TestCase):
self.nodes[JOINER].get_masterkey(),
self.nodes[COMMISSIONER].get_masterkey(),
)
joiner_messages = self.simulator.get_messages_sent_by(JOINER)
commissioner_messages = self.simulator.get_messages_sent_by(COMMISSIONER)
# 2 - N/A
# 3 - Joiner_1
msg = joiner_messages.next_mle_message(mle.CommandType.DISCOVERY_REQUEST)
command.check_discovery_request(msg, thread_version=self.nodes[JOINER].version)
request_src_addr = msg.mac_header.src_address
# 4 - Commissioner
msg = commissioner_messages.next_mle_message(mle.CommandType.DISCOVERY_RESPONSE)
command.check_discovery_response(msg,
request_src_addr,
steering_data=CheckType.CONTAIN,
thread_version=self.nodes[COMMISSIONER].version)
udp_port_set_by_commissioner = command.get_joiner_udp_port_in_discovery_response(msg)
# 5.2 - Joiner_1
msg = joiner_messages.next_dtls_message(dtls.ContentType.HANDSHAKE, dtls.HandshakeType.CLIENT_HELLO)
self.assertEqual(msg.get_dst_udp_port(), udp_port_set_by_commissioner)
# 5.3 - Commissioner
msg = commissioner_messages.next_dtls_message(dtls.ContentType.HANDSHAKE,
dtls.HandshakeType.HELLO_VERIFY_REQUEST)
commissioner_cookie = msg.dtls.body.cookie
# 5.4 - Joiner_1
msg = joiner_messages.next_dtls_message(dtls.ContentType.HANDSHAKE, dtls.HandshakeType.CLIENT_HELLO)
self.assertEqual(commissioner_cookie, msg.dtls.body.cookie)
self.assertEqual(msg.get_dst_udp_port(), udp_port_set_by_commissioner)
# 5.5 - Commissioner
commissioner_messages.next_dtls_message(dtls.ContentType.HANDSHAKE, dtls.HandshakeType.SERVER_HELLO)
commissioner_messages.next_dtls_message(dtls.ContentType.HANDSHAKE, dtls.HandshakeType.SERVER_KEY_EXCHANGE)
commissioner_messages.next_dtls_message(dtls.ContentType.HANDSHAKE, dtls.HandshakeType.SERVER_HELLO_DONE)
# 5.6 - Joiner_1
msg = joiner_messages.next_dtls_message(dtls.ContentType.HANDSHAKE, dtls.HandshakeType.CLIENT_KEY_EXCHANGE)
self.assertEqual(msg.get_dst_udp_port(), udp_port_set_by_commissioner)
msg = joiner_messages.next_dtls_message(dtls.ContentType.CHANGE_CIPHER_SPEC)
self.assertEqual(msg.get_dst_udp_port(), udp_port_set_by_commissioner)
# TODO(wgtdkp): It's required to verify DTLS FINISHED message here.
# Currently not handled as it is encrypted.
# 5.7 - Commissioner
commissioner_messages.next_dtls_message(dtls.ContentType.CHANGE_CIPHER_SPEC)
# TODO(wgtdkp): It's required to verify DTLS FINISHED message here.
# Currently not handled as it is encrypted.
# 5.8,9,10,11
# - Joiner_1
# - Joiner
command.check_joiner_commissioning_messages(joiner_messages.commissioning_messages)
# - Commissioner
command.check_commissioner_commissioning_messages(commissioner_messages.commissioning_messages)
# As commissioner is also joiner router
command.check_joiner_router_commissioning_messages(commissioner_messages.commissioning_messages)
self.nodes[JOINER].thread_start()
self.simulator.go(5)
self.assertEqual(self.nodes[JOINER].get_state(), 'router')
def verify(self, pv):
pkts = pv.pkts
pv.summary.show()
COMMISSIONER = pv.vars['COMMISSIONER']
COMMISSIONER_VERSION = pv.vars['COMMISSIONER_VERSION']
JOINER_VERSION = pv.vars['JOINER_VERSION']
# Step 3: Joiner sends MLE Discovery Request
# MLE Discovery Request message MUST have these values:
# - MLE Security Suite: 255 (No MLE Security)
# - Thread Discovery TLV
# Sub-TLVs:
# - Discovery Request TLV
# - Protocol Version: 2 or 3
# (depends on the Thread stack version in testing)
pkts.filter_mle_cmd(MLE_DISCOVERY_REQUEST).\
filter_LLARMA().\
filter(lambda p:
[THREAD_DISCOVERY_TLV] == p.mle.tlv.type and\
[NM_DISCOVERY_REQUEST_TLV] == p.thread_meshcop.tlv.type and\
p.thread_meshcop.tlv.discovery_req_ver == JOINER_VERSION
).\
must_next()
# Step 4: Commissioner sends MLE Discovery Response
# MLE Discovery Response message MUST have these values:
# - MLE Security Suite: 255 (No MLE Security)
# - Source Address in IEEE 802.15.4 header MUST be set to
# the MAC Extended Address (64-bit)
# - Destination Address in IEEE 802.15.4 header MUST be
# set to Discovery Request Source Address
# - Thread Discovery TLV
# Sub-TLVs:
# - Discovery Request TLV
# - Protocol Version: 2 or 3
# (depends on the Thread stack version in testing)
# - Extended PAN ID TLV
# - Joiner UDP Port TLV
# - Network Name TLV
# - Steering Data TLV
# - Commissioner UDP Port TLV (optional)
_rs_pkt = pkts.filter_wpan_src64(COMMISSIONER).\
filter_mle_cmd(MLE_DISCOVERY_RESPONSE).\
filter(lambda p: {
NM_EXTENDED_PAN_ID_TLV,
NM_NETWORK_NAME_TLV,
NM_STEERING_DATA_TLV,
NM_JOINER_UDP_PORT_TLV,
NM_DISCOVERY_RESPONSE_TLV
} <= set(p.thread_meshcop.tlv.type) and\
p.thread_meshcop.tlv.discovery_rsp_ver == COMMISSIONER_VERSION
).\
must_next()
# Step 5: Verify the following details occur in the exchange between
# Joiner and the Commissioner
# 1. UDP port (Specified by the Commissioner: in Discovery Response)
# is used as destination port for UDP datagrams from Joiner to
# the Commissioner.
# 2. Joiner sends an initial DTLS-ClientHello handshake record to the
# Commissioner
pkts.filter_wpan_dst64(COMMISSIONER).\
filter(lambda p:
p.dtls.handshake.type == [HANDSHAKE_CLIENT_HELLO] and\
p.udp.srcport in _rs_pkt.thread_meshcop.tlv.udp_port and\
p.udp.dstport in _rs_pkt.thread_meshcop.tlv.udp_port
).\
must_next()
# 3. The Commissioner receives the initial DTLS-ClientHello handshake
# record and sends a DTLS-HelloVerifyRequest handshake record Joiner
_pkt = pkts.filter_wpan_src64(COMMISSIONER).\
filter(lambda p: p.dtls.handshake.type == [HANDSHAKE_HELLO_VERIFY_REQUEST]).\
must_next()
_pkt.must_verify(lambda p: p.dtls.handshake.cookie is not None)
# 4. Joiner 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
pkts.filter_wpan_dst64(COMMISSIONER).\
filter(lambda p:
p.dtls.handshake.type == [HANDSHAKE_CLIENT_HELLO] and\
p.dtls.handshake.cookie == _pkt.dtls.handshake.cookie
).\
must_next()
# 5. Commissioner must correctly receive the subsequent DTLSClientHello
# handshake record and then send, in order, DTLSServerHello,
# DTLS-ServerKeyExchange and DTLSServerHelloDone handshake records to Joiner
pkts.filter_wpan_src64(COMMISSIONER).\
filter(lambda p:
p.dtls.handshake.type == [HANDSHAKE_SERVER_HELLO,
HANDSHAKE_SERVER_KEY_EXCHANGE,
HANDSHAKE_SERVER_HELLO_DONE]
).\
must_next()
# 6. Joiner receives the DTLS-ServerHello, DTLSServerKeyExchange and
# DTLS-ServerHelloDone handshake records and sends, in order,
# a DTLS-ClientKeyExchange handshake record,
# a DTLS-ChangeCipherSpec record and
# an encrypted DTLS-Finished handshake record to the Commissioner.
pkts.filter_wpan_dst64(COMMISSIONER).\
filter(lambda p:
p.dtls.handshake.type == [HANDSHAKE_CLIENT_KEY_EXCHANGE] and\
{
CONTENT_CHANGE_CIPHER_SPEC,
CONTENT_HANDSHAKE
} == set(p.dtls.record.content_type)
).\
must_next()
# 7. Commissioner receives the DTLS-ClientKeyExchange handshake record, the
# DTLS-ChangeCipherSpec record and the encrypted DTLS-Finished handshake record,
# and sends a DTLS-ChangeCipherSpec record and an encrypted DTLSFinished handshake
# record in that order to Joiner
pkts.filter_wpan_src64(COMMISSIONER).\
filter(lambda p: {
CONTENT_CHANGE_CIPHER_SPEC,
CONTENT_HANDSHAKE
} == set(p.dtls.record.content_type)
).\
must_next()
# 8. Joiner receives the DTLS-ChangeCipherSpec record and the encrypted DTLS-Finished
# handshake record and sends a JOIN_FIN.req message in an encrypted DTLS-ApplicationData
# record in a single UDP datagram to Commissioner.
pkts.filter_wpan_dst64(COMMISSIONER).\
filter(lambda p:
[CONTENT_APPLICATION_DATA] == p.dtls.record.content_type
).\
must_next()
# 9. Commissioner receives the encrypted DTLS-ApplicationData record and sends a
# JOIN_FIN.rsp message in an encrypted DTLS-ApplicationData record in a single
# UDP datagram to Joiner
pkts.filter_wpan_src64(COMMISSIONER).\
filter(lambda p:
[CONTENT_APPLICATION_DATA] == p.dtls.record.content_type
).\
must_next()
# 10. Commissioner sends an encrypted JOIN_ENT.ntf message to Joiner
# 11. Joiner receives the encrypted JOIN_ENT.ntf message and sends an encrypted
# JOIN_ENT.ntf dummy response to Commissioner
# Check Step 8 ~ 11 in test()
# 12. Joiner sends an encrypted DTLS-Alert record with a code of 0 (close_notify)
# to Commissioner
pkts.filter_wpan_dst64(COMMISSIONER).\
filter(lambda p:
[CONTENT_ALERT] == p.dtls.record.content_type
).\
must_next()
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/pktverify/consts.py
+2
View File
@@ -183,6 +183,7 @@ NM_JOINER_UDP_PORT_TLV = 18
NM_DELAY_TIMER_TLV = 52
NM_CHANNEL_MASK_TLV = 53
NM_ENERGY_LIST_TLV = 57
NM_DISCOVERY_REQUEST_TLV = 128
NM_DISCOVERY_RESPONSE_TLV = 129
# Diagnostic TLVs
@@ -218,6 +219,7 @@ HANDSHAKE_FINISHED = 20
CONTENT_CHANGE_CIPHER_SPEC = 20
CONTENT_ALERT = 21
CONTENT_HANDSHAKE = 22
CONTENT_APPLICATION_DATA = 23
# Network Data TLVs
NWD_HAS_ROUTER_TLV = 0
tests/scripts/thread-cert/pktverify/layer_fields.py
+4
View File
@@ -278,6 +278,7 @@ _LAYER_FIELDS = {
# MLE
'mle.cmd': _auto,
'mle.sec_suite': _hex,
'mle.tlv.type': _list(_dec),
'mle.tlv.len': _list(_dec),
'mle.tlv.mode.receiver_on_idle': _auto,
@@ -568,6 +569,8 @@ _LAYER_FIELDS = {
'thread_meshcop.tlv.chan_mask_page': _auto,
'thread_meshcop.tlv.chan_mask_len': _auto,
'thread_meshcop.tlv.chan_mask_mask': _bytes,
'thread_meshcop.tlv.discovery_req_ver': _auto,
'thread_meshcop.tlv.discovery_rsp_ver': _auto,
'thread_meshcop.tlv.energy_list': _list(_auto),
'thread_meshcop.tlv.pan_id': _auto,
'thread_meshcop.tlv.xpan_id': _bytes,
@@ -583,6 +586,7 @@ _LAYER_FIELDS = {
'thread_meshcop.tlv.state': _auto,
'thread_meshcop.tlv.steering_data': _list(_auto),
'thread_meshcop.tlv.unknown': _bytes,
'thread_meshcop.tlv.udp_port': _list(_auto),
'thread_meshcop.tlv.ba_locator': _auto,
'thread_meshcop.tlv.active_tstamp': _auto,
'thread_meshcop.tlv.ipv6_addr': _list(_ipv6_addr),
tests/scripts/thread-cert/pktverify/packet_verifier.py
+5
View File
@@ -165,6 +165,11 @@ class PacketVerifier(object):
logging.info("add extra var: %s = %s", k, v)
self._vars[k] = v
for i, topo in self.test_info.topology.items():
name = self.test_info.get_node_name(i)
if topo['version']:
self._vars[name + '_VERSION'] = {'1.1': 2, '1.2': 3}[topo['version']]
def verify_attached(self, child: str, parent: str = None, child_type: str = 'FTD', pkts=None) -> VerifyResult:
"""
Verify that the device attaches to the Thread network.