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

[scripts] test multicast routing across Thread and Backbone (#5852)

Browse Source 
- This commit fixes some issue in multicast forwarding from Thread to
  Backbone.
- It also adds a test to verify that multicast forwarding works across
  Thread and Backbone.
  - Test script tests/scripts/thread-cert/mcast6.py is added for Host
    to subscribe to a multicast address

The ping reply can not reach ROUTER1 since DUA feature is not
complete.
This commit is contained in:
Simon Lin
2020-12-09 08:49:07 -08:00
committed by GitHub
parent 80c180d60d
commit 9380ffd07c
8 changed files with 425 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/net/ip6.cpp
+37 -28
View File
@@ -533,7 +533,7 @@ void Ip6::HandleSendQueue(void)
}
}
otError Ip6::HandleOptions(Message &aMessage, Header &aHeader, bool aForward, bool &aReceive)
otError Ip6::HandleOptions(Message &aMessage, Header &aHeader, bool aIsOutbound, bool &aReceive)
{
otError error = OT_ERROR_NONE;
HopByHopHeader hbhHeader;
@@ -563,7 +563,7 @@ otError Ip6::HandleOptions(Message &aMessage, Header &aHeader, bool aForward, bo
switch (optionHeader.GetType())
{
case OptionMpl::kType:
SuccessOrExit(error = mMpl.ProcessOption(aMessage, aHeader.GetSource(), aForward, aReceive));
SuccessOrExit(error = mMpl.ProcessOption(aMessage, aHeader.GetSource(), aIsOutbound, aReceive));
break;
default:
@@ -890,7 +890,7 @@ otError Ip6::HandleExtensionHeaders(Message & aMessage,
MessageInfo &aMessageInfo,
Header & aHeader,
uint8_t & aNextHeader,
bool aForward,
bool aIsOutbound,
bool aFromNcpHost,
bool & aReceive)
{
@@ -904,7 +904,7 @@ otError Ip6::HandleExtensionHeaders(Message & aMessage,
switch (aNextHeader)
{
case kProtoHopOpts:
SuccessOrExit(error = HandleOptions(aMessage, aHeader, aForward, aReceive));
SuccessOrExit(error = HandleOptions(aMessage, aHeader, aIsOutbound, aReceive));
break;
case kProtoFragment:
@@ -915,7 +915,7 @@ otError Ip6::HandleExtensionHeaders(Message & aMessage,
break;
case kProtoDstOpts:
SuccessOrExit(error = HandleOptions(aMessage, aHeader, aForward, aReceive));
SuccessOrExit(error = HandleOptions(aMessage, aHeader, aIsOutbound, aReceive));
break;
case kProtoIp6:
@@ -1127,15 +1127,16 @@ otError Ip6::HandleDatagram(Message &aMessage, Netif *aNetif, const void *aLinkM
MessageInfo messageInfo;
Header header;
bool receive;
bool forward;
bool forwardThread;
bool forwardHost;
bool multicastPromiscuous;
bool shouldFreeMessage;
bool shouldForwardToHost;
uint8_t nextHeader;
start:
receive = false;
forward = false;
forwardThread = false;
forwardHost = false;
multicastPromiscuous = false;
shouldFreeMessage = true;
@@ -1158,7 +1159,7 @@ start:
if (header.GetDestination().IsMulticastLargerThanRealmLocal() &&
Get<ChildTable>().HasSleepyChildWithAddress(header.GetDestination()))
{
forward = true;
forwardThread = true;
}
#endif
@@ -1166,11 +1167,13 @@ start:
}
else
{
forward = true;
forwardThread = true;
netif = &Get<ThreadNetif>();
}
forwardHost = header.GetDestination().IsMulticastLargerThanRealmLocal();
if ((aNetif != nullptr || aMessage.GetMulticastLoop()) && netif->IsMulticastSubscribed(header.GetDestination()))
{
receive = true;
@@ -1182,17 +1185,24 @@ start:
}
else
{
// unicast
if (Get<ThreadNetif>().HasUnicastAddress(header.GetDestination()))
{
receive = true;
}
else if (!header.GetDestination().IsLinkLocal())
{
forward = true;
forwardThread = true;
}
else if (aNetif == nullptr)
{
forward = true;
forwardThread = true;
}
if (forwardThread && !ShouldForwardToThread(messageInfo, aFromNcpHost))
{
forwardThread = false;
forwardHost = true;
}
}
@@ -1200,11 +1210,9 @@ start:
// process IPv6 Extension Headers
nextHeader = static_cast<uint8_t>(header.GetNextHeader());
SuccessOrExit(error = HandleExtensionHeaders(aMessage, aNetif, messageInfo, header, nextHeader, forward,
SuccessOrExit(error = HandleExtensionHeaders(aMessage, aNetif, messageInfo, header, nextHeader, aNetif == nullptr,
aFromNcpHost, receive));
shouldForwardToHost = forward && !ShouldForwardToThread(messageInfo, aFromNcpHost);
// process IPv6 Payload
if (receive)
{
@@ -1233,31 +1241,32 @@ start:
error = ProcessReceiveCallback(aMessage, messageInfo, nextHeader, aFromNcpHost, Message::kCopyToUse);
if ((error == OT_ERROR_NONE || error == OT_ERROR_NO_ROUTE) && shouldForwardToHost)
if ((error == OT_ERROR_NONE || error == OT_ERROR_NO_ROUTE) && forwardHost)
{
forward = false;
forwardHost = false;
}
error =
HandlePayload(aMessage, messageInfo, nextHeader, (forward ? Message::kCopyToUse : Message::kTakeCustody));
shouldFreeMessage = forward;
error = HandlePayload(aMessage, messageInfo, nextHeader,
(forwardThread || forwardHost ? Message::kCopyToUse : Message::kTakeCustody));
shouldFreeMessage = forwardThread || forwardHost;
}
else if (multicastPromiscuous)
{
IgnoreError(ProcessReceiveCallback(aMessage, messageInfo, nextHeader, aFromNcpHost, Message::kCopyToUse));
}
if (forward)
if (forwardHost)
{
// try passing to host
error = ProcessReceiveCallback(aMessage, messageInfo, nextHeader, aFromNcpHost,
forwardThread ? Message::kCopyToUse : Message::kTakeCustody);
shouldFreeMessage = forwardThread;
}
if (forwardThread)
{
uint8_t hopLimit;
if (shouldForwardToHost)
{
// try passing to host
error = ProcessReceiveCallback(aMessage, messageInfo, nextHeader, aFromNcpHost, Message::kTakeCustody);
ExitNow(shouldFreeMessage = false);
}
if (aNetif != nullptr)
{
VerifyOrExit(mForwardingEnabled);
src/core/net/ip6.hpp
+2 -2
View File
@@ -341,7 +341,7 @@ private:
MessageInfo &aMessageInfo,
Header & aHeader,
uint8_t & aNextHeader,
bool aForward,
bool aIsOutbound,
bool aFromNcpHost,
bool & aReceive);
otError FragmentDatagram(Message &aMessage, uint8_t aIpProto);
@@ -356,7 +356,7 @@ private:
otError AddTunneledMplOption(Message &aMessage, Header &aHeader, MessageInfo &aMessageInfo);
otError InsertMplOption(Message &aMessage, Header &aHeader, MessageInfo &aMessageInfo);
otError RemoveMplOption(Message &aMessage);
otError HandleOptions(Message &aMessage, Header &aHeader, bool aForward, bool &aReceive);
otError HandleOptions(Message &aMessage, Header &aHeader, bool aIsOutbound, bool &aReceive);
otError HandlePayload(Message & aMessage,
MessageInfo & aMessageInfo,
uint8_t aIpProto,
tests/scripts/thread-cert/backbone/test_mlr_multicast_routing_across_thread_pans.py
+233
View File
@@ -0,0 +1,233 @@
#!/usr/bin/env python3
#
# Copyright (c) 2020, 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 test verifies that the basic MLR feature works.
#
import unittest
import config
import thread_cert
from pktverify.packet_verifier import PacketVerifier
CH1 = 11
CH2 = 22
PBBR1 = 1
ROUTER1 = 2
PBBR2 = 3
ROUTER2 = 4
HOST = 5
MA1 = 'ff05::1234:777a:1'
MA2 = 'ff05::1234:777a:2'
BBR_REGISTRATION_JITTER = 1
WAIT_REDUNDANCE = 3
class TestMlr(thread_cert.TestCase):
USE_MESSAGE_FACTORY = False
# Topology:
# --------(eth)-----------
# | | |
# PBBR HOST PBBR2
# | |
# ROUTER1 ROUTER2
#
TOPOLOGY = {
PBBR1: {
'name': 'PBBR1',
'allowlist': [ROUTER1],
'is_otbr': True,
'version': '1.2',
'router_selection_jitter': 1,
'bbr_registration_jitter': BBR_REGISTRATION_JITTER,
'channel': CH1,
},
ROUTER1: {
'name': 'ROUTER1',
'allowlist': [PBBR1],
'version': '1.2',
'router_selection_jitter': 1,
'channel': CH1,
},
PBBR2: {
'name': 'PBBR2',
'allowlist': [ROUTER2],
'is_otbr': True,
'version': '1.2',
'router_selection_jitter': 1,
'bbr_registration_jitter': BBR_REGISTRATION_JITTER,
'channel': CH2,
},
ROUTER2: {
'name': 'ROUTER2',
'allowlist': [PBBR2],
'version': '1.2',
'router_selection_jitter': 1,
'channel': CH2,
},
HOST: {
'name': 'Host',
'is_host': True
},
}
def test(self):
# Bring up Host
self.nodes[HOST].start()
self.nodes[HOST].add_ipmaddr(MA2, backbone=True)
# Bring up PBBR1
self.nodes[PBBR1].start()
self.simulator.go(5)
self.assertEqual('leader', self.nodes[PBBR1].get_state())
self.nodes[PBBR1].enable_backbone_router()
self.simulator.go(10)
self.assertTrue(self.nodes[PBBR1].is_primary_backbone_router)
self.nodes[PBBR1].add_prefix(config.DOMAIN_PREFIX, "parosD")
self.nodes[PBBR1].register_netdata()
self.simulator.go(WAIT_REDUNDANCE)
# Bring up ROUTER1
self.nodes[ROUTER1].start()
self.simulator.go(5)
self.assertEqual('router', self.nodes[ROUTER1].get_state())
self.nodes[ROUTER1].add_ipmaddr(MA1)
self.simulator.go(WAIT_REDUNDANCE)
# Bring up PBBR2
self.nodes[PBBR2].start()
self.simulator.go(5)
self.assertEqual('leader', self.nodes[PBBR2].get_state())
self.nodes[PBBR2].enable_backbone_router()
self.simulator.go(10)
self.assertTrue(self.nodes[PBBR2].is_primary_backbone_router)
self.nodes[PBBR2].add_prefix(config.DOMAIN_PREFIX, "parosD")
self.nodes[PBBR2].register_netdata()
self.simulator.go(WAIT_REDUNDANCE)
# Bring up ROUTER2
self.nodes[ROUTER2].start()
self.simulator.go(5)
self.assertEqual('router', self.nodes[ROUTER1].get_state())
self.nodes[ROUTER2].add_ipmaddr(MA1)
self.nodes[ROUTER2].add_ipmaddr(MA2)
self.simulator.go(WAIT_REDUNDANCE)
self.collect_ipaddrs()
self.collect_rloc16s()
# ping MA1 from Host could get replied from R1 and R2
self.assertTrue(self.nodes[HOST].ping(MA1, backbone=True, ttl=5))
self.simulator.go(WAIT_REDUNDANCE)
# ping MA2 from R1 could get replied from Host and R2
# TODO (DUA): ping reply can not reach ROUTER1 since DUA feature is not complete.
self.assertFalse(self.nodes[ROUTER1].ping(MA2))
self.simulator.go(WAIT_REDUNDANCE)
def verify(self, pv: PacketVerifier):
pkts = pv.pkts
pv.add_common_vars()
pv.summary.show()
PBBR1 = pv.vars['PBBR1']
PBBR1_ETH = pv.vars['PBBR1_ETH']
PBBR2 = pv.vars['PBBR2']
PBBR2_ETH = pv.vars['PBBR2_ETH']
ROUTER1 = pv.vars['ROUTER1']
ROUTER2 = pv.vars['ROUTER2']
HOST_ETH = pv.vars['Host_ETH']
HOST_BGUA = pv.vars['Host_BGUA']
ROUTER1_DUA = pv.vars['ROUTER1_DUA']
ROUTER2_DUA = pv.vars['ROUTER2_DUA']
ROUTER1_RLOC16 = pv.vars['ROUTER1_RLOC16']
#
# Verify Host ping MA1 to R1 and R2
#
# Host should send ping MA1 in the Backbone link
ping_ma1 = pkts.filter_eth_src(HOST_ETH).filter_ipv6_src_dst(HOST_BGUA, MA1).filter_ping_request().must_next()
with pkts.save_index():
# PBBR1 should forward ping-MA1 to Router1
pkts.filter_wpan_src64(PBBR1).filter_AMPLFMA().filter_ping_request(
identifier=ping_ma1.icmpv6.echo.identifier).must_next()
# Router1 should send ping reply
ping_reply_pkts = pkts.filter_ipv6_src_dst(
ROUTER1_DUA, HOST_BGUA).filter_ping_reply(identifier=ping_ma1.icmpv6.echo.identifier)
ping_reply_pkts.filter_wpan_src64(ROUTER1).must_next()
# PBBR1 should forward ping reply to the Backbone link
ping_reply_pkts.filter_eth_src(PBBR1_ETH).must_next()
# PBBR2 should forward ping-MA1 to Router2
pkts.filter_wpan_src64(PBBR2).filter_AMPLFMA().filter_ping_request(
identifier=ping_ma1.icmpv6.echo.identifier).must_next()
# Router2 should send ping reply
ping_reply_pkts = pkts.filter_ipv6_src_dst(
ROUTER2_DUA, HOST_BGUA).filter_ping_reply(identifier=ping_ma1.icmpv6.echo.identifier)
ping_reply_pkts.filter_wpan_src64(ROUTER2).must_next()
# PBBR2 should forward ping reply to the Backbone link
ping_reply_pkts.filter_eth_src(PBBR2_ETH).must_next()
#
# Verify R1 pings MA2 to Host and R2
#
# ROUTER1 should send the multicast ping request
ping_ma2 = pkts.filter_wpan_src64(ROUTER1).filter_AMPLFMA(
mpl_seed_id=ROUTER1_RLOC16).filter_ping_request().must_next()
# PBBR1 should forward the multicast ping request to the Backbone link
pkts.filter_eth_src(PBBR1_ETH).filter_ipv6_src_dst(
ROUTER1_DUA, MA2).filter_ping_request(identifier=ping_ma2.icmpv6.echo.identifier).must_next()
with pkts.save_index():
# Host should send ping reply to Router1
pkts.filter_eth_src(HOST_ETH).filter_ipv6_dst(ROUTER1_DUA).filter_ping_reply(
identifier=ping_ma2.icmpv6.echo.identifier).must_next()
# PBBR2 should forward the multicast ping request to Thread network at CH2
pkts.filter_wpan_src64(PBBR2).filter_AMPLFMA().filter_ping_request(
identifier=ping_ma2.icmpv6.echo.identifier).must_next()
# ROUTER2 should send ping reply back to ROUTER1
pkts.filter_wpan_src64(ROUTER2).filter_ipv6_src_dst(
ROUTER2_DUA, ROUTER1_DUA).filter_ping_reply(identifier=ping_ma2.icmpv6.echo.identifier).must_next()
if __name__ == '__main__':
unittest.main()
tests/scripts/thread-cert/mcast6.py
+128
View File
@@ -0,0 +1,128 @@
#!/usr/bin/env python3
#
# Copyright (c) 2020, 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 ctypes
import ctypes.util
import socket
import struct
import sys
import time
MYPORT = 8123
MYTTL = 1 # Increase to reach other networks
libc = ctypes.CDLL(ctypes.util.find_library('c'))
def if_nametoindex(name):
if not isinstance(name, str):
raise TypeError('name must be a string.')
ret = libc.if_nametoindex(name.encode('ascii'))
if not ret:
raise RuntimeError("Invalid Name")
return ret
def if_indextoname(index):
if not isinstance(index, int):
raise TypeError('index must be an int.')
libc.if_indextoname.argtypes = [ctypes.c_uint32, ctypes.c_char_p]
libc.if_indextoname.restype = ctypes.c_char_p
ifname = ctypes.create_string_buffer(32)
ifname = libc.if_indextoname(index, ifname)
if not ifname:
raise RuntimeError("Inavlid Index")
return ifname
def main():
args = sys.argv[1:]
is_sender = False
if args[0] == '-s':
is_sender = True
args.pop(0)
ifname, group = args
if is_sender:
sender(ifname, group)
else:
receiver(ifname, group)
def sender(ifname, group):
addrinfo = socket.getaddrinfo(group, None)[0]
s = socket.socket(addrinfo[0], socket.SOCK_DGRAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_BINDTODEVICE, (ifname + '\0').encode('ascii'))
# Set Time-to-live (optional)
ttl_bin = struct.pack('@i', MYTTL)
assert addrinfo[0] == socket.AF_INET6
s.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_MULTICAST_HOPS, ttl_bin)
while True:
data = repr(time.time())
s.sendto((data + '\0').encode('ascii'), (addrinfo[4][0], MYPORT))
time.sleep(1)
def receiver(ifname, group):
# Look up multicast group address in name server and find out IP version
addrinfo = socket.getaddrinfo(group, None)[0]
assert addrinfo[0] == socket.AF_INET6
# Create a socket
s = socket.socket(addrinfo[0], socket.SOCK_DGRAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_BINDTODEVICE, (ifname + '\0').encode('ascii'))
# Allow multiple copies of this program on one machine
# (not strictly needed)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
# Bind it to the port
s.bind(('', MYPORT))
group_bin = socket.inet_pton(addrinfo[0], addrinfo[4][0])
# Join group
interface_index = if_nametoindex(ifname)
mreq = group_bin + struct.pack('@I', interface_index)
s.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_JOIN_GROUP, mreq)
# Loop, printing any data we receive
while True:
data, sender = s.recvfrom(1500)
while data[-1:] == '\0':
data = data[:-1] # Strip trailing \0's
print(str(sender) + ' ' + repr(data))
if __name__ == '__main__':
main()
tests/scripts/thread-cert/node.py
+11
View File
@@ -1993,6 +1993,10 @@ class LinuxHost():
assert False, output
def add_ipmaddr_ether(self, ip: str):
cmd = f'python3 /app/third_party/openthread/repo/tests/scripts/thread-cert/mcast6.py {self.ETH_DEV} {ip} &'
self.bash(cmd)
def ping_ether(self, ipaddr, num_responses=1, size=None, timeout=5, ttl=None) -> int:
cmd = f'ping -6 {ipaddr} -I eth0 -c {num_responses} -W {timeout}'
if size is not None:
@@ -2026,6 +2030,13 @@ class LinuxHost():
else:
return super().ping(*args, **kwargs)
def add_ipmaddr(self, *args, **kwargs):
backbone = kwargs.pop('backbone', False)
if backbone:
return self.add_ipmaddr_ether(*args, **kwargs)
else:
return super().add_ipmaddr(*args, **kwargs)
def ip_neighbors_flush(self):
# clear neigh cache on linux
self.bash(f'ip -6 neigh list dev {self.ETH_DEV}')
tests/scripts/thread-cert/pktverify/bytes.py
+2 -2
View File
@@ -27,13 +27,13 @@
# POSSIBILITY OF SUCH DAMAGE.
#
import sys
from typing import Union
from typing import Union, Any
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']):
def __init__(self, s: Union[str, bytearray, 'Bytes', Any]):
if isinstance(s, str):
try:
s = Bytes._parse_compact(s)
tests/scripts/thread-cert/pktverify/layer_fields.py
+1
View File
@@ -385,6 +385,7 @@ _LAYER_FIELDS = {
'ipv6.opt.router_alert': _auto,
'ipv6.opt.padn': _str,
'ipv6.opt.length': _list(_auto),
'ipv6.opt.mpl.seed_id': _bytes,
'ipv6.opt.mpl.sequence': _auto,
'ipv6.opt.mpl.flag.v': _auto,
'ipv6.opt.mpl.flag.s': _auto,
tests/scripts/thread-cert/pktverify/packet_filter.py
+11 -6
View File
@@ -33,6 +33,7 @@ from typing import Optional, Callable, Tuple
from pktverify import consts, errors
from pktverify.addrs import EthAddr, ExtAddr, Ipv6Addr
from pktverify.bytes import Bytes
from pktverify.packet import Packet
from pktverify.utils import make_filter_func
@@ -359,11 +360,11 @@ class PacketFilter(object):
if role != 'REED':
tlv_set.add(consts.ROUTE64_TLV)
return self.filter_LLANMA().\
filter_mle_cmd(consts.MLE_ADVERTISEMENT).\
return self.filter_LLANMA(). \
filter_mle_cmd(consts.MLE_ADVERTISEMENT). \
filter(lambda p: tlv_set ==
set(p.mle.tlv.type) and\
p.ipv6.hlim == 255, **kwargs
set(p.mle.tlv.type) and \
p.ipv6.hlim == 255, **kwargs
)
def filter_coap(self, **kwargs):
@@ -528,8 +529,12 @@ class PacketFilter(object):
def filter_LLARMA(self, **kwargs):
return self.filter(lambda p: p.ipv6.dst == consts.LINK_LOCAL_ALL_ROUTERS_MULTICAST_ADDRESS, **kwargs)
def filter_AMPLFMA(self, **kwargs):
return self.filter(lambda p: p.ipv6.dst == consts.ALL_MPL_FORWARDERS_MA, **kwargs)
def filter_AMPLFMA(self, mpl_seed_id: int = None, **kwargs):
f = self.filter(lambda p: p.ipv6.dst == consts.ALL_MPL_FORWARDERS_MA, **kwargs)
if mpl_seed_id is not None:
mpl_seed_id = Bytes([mpl_seed_id >> 8, mpl_seed_id & 0xFF])
f = f.filter(lambda p: p.ipv6.opt.mpl.seed_id == mpl_seed_id)
return f
def filter_mle(self, **kwargs):
return self.filter(attrgetter('mle'), **kwargs)