mirror of
https://github.com/espressif/openthread.git
synced 2026-10-07 08:17:40 +00:00
[tests] improve CoAP TLV disambiguation in packet verification (#12514)
This commit implements a more robust mechanism to differentiate between overlapping TLV types in CoAP payloads, specifically for TLV type 8, which is used for both NM_STEERING_DATA_TLV (MeshCoP) and DG_IPV6_ADDRESS_LIST_TLV (Diagnostic). The logic previously relied solely on TLV length, which was fragile and caused 16-byte Steering Data TLVs to be misinterpreted. Key changes: - Enhanced CoapTlvParser.parse() and CoapLayer to accept and pass the CoapLayer instance as context during parsing. - Added a 'uri_path' property to CoapLayer for easier access to the CoAP URI path (recon). - Updated thread_coap_tlv_parse in verify_utils.py to use the URI path to correctly identify Diagnostic TLVs (if URI starts with '/d/') vs. MeshCoP or other Thread TLVs. - Updated verify_9_2_1.py monkey-patches to match the new parser signature.
This commit is contained in:
Jonathan Hui
@@ -41,14 +41,14 @@ from pktverify.null_field import nullField
|
||||
|
||||
|
||||
# Monkey-patch CoapTlvParser to parse MeshCoP TLVs in CoAP payload
|
||||
def meshcop_coap_tlv_parse(t, v):
|
||||
def meshcop_coap_tlv_parse(t, v, layer=None):
|
||||
kvs = []
|
||||
if t == consts.NM_COMMISSIONER_SESSION_ID_TLV:
|
||||
kvs.append(('commissioner_session_id', str(struct.unpack('>H', v)[0])))
|
||||
kvs.append(('comm_sess_id', str(struct.unpack('>H', v)[0])))
|
||||
elif t == consts.NM_STEERING_DATA_TLV:
|
||||
kvs.append(('steering_data', v.hex()))
|
||||
elif t == consts.NM_BORDER_AGENT_LOCATOR_TLV:
|
||||
kvs.append(('border_agent_locator', str(struct.unpack('>H', v)[0])))
|
||||
kvs.append(('border_agent_rloc16', hex(struct.unpack('>H', v)[0])))
|
||||
elif t == consts.TLV_REQUEST_TLV:
|
||||
kvs.append(('tlv_request', v.hex()))
|
||||
return kvs
|
||||
@@ -77,11 +77,11 @@ def verify(pv):
|
||||
from pktverify import layer_fields
|
||||
layer_fields._LAYER_FIELDS['coap.tlv.tlv_request'] = layer_fields._bytes
|
||||
|
||||
def new_parse(t, v):
|
||||
def new_parse(t, v, layer=None):
|
||||
if t in (consts.NM_COMMISSIONER_SESSION_ID_TLV, consts.NM_STEERING_DATA_TLV,
|
||||
consts.NM_BORDER_AGENT_LOCATOR_TLV, consts.TLV_REQUEST_TLV):
|
||||
return meshcop_coap_tlv_parse(t, v)
|
||||
return old_parse(t, v)
|
||||
return meshcop_coap_tlv_parse(t, v, layer=layer)
|
||||
return old_parse(t, v, layer=layer)
|
||||
|
||||
verify_utils.CoapTlvParser.parse = staticmethod(new_parse)
|
||||
|
||||
|
||||
@@ -49,11 +49,33 @@ from pktverify.bytes import Bytes
|
||||
|
||||
|
||||
# Monkey-patch CoapTlvParser to parse Thread TLVs in CoAP payload
|
||||
def thread_coap_tlv_parse(t, v):
|
||||
def thread_coap_tlv_parse(t, v, layer=None):
|
||||
kvs = []
|
||||
if t == consts.NL_TARGET_EID_TLV and len(v) == 16:
|
||||
|
||||
uri_path = None
|
||||
if layer is not None:
|
||||
uri_path = layer.uri_path
|
||||
|
||||
# If the URI starts with '/d/', it is likely a Diagnostic message.
|
||||
# Otherwise, we assume it's MeshCoP or other Thread TLVs.
|
||||
is_diag = uri_path is not None and uri_path.startswith('/d/')
|
||||
|
||||
# MeshCoP TLVs (often overlap with Diagnostic TLVs)
|
||||
if t == consts.NM_COMMISSIONER_SESSION_ID_TLV and len(v) == 2 and not is_diag:
|
||||
kvs.append(('comm_sess_id', str(struct.unpack('>H', v)[0])))
|
||||
elif t == consts.NM_STATE_TLV and len(v) == 1 and not is_diag:
|
||||
kvs.append(('state', str(v[0])))
|
||||
elif t == consts.NM_STEERING_DATA_TLV and not is_diag: # DG_IPV6_ADDRESS_LIST_TLV is 16*n
|
||||
kvs.append(('steering_data', v.hex()))
|
||||
elif t == consts.NM_BORDER_AGENT_LOCATOR_TLV and len(v) == 2 and not is_diag: # DG_MAC_COUNTERS_TLV is 4*n
|
||||
kvs.append(('border_agent_rloc16', hex(struct.unpack('>H', v)[0])))
|
||||
elif t == consts.NM_CHANNEL_TLV and len(v) == 3 and not is_diag: # DG_MAC_EXTENDED_ADDRESS_TLV is 8
|
||||
kvs.append(('channel', str(struct.unpack('>H', v[1:3])[0])))
|
||||
|
||||
# Other Thread TLVs
|
||||
elif t == consts.NL_TARGET_EID_TLV and len(v) == 16:
|
||||
kvs.append(('target_eid', str(Ipv6Addr(v))))
|
||||
elif t == consts.DG_MAC_EXTENDED_ADDRESS_TLV and len(v) == 8:
|
||||
elif t == consts.DG_MAC_EXTENDED_ADDRESS_TLV and len(v) == 8 and is_diag:
|
||||
kvs.append(('mac_addr', v.hex()))
|
||||
elif t == consts.DG_MAC_ADDRESS_TLV and len(v) == 2:
|
||||
kvs.append(('rloc16', hex(struct.unpack('>H', v)[0])))
|
||||
@@ -67,16 +89,18 @@ def thread_coap_tlv_parse(t, v):
|
||||
kvs.append(('status', str(v[0])))
|
||||
elif t == consts.NL_ROUTER_MASK_TLV:
|
||||
kvs.append(('router_mask', v.hex()))
|
||||
elif t == consts.DG_MAC_COUNTERS_TLV:
|
||||
elif t == consts.DG_MAC_COUNTERS_TLV and is_diag:
|
||||
# MAC counters are a list of 4-byte values
|
||||
for i in range(0, len(v), 4):
|
||||
val = struct.unpack('>I', v[i:i + 4])[0]
|
||||
kvs.append(('mac_counter', str(val)))
|
||||
if i + 4 <= len(v):
|
||||
val = struct.unpack('>I', v[i:i + 4])[0]
|
||||
kvs.append(('mac_counter', str(val)))
|
||||
elif t == consts.DG_MODE_TLV and len(v) == 1:
|
||||
kvs.append(('mode', hex(v[0])))
|
||||
elif t == consts.DG_IPV6_ADDRESS_LIST_TLV:
|
||||
elif t == consts.DG_IPV6_ADDRESS_LIST_TLV and is_diag:
|
||||
for i in range(0, len(v), 16):
|
||||
kvs.append(('ipv6_address', str(Ipv6Addr(v[i:i + 16]))))
|
||||
if i + 16 <= len(v):
|
||||
kvs.append(('ipv6_address', str(Ipv6Addr(v[i:i + 16]))))
|
||||
elif t == consts.DG_LEADER_DATA_TLV and len(v) == 8:
|
||||
# Leader data contains Partition ID (4), Weighting (1), Data Version (1), Stable Data Version (1), Leader Router ID (1)
|
||||
kvs.append(('partition_id', hex(struct.unpack('>I', v[0:4])[0])))
|
||||
@@ -85,7 +109,7 @@ def thread_coap_tlv_parse(t, v):
|
||||
# Route64 contains Router ID Sequence (1), and Router ID Mask (8), then link qualities
|
||||
kvs.append(('router_id_sequence', str(v[0])))
|
||||
kvs.append(('router_id_mask', v[1:9].hex()))
|
||||
elif t == consts.DG_CHILD_TABLE_TLV:
|
||||
elif t == consts.DG_CHILD_TABLE_TLV and is_diag:
|
||||
# Child table contains a list of child entries.
|
||||
# Each entry: [Timeout(5 bits), LQI(2 bits), Child ID(9 bits), Mode(8 bits)] -> total 3 bytes
|
||||
for i in range(0, len(v), 3):
|
||||
@@ -111,6 +135,11 @@ def apply_patches():
|
||||
layer_fields._LAYER_FIELDS['coap.tlv.child_id'] = layer_fields._list(layer_fields._auto)
|
||||
layer_fields._LAYER_FIELDS['coap.tlv.child_mode'] = layer_fields._list(layer_fields._auto)
|
||||
layer_fields._LAYER_FIELDS['coap.tlv.channel_pages'] = layer_fields._bytes
|
||||
layer_fields._LAYER_FIELDS['coap.tlv.steering_data'] = layer_fields._bytes
|
||||
layer_fields._LAYER_FIELDS['coap.tlv.comm_sess_id'] = layer_fields._auto
|
||||
layer_fields._LAYER_FIELDS['coap.tlv.state'] = layer_fields._auto
|
||||
layer_fields._LAYER_FIELDS['coap.tlv.border_agent_rloc16'] = layer_fields._auto
|
||||
layer_fields._LAYER_FIELDS['coap.tlv.channel'] = layer_fields._auto
|
||||
|
||||
def which_tshark_patch():
|
||||
default_path = '/tmp/thread-wireshark/tshark'
|
||||
|
||||
@@ -26,7 +26,7 @@
|
||||
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
|
||||
# POSSIBILITY OF SUCH DAMAGE.
|
||||
#
|
||||
from typing import Tuple, List
|
||||
from typing import Tuple, List, Optional
|
||||
|
||||
from pktverify.consts import COAP_CODE_POST, COAP_CODE_ACK
|
||||
from pktverify.layers import Layer
|
||||
@@ -35,7 +35,7 @@ from pktverify.layers import Layer
|
||||
class CoapTlvParser(object):
|
||||
|
||||
@staticmethod
|
||||
def parse(t, v: bytearray) -> List[Tuple[str, str]]:
|
||||
def parse(t, v: bytearray, layer: 'CoapLayer' = None) -> List[Tuple[str, str]]:
|
||||
assert isinstance(v, bytearray)
|
||||
return []
|
||||
|
||||
@@ -62,6 +62,16 @@ class CoapLayer(Layer):
|
||||
"""
|
||||
return self.code == COAP_CODE_ACK
|
||||
|
||||
@property
|
||||
def uri_path(self) -> Optional[str]:
|
||||
"""
|
||||
Returns the COAP URI path.
|
||||
"""
|
||||
try:
|
||||
return self.opt.uri_path_recon
|
||||
except AttributeError:
|
||||
return None
|
||||
|
||||
def __getattr__(self, name):
|
||||
super_attr = super().__getattr__(name)
|
||||
if name == 'tlv':
|
||||
@@ -74,7 +84,7 @@ class CoapLayer(Layer):
|
||||
|
||||
r = 0
|
||||
while True:
|
||||
t, tvs, r = self._parse_next_tlv(payload, r)
|
||||
t, tvs, r = self._parse_next_tlv(payload, r, layer=self)
|
||||
if t is None:
|
||||
break
|
||||
|
||||
@@ -85,7 +95,7 @@ class CoapLayer(Layer):
|
||||
self._add_field('coap.tlv.' + k, v)
|
||||
|
||||
@staticmethod
|
||||
def _parse_next_tlv(payload, read_pos) -> tuple:
|
||||
def _parse_next_tlv(payload, read_pos, layer=None) -> tuple:
|
||||
assert read_pos <= len(payload)
|
||||
if read_pos == len(payload):
|
||||
return None, None, read_pos
|
||||
@@ -93,5 +103,5 @@ class CoapLayer(Layer):
|
||||
t = payload[read_pos]
|
||||
len_ = payload[read_pos + 1]
|
||||
assert (len(payload) - read_pos - 2 >= len_)
|
||||
kvs = CoapTlvParser.parse(t, payload[read_pos + 2:read_pos + 2 + len_])
|
||||
kvs = CoapTlvParser.parse(t, payload[read_pos + 2:read_pos + 2 + len_], layer=layer)
|
||||
return t, kvs, read_pos + len_ + 2
|
||||
|
||||
Reference in New Issue
Block a user