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

[thread-cert] refactor case 9.2.4 using pktverify (#5412)

Browse Source 
1. add timestamp parse and check
2. add leader aloc parse and check
This commit is contained in:
Jing Ma
2020-09-22 09:41:39 -07:00
committed by GitHub
parent b5cc12c92c
commit f787275eee
7 changed files with 221 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/scripts/thread-cert/Cert_9_2_04_ActiveDataset.py
+151 -15
View File
@@ -30,6 +30,8 @@
import unittest
import thread_cert
from pktverify.consts import MLE_CHILD_ID_RESPONSE, MGMT_ACTIVE_SET_URI, MGMT_ACTIVE_GET_URI
from pktverify.packet_verifier import PacketVerifier
COMMISSIONER = 1
LEADER = 2
@@ -40,22 +42,26 @@ class Cert_9_2_04_ActiveDataset(thread_cert.TestCase):
TOPOLOGY = {
COMMISSIONER: {
'name': 'COMMISSIONER',
'active_dataset': {
'timestamp': 10,
'panid': 0xface,
'master_key': '000102030405060708090a0b0c0d0e0f'
'master_key': '00112233445566778899aabbccddeeff'
},
'mode': 'rsdn',
'router_selection_jitter': 1
'router_selection_jitter': 1,
'allowlist': [LEADER]
},
LEADER: {
'name': 'LEADER',
'active_dataset': {
'timestamp': 10,
'panid': 0xface,
'master_key': '000102030405060708090a0b0c0d0e0f'
'master_key': '00112233445566778899aabbccddeeff'
},
'mode': 'rsdn',
'router_selection_jitter': 1
'router_selection_jitter': 1,
'allowlist': [COMMISSIONER]
},
}
@@ -71,21 +77,26 @@ class Cert_9_2_04_ActiveDataset(thread_cert.TestCase):
self.nodes[COMMISSIONER].commissioner_start()
self.simulator.go(3)
self.collect_rlocs()
self.collect_leader_aloc(LEADER)
# Step 2
self.nodes[COMMISSIONER].send_mgmt_active_set(
active_timestamp=101,
channel_mask=0x001fffe0,
channel_mask=0x7fff800,
extended_panid='000db70000000000',
network_name='GRL',
)
self.simulator.go(3)
self.assertEqual(self.nodes[LEADER].get_network_name(), 'GRL')
self.nodes[COMMISSIONER].send_mgmt_active_get()
self.simulator.go(3)
# Step 6
# Attempt to set Channel TLV
self.nodes[COMMISSIONER].send_mgmt_active_set(
active_timestamp=102,
channel=18,
channel_mask=0x001fffe0,
channel_mask=0x7fff800,
extended_panid='000db70000000001',
network_name='threadcert',
)
@@ -96,7 +107,7 @@ class Cert_9_2_04_ActiveDataset(thread_cert.TestCase):
# Attempt to set Mesh Local Prefix TLV
self.nodes[COMMISSIONER].send_mgmt_active_set(
active_timestamp=103,
channel_mask=0x001ffee0,
channel_mask=0x7fff800,
extended_panid='000db70000000000',
mesh_local='fd00:0db7::',
network_name='UL',
@@ -108,11 +119,11 @@ class Cert_9_2_04_ActiveDataset(thread_cert.TestCase):
# Attempt to set Network Master Key TLV
self.nodes[COMMISSIONER].send_mgmt_active_set(
active_timestamp=104,
channel_mask=0x001fffe0,
channel_mask=0x7fff800,
extended_panid='000db70000000000',
master_key='00112233445566778899aabbccddeeff',
master_key='ffeeddccbbaa99887766554433221100',
mesh_local='fd00:0db7::',
network_name='UL',
network_name='GRL',
)
self.simulator.go(3)
self.assertEqual(self.nodes[LEADER].get_network_name(), 'GRL')
@@ -121,7 +132,7 @@ class Cert_9_2_04_ActiveDataset(thread_cert.TestCase):
# Attempt to set PAN ID TLV
self.nodes[COMMISSIONER].send_mgmt_active_set(
active_timestamp=105,
channel_mask=0x001fffe0,
channel_mask=0x7fff800,
extended_panid='000db70000000000',
master_key='00112233445566778899aabbccddeeff',
mesh_local='fd00:0db7::',
@@ -135,7 +146,7 @@ class Cert_9_2_04_ActiveDataset(thread_cert.TestCase):
# Invalid Commissioner Session ID
self.nodes[COMMISSIONER].send_mgmt_active_set(
active_timestamp=106,
channel_mask=0x001fffe0,
channel_mask=0x7fff800,
extended_panid='000db70000000000',
network_name='UL',
binary='0b02abcd',
@@ -147,7 +158,7 @@ class Cert_9_2_04_ActiveDataset(thread_cert.TestCase):
# Old Active Timestamp
self.nodes[COMMISSIONER].send_mgmt_active_set(
active_timestamp=101,
channel_mask=0x001fffe0,
channel_mask=0x01fff800,
extended_panid='000db70000000000',
network_name='UL',
)
@@ -158,7 +169,7 @@ class Cert_9_2_04_ActiveDataset(thread_cert.TestCase):
# Unexpected Steering Data TLV
self.nodes[COMMISSIONER].send_mgmt_active_set(
active_timestamp=107,
channel_mask=0x001fffe0,
channel_mask=0x7fff800,
extended_panid='000db70000000000',
network_name='UL',
binary='0806113320440000',
@@ -170,7 +181,7 @@ class Cert_9_2_04_ActiveDataset(thread_cert.TestCase):
# Undefined TLV
self.nodes[COMMISSIONER].send_mgmt_active_set(
active_timestamp=108,
channel_mask=0x001fffe0,
channel_mask=0x7fff800,
extended_panid='000db70000000000',
network_name='GRL',
binary='8202aa55',
@@ -182,6 +193,131 @@ class Cert_9_2_04_ActiveDataset(thread_cert.TestCase):
for ipaddr in ipaddrs:
self.assertTrue(self.nodes[LEADER].ping(ipaddr))
def verify(self, pv):
pkts = pv.pkts
pv.summary.show()
LEADER = pv.vars['LEADER']
LEADER_RLOC = pv.vars['LEADER_RLOC']
LEADER_ALOC = pv.vars['LEADER_ALOC']
COMMISSIONER = pv.vars['COMMISSIONER']
COMMISSIONER_RLOC = pv.vars['COMMISSIONER_RLOC']
# Step 1: Ensure the topology is formed correctly
pkts.filter_wpan_src64(LEADER).filter_wpan_dst64(COMMISSIONER).filter_mle_cmd(
MLE_CHILD_ID_RESPONSE).must_next()
# Step 2: Commissioner sends MGMT_ACTIVE_SET.req to Leader RLOC or Anycast Locator
pkts.filter_wpan_src64(COMMISSIONER).filter_ipv6_2dsts(
LEADER_RLOC, LEADER_ALOC).filter_coap_request(MGMT_ACTIVE_SET_URI).must_next().must_verify(
lambda p: p.thread_meshcop.tlv.xpan_id == '000db70000000000' and p.thread_meshcop.tlv.net_name == 'GRL'
and p.thread_meshcop.tlv.chan_mask_mask == '001fffe0' and p.thread_meshcop.tlv.active_tstamp == 101)
# Step 3: Leader MUST send MGMT_ACTIVE_SET.rsp to the Commissioner
pkts.filter_wpan_src64(LEADER).filter_ipv6_dst(COMMISSIONER_RLOC).filter_coap_ack(
MGMT_ACTIVE_SET_URI).must_next().must_verify(lambda p: p.thread_meshcop.tlv.state == 1)
# Step 4: Commissioner sends MGMT_ACTIVE_GET.req to Leader
pkts.filter_wpan_src64(COMMISSIONER).filter_ipv6_2dsts(
LEADER_RLOC, LEADER_ALOC).filter_coap_request(MGMT_ACTIVE_GET_URI).must_next()
# Step 5: The Leader MUST send MGMT_ACTIVE_GET.rsp to the Commissioner
pkts.filter_wpan_src64(LEADER).filter_ipv6_dst(COMMISSIONER_RLOC).filter_coap_ack(
MGMT_ACTIVE_GET_URI
).must_next().must_verify(
lambda p: p.thread_meshcop.tlv.active_tstamp == 101 and p.thread_meshcop.tlv.xpan_id == '000db70000000000'
and p.thread_meshcop.tlv.net_name == 'GRL' and p.thread_meshcop.tlv.chan_mask_mask == '001fffe0')
# Step 6: Commissioner sends MGMT_ACTIVE_SET.req to Leader RLOC or Anycast Locator
pkts.filter_wpan_src64(COMMISSIONER).filter_ipv6_2dsts(
LEADER_RLOC, LEADER_ALOC).filter_coap_request(MGMT_ACTIVE_SET_URI).must_next().must_verify(
lambda p: p.thread_meshcop.tlv.active_tstamp == 102 and p.thread_meshcop.tlv.xpan_id ==
'000db70000000001' and p.thread_meshcop.tlv.net_name == 'threadcert' and p.thread_meshcop.tlv.
chan_mask_mask == '001fffe0' and p.thread_meshcop.tlv.channel == 18)
# Step 7: Leader MUST send MGMT_ACTIVE_SET.rsp to the Commissioner
pkts.filter_wpan_src64(LEADER).filter_ipv6_dst(COMMISSIONER_RLOC).filter_coap_ack(
MGMT_ACTIVE_SET_URI).must_next().must_verify(lambda p: p.thread_meshcop.tlv.state == -1)
# Step 8: Commissioner sends MGMT_ACTIVE_SET.req to Leader RLOC or Leader Anycast Locator
pkts.filter_wpan_src64(COMMISSIONER).filter_ipv6_2dsts(
LEADER_RLOC, LEADER_ALOC).filter_coap_request(MGMT_ACTIVE_SET_URI).must_next().must_verify(
lambda p: p.thread_meshcop.tlv.active_tstamp == 103 and p.thread_meshcop.tlv.xpan_id ==
'000db70000000000' and p.thread_meshcop.tlv.net_name == 'UL' and p.thread_meshcop.tlv.chan_mask_mask ==
'001fffe0' and p.thread_meshcop.tlv.ml_prefix == 'fd000db700000000')
# Step 9: Leader MUST send MGMT_ACTIVE_SET.rsp to the Commissioner
pkts.filter_wpan_src64(LEADER).filter_ipv6_dst(COMMISSIONER_RLOC).filter_coap_ack(
MGMT_ACTIVE_SET_URI).must_next().must_verify(lambda p: p.thread_meshcop.tlv.state == -1)
# Step 10: Commissioner sends MGMT_ACTIVE_SET.req to Leader RLOC or Leader Anycast Locator
pkts.filter_wpan_src64(COMMISSIONER).filter_ipv6_2dsts(
LEADER_RLOC, LEADER_ALOC).filter_coap_request(MGMT_ACTIVE_SET_URI).must_next().must_verify(
lambda p: p.thread_meshcop.tlv.active_tstamp == 104 and p.thread_meshcop.tlv.xpan_id ==
'000db70000000000' and p.thread_meshcop.tlv.net_name == 'GRL' and p.thread_meshcop.tlv.master_key ==
'ffeeddccbbaa99887766554433221100' and p.thread_meshcop.tlv.chan_mask_mask == '001fffe0' and p.
thread_meshcop.tlv.ml_prefix == 'fd000db700000000')
# Step 11: Leader MUST send MGMT_ACTIVE_SET.rsp to the Commissioner
pkts.filter_wpan_src64(LEADER).filter_ipv6_dst(COMMISSIONER_RLOC).filter_coap_ack(
MGMT_ACTIVE_SET_URI).must_next().must_verify(lambda p: p.thread_meshcop.tlv.state == -1)
# Step 12: Commissioner sends MGMT_ACTIVE_SET.req to Leader RLOC or Leader Anycast Locator
pkts.filter_wpan_src64(COMMISSIONER).filter_ipv6_2dsts(
LEADER_RLOC, LEADER_ALOC).filter_coap_request(MGMT_ACTIVE_SET_URI).must_next().must_verify(
lambda p: p.thread_meshcop.tlv.active_tstamp == 105 and p.thread_meshcop.tlv.xpan_id ==
'000db70000000000' and p.thread_meshcop.tlv.net_name == 'UL' and p.thread_meshcop.tlv.master_key ==
'00112233445566778899aabbccddeeff' and p.thread_meshcop.tlv.pan_id == 0xafce and p.thread_meshcop.tlv.
chan_mask_mask == '001fffe0' and p.thread_meshcop.tlv.ml_prefix == 'fd000db700000000')
# Step 13: Leader MUST send MGMT_ACTIVE_SET.rsp to the Commissioner
pkts.filter_wpan_src64(LEADER).filter_ipv6_dst(COMMISSIONER_RLOC).filter_coap_ack(
MGMT_ACTIVE_SET_URI).must_next().must_verify(lambda p: p.thread_meshcop.tlv.state == -1)
# Step 14: Commissioner sends MGMT_ACTIVE_SET.req to Leader RLOC or Leader Anycast Locator
pkts.filter_wpan_src64(COMMISSIONER).filter_ipv6_2dsts(
LEADER_RLOC, LEADER_ALOC).filter_coap_request(MGMT_ACTIVE_SET_URI).must_next().must_verify(
lambda p: p.thread_meshcop.tlv.active_tstamp == 106 and p.thread_meshcop.tlv.xpan_id ==
'000db70000000000' and p.thread_meshcop.tlv.net_name == 'UL' and p.thread_meshcop.tlv.
commissioner_sess_id == 0xabcd and p.thread_meshcop.tlv.chan_mask_mask == '001fffe0')
# Step 15: Leader MUST send MGMT_ACTIVE_SET.rsp to the Commissioner
pkts.filter_wpan_src64(LEADER).filter_ipv6_dst(COMMISSIONER_RLOC).filter_coap_ack(
MGMT_ACTIVE_SET_URI).must_next().must_verify(lambda p: p.thread_meshcop.tlv.state == -1)
# Step 16: Commissioner sends MGMT_ACTIVE_SET.req to Leader RLOC or Leader Anycast Locator
pkts.filter_wpan_src64(COMMISSIONER).filter_ipv6_2dsts(
LEADER_RLOC, LEADER_ALOC
).filter_coap_request(MGMT_ACTIVE_SET_URI).must_next().must_verify(
lambda p: p.thread_meshcop.tlv.active_tstamp == 101 and p.thread_meshcop.tlv.xpan_id == '000db70000000000'
and p.thread_meshcop.tlv.net_name == 'UL' and p.thread_meshcop.tlv.chan_mask_mask == '001fff80')
# Step 17: Leader MUST send MGMT_ACTIVE_SET.rsp to the Commissioner
pkts.filter_wpan_src64(LEADER).filter_ipv6_dst(COMMISSIONER_RLOC).filter_coap_ack(
MGMT_ACTIVE_SET_URI).must_next().must_verify(lambda p: p.thread_meshcop.tlv.state == -1)
# Step 18: Commissioner sends MGMT_ACTIVE_SET.req to Leader RLOC or Leader Anycast Locator
pkts.filter_wpan_src64(COMMISSIONER).filter_ipv6_2dsts(
LEADER_RLOC, LEADER_ALOC).filter_coap_request(MGMT_ACTIVE_SET_URI).must_next().must_verify(
lambda p: p.thread_meshcop.tlv.active_tstamp == 107 and p.thread_meshcop.tlv.xpan_id ==
'000db70000000000' and p.thread_meshcop.tlv.net_name == 'UL' and p.thread_meshcop.tlv.steering_data ==
[0x113320440000] and p.thread_meshcop.tlv.chan_mask_mask == '001fffe0')
# Step 19: Leader MUST send MGMT_ACTIVE_SET.rsp to the Commissioner
pkts.filter_wpan_src64(LEADER).filter_ipv6_dst(COMMISSIONER_RLOC).filter_coap_ack(
MGMT_ACTIVE_SET_URI).must_next().must_verify(lambda p: p.thread_meshcop.tlv.state == 1)
# Step 20: Commissioner sends MGMT_ACTIVE_SET.req to Leader RLOC or Leader Anycast Locator
pkts.filter_wpan_src64(COMMISSIONER).filter_ipv6_2dsts(
LEADER_RLOC, LEADER_ALOC).filter_coap_request(MGMT_ACTIVE_SET_URI).must_next().must_verify(
lambda p: p.thread_meshcop.tlv.active_tstamp == 108 and p.thread_meshcop.tlv.xpan_id ==
'000db70000000000' and p.thread_meshcop.tlv.net_name == 'GRL' and p.thread_meshcop.tlv.unknown ==
'aa55' and p.thread_meshcop.tlv.chan_mask_mask == '001fffe0')
# Step 21: Leader MUST send MGMT_ACTIVE_SET.rsp to the Commissioner
pkts.filter_wpan_src64(LEADER).filter_ipv6_dst(COMMISSIONER_RLOC).filter_coap_ack(
MGMT_ACTIVE_SET_URI).must_next().must_verify(lambda p: p.thread_meshcop.tlv.state == 1)
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/node.py
+43
View File
@@ -1248,6 +1248,49 @@ class Node:
self.send_command(cmd)
self._expect('Done')
def send_mgmt_active_get(
self,
active_timestamp=None,
channel=None,
channel_mask=None,
extended_panid=None,
panid=None,
master_key=None,
mesh_local=None,
network_name=None,
binary=None,
):
cmd = 'dataset mgmtgetcommand active '
if active_timestamp is not None:
cmd += 'activetimestamp %d ' % active_timestamp
if channel is not None:
cmd += 'channel %d ' % channel
if channel_mask is not None:
cmd += 'channelmask %d ' % channel_mask
if extended_panid is not None:
cmd += 'extpanid %s ' % extended_panid
if panid is not None:
cmd += 'panid %d ' % panid
if master_key is not None:
cmd += 'masterkey %s ' % master_key
if mesh_local is not None:
cmd += 'localprefix %s ' % mesh_local
if network_name is not None:
cmd += 'networkname %s ' % self._escape_escapable(network_name)
if binary is not None:
cmd += 'binary %s ' % binary
self.send_command(cmd)
def send_mgmt_pending_set(
self,
pending_timestamp=None,
tests/scripts/thread-cert/pktverify/layer_fields.py
+11 -5
View File
@@ -27,6 +27,7 @@
# POSSIBILITY OF SUCH DAMAGE.
#
import sys
import time, datetime
from typing import Any, Union
from pyshark.packet.fields import LayerFieldsContainer, LayerField
@@ -64,10 +65,15 @@ def _auto(v: Union[LayerFieldsContainer, LayerField]):
if ':' in dv and '::' not in dv and dv.replace(':', '') == rv: # '88:00', '8800'
return int(rv, 16)
if dv.endswith(' CST'):
# e.x. 'Jan 1, 1970 08:00:00.000000000 CST', '0000000000000000'
# todo: check if the time is valid
return int(rv, 16)
# timestamp: 'Jan 1, 1970 08:00:00.000000000 CST', '0000000000000000'
# convert to seconds from 1970, ignore the nanosecond for now since
# there are integer seconds applied in the test cases
try:
time_str = datetime.datetime.strptime(dv, "%b %d, %Y %H:%M:%S.%f000 %Z")
time_in_sec = time.mktime(time_str.utctimetuple())
return int(time_in_sec)
except (ValueError, TypeError):
pass
try:
int(rv, 16)
@@ -520,7 +526,7 @@ _LAYER_FIELDS = {
"thread_meshcop.tlv.chan_mask": _str, # from ble
'thread_meshcop.tlv.chan_mask_page': _auto,
'thread_meshcop.tlv.chan_mask_len': _auto,
'thread_meshcop.tlv.chan_mask_mask': _auto,
'thread_meshcop.tlv.chan_mask_mask': _bytes,
'thread_meshcop.tlv.pan_id': _auto,
'thread_meshcop.tlv.xpan_id': _bytes,
'thread_meshcop.tlv.ml_prefix': _bytes,
tests/scripts/thread-cert/pktverify/packet_filter.py
+5
View File
@@ -443,6 +443,11 @@ class PacketFilter(object):
assert isinstance(addr, (str, Ipv6Addr))
return self.filter(lambda p: p.ipv6.dst == addr, **kwargs)
def filter_ipv6_2dsts(self, addr1, addr2, **kwargs):
assert isinstance(addr1, (str, Ipv6Addr))
assert isinstance(addr2, (str, Ipv6Addr))
return self.filter(lambda p: p.ipv6.dst == addr1 or p.ipv6.dst == addr2, **kwargs)
def filter_ipv6_src_dst(self, src_addr, dst_addr, **kwargs):
assert isinstance(src_addr, (str, Ipv6Addr))
assert isinstance(dst_addr, (str, Ipv6Addr))
tests/scripts/thread-cert/pktverify/packet_verifier.py
+3
View File
@@ -148,6 +148,9 @@ class PacketVerifier(object):
key = self.test_info.get_node_name(i) + '_RLOC'
self._vars[key] = rloc
if self.test_info.leader_aloc:
self._vars['LEADER_ALOC'] = self.test_info.leader_aloc
for k, v in self.test_info.extra_vars.items():
assert k not in self._vars, k
logging.info("add extra var: %s = %s", k, v)
tests/scripts/thread-cert/pktverify/test_info.py
+1
View File
@@ -54,6 +54,7 @@ class TestInfo(object):
self.rlocs = {int(k): Ipv6Addr(v) for k, v in test_info.get('rlocs', {}).items()}
self.rloc16s = self._convert_hex_values(self._convert_keys_to_ints(test_info.get('rloc16s', {})))
self.extra_vars = test_info.get('extra_vars', {})
self.leader_aloc = Ipv6Addr(test_info.get('leader_aloc')) if 'leader_aloc' in test_info else ''
def __str__(self):
macs = dict(self.extaddrs)
tests/scripts/thread-cert/thread_cert.py
+7
View File
@@ -269,6 +269,13 @@ class TestCase(NcpSupportMixin, unittest.TestCase):
for i, node in self.nodes.items():
test_info['rlocs'][i] = node.get_rloc()
def collect_leader_aloc(self, node):
if not self._do_packet_verification:
return
test_info = self._test_info
test_info['leader_aloc'] = self.nodes[node].get_addr_leader_aloc()
def collect_extra_vars(self, **vars):
if not self._do_packet_verification:
return