From 8d5e641668a0df9160500df3b173e9c0b6a0945c Mon Sep 17 00:00:00 2001 From: Simon Lin Date: Fri, 28 Aug 2020 10:11:16 +0800 Subject: [PATCH] [pktverify] misc enhancement (#5430) 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, ... --- .../thread-cert/Cert_5_5_02_LeaderReboot.py | 8 +- tests/scripts/thread-cert/pktverify/coap.py | 92 +------------------ .../thread-cert/pktverify/layer_fields.py | 14 ++- .../pktverify/test_layer_fields.py | 15 +-- 4 files changed, 26 insertions(+), 103 deletions(-) diff --git a/tests/scripts/thread-cert/Cert_5_5_02_LeaderReboot.py b/tests/scripts/thread-cert/Cert_5_5_02_LeaderReboot.py index f702f1f61..86edbebb8 100755 --- a/tests/scripts/thread-cert/Cert_5_5_02_LeaderReboot.py +++ b/tests/scripts/thread-cert/Cert_5_5_02_LeaderReboot.py @@ -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( diff --git a/tests/scripts/thread-cert/pktverify/coap.py b/tests/scripts/thread-cert/pktverify/coap.py index 17aa7d50d..1fc38692e 100644 --- a/tests/scripts/thread-cert/pktverify/coap.py +++ b/tests/scripts/thread-cert/pktverify/coap.py @@ -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): diff --git a/tests/scripts/thread-cert/pktverify/layer_fields.py b/tests/scripts/thread-cert/pktverify/layer_fields.py index 66a15eddc..c65cbe479 100644 --- a/tests/scripts/thread-cert/pktverify/layer_fields.py +++ b/tests/scripts/thread-cert/pktverify/layer_fields.py @@ -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': diff --git a/tests/scripts/thread-cert/pktverify/test_layer_fields.py b/tests/scripts/thread-cert/pktverify/test_layer_fields.py index d7ea80ec9..a598219ca 100644 --- a/tests/scripts/thread-cert/pktverify/test_layer_fields.py +++ b/tests/scripts/thread-cert/pktverify/test_layer_fields.py @@ -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