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

[pktverify] misc enhancement (#5430)

Browse Source 
This commit includes several enhancements to pktverify
- Add more fields of thread_meshcop and thread_address layers
- Remove COAP TLV manual parsing for now since it's not correctly
  implemented
  - Fix 5.5.2 which had already been using p.coap.tlv.status,
    p.coap.tlv.rloc16, ...
This commit is contained in:
Simon Lin
2020-08-27 19:11:16 -07:00
committed by GitHub
parent c3845e72ff
commit 8d5e641668
4 changed files with 26 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/scripts/thread-cert/Cert_5_5_02_LeaderReboot.py
+4 -4
View File
@@ -171,12 +171,12 @@ class Cert_5_5_2_LeaderReboot(thread_cert.TestCase):
#Step 13: Leader send an Address Solicit Request
_lpkts.filter_coap_request(ADDR_SOL_URI).must_next().must_verify(
lambda p: p.coap.tlv.ext_mac_addr and p.coap.tlv.rloc16 is not nullField and p.coap.tlv.status != 0)
lambda p: p.thread_address.tlv.ext_mac_addr and p.thread_address.tlv.status != 0)
#Step 14: Router_1 send an Address Solicit Response
_rpkts.filter_coap_ack(
ADDR_SOL_URI).must_next().must_verify(lambda p: p.coap.tlv.router_mask_assigned and p.coap.tlv.rloc16 is
not nullField and p.coap.tlv.status == 0)
_rpkts.filter_coap_ack(ADDR_SOL_URI).must_next().must_verify(
lambda p: p.thread_address.tlv.router_mask_assigned and p.thread_address.tlv.rloc16 is not nullField and p.
thread_address.tlv.status == 0)
#Step 15: Leader Send a Multicast Link Request
_lpkts.filter_mle_cmd(MLE_LINK_REQUEST).must_next().must_verify(
tests/scripts/thread-cert/pktverify/coap.py
+2 -90
View File
@@ -26,10 +26,8 @@
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import struct
from typing import Tuple, List
from pktverify.addrs import ExtAddr, Ipv6Addr
from pktverify.consts import COAP_CODE_POST, COAP_CODE_ACK
from pktverify.layers import Layer
@@ -37,95 +35,9 @@ from pktverify.layers import Layer
class CoapTlvParser(object):
@staticmethod
def _parse_0(v: bytearray) -> List[Tuple[str, str]]:
"""parse Target EID TLV"""
return [('target_eid', CoapTlvParser._parse_ipv6_address(v))]
@staticmethod
def _parse_1(v: bytearray) -> List[Tuple[str, str]]:
"""parse MAC Extended Address TLV"""
return [('ext_mac_addr', CoapTlvParser._parse_ext_mac_addr(v))]
@staticmethod
def _parse_2(v: bytearray) -> List[Tuple[str, str]]:
"""parse RLOC16 TLV"""
return [('rloc16', CoapTlvParser._parse_uint16(v))]
@staticmethod
def _parse_3(v: bytearray) -> List[Tuple[str, str]]:
"""parse ML-EID TLV"""
return [('ml_eid', CoapTlvParser._parse_ext_mac_addr(v))]
@staticmethod
def _parse_4(v: bytearray) -> List[Tuple[str, str]]:
"""parse Status TLV"""
assert len(v) == 1
return [('status', hex(v[0]))]
@staticmethod
def _parse_6(v: bytearray) -> List[Tuple[str, str]]:
"""parse Time Since Last Transaction TLV"""
return [('last_transaction_time', CoapTlvParser._parse_uint32(v))]
@staticmethod
def _parse_7(v: bytearray) -> List[Tuple[str, str]]:
"""parse Router Mask TLV"""
assert len(v) == 9
return [
('router_mask_id_seq', hex(v[0])),
('router_mask_assigned', CoapTlvParser._parse_uint64(v[1:])),
]
@staticmethod
def _parse_10(v: bytearray) -> List[Tuple[str, str]]:
"""parse Thread Network Data TLV"""
# TODO: Thread Network Data can not be parsed by COAP TLVs yet
return []
@staticmethod
def _parse_12(v: bytearray) -> List[Tuple[str, str]]:
"""parse Network Name TLV"""
return [('net_name', CoapTlvParser._parse_utf8_str(v))]
@staticmethod
def _parse_uint16(v: bytearray) -> str:
assert len(v) == 2
return hex(v[0] * 256 + v[1])
@staticmethod
def _parse_uint32(v: bytearray) -> str:
assert len(v) == 4
return hex(struct.unpack(">I", v)[0])
@staticmethod
def _parse_uint64(v: bytearray) -> str:
assert len(v) == 8
return hex(struct.unpack(">Q", v)[0])
@staticmethod
def _parse_ipv6_address(s: bytearray):
assert len(s) == 16
a = Ipv6Addr(s)
return a.format_hextets()
@staticmethod
def _parse_utf8_str(v: bytearray) -> str:
return v.decode('utf-8')
@staticmethod
def parse(t, v: bytearray) -> str:
def parse(t, v: bytearray) -> List[Tuple[str, str]]:
assert isinstance(v, bytearray)
try:
parse_func = getattr(CoapTlvParser, f'_parse_{t}')
except AttributeError:
raise NotImplementedError(f"Please implement _parse_{t} for COAP TLV: type={t}")
return parse_func(v)
@staticmethod
def _parse_ext_mac_addr(v: bytearray) -> str:
assert len(v) == 8
return ExtAddr(v).format_octets()
return []
class CoapLayer(Layer):
tests/scripts/thread-cert/pktverify/layer_fields.py
+12 -2
View File
@@ -67,7 +67,7 @@ def _auto(v: Union[LayerFieldsContainer, LayerField]):
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)
return int(rv, 16)
try:
int(rv, 16)
@@ -486,8 +486,13 @@ _LAYER_FIELDS = {
'dtls.alert_message.desc': _auto,
# thread_address
'thread_address.tlv.len': _list(_auto),
'thread_address.tlv.type': _list(_auto),
'thread_address.tlv.status': _auto,
'thread_address.tlv.ext_mac_addr': _ext_addr,
'thread_address.tlv.router_mask_id_seq': _auto,
'thread_address.tlv.router_mask_assigned': _bytes,
'thread_address.tlv.rloc16': _hex,
# thread bl
'thread_bl.tlv.type': _list(_auto),
@@ -504,6 +509,7 @@ _LAYER_FIELDS = {
'thread_meshcop.tlv.type': _list(_auto),
'thread_meshcop.tlv.len8': _list(_auto),
'thread_meshcop.tlv.net_name': _str, # from thread_bl
'thread_meshcop.tlv.commissioner_id': _str,
'thread_meshcop.tlv.commissioner_sess_id': _auto, # from mle
"thread_meshcop.tlv.channel_page": _auto, # from ble
"thread_meshcop.tlv.channel": _auto, # from ble
@@ -522,7 +528,11 @@ _LAYER_FIELDS = {
'thread_meshcop.tlv.sec_policy_r': _auto,
'thread_meshcop.tlv.sec_policy_c': _auto,
'thread_meshcop.tlv.sec_policy_b': _auto,
'thread_meshcop.tlv.state': _auto,
'thread_meshcop.tlv.steering_data': _list(_auto),
'thread_meshcop.tlv.unknown': _bytes,
'thread_meshcop.tlv.ba_locator': _auto,
'thread_meshcop.tlv.active_tstamp': _auto,
# THREAD NWD
'thread_nwd.tlv.type': _list(_auto),
@@ -647,7 +657,7 @@ def check_layer_field_exists(packet, field_uri):
def _get_candidate_layers(packet, layer_name):
if layer_name == 'thread_meshcop':
candidate_layer_names = ['mle', 'coap', 'thread_bl']
candidate_layer_names = ['thread_meshcop', 'mle', 'coap', 'thread_bl']
elif layer_name == 'thread_nwd':
candidate_layer_names = ['mle', 'thread_address']
elif layer_name == 'wpan':
tests/scripts/thread-cert/pktverify/test_layer_fields.py
+8 -7
View File
@@ -30,7 +30,6 @@
#
import logging
import sys
import unittest
from pktverify import layer_fields
@@ -49,7 +48,9 @@ class TestLayerFields(unittest.TestCase):
logging.basicConfig(level=logging.DEBUG)
pkts = PcapReader.read("test.pcap")
for p in pkts:
for i, p in enumerate(pkts):
logging.info("check packet #%d", i + 1)
for layer_name in VALID_LAYER_NAMES:
if layer_name == 'lowpan': # we already checked 6lowpan
@@ -68,12 +69,13 @@ class TestLayerFields(unittest.TestCase):
except Exception:
layer.show()
raise
self._check_missing_fields(p, layer_name, getattr(p._packet, layer_name))
else:
if layer_name in REAL_LAYER_NAMES:
self.assertFalse(layer)
for layer in p._packet.layers:
self._check_missing_fields(p, layer.layer_name, layer)
def _test_coap(self, p):
coap = p.coap
self.assertIsInstance(coap.version, int)
@@ -306,7 +308,7 @@ class TestLayerFields(unittest.TestCase):
if f.startswith('_ws') or f.startswith('data'):
continue
logging.info('_check_missing_fields: %s = %r' % (f, _layer._all_fields[f]))
logging.info('_check_missing_fields in layer %s: %s = %r' % (layer_name, f, _layer._all_fields[f]))
if f in {
'', 'icmpv6.checksum.status', 'ip.ttl.lncb', 'wpan.aux_sec.key_source.bytes', 'wpan.src64.origin'
}:
@@ -349,6 +351,5 @@ class TestLayerFields(unittest.TestCase):
raise NotImplementedError(parser)
except Exception:
logging.info('checking [%s] %s=%r, %r, %r (%d)' %
(layer_name, f, v, v.get_default_value(), v.raw_value, len(v.fields)),
file=sys.stderr)
(layer_name, f, v, v.get_default_value(), v.raw_value, len(v.fields)))
raise