[scripts] add packet verification framework (#4428)

This commit introduces the packet verification (PV) framework for
certification tests.

- Add packet verification framework code

- Implement packet verification for cert 5.1.7 as a minimal
  example. There will be more 1.1/1.2 tests with PV submitted in the
  future.

- Download pre-built thread-wireshark binaries from
  openthread/wireshark/releases for packet dissecting (used by
  pyshark)

- Added a Github Action job for Packet Verification
This commit is contained in:
Simon Lin
2020-08-11 09:26:28 -07:00
committed by GitHub
parent 6dddd555e4
commit ecd9436427
29 changed files with 4132 additions and 16 deletions
@@ -31,6 +31,8 @@ import unittest
import config
import thread_cert
from pktverify.consts import MLE_PARENT_RESPONSE, MLE_CHILD_ID_RESPONSE
from pktverify.packet_verifier import PacketVerifier
LEADER = 1
ROUTER = 2
@@ -40,11 +42,13 @@ SED1 = 7
class Cert_5_1_07_MaxChildCount(thread_cert.TestCase):
TOPOLOGY = {
LEADER: {
'name': 'LEADER',
'mode': 'rsdn',
'panid': 0xface,
'whitelist': [ROUTER]
},
ROUTER: {
'name': 'ROUTER',
'max_children': 10,
'mode': 'rsdn',
'panid': 0xface,
@@ -52,6 +56,7 @@ class Cert_5_1_07_MaxChildCount(thread_cert.TestCase):
'whitelist': [LEADER, 3, 4, 5, 6, SED1, 8, 9, 10, 11, 12]
},
3: {
'name': 'MED1',
'is_mtd': True,
'mode': 'rsn',
'panid': 0xface,
@@ -59,6 +64,7 @@ class Cert_5_1_07_MaxChildCount(thread_cert.TestCase):
'whitelist': [ROUTER]
},
4: {
'name': 'MED2',
'is_mtd': True,
'mode': 'rsn',
'panid': 0xface,
@@ -66,6 +72,7 @@ class Cert_5_1_07_MaxChildCount(thread_cert.TestCase):
'whitelist': [ROUTER]
},
5: {
'name': 'MED3',
'is_mtd': True,
'mode': 'rsn',
'panid': 0xface,
@@ -73,6 +80,7 @@ class Cert_5_1_07_MaxChildCount(thread_cert.TestCase):
'whitelist': [ROUTER]
},
6: {
'name': 'MED4',
'is_mtd': True,
'mode': 'rsn',
'panid': 0xface,
@@ -80,6 +88,7 @@ class Cert_5_1_07_MaxChildCount(thread_cert.TestCase):
'whitelist': [ROUTER]
},
SED1: {
'name': 'SED1',
'is_mtd': True,
'mode': 's',
'panid': 0xface,
@@ -87,6 +96,7 @@ class Cert_5_1_07_MaxChildCount(thread_cert.TestCase):
'whitelist': [ROUTER]
},
8: {
'name': 'SED2',
'is_mtd': True,
'mode': 's',
'panid': 0xface,
@@ -94,6 +104,7 @@ class Cert_5_1_07_MaxChildCount(thread_cert.TestCase):
'whitelist': [ROUTER]
},
9: {
'name': 'SED3',
'is_mtd': True,
'mode': 's',
'panid': 0xface,
@@ -101,6 +112,7 @@ class Cert_5_1_07_MaxChildCount(thread_cert.TestCase):
'whitelist': [ROUTER]
},
10: {
'name': 'SED4',
'is_mtd': True,
'mode': 's',
'panid': 0xface,
@@ -108,6 +120,7 @@ class Cert_5_1_07_MaxChildCount(thread_cert.TestCase):
'whitelist': [ROUTER]
},
11: {
'name': 'SED5',
'is_mtd': True,
'mode': 's',
'panid': 0xface,
@@ -115,6 +128,7 @@ class Cert_5_1_07_MaxChildCount(thread_cert.TestCase):
'whitelist': [ROUTER]
},
12: {
'name': 'SED6',
'is_mtd': True,
'mode': 's',
'panid': 0xface,
@@ -137,6 +151,9 @@ class Cert_5_1_07_MaxChildCount(thread_cert.TestCase):
self.simulator.go(7)
self.assertEqual(self.nodes[i].get_state(), 'child')
self.collect_rloc16s()
self.collect_ipaddrs()
ipaddrs = self.nodes[SED1].get_addrs()
for addr in ipaddrs:
if addr[0:4] != 'fe80' and 'ff:fe00' not in addr:
@@ -150,6 +167,44 @@ class Cert_5_1_07_MaxChildCount(thread_cert.TestCase):
self.assertTrue(self.nodes[LEADER].ping(addr, size=106))
break
def verify(self, pv: PacketVerifier):
pkts = pv.pkts
pv.summary.show()
ROUTER = pv.vars['ROUTER']
router_pkts = pkts.filter_wpan_src64(ROUTER)
# Step 1: The DUT MUST send properly formatted MLE Parent Response
# and MLE Child ID Response to each child.
for i in range(1, 7):
_pkts = router_pkts.copy().filter_wpan_dst64(pv.vars['SED%d' % i])
_pkts.filter_mle_cmd(MLE_PARENT_RESPONSE).must_next()
_pkts.filter_mle_cmd(MLE_CHILD_ID_RESPONSE).must_next()
for i in range(1, 5):
_pkts = router_pkts.copy().filter_wpan_dst64(pv.vars['MED%d' % i])
_pkts.filter_mle_cmd(MLE_PARENT_RESPONSE).must_next()
_pkts.filter_mle_cmd(MLE_CHILD_ID_RESPONSE).must_next()
# Step 2:The DUT MUST properly forward ICMPv6 Echo Requests to all MED children
# The DUT MUST properly forward ICMPv6 Echo Replies to the Leader
leader_rloc16 = pv.vars['LEADER_RLOC16']
for i in range(1, 5):
rloc16 = pv.vars['MED%d_RLOC16' % i]
_pkts = router_pkts.copy()
p = _pkts.filter('wpan.dst16 == {rloc16}', rloc16=rloc16).filter_ping_request().must_next()
_pkts.filter('wpan.dst16 == {rloc16}',
rloc16=leader_rloc16).filter_ping_reply(identifier=p.icmpv6.echo.identifier).must_next()
# Step 3: The DUT MUST properly forward ICMPv6 Echo Requests to all SED children
# The DUT MUST properly forward ICMPv6 Echo Replies to the Leader
for i in range(1, 7):
rloc16 = pv.vars['SED%d_RLOC16' % i]
_pkts = router_pkts.copy()
p = _pkts.filter('wpan.dst16 == {rloc16}', rloc16=rloc16).filter_ping_request().must_next()
_pkts.filter('wpan.dst16 == {rloc16}',
rloc16=leader_rloc16).filter_ping_reply(identifier=p.icmpv6.echo.identifier).must_next()
if __name__ == '__main__':
unittest.main()
+20
View File
@@ -161,6 +161,26 @@ EXTRA_DIST = \
test_service.py \
thread_cert.py \
tlvs_parsing.py \
thread_cert.py \
pktverify/__init__.py \
pktverify/addrs.py \
pktverify/bytes.py \
pktverify/coap.py \
pktverify/consts.py \
pktverify/decorators.py \
pktverify/errors.py \
pktverify/layer_fields.py \
pktverify/layer_fields_container.py \
pktverify/layers.py \
pktverify/null_field.py \
pktverify/packet.py \
pktverify/packet_filter.py \
pktverify/packet_verifier.py \
pktverify/pcap_reader.py \
pktverify/summary.py \
pktverify/test_info.py \
pktverify/utils.py \
pktverify/verify_result.py \
$(NULL)
check_PROGRAMS = \
+2 -1
View File
@@ -43,8 +43,9 @@ import binascii
class Node:
def __init__(self, nodeid, is_mtd=False, simulator=None, version=None, is_bbr=False):
def __init__(self, nodeid, is_mtd=False, simulator=None, name=None, version=None, is_bbr=False):
self.nodeid = nodeid
self.name = name or ('Node%d' % nodeid)
self.verbose = int(float(os.getenv('VERBOSE', 0)))
self.node_type = os.getenv('NODE_TYPE', 'sim')
self.env_version = os.getenv('THREAD_VERSION', '1.1')
@@ -0,0 +1,33 @@
#!/usr/bin/env python3
#
# Copyright (c) 2019, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import sys
# make sure we are using Python3.6+
assert sys.version_info.major == 3 and sys.version_info.minor >= 6
@@ -0,0 +1,189 @@
#!/usr/bin/env python3
#
# Copyright (c) 2019, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import ipaddress
from typing import Union
from pktverify.bytes import Bytes
class EthAddr(Bytes):
"""
Represents an Ethernet address.
"""
def __init__(self, addr: Union[str, bytearray, 'Bytes']):
super().__init__(addr)
if len(self) not in (6, 8):
raise ValueError((addr, self))
def __repr__(self):
return '%s(%r)' % (self.__class__.__name__, self.format_octets())
def __str__(self):
return self.format_octets()
class ExtAddr(Bytes):
"""
Represents an WPAN Extended address.
"""
def __init__(self, addr: Union[str, bytearray, 'Bytes']):
super().__init__(addr)
if len(self) != 8:
raise ValueError((addr, self))
def __repr__(self):
return '%s(%r)' % (self.__class__.__name__, self.format_octets())
def __str__(self):
return self.format_octets()
class Ipv6Addr(Bytes):
"""
Represents an Ip6 address.
"""
def __init__(self, addr: Union[str, bytearray, 'Bytes']):
if isinstance(addr, str):
# try to parse compacted ipv6 address
try:
addr = Ipv6Addr._expand(addr)
except ipaddress.AddressValueError:
pass
super().__init__(addr)
if len(self) != 16:
raise ValueError((addr, self))
self._addr = ipaddress.IPv6Address(self)
def __repr__(self):
return '%s(%r)' % (self.__class__.__name__, self.format_hextets())
def __str__(self):
return self.format_hextets()
@staticmethod
def _expand(addr) -> str:
assert isinstance(addr, str)
a = ipaddress.IPv6Address(addr)
return a.exploded
@property
def is_global(self) -> bool:
"""
Returns if the Ip6 address is global.
"""
if self._addr.is_global:
return True
if self._addr.is_link_local or self._addr.is_multicast or self._addr.is_loopback:
return False
return True
@property
def is_dua(self) -> bool:
"""
Returns if the Ip6 address is Domain Unicast Address.
"""
from pktverify import consts
return self.startswith(consts.DOMAIN_PREFIX)
@property
def is_backbone(self) -> bool:
"""
Returns if the Ip6 address is Backbone address.
"""
from pktverify import consts
return self.startswith(consts.BACKBONE_IPV6_PREFIX)
@property
def is_link_local(self) -> bool:
"""
Returns if the Ip6 address is link local.
"""
return self._addr.is_link_local
@property
def is_multicast(self) -> bool:
"""
Returns if the Ip6 address is multicast.
"""
return self._addr.is_multicast
@property
def is_mleid(self) -> bool:
"""
Returns if the Ip6 address is ML-EID.
"""
from pktverify.consts import DEFAULT_MESH_LOCAL_PREFIX
return self.startswith(DEFAULT_MESH_LOCAL_PREFIX)
if __name__ == '__main__':
a = EthAddr("010203040506")
assert a == EthAddr("01:02:03:04:05:06")
assert a == EthAddr("0102:0304:0506")
assert eval(repr(a)) == a
assert a == str(a)
assert str(a) == a
print(a, repr(a))
assert isinstance(a[:], Bytes)
assert a[:3] == "010203"
assert a[3:] == "040506"
a = ExtAddr("0102030405060708")
assert a == ExtAddr("01:02:03:04:05:06:07:08")
assert a == ExtAddr("0102:0304:0506:0708")
assert eval(repr(a)) == a
assert a == str(a)
assert str(a) == a
print(a, repr(a))
assert isinstance(a[:], Bytes)
assert a[:4] == "01020304"
assert a[4:] == "05060708"
a = Ipv6Addr("00112233445566778899aabbccddeeff")
assert a == Ipv6Addr("00:11:22:33:44:55:66:77:88:99:aa:bb:cc:dd:ee:ff")
assert a == Ipv6Addr("0011:2233:4455:6677:8899:aabb:ccdd:eeff")
assert eval(repr(a)) == a
assert a == str(a)
assert str(a) == a
print(a, repr(a))
assert isinstance(a[:], Bytes)
assert a[:4] == "00112233"
assert a[-4:] == "ccddeeff"
assert Ipv6Addr("fd00:db8::ff:fe00:8001") == Ipv6Addr("fd00:0db8:0000:0000:0000:00ff:fe00:8001")
print(Ipv6Addr("fdde:ad00:beef:0:9d87:85f0:3358:3fff"))
@@ -0,0 +1,159 @@
#!/usr/bin/env python3
#
# Copyright (c) 2019, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import sys
from typing import Union
class Bytes(bytearray):
"""Bytes represents a byte array which is able to handle strings of flexible formats"""
def __init__(self, s: Union[str, bytearray, 'Bytes']):
if isinstance(s, str):
try:
s = Bytes._parse_compact(s)
except ValueError:
try:
s = Bytes._parse_octets(s)
except ValueError:
s = Bytes._parse_hextets(s)
super().__init__(s)
def __hash__(self):
return hash(bytes(self))
def __repr__(self):
return '%s(%r)' % (self.__class__.__name__, self.format_compact())
def format_compact(self) -> str:
"""
Converts the Bytes to a compact string (without ":").
"""
return ''.join('%02x' % b for b in self)
def format_octets(self) -> str:
"""
Converts the Bytes to a string of octets separated by ":".
"""
return ':'.join('%02x' % b for b in self)
def format_hextets(self) -> str:
"""
Converts the Bytes to a string of hextets separated by ":"
"""
assert len(self) % 2 == 0, self.format_octets()
return ':'.join('%04x' % (self[i] * 256 + self[i + 1]) for i in range(0, len(self), 2))
__str__ = format_octets
@staticmethod
def _parse_compact(s: str) -> bytearray:
try:
assert len(s) % 2 == 0
return bytearray(int(s[i:i + 2], 16) for i in range(0, len(s), 2))
except Exception:
raise ValueError(s)
@staticmethod
def _parse_octets(s: str) -> bytearray:
try:
assert len(s) % 3 == 2 or not s
if not s:
return bytearray(b"")
return bytearray(int(x, 16) for x in s.split(':'))
except Exception:
raise ValueError(s)
@staticmethod
def _parse_hextets(s) -> bytearray:
try:
assert len(s) % 5 == 4 or not s
if not s:
return bytearray(b"")
return bytearray(int(x[i:i + 2], 16) for x in s.split(':') for i in (0, 2))
except Exception:
raise ValueError(s)
def __getitem__(self, item) -> Union['Bytes', int]:
"""
Get self[item].
:param item: index or slice to retrieve
:return: the byte value at specified index or sub `Bytes` if item is slice
"""
x = super().__getitem__(item)
if isinstance(x, bytearray):
return Bytes(x)
else:
return x
def __eq__(self, other: Union[str, 'Bytes']):
"""
Check if bytes is equal to other.
"""
if other is None:
return False
elif not isinstance(other, Bytes):
other = self.__class__(other)
eq = super().__eq__(other)
print("[%r %s %r]" % (self, "==" if eq else "!=", other), file=sys.stderr)
return eq
if __name__ == '__main__':
# some simple tests
x = Bytes(b"\x01\x02\x03\x04")
assert eval(repr(x)) == x, repr(x) # representation of Bytes should be able to be evaluated back
assert x == str(x), (x, str(x))
assert x.format_compact() == "01020304", x.format_compact()
assert x.format_octets() == "01:02:03:04", x.format_octets()
assert x.format_hextets() == "0102:0304", x.format_hextets()
assert Bytes._parse_compact("") == Bytes(b"")
assert Bytes._parse_compact('01020304') == x
assert Bytes._parse_octets("") == Bytes(b"")
assert Bytes._parse_octets('01:02:03:04') == x
assert Bytes._parse_hextets("") == Bytes(b"")
assert Bytes._parse_hextets('0102:0304') == x
assert isinstance(x[:2], Bytes)
assert isinstance(x[-2:], Bytes)
assert x[:2] == Bytes(b'\x01\x02')
assert x[-2:] == Bytes(b'\x03\x04')
# should also parse string formats
assert Bytes("01020304") == Bytes(b"\x01\x02\x03\x04")
assert Bytes("01:02:03:04") == Bytes(b"\x01\x02\x03\x04")
assert Bytes("0102:0304") == Bytes(b"\x01\x02\x03\x04")
+185
View File
@@ -0,0 +1,185 @@
#!/usr/bin/env python3
#
# Copyright (c) 2019, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# 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
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:
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()
class CoapLayer(Layer):
"""
Represents a COAP layer.
"""
def __init__(self, packet, layer_name):
super().__init__(packet, layer_name)
@property
def is_post(self) -> bool:
"""
Returns if the COAP layer is using code POST.
"""
return self.code == COAP_CODE_POST
@property
def is_ack(self) -> bool:
"""
Returns if the COAP layer is using code ACK.
"""
return self.code == COAP_CODE_ACK
def __getattr__(self, name):
super_attr = super().__getattr__(name)
if name == 'tlv':
self._parse_coap_payload()
return super_attr
def _parse_coap_payload(self):
payload = self.payload
r = 0
while True:
t, tvs, r = self._parse_next_tlv(payload, r)
if t is None:
break
self._add_field('coap.tlv.type', hex(t))
for k, v in tvs:
assert isinstance(k, str), (t, k, v)
assert isinstance(v, str), (t, k, v)
self._add_field('coap.tlv.' + k, v)
@staticmethod
def _parse_next_tlv(payload, read_pos) -> tuple:
assert read_pos <= len(payload)
if read_pos == len(payload):
return None, None, read_pos
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_])
return t, kvs, read_pos + len_ + 2
@@ -0,0 +1,146 @@
#!/usr/bin/env python3
#
# Copyright (c) 2019, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
from pktverify.addrs import Ipv6Addr
from pktverify.bytes import Bytes
DOMAIN_PREFIX = Bytes('fd00:7d03:7d03:7d03')
BACKBONE_IPV6_PREFIX = Bytes('2001:0db8:0001:0000')
LINK_LOCAL_All_THREAD_NODES_MULTICAST_ADDRESS = Ipv6Addr('ff32:40:fdde:ad00:beef:0:0:1')
REALM_LOCAL_All_THREAD_NODES_MULTICAST_ADDRESS = Ipv6Addr('ff33:40:fdde:ad00:beef:0:0:1')
REALM_LOCAL_ALL_ROUTERS_ADDRESS = Ipv6Addr('ff03::2')
LINK_LOCAL_ALL_NODES_MULTICAST_ADDRESS = Ipv6Addr('ff02::1')
LINK_LOCAL_ALL_ROUTERS_MULTICAST_ADDRESS = Ipv6Addr('ff02::2')
LINK_LOCAL_ALL_BBRS_MULTICAST_ADDRESS = Ipv6Addr('ff32:40:fd00:7d03:7d03:7d03:0:3')
# MA in Test Plan, make sure these are same as ../config.py
MA1 = Ipv6Addr('ff04::1234:777a:1')
MA1g = Ipv6Addr('ff0e::1234:777a:1')
MA2 = Ipv6Addr('ff05::1234:777a:1')
MA3 = Ipv6Addr('ff0e::1234:777a:3')
MA4 = Ipv6Addr('ff05::1234:777a:4')
MA5 = Ipv6Addr('ff03::1234:777a:5')
MA6 = Ipv6Addr('ff02::1')
MAe1 = Ipv6Addr('fd0e::1234:777a:1')
MAe2 = Ipv6Addr('::')
MAe3 = Ipv6Addr('cafe::e0ff')
ALL_MPL_FORWARDERS_MA = Ipv6Addr('ff03::fc')
LINK_LOCAL_PREFIX = Bytes("fe80")
DEFAULT_MESH_LOCAL_PREFIX = Bytes("fd00:0db8:0000:0000")
# COAP methods
COAP_CODE_POST = 2
COAP_CODE_ACK = 68
MLE_LINK_REQUEST = 0
MLE_LINK_ACCEPT = 1
MLE_LINK_ACCEPT_AND_REQUEST = 2
MLE_ADVERTISEMENT = 4
MLE_DATA_RESPONSE = 8
MLE_PARENT_REQUEST = 9
MLE_PARENT_RESPONSE = 10
MLE_CHILD_ID_REQUEST = 11
MLE_CHILD_ID_RESPONSE = 12
# DUA related constants
ADDRESS_QUERY_INITIAL_RETRY_DELAY = 15
ADDRESS_QUERY_MAX_RETRY_DELAY = 8
ADDRESS_QUERY_TIMEOUT = 3
ADVERTISEMENT_I_MAX = 32
ADVERTISEMENT_I_MIN = 1
CONTEXT_ID_REUSE_DELAY = 48
DATA_RESUBMIT_DELAY = 300
DUA_DAD_PERIOD = 100
DUA_DAD_QUERY_TIMEOUT = 1.0
DUA_DAD_REPEATS = 2
DUA_RECENT_TIME = 20
FAILED_ROUTER_TRANSMISSIONS = 4
ID_REUSE_DELAY = 100
ID_SEQUENCE_PERIOD = 10
INFINITE_COST_TIMEOUT = 90
REAL_LAYER_NAMES = {
'mle',
'coap',
'wpan',
'eth',
'tcp',
'udp',
'ip',
'ipv6',
'icmpv6',
'6lowpan',
'arp',
'thread_bl',
'thread_address',
'thread_nm',
'ssdp',
'dns',
'igmp',
'mdns',
}
FAKE_LAYER_NAMES = {'thread_nwd', 'thread_meshcop'}
VALID_LAYER_NAMES = REAL_LAYER_NAMES | FAKE_LAYER_NAMES
AUTO_SEEK_BACK_MAX_DURATION = 0.01
# Wireshark configs
WIRESHARK_OVERRIDE_PREFS = {
'6lowpan.context0': 'fd00:db8::/64',
'6lowpan.context1': 'fd00:7d03:7d03:7d03::/64',
'wpan.802154_fcs_ok': 'FALSE',
'wpan.802154_sec_suite': 'AES-128 Encryption, 32-bit Integrity Protection',
'thread.thr_seq_ctr': '00000000',
'uat:ieee802154_keys': '"00112233445566778899aabbccddeeff","1","Thread hash"',
}
WIRESHARK_DECODE_AS_ENTRIES = {
'udp.port==61631': 'coap',
}
TIMEOUT_JOIN_NETWORK = 10
TIMEOUT_DUA_REGISTRATION = 10
TIMEOUT_DUA_DAD = 15
TIMEOUT_HOST_READY = 10
TIMEOUT_CHILD_DETACH = 120
TIMEOUT_REGISTER_MA = 5
if __name__ == '__main__':
from pktverify.addrs import Ipv6Addr
assert Ipv6Addr("fe80:0000:0000:0000:0200:0000:0000:0004").startswith(LINK_LOCAL_PREFIX)
assert Ipv6Addr("fd00:0db8:0000:0000:0000:00ff:fe00:8001").startswith(DEFAULT_MESH_LOCAL_PREFIX)
@@ -0,0 +1,62 @@
#!/usr/bin/env python3
#
# Copyright (c) 2019, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
from functools import wraps
def cached(f):
"""
Decorator to convert a function to cache its return value when it's called by the first time.
:param f: The function to decorate.
:return: The caching function.
"""
cache_key = '_once_' + f.__name__
@wraps(f)
def once_f(self):
try:
v = object.__getattribute__(self, cache_key) # can not use getattr, will trigger __getattr__ wrongly
except AttributeError:
v = f(self)
setattr(self, cache_key, v)
return v
return once_f
def cached_property(f):
"""
Decorator for declaring a property that caches its value.
:param f: The property getter function.
:return: The caching property.
"""
return property(cached(f))
@@ -0,0 +1,64 @@
#!/usr/bin/env python3
#
# Copyright (c) 2019, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
class Error(Exception):
"""
Base error class for all Packet Verification errors.
"""
pass
class PacketNotFound(Error):
"""
Represents an error that the packet was not found.
"""
def __init__(self, start_index, stop_index):
self._start_index = start_index
self._stop_index = stop_index
class UnexpectedPacketFound(Error):
"""
Represents an error that the packet was found unexpectedly.
"""
def __init__(self, idx, pkt):
self._idx = idx
self._pkt = pkt
class VerifyFailed(Error):
"""
Represents an error that the packet failed to pass verification criteria.
"""
def __init__(self, pkt):
self._pkt = pkt
@@ -0,0 +1,658 @@
#!/usr/bin/env python3
#
# Copyright (c) 2019, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import sys
from typing import Any, Union
from pyshark.packet.fields import LayerFieldsContainer, LayerField
from pyshark.packet.packet import Packet as RawPacket
from pktverify.addrs import EthAddr, ExtAddr, Ipv6Addr
from pktverify.bytes import Bytes
from pktverify.consts import VALID_LAYER_NAMES
from pktverify.null_field import nullField
def _auto(v: Union[LayerFieldsContainer, LayerField]):
"""parse the layer field automatically according to its format"""
assert not isinstance(v, LayerFieldsContainer) or len(v.fields) == 1, v.fields
dv = v.get_default_value()
rv = v.raw_value
if dv.startswith('0x'):
return int(dv, 16)
try:
if dv == rv:
return int(dv)
elif int(dv) == int(rv, 16):
return int(dv)
except (ValueError, TypeError):
pass
if rv is None:
try:
return int(dv)
except (ValueError, TypeError):
pass
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)
try:
int(rv, 16)
return int(dv)
except Exception:
pass
raise ValueError((v, v.get_default_value(), v.raw_value))
def _payload(v: Union[LayerFieldsContainer, LayerField]) -> bytearray:
"""parse the layer field as a bytearray"""
assert not isinstance(v, LayerFieldsContainer) or len(v.fields) == 1
hex_value = v.raw_value
assert len(hex_value) % 2 == 0
s = bytearray()
for i in range(0, len(hex_value), 2):
s.append(int(hex_value[i:i + 2], 16))
return s
def _hex(v: Union[LayerFieldsContainer, LayerField]) -> int:
"""parse the layer field as a hex string"""
# split v into octets and reverse the order
assert not isinstance(v, LayerFieldsContainer) or len(v.fields) == 1
return int(v.get_default_value(), 16)
def _raw_hex(v: Union[LayerFieldsContainer, LayerField]) -> int:
"""parse the layer field as a raw hex string"""
# split v into octets and reverse the order
assert not isinstance(v, LayerFieldsContainer) or len(v.fields) == 1
iv = v.hex_value
try:
int(v.get_default_value())
assert int(v.get_default_value()) == iv, (v.get_default_value(), v.raw_value)
except ValueError:
pass
try:
int(v.get_default_value(), 16)
assert int(v.get_default_value(), 16) == iv, (v.get_default_value(), v.raw_value)
except ValueError:
pass
return iv
def _raw_hex_rev(v: Union[LayerFieldsContainer, LayerField]) -> int:
"""parse the layer field as a reversed raw hex string"""
# split v into octets and reverse the order
assert not isinstance(v, LayerFieldsContainer) or len(v.fields) == 1
rv = v.raw_value
octets = [rv[i:i + 2] for i in range(0, len(rv), 2)]
iv = int(''.join(reversed(octets)), 16)
try:
int(v.get_default_value())
assert int(v.get_default_value()) == iv, (v.get_default_value(), v.raw_value)
except ValueError:
pass
try:
int(v.get_default_value(), 16)
assert int(v.get_default_value(), 16) == iv, (v.get_default_value(), v.raw_value)
except ValueError:
pass
return iv
def _dec(v: Union[LayerFieldsContainer, LayerField]) -> int:
"""parse the layer field as a decimal"""
assert not isinstance(v, LayerFieldsContainer) or len(v.fields) == 1
return int(v.get_default_value())
def _float(v: Union[LayerFieldsContainer, LayerField]) -> float:
"""parse the layer field as a float"""
assert not isinstance(v, LayerFieldsContainer) or len(v.fields) == 1
return float(v.get_default_value())
def _str(v: Union[LayerFieldsContainer, LayerField]) -> str:
"""parse the layer field as a string"""
assert not isinstance(v, LayerFieldsContainer) or len(v.fields) == 1
return str(v.get_default_value())
def _bytes(v: Union[LayerFieldsContainer, LayerField]) -> Bytes:
"""parse the layer field as raw bytes"""
assert not isinstance(v, LayerFieldsContainer) or len(v.fields) == 1
return Bytes(v.raw_value)
def _ext_addr(v: Union[LayerFieldsContainer, LayerField]) -> ExtAddr:
"""parse the layer field as an extended address"""
assert not isinstance(v, LayerFieldsContainer) or len(v.fields) == 1
return ExtAddr(v.get_default_value())
def _ipv6_addr(v: Union[LayerFieldsContainer, LayerField]) -> Ipv6Addr:
"""parse the layer field as an IPv6 address"""
assert not isinstance(v, LayerFieldsContainer) or len(v.fields) == 1
return Ipv6Addr(v.get_default_value())
def _eth_addr(v: Union[LayerFieldsContainer, LayerField]) -> EthAddr:
"""parse the layer field as an Ethernet MAC address"""
assert not isinstance(v, LayerFieldsContainer) or len(v.fields) == 1, v.fields
return EthAddr(v.get_default_value())
class _first(object):
"""parse the first layer field"""
def __init__(self, sub_parse):
self._sub_parse = sub_parse
def __call__(self, v: Union[LayerFieldsContainer, LayerField]):
return self._sub_parse(v.fields[0])
class _list(object):
"""parse all layer fields into a list"""
def __init__(self, sub_parse):
self._sub_parse = sub_parse
def __call__(self, v: Union[LayerFieldsContainer, LayerField]):
return [self._sub_parse(f) for f in v.fields]
_LAYER_FIELDS = {
# WPAN
'wpan.fcf': _raw_hex_rev,
'wpan.security': _auto,
'wpan.frame_type': _auto,
'wpan.pending': _auto,
'wpan.ack_request': _auto,
'wpan.pan_id_compression': _auto,
'wpan.seqno_suppression': _auto,
'wpan.ie_present': _auto,
'wpan.dst_addr_mode': _auto,
'wpan.version': _auto,
'wpan.src_addr_mode': _auto,
'wpan.dst_pan': _auto,
'wpan.seq_no': _auto,
'wpan.src16': _auto,
'wpan.dst16': _auto,
'wpan.src64': _ext_addr,
'wpan.dst64': _ext_addr,
'wpan.fcs': _raw_hex_rev,
'wpan.fcs_ok': _auto,
'wpan.frame_length': _dec,
'wpan.key_number': _auto,
'wpan.aux_sec.sec_suite': _auto,
'wpan.aux_sec.security_control_field': _auto,
'wpan.aux_sec.sec_level': _auto,
'wpan.aux_sec.key_id_mode': _auto,
'wpan.aux_sec.frame_counter_suppression': _auto,
'wpan.aux_sec.asn_in_nonce': _auto,
'wpan.aux_sec.reserved': _auto,
'wpan.aux_sec.frame_counter': _auto,
'wpan.aux_sec.key_source': _auto,
'wpan.aux_sec.key_index': _auto,
'wpan.aux_sec.hdr': _str,
'wpan.mic': _auto,
'wpan.channel': _auto,
# MLE
'mle.cmd': _auto,
'mle.tlv.type': _list(_dec),
'mle.tlv.len': _list(_dec),
'mle.tlv.mode.receiver_on_idle': _auto,
'mle.tlv.mode.reserved1': _auto,
'mle.tlv.mode.reserved2': _auto,
'mle.tlv.mode.device_type_bit': _auto,
'mle.tlv.mode.network_data': _auto,
'mle.tlv.challenge': _bytes,
'mle.tlv.scan_mask.r': _auto,
'mle.tlv.scan_mask.e': _auto,
'mle.tlv.version': _auto,
'mle.tlv.source_addr': _auto,
'mle.tlv.active_tstamp': _auto,
'mle.tlv.leader_data.partition_id': _auto,
'mle.tlv.leader_data.weighting': _auto,
'mle.tlv.leader_data.data_version': _auto,
'mle.tlv.leader_data.stable_data_version': _auto,
'mle.tlv.leader_data.router_id': _auto,
'mle.tlv.route64.nbr_out': _list(_auto),
'mle.tlv.route64.nbr_in': _list(_auto),
'mle.tlv.route64.id_seq': _auto,
'mle.tlv.route64.id_mask': _auto,
'mle.tlv.route64.cost': _list(_auto),
'mle.tlv.response': _bytes,
'mle.tlv.mle_frm_cntr': _auto,
'mle.tlv.ll_frm_cntr': _auto,
'mle.tlv.link_margin': _auto,
'mle.tlv.conn.sed_dgram_cnt': _auto,
'mle.tlv.conn.sed_buf_size': _auto,
'mle.tlv.conn.lq3': _auto,
'mle.tlv.conn.lq2': _auto,
'mle.tlv.conn.lq1': _auto,
'mle.tlv.conn.leader_cost': _auto,
'mle.tlv.conn.id_seq': _auto,
'mle.tlv.conn.flags.pp': _auto,
'mle.tlv.conn.active_rtrs': _auto,
'mle.tlv.timeout': _auto,
'mle.tlv.addr16': _auto,
# IP
'ip.version': _auto,
'ip.src': _str,
'ip.src_host': _str,
'ip.dst': _str,
'ip.dst_host': _str,
'ip.ttl': _auto,
'ip.proto': _auto,
'ip.len': _auto,
'ip.id': _auto,
'ip.host': _list(_str),
'ip.hdr_len': _dec,
'ip.frag_offset': _auto,
'ip.flags.rb': _auto,
'ip.flags.mf': _auto,
'ip.flags.df': _auto,
'ip.dsfield.ecn': _auto,
'ip.dsfield.dscp': _auto,
'ip.checksum.status': _auto,
'ip.addr': _list(_str),
'ip.options.routeralert': _bytes,
'ip.opt.type.number': _auto,
'ip.opt.type.copy': _auto,
'ip.opt.type.class': _auto,
'ip.opt.ra': _auto,
'ip.opt.len': _auto,
# UDP
'udp.stream': _auto,
'udp.srcport': _auto,
'udp.dstport': _auto,
'udp.length': _auto,
'udp.port': _list(_dec),
'udp.checksum.status': _auto,
# IPv6
'ipv6.version': _auto,
'ipv6.src': _ipv6_addr,
'ipv6.src_host': _ipv6_addr,
'ipv6.dst': _ipv6_addr,
'ipv6.dst_host': _ipv6_addr,
'ipv6.addr': _list(_ipv6_addr),
'ipv6.tclass.dscp': _auto,
'ipv6.tclass.ecn': _auto,
'ipv6.flow': _auto,
'ipv6.hlim': _auto,
'ipv6.nxt': _auto,
'ipv6.hopopts.len': _auto,
'ipv6.hopopts.nxt': _auto,
'ipv6.hopopts.len_oct': _dec,
'ipv6.host': _list(_ipv6_addr),
'ipv6.plen': _auto,
'ipv6.opt.type.rest': _list(_auto),
'ipv6.opt.type.change': _list(_auto),
'ipv6.opt.type.action': _list(_auto),
'ipv6.opt.router_alert': _auto,
'ipv6.opt.padn': _str,
'ipv6.opt.length': _list(_auto),
'ipv6.opt.mpl.sequence': _auto,
'ipv6.opt.mpl.flag.v': _auto,
'ipv6.opt.mpl.flag.s': _auto,
'ipv6.opt.mpl.flag.rsv': _auto,
'ipv6.opt.mpl.flag.m': _auto,
# Eth
'eth.src': _eth_addr,
'eth.src_resolved': _eth_addr,
'eth.dst': _eth_addr,
'eth.dst_resolved': _eth_addr,
'eth.type': _auto,
'eth.addr': _list(_eth_addr),
'eth.addr_resolved': _list(_eth_addr),
'eth.ig': _list(_auto),
'eth.lg': _list(_auto),
# 6LOWPAN
'6lowpan.src': _ipv6_addr,
'6lowpan.dst': _ipv6_addr,
'6lowpan.udp.src': _auto,
'6lowpan.udp.dst': _auto,
'6lowpan.udp.checksum': _auto,
'6lowpan.frag.offset': _auto,
'6lowpan.frag.tag': _auto,
'6lowpan.frag.size': _auto,
'6lowpan.pattern': _list(_auto),
'6lowpan.hops': _auto,
'6lowpan.padding': _auto,
'6lowpan.next': _auto,
'6lowpan.flow': _auto,
'6lowpan.ecn': _auto,
'6lowpan.iphc.tf': _auto,
'6lowpan.iphc.m': _auto,
'6lowpan.iphc.nh': _auto,
'6lowpan.iphc.hlim': _auto,
'6lowpan.iphc.cid': _auto,
'6lowpan.iphc.sac': _auto,
'6lowpan.iphc.sam': _auto,
'6lowpan.iphc.dac': _auto,
'6lowpan.iphc.dam': _auto,
'6lowpan.iphc.sci': _auto,
'6lowpan.iphc.dci': _auto,
'6lowpan.iphc.sctx.prefix': _ipv6_addr,
'6lowpan.iphc.dctx.prefix': _ipv6_addr,
'6lowpan.mesh.v': _auto,
'6lowpan.nhc.pattern': _list(_auto),
'6lowpan.nhc.udp.checksum': _auto,
'6lowpan.nhc.udp.ports': _auto,
'6lowpan.nhc.ext.nh': _auto,
'6lowpan.nhc.ext.length': _auto,
'6lowpan.nhc.ext.eid': _auto,
'6lowpan.reassembled.length': _auto,
'6lowpan.fragments': _str,
'6lowpan.fragment.count': _auto,
'6lowpan.mesh.orig16': _auto,
'6lowpan.mesh.hops8': _auto,
'6lowpan.mesh.hops': _auto,
'6lowpan.mesh.f': _auto,
'6lowpan.mesh.dest16': _auto,
# ICMPv6
'icmpv6.type': _first(_auto),
'icmpv6.code': _first(_auto),
'icmpv6.checksum': _first(_auto),
'icmpv6.reserved': _raw_hex,
'icmpv6.resptime': _float,
'icmpv6.resp_to': _auto,
'icmpv6.mldr.nb_mcast_records': _auto,
'icmpv6.nd.ra.cur_hop_limit': _auto,
'icmpv6.nd.ns.target_address': _ipv6_addr,
'icmpv6.nd.na.target_address': _ipv6_addr,
'icmpv6.nd.na.flag.s': _auto,
'icmpv6.nd.na.flag.o': _auto,
'icmpv6.nd.na.flag.r': _auto,
'icmpv6.nd.na.flag.rsv': _auto,
'icmpv6.mldr.mar.record_type': _list(_auto),
'icmpv6.mldr.mar.aux_data_len': _list(_auto),
'icmpv6.mldr.mar.nb_sources': _list(_auto),
'icmpv6.mldr.mar.multicast_address': _list(_ipv6_addr),
'icmpv6.opt.type': _list(_auto),
'icmpv6.opt.nonce': _bytes,
'icmpv6.opt.linkaddr': _eth_addr,
'icmpv6.opt.src_linkaddr': _eth_addr,
'icmpv6.opt.target_linkaddr': _eth_addr,
'icmpv6.opt.route_lifetime': _auto,
'icmpv6.opt.route_info.flag.route_preference': _auto,
'icmpv6.opt.route_info.flag.reserved': _auto,
'icmpv6.opt.prefix.valid_lifetime': _auto,
'icmpv6.opt.prefix.preferred_lifetime': _auto,
'icmpv6.opt.prefix.length': _list(_auto),
'icmpv6.opt.prefix.flag.reserved': _auto,
'icmpv6.opt.prefix.flag.r': _auto,
'icmpv6.opt.prefix.flag.l': _auto,
'icmpv6.opt.prefix.flag.a': _auto,
'icmpv6.opt.length': _list(_auto),
'icmpv6.opt.reserved': _str,
'icmpv6.nd.ra.router_lifetime': _auto,
'icmpv6.nd.ra.retrans_timer': _auto,
'icmpv6.nd.ra.reachable_time': _auto,
'icmpv6.nd.ra.flag.rsv': _auto,
'icmpv6.nd.ra.flag.prf': _auto,
'icmpv6.nd.ra.flag.p': _auto,
'icmpv6.nd.ra.flag.o': _auto,
'icmpv6.nd.ra.flag.m': _auto,
'icmpv6.nd.ra.flag.h': _auto,
'icmpv6.echo.sequence_number': _auto,
'icmpv6.echo.identifier': _auto,
# COAP
'coap.code': _auto,
'coap.version': _auto,
'coap.type': _auto,
'coap.mid': _auto,
'coap.token_len': _auto,
'coap.token': _auto,
'coap.opt.uri_path': _list(_str),
'coap.opt.name': _list(_str),
'coap.opt.length': _list(_auto),
'coap.opt.uri_path_recon': _str,
'coap.payload': _payload,
'coap.payload_length': _auto,
'coap.payload_desc': _str,
'coap.opt.end_marker': _auto,
'coap.opt.desc': _list(_str),
'coap.opt.delta': _list(_auto),
'coap.response_to': _auto,
'coap.response_time': _float,
# COAP TLVS
'coap.tlv.type': _list(_auto),
'coap.tlv.status': _auto,
'coap.tlv.target_eid': _ipv6_addr,
'coap.tlv.ml_eid': _ext_addr,
'coap.tlv.last_transaction_time': _auto,
'coap.tlv.rloc16': _auto,
'coap.tlv.net_name': _str,
'coap.tlv.ext_mac_addr': _ext_addr,
'coap.tlv.router_mask_assigned': _auto,
'coap.tlv.router_mask_id_seq': _auto,
# thread_address
'thread_address.tlv.type': _list(_auto),
'thread_address.tlv.status': _auto,
# thread bl
'thread_bl.tlv.type': _list(_auto),
'thread_bl.tlv.len': _list(_auto),
'thread_bl.tlv.target_eid': _ipv6_addr,
'thread_bl.tlv.ml_eid': _ext_addr,
'thread_bl.tlv.last_transaction_time': _auto,
# THEAD NM
'thread_nm.tlv.type': _list(_auto),
'thread_nm.tlv.ml_eid': _ext_addr,
'thread_nm.tlv.target_eid': _ipv6_addr,
# thread_meshcop is not a real layer
'thread_meshcop.len_size_mismatch': _str,
'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_sess_id': _auto, # from mle
"thread_meshcop.tlv.channel_page": _auto, # from ble
"thread_meshcop.tlv.channel": _auto, # from ble
"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.pan_id': _auto,
'thread_meshcop.tlv.xpan_id': _bytes,
'thread_meshcop.tlv.ml_prefix': _bytes,
'thread_meshcop.tlv.master_key': _bytes,
'thread_meshcop.tlv.pskc': _bytes,
'thread_meshcop.tlv.sec_policy_rot': _auto,
'thread_meshcop.tlv.sec_policy_o': _auto,
'thread_meshcop.tlv.sec_policy_n': _auto,
'thread_meshcop.tlv.sec_policy_r': _auto,
'thread_meshcop.tlv.sec_policy_c': _auto,
'thread_meshcop.tlv.sec_policy_b': _auto,
'thread_meshcop.tlv.unknown': _bytes,
# THREAD NWD
'thread_nwd.tlv.type': _list(_auto),
'thread_nwd.tlv.len': _list(_auto),
'thread_nwd.tlv.stable': _list(_auto),
'thread_nwd.tlv.service.t': _auto,
'thread_nwd.tlv.service.s_id': _auto,
'thread_nwd.tlv.service.s_data_len': _auto,
'thread_nwd.tlv.service.s_data.seqno': _auto,
'thread_nwd.tlv.service.s_data.rrdelay': _auto,
'thread_nwd.tlv.service.s_data.mlrtimeout': _auto,
'thread_nwd.tlv.server.16': _auto,
'thread_nwd.tlv.border_router.16': _auto,
'thread_nwd.tlv.sub_tlvs': _list(_str),
'thread_nwd.tlv.prefix.length': _auto,
'thread_nwd.tlv.prefix.domain_id': _auto,
'thread_nwd.tlv.border_router.pref': _auto,
'thread_nwd.tlv.border_router.flag.s': _auto,
'thread_nwd.tlv.border_router.flag.r': _auto,
'thread_nwd.tlv.border_router.flag.p': _auto,
'thread_nwd.tlv.border_router.flag.o': _auto,
'thread_nwd.tlv.border_router.flag.n': _auto,
'thread_nwd.tlv.border_router.flag.dp': _auto,
'thread_nwd.tlv.border_router.flag.d': _auto,
'thread_nwd.tlv.border_router.flag.c': _auto,
'thread_nwd.tlv.6co.flag.reserved': _auto,
'thread_nwd.tlv.6co.flag.cid': _auto,
'thread_nwd.tlv.6co.flag.c': _auto,
'thread_nwd.tlv.6co.context_length': _auto,
}
_layer_containers = set()
for key in _LAYER_FIELDS:
assert key.strip() == key and ' ' not in key, key
secs = key.split('.')
assert len(secs) >= 2
assert secs[0] in VALID_LAYER_NAMES, secs[0]
for i in range(len(secs) - 2):
path = secs[0] + '.' + '.'.join(secs[1:i + 2])
assert path not in _LAYER_FIELDS, '%s can not be both field and path' % path
_layer_containers.add(path)
def is_layer_field(uri: str) -> bool:
"""
Returns if the URI is a valid layer field.
:param uri: The layer field URI.
"""
return uri in _LAYER_FIELDS
def is_layer_field_container(uri: str) -> bool:
"""
Returns if the URI is a valid layer field container.
:param uri: The layer field container URI.
"""
return uri in _layer_containers
def get_layer_field(packet: RawPacket, field_uri: str) -> Any:
"""
Get a given layer field from the packet.
:param packet: The packet.
:param field_uri: The layer field URI.
:return: The specified layer field.
"""
assert isinstance(packet, RawPacket)
secs = field_uri.split('.')
layer_name = secs[0]
if is_layer_field(field_uri):
candidate_layers = _get_candidate_layers(packet, layer_name)
for layer in candidate_layers:
v = layer.get_field(field_uri)
if v is not None:
try:
v = _LAYER_FIELDS[field_uri](v)
print("[%s = %r] " % (field_uri, v), file=sys.stderr)
return v
except Exception as ex:
raise ValueError('can not parse field %s = %r' % (field_uri,
(v.get_default_value(), v.raw_value))) from ex
print("[%s = %s] " % (field_uri, "null"), file=sys.stderr)
return nullField
elif is_layer_field_container(field_uri):
from pktverify.layer_fields_container import LayerFieldsContainer
return LayerFieldsContainer(packet, field_uri)
else:
raise NotImplementedError('Field %s is not valid, please add it to `_LAYER_FIELDS`' % field_uri)
def check_layer_field_exists(packet, field_uri):
"""
Check if a given layer field URI exists in the packet.
:param packet: The packet to check.
:param field_uri: The layer field URI.
:return: Whether the layer field URI exists in the packet.
"""
assert isinstance(packet, RawPacket)
secs = field_uri.split('.')
layer_name = secs[0]
if not is_layer_field(field_uri) and not is_layer_field_container(field_uri):
raise NotImplementedError('%s is neither a field or field container' % field_uri)
candidate_layers = _get_candidate_layers(packet, layer_name)
for layer in candidate_layers:
for k, v in layer._all_fields.items():
if k == field_uri or k.startswith(field_uri + '.'):
return True
return False
def _get_candidate_layers(packet, layer_name):
if layer_name == 'thread_meshcop':
candidate_layer_names = ['mle', 'coap', 'thread_bl']
elif layer_name == 'thread_nwd':
candidate_layer_names = ['mle', 'thread_address']
elif layer_name == 'wpan':
candidate_layer_names = ['wpan', 'mle']
elif layer_name == 'ip':
candidate_layer_names = ['ip', 'ipv6']
else:
candidate_layer_names = [layer_name]
layers = []
for ln in candidate_layer_names:
if hasattr(packet, ln):
layers.append(getattr(packet, ln))
return layers
@@ -0,0 +1,84 @@
#!/usr/bin/env python3
#
# Copyright (c) 2019, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
from pyshark.packet.packet import Packet as RawPacket
from pktverify.layer_fields import get_layer_field, check_layer_field_exists
class LayerFieldsContainer(object):
"""
Represents a layer field container.
"""
def __init__(self, packet: RawPacket, path: str):
assert isinstance(packet, RawPacket)
assert isinstance(path, str)
self._packet = packet
self._path = path
def __getattr__(self, name):
subpath = self._path + '.' + name
v = get_layer_field(self._packet, subpath)
setattr(self, name, v)
return v
def has(self, subpath):
"""
Returns if the layer field container has a sub layer field or container.
:param subpath: The sub path to the layer field or container.
"""
subpath = self._path + '.' + subpath
return check_layer_field_exists(self._packet, subpath) is not None
def __getitem__(self, item):
return getattr(self, item)
@property
def full_path(self):
"""
Returns the full path to this layer field container.
"""
return self._path
@property
def field_path(self):
"""
Returns the field that references to this layer field container.
"""
secs = self._path.split('.')
assert len(secs) >= 2
return '.'.join(secs[1:])
def __bool__(self):
"""
Returns if this layer field container exists in the packet.
"""
return check_layer_field_exists(self._packet, self._path)
@@ -0,0 +1,181 @@
#!/usr/bin/env python3
#
# Copyright (c) 2019, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import logging
from typing import Optional
from pyshark.packet.fields import LayerField, LayerFieldsContainer
from pyshark.packet.layer import Layer as RawLayer
from pyshark.packet.packet import Packet as RawPacket
from pktverify.layer_fields import get_layer_field, check_layer_field_exists
class Layer(object):
"""
Represents a layer of a packet.
"""
def __init__(self, packet: RawPacket, layer_name: str):
assert isinstance(packet, RawPacket)
assert isinstance(layer_name, str)
self._packet = packet
self._layer_name = layer_name
@property
def _layer(self) -> Optional[RawLayer]:
try:
return getattr(self._packet, self._layer_name)
except AttributeError:
return None
@property
def layer_name(self) -> str:
"""
Returns the layer name.
"""
return self._layer_name
def show(self):
"""
Print the layer information.
"""
print(self._layer)
def has(self, name) -> bool:
"""
Returns if the layer has a given field.
:param name: The field name.
"""
path = '%s.%s' % (self.layer_name, name)
return check_layer_field_exists(self._packet, path)
def __bool__(self):
"""
Returns if this layer exists in the packet.
"""
layer_exists = hasattr(self._packet, self._layer_name)
return layer_exists
def __getattr__(self, name):
"""
Returns the layer field or container of a given field name.
:param name: The name of layer field or container.
"""
path = '%s.%s' % (self.layer_name, name)
v = get_layer_field(self._packet, path)
assert not isinstance(v, (LayerField, LayerFieldsContainer)), '%s = %s(%r)' % (path, v.__class__.__name__, v)
setattr(self, name, v)
return v
def _add_field(self, key: str, val: str):
logging.debug("layer %s add field: %s = %s", self.layer_name, key, val)
field = LayerField(name=key, value=val)
all_fields = self._layer._all_fields
if key not in all_fields:
all_fields[key] = LayerFieldsContainer(main_field=field)
else:
all_fields[key].fields.append(field)
class ThreadMeshcopLayer(Layer):
"""
Represents the Thread MeshCop layer of a packet.
"""
def __bool__(self):
raise NotImplementedError("thread_meshcop is not a real layer, please do not check as bool")
class ThreadNetworkDataLayer(Layer):
"""
Represents the Thread NetworkData layer of a packet.
"""
def __bool__(self):
raise NotImplementedError("thread_nwd is not a real layer, please do not check as bool")
class Icmpv6Layer(Layer):
"""
Represents the ICMPv6 layer of a packet.
"""
@property
def is_ping(self) -> bool:
"""
Returns if the ICMPv6 layer is a Ping Request or Reply.
"""
return self.type in (128, 129)
@property
def is_ping_request(self) -> bool:
"""
Returns if the ICMPv6 layer is a Ping Request.
"""
return self.type == 128
@property
def is_ping_reply(self) -> bool:
"""
Returns if the ICMPv6 layer is a Ping Reply.
"""
return self.type == 129
@property
def is_neighbor_advertisement(self) -> bool:
"""
Returns if the ICMPv6 layer is a Neighbor Advertisement.
"""
return self.type == 136
@property
def is_neighbor_solicitation(self) -> bool:
"""
Returns if the ICMPv6 layer is a Neighbor Solicitation.
"""
return self.type == 135
@property
def is_router_advertisement(self) -> bool:
"""
Returns if the ICMPv6 layer is a Router Advertisement.
"""
return self.type == 134
class WpanLayer(Layer):
"""
Represents the WPAN layer of a packet.
"""
@property
def is_ack(self) -> bool:
return self.frame_type == 0x2
@@ -0,0 +1,121 @@
#!/usr/bin/env python3
#
# Copyright (c) 2019, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
nullField = None
class NullField(object):
"""
Represents a null field that does not exists.
"""
def __new__(cls, *args, **kwargs):
global nullField
if nullField is None:
nullField = object.__new__(cls, *args, **kwargs)
return nullField
def __init__(self):
assert self is nullField
def __bool__(self):
"""
NullField is always treated as False.
"""
return False
def __getattr__(self, item):
"""
Any sub field of the NullField is NullField itself.
"""
return self
def __setattr__(self, key, value):
pass
def __len__(self) -> 0:
return 0
def __eq__(self, other):
"""
NullField is always not equal to any other value.
"""
return False
def __ne__(self, other):
return True
def __lt__(self, other):
"""
Comparing NullField to any other value gets False.
"""
return False
def __le__(self, other):
"""
Comparing NullField to any other value gets False.
"""
return False
def __gt__(self, other):
"""
Comparing NullField to any other value gets False.
"""
return False
def __ge__(self, other):
"""
Comparing NullField to any other value gets False.
"""
return False
def __str__(self):
return "nullField"
def __repr__(self):
return 'nullField'
NullField()
if __name__ == '__main__':
assert nullField is NullField()
assert not nullField, repr(nullField)
assert nullField != nullField, repr(nullField)
assert nullField != 0
assert not (nullField > 1)
assert not (nullField < 1)
assert not (nullField < nullField)
assert not (nullField > nullField)
assert bool(nullField) is False
assert nullField != ""
assert nullField != None # noqa
assert nullField is not None
@@ -0,0 +1,148 @@
#!/usr/bin/env python3
#
# Copyright (c) 2019, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import logging
import sys
from typing import Iterable, List, Union, Callable
from pyshark.packet.layer import Layer as RawLayer
from pyshark.packet.packet import Packet as RawPacket
from pktverify import errors
from pktverify.addrs import EthAddr
from pktverify.coap import CoapLayer
from pktverify.consts import VALID_LAYER_NAMES
from pktverify.decorators import cached_property
from pktverify.layers import Layer, ThreadMeshcopLayer, Icmpv6Layer, WpanLayer, ThreadNetworkDataLayer
from pktverify.utils import make_filter_func
class Packet(object):
def __init__(self, packet: RawPacket):
self._packet = packet
self._strip_wpan_eth_wrapper(packet)
def __str__(self) -> str:
return str(self._packet)
def __repr__(self) -> str:
return repr(self._packet)
def __dir__(self) -> Iterable[str]:
return dir(self._packet)
def _strip_wpan_eth_wrapper(self, packet: RawPacket):
if not hasattr(packet, 'wpan'):
return
for layer in packet.layers:
if layer.layer_name == 'eth':
packet.layers.remove(layer)
eth_src = EthAddr(layer.get_field('eth.src'))
eth_dst = EthAddr(layer.get_field('eth.dst'))
logging.debug("stripping eth: src=%s, dst=%s", eth_src, eth_dst)
channel = eth_src[5]
self.wpan._add_field('wpan.channel', hex(channel))
return
@property
def layers(self) -> Iterable[RawLayer]:
for l in self._packet.layers:
if l.layer_name != 'data':
yield getattr(self, l.layer_name)
@property
def layer_names(self) -> List[str]:
return [l.layer_name for l in self._packet.layers if l.layer_name != 'data']
@cached_property
def wpan(self) -> WpanLayer:
return WpanLayer(self._packet, 'wpan')
@cached_property
def coap(self) -> CoapLayer:
return CoapLayer(self._packet, 'coap')
@cached_property
def icmpv6(self) -> Icmpv6Layer:
return Icmpv6Layer(self._packet, 'icmpv6')
@cached_property
def thread_meshcop(self) -> ThreadMeshcopLayer:
return ThreadMeshcopLayer(self._packet, 'thread_meshcop')
@cached_property
def thread_nwd(self) -> ThreadNetworkDataLayer:
return ThreadNetworkDataLayer(self._packet, 'thread_nwd')
def __getattr__(self, layer_name: str) -> Layer:
real_layer_name = layer_name
if layer_name == 'lowpan':
real_layer_name = '6lowpan'
assert real_layer_name in VALID_LAYER_NAMES, '%s is not a valid layer name' % real_layer_name
_layer = getattr(self._packet, real_layer_name, None)
assert _layer is None or isinstance(_layer, RawLayer)
layer = Layer(self._packet, real_layer_name)
setattr(self, layer_name, layer)
if real_layer_name != layer_name:
setattr(self, real_layer_name, layer)
return layer
def verify(self, func: Union[str, Callable], **vars) -> bool:
print("\n>>> verifying packet:", file=sys.stderr, flush=False)
func = make_filter_func(func, **vars)
ok = func(self)
print("\t=> %s" % ok, file=sys.stderr)
return ok
def must_verify(self, func: Union[str, Callable], **vars):
if not self.verify(func, **vars):
raise errors.VerifyFailed(self)
def must_not_verify(self, func: Union[str, Callable], **vars):
if self.verify(func, **vars):
raise errors.VerifyFailed(self)
@property
def sniff_timestamp(self) -> float:
return float(self._packet.sniff_timestamp)
def show(self):
self._packet.show()
def debug_fields(self):
for layer in self._packet.layers:
print("### Layer %s ###" % layer.layer_name)
for k, v in layer._all_fields.items():
print("\t\t%r = %r" % (k, v))
@@ -0,0 +1,466 @@
#
# Copyright (c) 2019, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import sys
from operator import attrgetter
from typing import Optional, Callable, Tuple
from pktverify import consts, errors
from pktverify.addrs import EthAddr, ExtAddr, Ipv6Addr
from pktverify.packet import Packet
from pktverify.utils import make_filter_func
WPAN, ETH = 0, 1
class _SavedIndex(object):
__slots__ = ('_pkts', '_saved_index')
def __init__(self, pkts):
self._pkts = pkts
self._saved_index = pkts.index
def __enter__(self):
pass
def __exit__(self, exc_type, exc_val, exc_tb):
self._pkts.index = self._saved_index
def _always_true(p):
return True
class PacketFilter(object):
"""
Represents a range of packets that are filtered by given filter
"""
def __init__(self,
pkts,
start=(0, 0),
stop=None,
*,
index=None,
filter_func: Optional[Callable] = None,
parent: Optional['PacketFilter'] = None):
if stop is None:
stop = (len(pkts), len(pkts))
self._pkts = pkts
self._start_index = start
self._stop_index = stop
self._index = index if index is not None else self._start_index
self._last_index = -1
self._filter_func = filter_func or _always_true
self._parent = parent
self._check_type_ok()
def _check_type_ok(self):
assert self._last_index == -1 or 0 <= self._last_index < len(self._pkts)
assert isinstance(self._start_index, tuple) and len(self._start_index) == 2, self._start_index
assert isinstance(self._stop_index, tuple) and len(self._stop_index) == 2, self._stop_index
assert isinstance(self._index, tuple) and len(self._index) == 2, self._index
self._check_idx_range_ok((0, 0), self._start_index)
self._check_idx_range_ok(self._start_index, self._index)
self._check_idx_range_ok(self._index, self._stop_index)
self._check_idx_range_ok(self._stop_index, (len(self._pkts), len(self._pkts)))
def _check_idx_range_ok(self, start, stop):
assert start[0] <= stop[0], (start, stop)
assert start[1] <= stop[1], (start, stop)
@property
def index(self) -> Tuple[int, int]:
"""
:return: the current index (which is a tuple)
"""
return self._index
@index.setter
def index(self, index: Tuple[int, int]):
"""
Set the current index
:param index: the index tuple to set
"""
assert isinstance(index, tuple) and len(index) == 2, index
self._check_type_ok()
self._index = index
self._check_type_ok()
def __len__(self):
"""
:return: length of packets
"""
return len(self._pkts)
def save_index(self):
"""
Save the current index to be restored.
:return: a context that saves the current index when entering, and restores when exiting
"""
return _SavedIndex(self)
@property
def start_index(self) -> Tuple[int, int]:
"""
:return: the start index tuple
"""
return self._start_index
@property
def stop_index(self) -> Tuple[int, int]:
"""
:return: the stop index tuple
"""
return self._stop_index
def filter(self, func, cascade=True, **vars) -> 'PacketFilter':
"""
Create a new PacketFilter based on this packet filter with given filter func
:param func: a callable that returns a bool (e.x. lambda p: xxx) or a filter string
:param cascade: True if calling next in the new filter will also set index for this filter, False otherwise
:param vars: variables for filter string
:return: a new PacketFilter
"""
print('\n>>> filtering in range %s~%s%s:' %
(self._index, self._stop_index, "<end>" if self._stop_index == len(self._pkts) else "<stop>"),
file=sys.stderr)
func = make_filter_func(func, **vars)
self._check_type_ok()
return PacketFilter(self._pkts,
self._index,
self._stop_index,
filter_func=lambda p: self._filter_func(p) and func(p),
parent=self if cascade else None)
def filter_if(self, cond: bool, *args, **kwargs) -> 'PacketFilter':
"""
Create a filter using given arguments if `cond` is true.
:param cond: the condition to be checked
:param args: arguments for filter func
:param kwargs: arguments for filter func
:return: a sub filter using given arguments if cond is true, or self otherwise
"""
if cond:
return self.filter(*args, **kwargs)
else:
return self
@property
def last_index(self) -> Tuple[int, int]:
return self._last_index
def last(self) -> Packet:
"""
:return: the last packet found
"""
if self._last_index >= 0:
return self._pkts[self._last_index]
else:
raise errors.PacketNotFound(self.index, self._stop_index)
def next(self) -> Optional[Packet]:
"""
Find the next packet starting from the current index to the stop index that matches the current filter.
:return: the next matching packet, or None if packet not found
"""
self._check_type_ok()
idx = min(self._index)
stop_idx = max(self._stop_index)
while idx < stop_idx:
p = self._pkts[idx]
sys.stderr.write('#%d %s' % (idx + 1, '\n' if idx % 40 == 39 else ''))
if self._filter_func(p):
if p.wpan and not (self._index[0] <= idx < self._stop_index[0]): # wpan matched but not in range
pass
elif p.eth and not (self._index[1] <= idx < self._stop_index[1]): # eth matched but not in range
pass
else:
self._on_found_next(idx, p)
print("\n>>> found packet at #%d!" % (idx + 1,), file=sys.stderr)
return p
idx += 1
return None
def must_next(self) -> Packet:
"""
Call .next(), raise error if packet is not found.
:return: the next matching packet
"""
p = self.next()
if p is not None:
return p
else:
raise errors.PacketNotFound(self.index, self._stop_index)
def must_not_next(self) -> None:
"""
Call .next(), raise error if packet is found
"""
p = self.next()
if p is None:
return
else:
raise errors.UnexpectedPacketFound(self.index, p)
def _on_found_next(self, idx: int, p: Packet):
assert self._pkts[idx] is p
assert idx >= min(self._index)
assert not p.wpan or idx >= self._index[0]
assert not p.eth or idx >= self._index[1], (self._index, idx)
if p.wpan:
wpan_idx = idx + 1
eth_idx = max(self._index[1],
self._find_prev_packet(idx + 1, p.sniff_timestamp - consts.AUTO_SEEK_BACK_MAX_DURATION, ETH))
else:
eth_idx = idx + 1
wpan_idx = max(
self._index[0],
self._find_prev_packet(idx + 1, p.sniff_timestamp - consts.AUTO_SEEK_BACK_MAX_DURATION, WPAN))
# make sure index never go back
assert wpan_idx >= self._index[0]
assert eth_idx >= self._index[1]
print('\n>>>_on_found_next %d %s => %s' % (idx, self._index, (wpan_idx, eth_idx)), file=sys.stderr)
self._set_found_index(idx, (wpan_idx, eth_idx))
def _find_prev_packet(self, idx, min_sniff_timestamp, pkttype):
assert pkttype in (WPAN, ETH)
prev_idx = idx
while idx > 0 and self._pkts[idx - 1].sniff_timestamp >= min_sniff_timestamp:
idx -= 1
if pkttype == WPAN and self._pkts[idx].wpan:
prev_idx = idx
elif pkttype == ETH and self._pkts[idx].eth:
prev_idx = idx
return prev_idx
def __iter__(self):
for pkt in self._pkts:
yield pkt
def range(self, start, stop=None, cascade=True) -> 'PacketFilter':
"""
Create a new PacketFilter using the specified start and stop index tuples
:param start: the new start index tuple
:param stop: the new stop index tuple
:param cascade: True if calling next in the new filter will also set index for this filter, False otherwise
:return: a new PacketFilter with new start and stop range
"""
if stop is None:
stop = self._stop_index
assert self._start_index <= start <= self._stop_index
assert self._start_index <= stop <= self._stop_index
return PacketFilter(self._pkts, start, stop, filter_func=self._filter_func, parent=self if cascade else None)
def copy(self) -> 'PacketFilter':
"""
:return: a copy of the current PacketFilter
"""
return PacketFilter(self._pkts, self._index, self._stop_index, filter_func=self._filter_func, parent=None)
def __getitem__(self, index: int) -> Packet:
"""
:param index: the packet index (not tuple!)
:return: the packet at the specified index
"""
assert isinstance(index, int), index
return self._pkts[index]
def seek_back(self, max_duration: float, *, eth=False, wpan=False) -> 'PacketFilter':
"""
Move the current index back in time within the specified max duration. Either eth or wpan must be True.
:param max_duration: the max duration to move back
:param eth: True if eth index can be moved back
:param wpan: True if wpan index can be moved back
:return: self
"""
assert eth or wpan, "must have eth or wpan"
wpan_idx = self._index[0]
if wpan and wpan_idx < len(self._pkts):
wpan_idx = self._find_prev_packet(wpan_idx, self._pkts[wpan_idx].sniff_timestamp - max_duration, WPAN)
wpan_idx = max(self._start_index[0], wpan_idx)
eth_idx = self._index[1]
if eth and eth_idx < len(self._pkts):
eth_idx = self._find_prev_packet(eth_idx, self._pkts[eth_idx].sniff_timestamp - max_duration, ETH)
eth_idx = max(self._start_index[1], eth_idx)
print("\n>>> back %s wpan=%s, eth=%s: index %s => %s" % (max_duration, wpan, eth, self._index,
(wpan_idx, eth_idx)),
file=sys.stderr)
self._index = (wpan_idx, eth_idx)
self._check_type_ok()
return self
def _set_found_index(self, last_index: Tuple[int, int], index: Tuple[int, int]):
self._last_index = last_index
self._index = index
self._check_type_ok()
if self._parent is not None:
self._parent._set_found_index(last_index, index)
def filter_coap(self, **kwargs):
"""
Create a new PacketFilter to filter COAP packets.
:param kwargs: Extra arguments for `filter`.
:return: The new PacketFilter to filter COAP packets.
"""
return self.filter(attrgetter('coap'), **kwargs)
def filter_coap_request(self, uri_path, port=None, **kwargs):
"""
Create a new PacketFilter to filter COAP Request packets.
:param uri_path: The COAP URI path to filter.
:param port: The UDP port to filter if specified.
:param kwargs: Extra arguments for `filter`.
:return: The new PacketFilter to filter COAP Request packets.
"""
assert isinstance(uri_path, str), uri_path
assert port is None or isinstance(port, int), port
return self.filter(
lambda p: (p.coap.is_post and p.coap.opt.uri_path_recon == uri_path and
(port is None or p.udp.dstport == port)), **kwargs)
def filter_coap_ack(self, uri_path, port=None, **kwargs):
"""
Create a new PacketFilter for filter COAP ACK packets.
:param uri_path: The COAP URI path to filter.
:param port: The UDP port to filter if specified.
:param kwargs: Extra arguments for `filter`.
:return: The new PacketFilter to filter COAP ACK packets.
"""
assert isinstance(uri_path, str), uri_path
assert port is None or isinstance(port, int), port
return self.filter(
lambda p: (p.coap.is_ack and p.coap.opt.uri_path_recon == uri_path and
(port is None or p.udp.dstport == port)), **kwargs)
def filter_wpan(self, **kwargs):
"""
Create a new PacketFilter for filter WPAN packets.
:param kwargs: Extra arguments for `filter`.
:return: The new PacketFilter to filter WPAN packets.
"""
return self.filter(attrgetter('wpan'), **kwargs)
def filter_wpan_channel(self, channel: int, **kwargs):
"""
Create a new PacketFilter for filter WPAN packets of a given channel.
:param channel: The channel to filter.
:param kwargs: Extra arguments for `filter`.
:return: The new PacketFilter to filter WPAN packets.
"""
return self.filter(lambda p: p.wpan.channel == channel, **kwargs)
def filter_wpan_src64(self, addr, **kwargs):
assert isinstance(addr, (str, ExtAddr)), addr
return self.filter(lambda p: p.wpan.src64 == addr, **kwargs)
def filter_wpan_dst64(self, addr, **kwargs):
assert isinstance(addr, (str, ExtAddr)), addr
return self.filter(lambda p: p.wpan.dst64 == addr, **kwargs)
def filter_ping_request(self, **kwargs):
return self.filter(lambda p: p.icmpv6.is_ping_request, **kwargs)
def filter_ping_reply(self, **kwargs):
identifier = kwargs.pop('identifier', None)
return self.filter(
lambda p: (p.icmpv6.is_ping_reply and (identifier is None or p.icmpv6.echo.identifier == identifier)),
**kwargs)
def filter_eth(self, **kwargs):
return self.filter(attrgetter('eth'), **kwargs)
def filter_eth_src(self, addr, **kwargs):
assert isinstance(addr, (str, EthAddr))
return self.filter(lambda p: p.eth.src == addr, **kwargs)
def filter_ipv6_dst(self, addr, **kwargs):
assert isinstance(addr, (str, Ipv6Addr))
return self.filter(lambda p: p.ipv6.dst == addr, **kwargs)
def filter_LLANMA(self, **kwargs):
return self.filter(lambda p: p.ipv6.dst == consts.LINK_LOCAL_ALL_NODES_MULTICAST_ADDRESS, **kwargs)
def filter_LLABMA(self, **kwargs):
return self.filter(lambda p: p.ipv6.dst == consts.LINK_LOCAL_ALL_BBRS_MULTICAST_ADDRESS, **kwargs)
def filter_mle(self, **kwargs):
return self.filter(attrgetter('mle'), **kwargs)
def filter_mle_cmd(self, cmd, **kwargs):
assert isinstance(cmd, int), cmd
return self.filter(lambda p: p.mle.cmd == cmd, **kwargs)
def filter_icmpv6(self, **kwargs):
return self.filter(attrgetter('icmpv6'), **kwargs)
def filter_icmpv6_nd_ns(self, target_address: Ipv6Addr):
return self.filter(lambda p:
(p.icmpv6.is_neighbor_solicitation and p.icmpv6.nd.ns.target_address == target_address))
def filter_icmpv6_nd_na(self, target_address: Ipv6Addr):
return self.filter(lambda p:
(p.icmpv6.is_neighbor_advertisement and p.icmpv6.nd.na.target_address == target_address))
def filter_has_bbr_dataset(self):
return self.filter("""
thread_nwd.tlv.server.has('16')
and thread_nwd.tlv.service.s_data.seqno is not null
and thread_nwd.tlv.service.s_data.rrdelay is not null
and thread_nwd.tlv.service.s_data.mlrtimeout is not null
""")
@@ -0,0 +1,449 @@
#!/usr/bin/env python3
#
# Copyright (c) 2019, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import logging
from typing import Tuple
from pktverify import consts
from pktverify.consts import DUA_RECENT_TIME, MLE_CHILD_ID_REQUEST, MLE_ADVERTISEMENT, MLE_CHILD_ID_RESPONSE
from pktverify.pcap_reader import PcapReader
from pktverify.summary import Summary
from pktverify.test_info import TestInfo
from pktverify.verify_result import VerifyResult
class PacketVerifier(object):
"""
Base class for packet verifiers that runs the packet verification process
"""
NET_NAME = "OpenThread"
MC_PORT = 49191
MM_PORT = 61631
LLANMA = 'ff02::1' # Link-Local All Nodes multicast address
LLARMA = 'ff02::2' # Link-Local All Routers multicast address
RLANMA = 'ff03::1' # realm-local all-nodes multicast address
RLARMA = 'ff03::2' # realm-local all-routers multicast address
RLAMFMA = 'ff03::fc' # realm-local ALL_MPL_FORWARDERS address
LLABMA = 'ff32:40:fd00:7d03:7d03:7d03:0:3' # Link-Local All BBRs multicast address
def __init__(self, test_info_path):
logging.basicConfig(level=logging.INFO,
format='File "%(pathname)s", line %(lineno)d, in %(funcName)s\n'
'%(asctime)s - %(levelname)s - %(message)s')
ti = TestInfo(test_info_path)
pkts = PcapReader.read(ti.pcap_path)
print('loaded %d packets from %s' % (len(pkts), ti.pcap_path))
self.pkts = pkts
self.test_info = ti
self.summary = Summary(pkts, ti)
self._vars = {}
self._add_initial_vars()
def add_vars(self, **vars):
"""
Add new variables.
:param vars: The new variables.
"""
self._vars.update(vars)
@property
def vars(self):
"""
:return: the dict of all variables
"""
return self._vars
def add_common_vars(self):
"""
Add common variables that is needed by many test cases.
"""
self.add_vars(
NET_NAME=PacketVerifier.NET_NAME,
MM_PORT=PacketVerifier.MM_PORT,
MC_PORT=PacketVerifier.MC_PORT,
LLANMA=PacketVerifier.LLANMA, # Link-Local All Nodes multicast address
LLARMA=PacketVerifier.LLARMA, # Link-Local All Routers multicast address
RLANMA=PacketVerifier.RLANMA, # realm-local all-nodes multicast address
RLARMA=PacketVerifier.RLARMA, # realm-local all-routers multicast address
RLAMFMA=PacketVerifier.RLAMFMA, # realm-local ALL_MPL_FORWARDERS address
LLABMA=PacketVerifier.LLABMA, # Link-Local All BBRs multicast address
MA1=consts.MA1,
MA2=consts.MA2,
MA3=consts.MA3,
MA4=consts.MA4,
MA5=consts.MA5,
MA6=consts.MA6,
MA1g=consts.MA1g,
MAe1=consts.MAe1,
MAe2=consts.MAe2,
MAe3=consts.MAe3,
)
def _add_initial_vars(self):
for i, addr in self.test_info.extaddrs.items():
name = self.test_info.get_node_name(i)
self._vars[name] = addr
for i, addr in self.test_info.ethaddrs.items():
name = self.test_info.get_node_name(i) + '_ETH'
self._vars[name] = addr
for i, addrs in self.test_info.ipaddrs.items():
name = self.test_info.get_node_name(i)
for addr in addrs:
if addr.is_dua:
key = name + '_DUA'
elif addr.is_backbone:
key = name + '_BBA'
elif addr.is_link_local:
key = name + '_LLA'
else:
logging.warning("IPv6 address ignored: name=%s, addr=%s, is_global=%s, is_link_local=%s", name,
addr, addr.is_global, addr.is_link_local)
continue
if key in self._vars:
logging.warning("duplicate IPv6 address type: name=%s, addr=%s,%s", name, addr, self._vars[key])
continue
self._vars[key] = addr
for i, addr in self.test_info.mleids.items():
name = self.test_info.get_node_name(i)
self._vars[name + '_MLEID'] = addr
for i, rloc16 in self.test_info.rloc16s.items():
key = self.test_info.get_node_name(i) + '_RLOC16'
self._vars[key] = rloc16
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)
self._vars[k] = v
def verify_dua_registration(self,
td: str,
bbr: str,
pkts=None,
dua_deadline=None,
DAD=False,
sbbr=None) -> VerifyResult:
"""
Run the packet verification for the while DAD registration, including optional steps
This is commonly used in many test cases.
:param pkts: The packet filter to verify, or self.pkts if None
:param td: TB's name.
:param bbr: BBR's name.
"""
assert self.is_wpan_device(td)
assert self.is_wpan_device(bbr) and self.is_eth_device(bbr), bbr
if pkts is None:
pkts = self.pkts
logging.info("verifying DUA registration from %s to %s ...", td, bbr)
result = VerifyResult()
TD = self.vars[td]
BBR = self.vars[bbr]
BBR_ETH = self.vars[bbr + '_ETH']
if sbbr:
SBBR_ETH = self.vars[sbbr + '_ETH']
# BBR_DUA = self.vars[bbr + '_DUA']
p = pkts.filter_wpan_src64(TD) \
.filter_coap_request("/n/dr", port=self.MM_PORT) \
.must_next()
result.record_last("/n/dr", pkts)
if dua_deadline is not None:
p.must_verify(lambda p: p.sniff_timestamp <= dua_deadline)
idx_after_n_dr = pkts.index
p.must_verify(lambda p: p.coap.tlv.target_eid and p.coap.tlv.ml_eid)
DUA = p.coap.tlv.target_eid
MLEID = p.coap.tlv.ml_eid
self.add_vars(**{td + "_DUA": DUA, td + "_MLEID": MLEID})
logging.info(f"DUA={DUA}, MLEID={MLEID}")
if DAD:
# DAD (DUA_DATA_REPEAT+1) TIMES
before_dad_index = pkts.index
after_dad_index = before_dad_index
with pkts.save_index():
for i in range(consts.DUA_DAD_REPEATS + 1):
# Step 3: PBBR - Performs DAD on the backbone link - Multicasts a BB.qry CoAP request
filter = pkts.filter_eth_src(BBR_ETH) \
.filter_LLABMA() \
.filter_coap_request("/b/bq") \
.filter(lambda p: p.coap.tlv.target_eid == DUA)
p = filter.must_next()
after_dad_index = self.max_index(after_dad_index, pkts.index)
with pkts.save_index():
# try to find the next multicast
# start_index = pkts.index
if filter.next():
pe = pkts.last()
time_gap = pe.sniff_timestamp - p.sniff_timestamp
# PBBR Waits for DUA_DAD_QUERY_TIMEOUT: Verify that DUA_DAD_QUERY_TIMEOUT time passes
assert time_gap >= consts.DUA_DAD_QUERY_TIMEOUT - 0.01, time_gap
# SBBR: Does not respond: SBBR does not respond to the BB.qry message.
if sbbr is not None:
dad_pkts_range = pkts.range(before_dad_index, after_dad_index, cascade=False)
dad_pkts_range.filter_eth_src(SBBR_ETH) \
.filter_LLABMA() \
.filter_coap_ack("/b/bq") \
.must_not_next()
# BBR updates the corresponding entry in its DUA device table.
# No pass criteria
# Step 7: BBR informs other BBRs on the network of the DUA registration.
# FIXME: Test plan requires that last_transaction_time <= 3,
# however real OT implementation can have last_transaction_time == 4
expected_last_transaction_time = 0 if not DAD else 4
pkts.filter_eth_src(BBR_ETH) \
.filter_LLABMA() \
.filter_coap_request("/b/ba") \
.filter("coap.tlv.target_eid == {DUA}", DUA=DUA) \
.must_next() \
.must_verify("""
coap.tlv.ml_eid == {MLEID}
and coap.tlv.last_transaction_time <= {expected_last_transaction_time}
and coap.tlv.net_name == {NET_NAME}
""", MLEID=MLEID, NET_NAME=self.NET_NAME,
expected_last_transaction_time=expected_last_transaction_time)
idx1 = pkts.index
# SBBR receives PRO_BB.ntf and optionally updates the corresponding entry
# in its Backup DUA Devices Table. No pass criteria.
# BBR announces itself as the new ND proxy for the roaming device
pkts.seek_back(0.2, eth=True) \
.filter_eth_src(BBR_ETH) \
.filter_LLANMA() \
.filter_icmpv6_nd_na(DUA) \
.must_next() \
.must_verify("""
icmpv6.nd.na.flag.s == 0
and icmpv6.nd.na.flag.o == 1
and icmpv6.nd.na.flag.r == 1
and icmpv6.opt.target_linkaddr == {BBR_ETH}
""", BBR_ETH=BBR_ETH)
idx2 = pkts.index
# BBR responds to the DUA registration
pkts.index = idx_after_n_dr # reset index to just after /n/dr request
pkts.filter_wpan_src64(BBR) \
.filter_coap_ack("/n/dr") \
.filter("coap.tlv.target_eid == {DUA}", DUA=DUA) \
.must_next() \
.must_verify("""
coap.tlv.target_eid == {DUA}
and coap.tlv.status == 0
""", DUA=DUA)
pkts.index = self.max_index(idx1, idx2, pkts.index)
# BBR optionally repeats the unsolicited neighbor advertisement.
# Optional 1 or 2 times
with pkts.save_index():
filter = pkts.filter_eth_src(BBR_ETH) \
.filter_LLANMA() \
.filter_icmpv6_nd_na(DUA)
for i in range(2):
if not filter.next():
break
filter.last().must_verify("""
icmpv6.nd.na.flag.s == 0
and icmpv6.nd.na.flag.o == 1
and icmpv6.nd.na.flag.r == 1
and icmpv6.opt.target_linkaddr == {BBR_ETH}
""",
BBR_ETH=BBR_ETH)
# BBR Optionally repeats the DUA registration notification
# Optional
with pkts.save_index():
filter = pkts.filter_eth_src(BBR_ETH) \
.filter_LLABMA() \
.filter_coap_request("/b/ba") \
.filter("coap.tlv.target_eid == {DUA}", DUA=DUA)
if filter.next():
p = filter.last()
p.must_verify("""
coap.tlv.ml_eid == {MLEID}
and coap.tlv.net_name == {NET_NAME}
and 3 < coap.tlv.last_transaction_time < DUA_RECENT_TIME
""",
DUA_RECENT_TIME=DUA_RECENT_TIME,
MLEID=MLEID,
NET_NAME=self.NET_NAME)
if sbbr is not None:
# SBBR: Does not respond to the ND Neighbor Solicitation message.
SBBR_ETH = self.vars[sbbr + '_ETH']
pkts_in_range = pkts.range(idx_after_n_dr, pkts.index, cascade=False)
pkts_in_range.filter_eth_src(SBBR_ETH).filter_LLANMA().filter_icmpv6_nd_na(DUA).must_not_next()
return result
def verify_attached(self, name: str, pkts=None) -> VerifyResult:
"""
Verify that the device attaches to the Thread network.
:param name: The device name.
"""
result = VerifyResult()
assert self.is_wpan_device(name), name
pkts = pkts or self.pkts
extaddr = self.vars[name]
src_pkts = pkts.filter_wpan_src64(extaddr)
src_pkts.filter_mle_cmd(MLE_CHILD_ID_REQUEST).must_next() # Child Id Request
result.record_last('child_id_request', pkts)
dst_pkts = pkts.filter_wpan_dst64(extaddr)
dst_pkts.filter_mle_cmd(MLE_CHILD_ID_RESPONSE).must_next() # Child Id Response
result.record_last('child_id_response', pkts)
with pkts.save_index():
src_pkts.filter_mle_cmd(MLE_ADVERTISEMENT).must_next() # MLE Advertisement
result.record_last('mle_advertisement', pkts)
logging.info(f"verify attached: d={name}, result={result}")
return result
def verify_ping(self, src: str, dst: str, bbr: str = None, pkts: 'PacketVerifier' = None) -> VerifyResult:
"""
Verify the ping process.
:param src: The source device name.
:param dst: The destination device name.
:param bbr: The Backbone Router name.
If specified, this method also verifies that the ping request and reply be forwarded by the Backbone Router.
:param pkts: The PacketFilter to search.
:return: The verification result.
"""
if bbr:
assert not (self.is_wpan_device(src) and self.is_wpan_device(dst)), \
f"both {src} and {dst} are WPAN devices"
assert not (self.is_eth_device(src) and self.is_eth_device(dst)), \
f"both {src} and {dst} are ETH devices"
if pkts is None:
pkts = self.pkts
src_dua = self.vars[src + '_DUA']
dst_dua = self.vars[dst + '_DUA']
if bbr:
bbr_ext = self.vars[bbr]
bbr_eth = self.vars[bbr + '_ETH']
result = VerifyResult()
ping_req = pkts.filter_ping_request().filter_ipv6_dst(dst_dua)
if self.is_eth_device(src):
p = ping_req.filter_eth_src(self.vars[src + '_ETH']).must_next()
else:
p = ping_req.filter_wpan_src64(self.vars[src]).must_next()
# pkts.last().show()
ping_id = p.icmpv6.echo.identifier
logging.info("verify_ping: ping_id=%x", ping_id)
result.record_last('ping_request', pkts)
ping_req = ping_req.filter(lambda p: p.icmpv6.echo.identifier == ping_id)
# BBR unicasts the ping packet to TD.
if bbr:
if self.is_eth_device(src):
ping_req.filter_wpan_src64(bbr_ext).must_next()
else:
ping_req.filter_eth_src(bbr_eth).must_next()
ping_reply = pkts.filter_ping_reply().filter_ipv6_dst(src_dua).filter(
lambda p: p.icmpv6.echo.identifier == ping_id)
# TD receives ping packet and responds back to Host via SBBR.
if self.is_wpan_device(dst):
ping_reply.filter_wpan_src64(self.vars[dst]).must_next()
else:
ping_reply.filter_eth_src(self.vars[dst + '_ETH']).must_next()
result.record_last('ping_reply', pkts)
if bbr:
# SBBR forwards the ping response packet to Host.
if self.is_wpan_device(dst):
ping_reply.filter_eth_src(bbr_eth).must_next()
else:
ping_reply.filter_wpan_src64(bbr_ext).must_next()
return result
def is_wpan_device(self, name: str) -> bool:
"""
Returns if the device is an WPAN device.
:param name: The device name.
Note that device can be both a WPAN device and an Ethernet device.
"""
assert isinstance(name, str), name
return name in self.test_info.extaddrs
def is_eth_device(self, name: str) -> bool:
"""
Returns if the device s an Ethernet device.
:param name: The device name.
Note that device can be both a WPAN device and an Ethernet device.
"""
assert isinstance(name, str), name
return name in self.test_info.ethaddrs
def max_index(self, *indexes: Tuple[int, int]) -> Tuple[int, int]:
wpan_idx = 0
eth_idx = 0
for wi, ei in indexes:
wpan_idx = max(wpan_idx, wi)
eth_idx = max(eth_idx, ei)
return wpan_idx, eth_idx
@@ -0,0 +1,76 @@
#!/usr/bin/env python3
#
# Copyright (c) 2019, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 'AS IS'
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import logging
import os
import subprocess
from typing import Optional
import pyshark
from pktverify import consts, utils
from pktverify.packet import Packet
from pktverify.packet_filter import PacketFilter
class PcapReader(object):
"""
Implements Pcap reading utilities.
"""
@classmethod
def read(cls, filename: str, tshark_path: Optional[str] = None) -> PacketFilter:
"""
Read packets from a given Pcap file.
:param filename: The Pcap file.
:param tshark_path: The optional path to the `tshark`.
:return: A PacketFilter containing all packets of the Pcap file.
"""
if tshark_path is None:
tshark_path = utils.which_tshark()
logging.info("Using tshark path: %s", tshark_path)
subprocess.check_call(f"{tshark_path} -v", shell=True)
os.system(f"ls -l {filename}")
filecap = pyshark.FileCapture(filename,
tshark_path=tshark_path,
override_prefs=consts.WIRESHARK_OVERRIDE_PREFS,
decode_as=consts.WIRESHARK_DECODE_AS_ENTRIES)
filecap.load_packets()
return PacketFilter(tuple(map(Packet, filecap._packets)))
if __name__ == '__main__':
pkts = PcapReader.read("../../../../merged.pcap")
print(len(pkts), 'packets loaded')
for p in pkts:
if not p.mle:
break
print("mle.cmd=%r" % p.mle.cmd)
@@ -0,0 +1,188 @@
#!/usr/bin/env python3
#
# Copyright (c) 2019, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import logging
import operator
import sys
from pktverify import consts
from pktverify.test_info import TestInfo
class NodeSummary(object):
"""
Represents a summary of a node.
"""
def __init__(self, role, extaddr):
self._role = role
self._extaddr = extaddr
self._ipaddrs = {}
@property
def role(self):
return self._role
@property
def extaddr(self):
return self._extaddr
@property
def ipaddr_link_local(self):
for a, _ in self._iter_ipaddrs_rev():
if a.is_link_local:
return a
return None
@property
def ipaddr_mleid(self):
for a, _ in self._iter_ipaddrs_rev():
if a.is_mleid:
return a
return None
def _iter_ipaddrs_rev(self):
return sorted(self._ipaddrs.items(), key=operator.itemgetter(1), reverse=True)
def add_ipaddr(self, ipaddr, index):
if ipaddr not in self._ipaddrs:
self._ipaddrs[ipaddr] = index
def __str__(self):
return "[node {role} extaddr {extaddr} ipaddrs {ipaddrs}]".format(
role=self._role,
extaddr=self.extaddr,
ipaddrs=", ".join(map(str, sorted(self._ipaddrs))),
)
__repr__ = __str__
class Summary(object):
"""
Represents a summary of the test.
"""
def __init__(self, pkts, test_info: TestInfo):
self._pkts = pkts
self._test_info = test_info
self._leader_id = None
self._analyze()
def iterroles(self):
return self._role_to_node.items()
def _analyze(self):
self._analyze_test_info()
with self._pkts.save_index():
for f in [
self._analyze_leader,
self._analyze_packets,
]:
self._pkts.index = (0, 0)
f()
def _analyze_test_info(self):
self._role_to_node = {}
self._extaddr_to_node = {}
for role, extaddr in self._test_info.extaddrs.items():
assert role not in self._role_to_node
assert extaddr not in self._extaddr_to_node
node = NodeSummary(role, extaddr)
self._role_to_node[role] = node
self._extaddr_to_node[extaddr] = node
def _analyze_leader(self):
for p in self._pkts:
if p.mle.cmd in [consts.MLE_DATA_RESPONSE, consts.MLE_ADVERTISEMENT]:
p.mle.__getattr__('tlv')
p.mle.__getattr__('tlv.leader_data')
p.mle.__getattr__('tlv.leader_data.router_id')
tlv = p.mle.tlv
if tlv.leader_data:
self._leader_id = tlv.leader_data.router_id
logging.info("leader found in pcap: %d", self._leader_id)
break
else:
logging.warning("leader not found in pcap")
def _analyze_packets(self):
for i, p in enumerate(self._pkts):
extaddr, src = None, None
# each packet should be either wpan or eth
assert (p.wpan and not p.eth) or (p.eth and not p.wpan)
if p.wpan:
# it is a 802.15.4 packet
extaddr = p.wpan.src64
if p.ipv6:
# it is a IPv6 packet
src = p.ipv6.src
if extaddr and src:
if extaddr in self._extaddr_to_node:
role_sum = self._extaddr_to_node[extaddr]
role_sum.add_ipaddr(src, i)
else:
logging.warn("Extaddr %s is not in the testbed", extaddr)
def show(self):
show_roles = "\n\t\t".join(map(str, self._role_to_node.values()))
sys.stderr.write("""{header}
Pcap Summary:
packets = {num_packets}
roles = {num_roles}
{show_roles}
{tailer}
""".format(
header='>' * 120,
num_packets=len(self._pkts),
num_roles=len(self._role_to_node),
show_roles=show_roles,
tailer='<' * 120,
))
def ipaddr_mleid_by_role(self, role):
node = self._role_to_node[role]
return node.ipaddr_mleid
def ipaddr_link_local_by_role(self, role):
node = self._role_to_node[role]
return node.ipaddr_link_local
def role(self, r):
return self._role_to_node[r]
@@ -0,0 +1,87 @@
#!/usr/bin/env python3
#
# Copyright (c) 2019, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import json
import os
from typing import Dict
from pktverify.addrs import EthAddr, ExtAddr, Ipv6Addr
class TestInfo(object):
"""
Represents the test information.
"""
def __init__(self, filename):
self.filename = filename
with open(filename, 'rt') as fd:
test_info = json.loads(fd.read())
self.testcase = test_info.get('testcase', '')
self._pcap = test_info.get('pcap', 'current.pcap')
self.topology = self._convert_keys_to_ints(test_info['topology'])
self.extaddrs = {int(k): ExtAddr(v) for k, v in test_info.get('extaddrs', {}).items()}
self.ethaddrs = {int(k): EthAddr(v) for k, v in test_info.get('ethaddrs', {}).items()}
self.ipaddrs = {int(k): [Ipv6Addr(x) for x in l] for k, l in test_info.get('ipaddrs', {}).items()}
self.mleids = {int(k): Ipv6Addr(v) for k, v in test_info.get('mleids', {}).items()}
self.rloc16s = self._convert_hex_values(self._convert_keys_to_ints(test_info.get('rloc16s', {})))
self.extra_vars = test_info.get('extra_vars', {})
def __str__(self):
macs = dict(self.extaddrs)
macs.update({k + '_ETH': v for k, v in self.ethaddrs.items()})
macs = ",\n\t".join("%s=%s" % (k, v) for k, v in macs.items())
return "TestInfo<{case}|{pcap}|\n\t{macs}>".format(case=self.testcase, pcap=self._pcap, macs=macs)
@property
def pcap_path(self) -> str:
"""
:return: The path to the Pcap file.
"""
dir = os.path.dirname(self.filename)
return os.path.join(dir, self._pcap)
@staticmethod
def _convert_keys_to_ints(d: Dict[str, any]) -> Dict[int, any]:
return {int(k): v for k, v in d.items()}
@staticmethod
def _convert_hex_values(d: Dict[any, str]) -> Dict[any, int]:
return {k: int(v, 16) for k, v in d.items()}
def get_node_name(self, node_id: int) -> str:
"""
Gets the name of the device.
:param node_id: The device ID.
:return: The name of the device.
"""
return self.topology[node_id].get('name', 'Node%d' % node_id)
@@ -0,0 +1,354 @@
#!/usr/bin/env python3
#
# Copyright (c) 2019, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
# This is a test script for checking layer fields against a given test.pcap.
#
import logging
import sys
import unittest
from pktverify import layer_fields
from pktverify.addrs import EthAddr, ExtAddr, Ipv6Addr
from pktverify.bytes import Bytes
from pktverify.consts import REAL_LAYER_NAMES, VALID_LAYER_NAMES
from pktverify.layer_fields_container import LayerFieldsContainer
from pktverify.null_field import nullField
from pktverify.packet import Packet
from pktverify.pcap_reader import PcapReader
class TestLayerFields(unittest.TestCase):
def test(self):
logging.basicConfig(level=logging.DEBUG)
pkts = PcapReader.read("test.pcap")
for p in pkts:
for layer_name in VALID_LAYER_NAMES:
if layer_name == 'lowpan': # we already checked 6lowpan
continue
layer = getattr(p, layer_name)
if hasattr(p._packet, layer_name):
if layer_name in REAL_LAYER_NAMES:
self.assertTrue(layer)
checker = getattr(self, '_test_' + layer_name, None)
if checker is None:
continue
try:
checker(p)
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)
def _test_coap(self, p):
coap = p.coap
self.assertIsInstance(coap.version, int)
self.assertIsInstance(coap.type, int)
self.assertIsInstance(coap.token_len, int)
self.assertIsInstance(coap.code, int)
self.assertIsInstance(coap.mid, int)
self.assertIsInstanceOrNull(coap.token, int)
self.assertIsInstanceOrNull(coap.opt.uri_path_recon, str)
self.assertIsInstanceOrNull(coap.payload, bytearray)
print(p.coap.tlv.type, p.coap.tlv)
assert isinstance(coap.tlv, LayerFieldsContainer), repr(coap.tlv)
self.assertIsInstanceOrNull(coap.tlv.type, list)
def _test_mle(self, p):
mle = p.mle
self._must_have_wpan_aux_sec(p)
self.assertIsInstance(mle.cmd, int)
self.assertIsInstanceOrNull(mle.tlv.mode.receiver_on_idle, int)
self.assertIsInstanceOrNull(mle.tlv.mode.reserved1, int)
self.assertIsInstanceOrNull(mle.tlv.mode.reserved2, int)
self.assertIsInstanceOrNull(mle.tlv.mode.device_type_bit, int)
self.assertIsInstanceOrNull(mle.tlv.mode.network_data, int)
self.assertIsInstanceOrNull(mle.tlv.challenge, Bytes)
self.assertIsInstanceOrNull(mle.tlv.scan_mask.r, int)
self.assertIsInstanceOrNull(mle.tlv.scan_mask.e, int)
self.assertIsInstanceOrNull(mle.tlv.version, int)
self.assertIsInstanceOrNull(mle.tlv.source_addr, int)
self.assertIsInstanceOrNull(mle.tlv.active_tstamp, int)
self.assertIsInstanceOrNull(mle.tlv.leader_data.partition_id, int)
self.assertIsInstanceOrNull(mle.tlv.leader_data.weighting, int)
self.assertIsInstanceOrNull(mle.tlv.leader_data.data_version, int)
self.assertIsInstanceOrNull(mle.tlv.leader_data.stable_data_version, int)
self.assertIsInstanceOrNull(mle.tlv.leader_data.router_id, int)
def _test_wpan(self, p):
wpan = p.wpan
self.assertIsInstance(wpan.fcf, int)
self.assertIsInstance(wpan.fcs, int)
self.assertIsInstance(wpan.security, int)
self.assertIsInstance(wpan.pending, int)
self.assertIsInstance(wpan.ack_request, int)
self.assertIsInstance(wpan.pan_id_compression, int)
self.assertIsInstance(wpan.seqno_suppression, int)
self.assertIsInstance(wpan.ie_present, int)
self.assertIsInstance(wpan.dst_addr_mode, int)
self.assertIsInstance(wpan.version, int)
self.assertIsInstance(wpan.src_addr_mode, int)
self.assertIsInstance(wpan.seq_no, int)
if not wpan.is_ack:
self.assertIsInstanceOrNull(wpan.dst_pan, int)
self.assertIsInstanceOrNull(wpan.dst16, int)
self.assertIsInstanceOrNull(wpan.src16, int)
self.assertIsInstanceOrNull(wpan.src64, ExtAddr)
self.assertIsInstanceOrNull(wpan.dst64, ExtAddr)
if wpan.aux_sec:
self._must_have_wpan_aux_sec(p)
def _must_have_wpan_aux_sec(self, p):
wpan = p.wpan
self.assertIsInstanceOrNull(wpan.aux_sec.sec_suite, int)
self.assertIsInstanceOrNull(wpan.aux_sec.security_control_field, int)
self.assertIsInstanceOrNull(wpan.aux_sec.sec_level, int)
self.assertIsInstanceOrNull(wpan.aux_sec.key_id_mode, int)
self.assertIsInstanceOrNull(wpan.aux_sec.frame_counter_suppression, int)
self.assertIsInstanceOrNull(wpan.aux_sec.asn_in_nonce, int)
self.assertIsInstanceOrNull(wpan.aux_sec.reserved, int)
self.assertIsInstanceOrNull(wpan.aux_sec.frame_counter, int)
self.assertIsInstanceOrNull(wpan.aux_sec.key_source, int)
self.assertIsInstanceOrNull(wpan.aux_sec.key_index, int)
def assertIsInstanceOrNull(self, field, type):
if field is not nullField:
self.assertIsInstance(field, type)
def _test_thread_bl(self, p):
thread_bl = p.thread_bl
self.assertTrue(thread_bl)
self.assertIsInstanceOrNull(thread_bl.tlv.target_eid, Ipv6Addr)
self.assertIsInstanceOrNull(thread_bl.tlv.ml_eid, ExtAddr)
self.assertIsInstanceOrNull(thread_bl.tlv.last_transaction_time, int)
self.assertIsInstanceOrNull(p.thread_meshcop.tlv.net_name, str)
def _test_thread_meshcop(self, p: Packet):
thread_meshcop = p.thread_meshcop
for layer in sorted(p.layers, key=lambda l: l.layer_name):
if 'thread_meshcop.tlv.commissioner_sess_id' in layer._layer._all_fields:
self.assertIsInstance(thread_meshcop.tlv.commissioner_sess_id, int)
if 'thread_meshcop.tlv.net_name' in layer._layer._all_fields:
self.assertIsInstance(thread_meshcop.tlv.net_name, str)
if 'thread_meshcop.tlv.channel_page' in layer._layer._all_fields:
self.assertIsInstance(thread_meshcop.tlv.channel_page, int)
if 'thread_meshcop.tlv.channel' in layer._layer._all_fields:
self.assertIsInstance(thread_meshcop.tlv.channel, int)
if 'thread_meshcop.tlv.chan_mask_page' in layer._layer._all_fields:
self.assertIsInstance(thread_meshcop.tlv.chan_mask_page, int)
if 'thread_meshcop.tlv.chan_mask_len' in layer._layer._all_fields:
self.assertIsInstance(thread_meshcop.tlv.chan_mask_len, int)
if 'thread_meshcop.tlv.chan_mask_mask' in layer._layer._all_fields:
self.assertIsInstance(thread_meshcop.tlv.chan_mask_mask, int)
if 'thread_meshcop.tlv.panid' in layer._layer._all_fields:
self.assertIsInstance(thread_meshcop.tlv.panid, int)
if 'thread_meshcop.tlv.ml_prefix' in layer._layer._all_fields:
self.assertIsInstance(thread_meshcop.tlv.ml_prefix, Bytes)
if 'thread_meshcop.tlv.master_key' in layer._layer._all_fields:
self.assertIsInstance(thread_meshcop.tlv.master_key, Bytes)
if 'thread_meshcop.tlv.pskc' in layer._layer._all_fields:
self.assertIsInstance(thread_meshcop.tlv.pskc, Bytes)
if 'thread_meshcop.tlv.sec_policy_rot' in layer._layer._all_fields:
self.assertIsInstance(thread_meshcop.tlv.sec_policy_rot, int)
if 'thread_meshcop.tlv.sec_policy_o' in layer._layer._all_fields:
self.assertIsInstance(thread_meshcop.tlv.sec_policy_o, int)
if 'thread_meshcop.tlv.sec_policy_n' in layer._layer._all_fields:
self.assertIsInstance(thread_meshcop.tlv.sec_policy_n, int)
if 'thread_meshcop.tlv.sec_policy_r' in layer._layer._all_fields:
self.assertIsInstance(thread_meshcop.tlv.sec_policy_r, int)
if 'thread_meshcop.tlv.sec_policy_c' in layer._layer._all_fields:
self.assertIsInstance(thread_meshcop.tlv.sec_policy_c, int)
if 'thread_meshcop.tlv.sec_policy_b' in layer._layer._all_fields:
self.assertIsInstance(thread_meshcop.tlv.sec_policy_b, int)
if 'thread_meshcop.tlv.pan_id' in layer._layer._all_fields:
self.assertIsInstance(thread_meshcop.tlv.pan_id, int)
if 'thread_meshcop.tlv.xpan_id' in layer._layer._all_fields:
self.assertIsInstance(thread_meshcop.tlv.xpan_id, Bytes)
for field in layer._layer._all_fields:
if field.startswith('thread_meshcop') and not layer_fields.is_layer_field(field) and field not in (
'thread_meshcop.tlv', 'thread_meshcop.tlv.type', 'thread_meshcop.tlv.len8'):
print('found %s = %s in layer %s' % (
field,
layer._layer.get_field(field),
layer.layer_name,
))
def _test_icmpv6(self, p):
icmpv6 = p.icmpv6
self.assertTrue(p.icmpv6)
self.assertIsInstance(icmpv6.type, int)
self.assertIsInstance(icmpv6.code, int)
self.assertIsInstance(icmpv6.checksum, int)
self.assertIsInstanceOrNull(icmpv6.reserved, int)
self.assertIsInstanceOrNull(icmpv6.nd.na.flag.s, int)
self.assertIsInstanceOrNull(icmpv6.nd.na.flag.o, int)
self.assertIsInstanceOrNull(icmpv6.nd.na.flag.r, int)
self.assertIsInstanceOrNull(icmpv6.nd.na.flag.rsv, int)
self.assertIsInstanceOrNull(icmpv6.nd.ra.cur_hop_limit, int)
self.assertIsInstanceOrNull(icmpv6.mldr.nb_mcast_records, int)
self.assertIsInstanceOrNull(icmpv6.nd.ns.target_address, Ipv6Addr)
self.assertIsInstanceOrNull(icmpv6.mldr.mar.multicast_address, list)
def get_field(self, p: Packet, f):
secs = f.split('.')
assert len(secs) >= 2
v = p
for sec in secs:
v = getattr(v, sec)
return v
def _test_6lowpan(self, p):
lowpan = p.lowpan
assert lowpan is getattr(p, '6lowpan')
self.assertIsInstanceOrNull(lowpan.src, Ipv6Addr)
self.assertIsInstanceOrNull(lowpan.dst, Ipv6Addr)
self.assertIsInstanceOrNull(lowpan.udp.src, int)
self.assertIsInstanceOrNull(lowpan.udp.dst, int)
self.assertIsInstanceOrNull(lowpan.udp.checksum, int)
self.assertIsInstanceOrNull(lowpan.frag.size, int)
self.assertIsInstanceOrNull(lowpan.frag.tag, int)
self.assertIsInstanceOrNull(lowpan.frag.offset, int)
self.assertIsInstanceOrNull(lowpan.nhc.pattern, list)
self.assertIsInstanceOrNull(lowpan.nhc.udp.checksum, int)
self.assertIsInstanceOrNull(lowpan.nhc.udp.ports, int)
self.assertIsInstanceOrNull(lowpan.pattern, list)
self.assertIsInstanceOrNull(lowpan.iphc.tf, int)
self.assertIsInstanceOrNull(lowpan.iphc.nh, int)
self.assertIsInstanceOrNull(lowpan.iphc.hlim, int)
self.assertIsInstanceOrNull(lowpan.iphc.cid, int)
self.assertIsInstanceOrNull(lowpan.iphc.sac, int)
self.assertIsInstanceOrNull(lowpan.iphc.sam, int)
self.assertIsInstanceOrNull(lowpan.iphc.m, int)
self.assertIsInstanceOrNull(lowpan.iphc.dac, int)
self.assertIsInstanceOrNull(lowpan.iphc.dam, int)
self.assertIsInstanceOrNull(lowpan.iphc.sctx.prefix, Bytes)
self.assertIsInstanceOrNull(lowpan.iphc.dctx.prefix, Bytes)
def _test_ip(self, p):
pass
def _test_ipv6(self, p):
pass
def _test_udp(self, p):
pass
def _test_eth(self, p):
eth = p.eth
self.assertIsInstance(eth.src, EthAddr)
self.assertIsInstance(eth.dst, EthAddr)
self.assertIsInstance(eth.type, int)
def _check_missing_fields(self, p, layer_name, _layer):
for f in sorted(_layer._all_fields.keys(), reverse=True):
if f.startswith('_ws') or f.startswith('data'):
continue
logging.info('_check_missing_fields: %s = %r' % (f, _layer._all_fields[f]))
if f in {
'', 'icmpv6.checksum.status', 'ip.ttl.lncb', 'wpan.aux_sec.key_source.bytes', 'wpan.src64.origin'
}:
# TODO: handle these fields
continue
v = _layer._all_fields[f]
if layer_fields.is_layer_field_container(f):
continue
try:
rv = self.get_field(p, f)
self.assertIsNot(rv, nullField)
parser = layer_fields._LAYER_FIELDS[f]
if isinstance(parser, layer_fields._first):
parser = parser._sub_parse
if parser in (layer_fields._raw_hex, layer_fields._hex, layer_fields._raw_hex_rev, layer_fields._dec,
layer_fields._auto):
self.assertIsInstance(rv, int)
elif isinstance(parser, layer_fields._list):
self.assertIsInstance(rv, list)
elif parser is layer_fields._ipv6_addr:
self.assertIsInstance(rv, Ipv6Addr)
elif parser is layer_fields._eth_addr:
self.assertIsInstance(rv, EthAddr)
elif parser is layer_fields._ext_addr:
self.assertIsInstance(rv, ExtAddr)
elif parser is layer_fields._str:
self.assertIsInstance(rv, str)
elif parser is layer_fields._bytes:
self.assertIsInstance(rv, Bytes)
elif parser is layer_fields._payload:
self.assertIsInstance(rv, bytearray)
elif parser is layer_fields._float:
self.assertIsInstance(rv, float)
else:
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)
raise
@@ -0,0 +1,152 @@
#!/usr/bin/env python3
#
# Copyright (c) 2019, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
import logging
import os
import subprocess
import sys
from typing import Callable, Union
from pktverify.addrs import EthAddr, ExtAddr, Ipv6Addr
from pktverify.bytes import Bytes
from pktverify.null_field import nullField
def make_filter_func(func: Union[str, Callable], **vars) -> Callable:
"""
Convert the filter to a callable function if it's a string.
:param func: The filter string or callable.
:param vars: The variables.
:return: The filter callable.
"""
if isinstance(func, str):
# if func is a string, compile it to a function
func = func.format_map({k: repr(v) for k, v in vars.items()}).strip()
print("\t%s" % func, file=sys.stderr)
code = compile('(\n' + func + '\n)', func, "eval")
def func(p):
return eval(
code, None, {
'p': p,
'coap': p.coap,
'wpan': p.wpan,
'mle': p.mle,
'ipv6': p.ipv6,
'lowpan': p.lowpan,
'eth': p.eth,
'icmpv6': p.icmpv6,
'udp': p.udp,
'thread_bl': p.thread_bl,
'thread_meshcop': p.thread_meshcop,
'Bytes': Bytes,
'ExtAddr': ExtAddr,
'Ipv6Addr': Ipv6Addr,
'EthAddr': EthAddr,
'thread_nm': p.thread_nm,
'thread_nwd': p.thread_nwd,
'null': nullField,
})
else:
assert not vars, 'can not provide vars for non-str filter: %r %r' % (func, vars)
assert callable(func)
return func
def _install_travis_thread_wireshark():
logging.info("downloading thread-wireshark from https://github.com/openthread/wireshark/releases ...")
download_url = 'https://github.com/openthread/wireshark/releases/download/ot-pktverify-20200727/thread-wireshark.tar.gz'
save_file = '/tmp/thread-wireshark.tar.gz'
subprocess.check_call(f'curl -L {download_url} -o {save_file}', shell=True)
subprocess.check_call(f'tar -C /tmp -xvzf {save_file}', shell=True)
assert os.path.isdir('/tmp/thread-wireshark')
def _setup_wireshark_disabled_protos():
home = os.environ['HOME']
wireshark_config_dir = os.path.join(home, '.config', 'wireshark')
os.makedirs(wireshark_config_dir, exist_ok=True)
disabled_protos_path = os.path.join(wireshark_config_dir, 'disabled_protos')
# read current disabled protos
try:
with open(disabled_protos_path, 'rt') as fd:
disabled_protos = set(l.strip() for l in fd if l.strip() != '')
except FileNotFoundError:
disabled_protos = set()
old_disabled_protos_num = len(disabled_protos)
disabled_protos.add('lwm')
disabled_protos.add('prp')
disabled_protos.add('stcsig')
disabled_protos.add('transum')
disabled_protos.add('zbee_nwk')
disabled_protos.add('zbee_nwk_gp')
if len(disabled_protos) > old_disabled_protos_num:
logging.info(f"set disabled_protos = {' '.join(disabled_protos)}")
with open(disabled_protos_path, 'wt') as fd:
fd.write('\n'.join(sorted(disabled_protos)))
fd.write('\n')
def get_wireshark_dir() -> str:
"""
:return: The path to wireshark directory.
"""
dir = '/tmp/thread-wireshark'
if not os.path.exists(dir):
_install_travis_thread_wireshark()
_setup_wireshark_disabled_protos()
return dir
def which_tshark() -> str:
"""
:return: The path to `tshark` executable.
"""
return os.path.join(get_wireshark_dir(), 'tshark')
def which_dumpcap() -> str:
"""
:return: The path to `dumpcap` executable.
"""
return os.path.join(get_wireshark_dir(), 'dumpcap')
def which_mergecap() -> str:
"""
:return: The path to `mergecap` executable.
"""
return os.path.join(get_wireshark_dir(), 'mergecap')
@@ -0,0 +1,85 @@
#!/usr/bin/env python3
#
# Copyright (c) 2019, The OpenThread Authors.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 1. Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
# 2. Redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution.
# 3. Neither the name of the copyright holder nor the
# names of its contributors may be used to endorse or promote products
# derived from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
from pktverify.packet import Packet
from pktverify.packet_filter import PacketFilter
class VerifyResult(object):
"""
Represents a verification result of a packet.
"""
__slots__ = ('_packet_found', '_packet_indexes', '_seek_indexes')
def __init__(self):
self._packet_found = {}
self._packet_indexes = {}
self._seek_indexes = {}
def record_last(self, name: str, pkts: PacketFilter) -> None:
"""
Record the information of the last found packet.
:param name: The record name.
:param pkts: The packet filter.
"""
assert name not in self._packet_found, f'duplicate name: {name}'
self._packet_found[name] = pkts.last()
self._packet_indexes[name] = pkts.last_index
self._seek_indexes[name] = pkts.index
def packet_index(self, name: str) -> int:
"""
Returns a recorded packet index.
:param name: The record name.
:return: The packet index.
"""
return self._packet_indexes[name]
def packet(self, name: str) -> Packet:
"""
Returns the recorded packet.
:param name: The record name.
:return: The packet.
"""
return self._packet_found[name]
def seek_index(self, name: str) -> tuple:
"""
Returns the recorded seek index.
:param name: The record name.
:return: The seek index.
"""
return self._seek_indexes[name]
def __str__(self):
return "VerifyResult%s" % self._packet_found
@@ -1,3 +1,4 @@
ipaddress
pexpect
pycryptodome
pyshark==0.4.2.11
+14 -3
View File
@@ -112,7 +112,13 @@ class RealTime(BaseSimulator):
time.sleep(duration)
def stop(self):
pass
if self.is_running:
# self._sniffer.stop() # FIXME: seems it blocks forever
self._sniffer = None
@property
def is_running(self):
return self._sniffer is not None
class VirtualTime(BaseSimulator):
@@ -170,8 +176,13 @@ class VirtualTime(BaseSimulator):
self.stop()
def stop(self):
self.sock.close()
self.sock = None
if self.sock:
self.sock.close()
self.sock = None
@property
def is_running(self):
return self.sock is not None
def _add_message(self, nodeid, message_obj):
addr = ('127.0.0.1', self.port + nodeid)
+118 -3
View File
@@ -27,14 +27,23 @@
# POSSIBILITY OF SUCH DAMAGE.
#
import json
import os
import subprocess
import sys
import time
import unittest
import config
import debug
from node import Node
PACKET_VERIFICATION = int(os.getenv('PACKET_VERIFICATION', 0))
if PACKET_VERIFICATION:
from pktverify.addrs import ExtAddr
from pktverify.packet_verifier import PacketVerifier
PORT_OFFSET = int(os.getenv('PORT_OFFSET', "0"))
DEFAULT_PARAMS = {
@@ -73,6 +82,11 @@ class TestCase(NcpSupportMixin, unittest.TestCase):
TOPOLOGY = None
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._start_time = None
self._do_packet_verification = PACKET_VERIFICATION and hasattr(self, 'verify')
def setUp(self):
"""Create simulator, nodes and apply configurations.
"""
@@ -81,18 +95,21 @@ class TestCase(NcpSupportMixin, unittest.TestCase):
self.simulator = config.create_default_simulator()
self.nodes = {}
initial_topology = {}
self._initial_topology = initial_topology = {}
for i, params in self.TOPOLOGY.items():
if params:
params = dict(DEFAULT_PARAMS, **params)
else:
params = DEFAULT_PARAMS.copy()
initial_topology[i] = params
self.nodes[i] = Node(
i,
params['is_mtd'],
simulator=self.simulator,
name=params.get('name'),
version=params['version'],
is_bbr=params['is_bbr'],
)
@@ -159,6 +176,7 @@ class TestCase(NcpSupportMixin, unittest.TestCase):
self.nodes[i].enable_whitelist()
self._inspector = debug.Inspector(self)
self._collect_test_info_after_setup()
def inspect(self):
self._inspector.inspect()
@@ -166,13 +184,25 @@ class TestCase(NcpSupportMixin, unittest.TestCase):
def tearDown(self):
"""Destroy nodes and simulator.
"""
if self._do_packet_verification and os.uname().sysname != "Linux":
raise NotImplementedError(
f'{self.testcase_name}: Packet Verification not available on {os.uname().sysname} (Linux only).')
if self._do_packet_verification:
time.sleep(3)
for node in list(self.nodes.values()):
node.stop()
node.destroy()
self.simulator.stop()
del self.nodes
del self.simulator
if self._do_packet_verification:
self._test_info['pcap'] = self._get_pcap_filename()
test_info_path = self._output_test_info()
os.environ['LD_LIBRARY_PATH'] = '/tmp/thread-wireshark'
self._verify_packets(test_info_path)
def flush_all(self):
"""Flush away all captured messages of all nodes.
@@ -196,3 +226,88 @@ class TestCase(NcpSupportMixin, unittest.TestCase):
Clean up node files in tmp directory
"""
os.system(f"rm -f tmp/{PORT_OFFSET}_*.flash tmp/{PORT_OFFSET}_*.data tmp/{PORT_OFFSET}_*.swap")
def _verify_packets(self, test_info_path: str):
pv = PacketVerifier(test_info_path)
pv.add_common_vars()
self.verify(pv)
print("Packet verification passed: %s" % test_info_path, file=sys.stderr)
@property
def testcase_name(self):
return os.path.splitext(os.path.basename(sys.argv[0]))[0]
def collect_ipaddrs(self):
if not self._do_packet_verification:
return
test_info = self._test_info
for i, node in self.nodes.items():
ipaddrs = node.get_addrs()
test_info['ipaddrs'][i] = ipaddrs
mleid = node.get_mleid()
test_info['mleids'][i] = mleid
def collect_rloc16s(self):
if not self._do_packet_verification:
return
test_info = self._test_info
test_info['rloc16s'] = {}
for i, node in self.nodes.items():
test_info['rloc16s'][i] = '0x%04x' % node.get_addr16()
def collect_extra_vars(self, **vars):
if not self._do_packet_verification:
return
for k in vars.keys():
assert isinstance(k, str), k
test_vars = self._test_info.setdefault("extra_vars", {})
test_vars.update(vars)
def _collect_test_info_after_setup(self):
"""
Collect test info after setUp
"""
if not self._do_packet_verification:
return
test_info = self._test_info = {
'testcase': self.testcase_name,
'start_time': time.ctime(self._start_time),
'pcap': '',
'extaddrs': {},
'ethaddrs': {},
'ipaddrs': {},
'mleids': {},
'topology': self._initial_topology,
}
for i, node in self.nodes.items():
extaddr = node.get_addr64()
test_info['extaddrs'][i] = ExtAddr(extaddr).format_octets()
def _output_test_info(self):
"""
Output test info to json file after tearDown
"""
filename = f'{self.testcase_name}.json'
with open(filename, 'wt') as ofd:
ofd.write(json.dumps(self._test_info, indent=1, sort_keys=True))
return filename
def _get_pcap_filename(self):
current_pcap = os.getenv('TEST_NAME', 'current') + '.pcap'
return os.path.abspath(current_pcap)
def assure_run_ok(self, cmd, shell=False):
if not shell and isinstance(cmd, str):
cmd = cmd.split()
proc = subprocess.run(cmd, stdout=sys.stdout, stderr=sys.stderr, shell=shell)
print(">>> %s => %d" % (cmd, proc.returncode), file=sys.stderr)
proc.check_returncode()