[tests] add traffic analysis for Cert_5_3_08 as TestPlan (#2484)

This commit adds the method set_lowpan_context to set the lowpan context of sniffer.

tests/scripts/thread-cert/Cert_5_3_08_ChildAddressSet.py

@@ -30,81 +30,158 @@
 import time
 import unittest
 
+import config
+import command
+import ipv6
+import mle
 import node
 
-LEADER = 1
-ED1 = 2
-ED2 = 3
-ED3 = 4
-ED4 = 5
+DUT_LEADER = 1
+BR = 2
+MED1 = 3
+MED2 = 4
 
-MTDS = [ED1, ED2, ED3, ED4]
+MTDS = [MED1, MED2]
 
 class Cert_5_3_8_ChildAddressSet(unittest.TestCase):
     def setUp(self):
+
         self.nodes = {}
-        for i in range(1,6):
+        for i in range(1, 5):
             self.nodes[i] = node.Node(i, (i in MTDS))
 
-        self.nodes[LEADER].set_panid(0xface)
-        self.nodes[LEADER].set_mode('rsdn')
-        self.nodes[LEADER].add_whitelist(self.nodes[ED1].get_addr64())
-        self.nodes[LEADER].add_whitelist(self.nodes[ED2].get_addr64())
-        self.nodes[LEADER].add_whitelist(self.nodes[ED3].get_addr64())
-        self.nodes[LEADER].add_whitelist(self.nodes[ED4].get_addr64())
-        self.nodes[LEADER].enable_whitelist()
+        self.nodes[DUT_LEADER].set_panid(0xface)
+        self.nodes[DUT_LEADER].set_mode('rsdn')
+        self.nodes[DUT_LEADER].add_whitelist(self.nodes[BR].get_addr64())
+        self.nodes[DUT_LEADER].add_whitelist(self.nodes[MED1].get_addr64())
+        self.nodes[DUT_LEADER].add_whitelist(self.nodes[MED2].get_addr64())
+        self.nodes[DUT_LEADER].enable_whitelist()
 
-        self.nodes[ED1].set_panid(0xface)
-        self.nodes[ED1].set_mode('rsn')
-        self.nodes[ED1].add_whitelist(self.nodes[LEADER].get_addr64())
-        self.nodes[ED1].enable_whitelist()
+        self.nodes[BR].set_panid(0xface)
+        self.nodes[BR].set_mode('rsdn')
+        self.nodes[BR].add_whitelist(self.nodes[DUT_LEADER].get_addr64())
+        self.nodes[BR].enable_whitelist()
+        self.nodes[BR].set_router_selection_jitter(1)
 
-        self.nodes[ED2].set_panid(0xface)
-        self.nodes[ED2].set_mode('rsn')
-        self.nodes[ED2].add_whitelist(self.nodes[LEADER].get_addr64())
-        self.nodes[ED2].enable_whitelist()
+        for i in MTDS:
+            self.nodes[i].set_panid(0xface)
+            self.nodes[i].set_mode('rsn')
+            self.nodes[i].add_whitelist(self.nodes[DUT_LEADER].get_addr64())
+            self.nodes[i].enable_whitelist()
 
-        self.nodes[ED3].set_panid(0xface)
-        self.nodes[ED3].set_mode('rsn')
-        self.nodes[ED3].add_whitelist(self.nodes[LEADER].get_addr64())
-        self.nodes[ED3].enable_whitelist()
-
-        self.nodes[ED4].set_panid(0xface)
-        self.nodes[ED4].set_mode('rsn')
-        self.nodes[ED4].add_whitelist(self.nodes[LEADER].get_addr64())
-        self.nodes[ED4].enable_whitelist()
+        self.sniffer = config.create_default_thread_sniffer()
+        self.sniffer.start()
 
     def tearDown(self):
+        self.sniffer.stop()
+        del self.sniffer
+
         for node in list(self.nodes.values()):
             node.stop()
         del self.nodes
 
     def test(self):
-        self.nodes[LEADER].start()
-        self.nodes[LEADER].set_state('leader')
-        self.assertEqual(self.nodes[LEADER].get_state(), 'leader')
+        self.nodes[DUT_LEADER].start()
+        self.nodes[DUT_LEADER].set_state('leader')
+        self.assertEqual(self.nodes[DUT_LEADER].get_state(), 'leader')
 
-        self.nodes[ED1].start()
+        self.nodes[BR].start()
         time.sleep(5)
-        self.assertEqual(self.nodes[ED1].get_state(), 'child')
+        self.assertEqual(self.nodes[BR].get_state(), 'router')
 
-        self.nodes[ED2].start()
-        time.sleep(5)
-        self.assertEqual(self.nodes[ED2].get_state(), 'child')
+        # 1 BR: Configure BR to be a DHCPv6 server
+        self.nodes[BR].add_prefix('2001::/64', 'pdros')
+        self.nodes[BR].add_prefix('2002::/64', 'pdros')
+        self.nodes[BR].add_prefix('2003::/64', 'pdros')
+        self.nodes[BR].register_netdata()
 
-        self.nodes[ED3].start()
-        time.sleep(5)
-        self.assertEqual(self.nodes[ED3].get_state(), 'child')
+        # Set lowpan context of sniffer
+        self.sniffer.set_lowpan_context(1, '2001::/64')
+        self.sniffer.set_lowpan_context(2, '2002::/64')
+        self.sniffer.set_lowpan_context(3, '2003::/64')
 
-        self.nodes[ED4].start()
-        time.sleep(5)
-        self.assertEqual(self.nodes[ED4].get_state(), 'child')
+        # 2 DUT_LEADER: Verify LEADER sent an Advertisement
+        leader_messages = self.sniffer.get_messages_sent_by(DUT_LEADER)
+        msg = leader_messages.next_mle_message(mle.CommandType.ADVERTISEMENT)
+        command.check_mle_advertisement(msg)
 
-        for i in range(2,6):
-            addrs = self.nodes[i].get_addrs()
-            for addr in addrs:
-                if addr[0:4] != 'fe80':
-                    self.assertTrue(self.nodes[LEADER].ping(addr))
+        # 3 MED1, MED2: MED1 and MED2 attach to DUT_LEADER
+        for i in MTDS:
+            self.nodes[i].start()
+            time.sleep(5)
+            self.assertEqual(self.nodes[i].get_state(), 'child')
+
+        # 4 MED1: MED1 send an ICMPv6 Echo Request to the MED2 ML-EID
+        med2_ml_eid = self.nodes[MED2].get_ip6_address(config.ADDRESS_TYPE.ML_EID)
+        self.assertTrue(med2_ml_eid != None)
+        self.assertTrue(self.nodes[MED1].ping(med2_ml_eid))
+
+        # Verify DUT_LEADER didn't generate an Address Query Request
+        leader_messages = self.sniffer.get_messages_sent_by(DUT_LEADER)
+        msg = leader_messages.next_coap_message('0.02', '/a/aq', False)
+        assert msg is None, "Error: The DUT_LEADER sent an unexpected Address Query Request"
+
+        # Wait for sniffer got packets
+        time.sleep(1)
+
+        # Verify MED2 sent an ICMPv6 Echo Reply
+        med2_messages = self.sniffer.get_messages_sent_by(MED2)
+        msg = med2_messages.get_icmp_message(ipv6.ICMP_ECHO_RESPONSE)
+        assert msg is not None, "Error: The MED2 didn't send ICMPv6 Echo Reply to MED1"
+
+        # 5 MED1: MED1 send an ICMPv6 Echo Request to the MED2 2001::GUA
+        addr = self.nodes[MED2].get_addr("2001::/64")
+        self.assertTrue(addr is not None)
+        self.assertTrue(self.nodes[MED1].ping(addr))
+
+        # Verify DUT_LEADER didn't generate an Address Query Request
+        leader_messages = self.sniffer.get_messages_sent_by(DUT_LEADER)
+        msg = leader_messages.next_coap_message('0.02', '/a/aq', False)
+        assert msg is None, "Error: The DUT_LEADER sent an unexpected Address Query Request"
+
+        # Wait for sniffer got packets
+        time.sleep(1)
+
+        # Verify MED2 sent an ICMPv6 Echo Reply
+        med2_messages = self.sniffer.get_messages_sent_by(MED2)
+        msg = med2_messages.get_icmp_message(ipv6.ICMP_ECHO_RESPONSE)
+        assert msg is not None, "Error: The MED2 didn't send ICMPv6 Echo Reply to MED1"
+
+        # 6 MED1: MED1 send an ICMPv6 Echo Request to the MED2 2002::GUA
+        addr = self.nodes[MED2].get_addr("2002::/64")
+        self.assertTrue(addr is not None)
+        self.assertTrue(self.nodes[MED1].ping(addr))
+
+        # Verify DUT_LEADER didn't generate an Address Query Request
+        leader_messages = self.sniffer.get_messages_sent_by(DUT_LEADER)
+        msg = leader_messages.next_coap_message('0.02', '/a/aq', False)
+        assert msg is None, "Error: The DUT_LEADER sent an unexpected Address Query Request"
+
+        # Wait for sniffer got packets
+        time.sleep(1)
+
+        # Verify MED2 sent an ICMPv6 Echo Reply
+        med2_messages = self.sniffer.get_messages_sent_by(MED2)
+        msg = med2_messages.get_icmp_message(ipv6.ICMP_ECHO_RESPONSE)
+        assert msg is not None, "Error: The MED2 didn't send ICMPv6 Echo Reply to MED1"
+
+        # 7 MED1: MED1 send an ICMPv6 Echo Request to the MED2 2003::GUA
+        addr = self.nodes[MED2].get_addr("2003::/64")
+        self.assertTrue(addr is not None)
+        self.assertTrue(self.nodes[MED1].ping(addr))
+
+        # Verify DUT_LEADER didn't generate an Address Query Request
+        leader_messages = self.sniffer.get_messages_sent_by(DUT_LEADER)
+        msg = leader_messages.next_coap_message('0.02', '/a/aq', False)
+        assert msg is None, "Error: The DUT_LEADER sent an unexpected Address Query Request"
+
+        # Wait for sniffer got packets
+        time.sleep(1)
+
+        # Verify MED2 sent an ICMPv6 Echo Reply
+        med2_messages = self.sniffer.get_messages_sent_by(MED2)
+        msg = med2_messages.get_icmp_message(ipv6.ICMP_ECHO_RESPONSE)
+        assert msg is not None, "Error: The MED2 didn't send ICMPv6 Echo Reply to MED1"
 
 if __name__ == '__main__':
     unittest.main()

tests/scripts/thread-cert/Makefile.am

@@ -265,6 +265,7 @@ TESTS                                                              = \
 XFAIL_NCP_TESTS                                                    = \
     test_service.py                                                  \
     Cert_5_2_05_AddressQuery.py                                      \
+    Cert_5_3_08_ChildAddressSet.py                                   \
     Cert_5_3_09_AddressQuery.py                                      \
     Cert_8_1_01_Commissioning.py                                     \
     Cert_8_1_02_Commissioning.py                                     \

tests/scripts/thread-cert/config.py

@@ -195,7 +195,7 @@ def create_deafult_network_tlvs_factories():
         network_layer.TlvType.ROUTER_MASK: network_layer.RouterMaskFactory(),
         network_layer.TlvType.ND_OPTION: network_layer.NdOptionFactory(),
         network_layer.TlvType.ND_DATA: network_layer.NdDataFactory(),
-        network_layer.TlvType.THREAD_NETWORK_DATA: network_layer.ThreadNetworkDataFactory(create_default_network_data_tlvs_factory),
+        network_layer.TlvType.THREAD_NETWORK_DATA: network_layer.ThreadNetworkDataFactory(create_default_network_data_tlvs_factory()),
 
         # Routing information are distributed in a Thread network by MLE Routing TLV
         # which is in fact MLE Route64 TLV. Thread specificaton v1.1. - Chapter 5.20
@@ -216,7 +216,8 @@ def create_default_uri_path_based_payload_factories():
         "/a/aq": network_layer_tlvs_factory,
         "/a/ar": network_layer_tlvs_factory,
         "/a/ae": network_layer_tlvs_factory,
-        "/a/an": network_layer_tlvs_factory
+        "/a/an": network_layer_tlvs_factory,
+        "/a/sd": network_layer_tlvs_factory
     }
 
 

tests/scripts/thread-cert/lowpan.py

@@ -731,6 +731,9 @@ class LowpanIpv6HeaderFactory:
         elif iphc.m == self.IPHC_M_YES:
             return self._decompress_multicast_dst_addr(iphc, dci, data)
 
+    def set_lowpan_context(self, cid, prefix):
+        self._context_manager[cid] = Context(prefix)
+
     def parse(self, data, message_info):
         iphc = LowpanIPHC.from_bytes(bytearray(data.read(2)))
 
@@ -776,6 +779,9 @@ class LowpanDecompressor:
     def _is_next_header_compressed(self, header):
         return (header.next_header is None)
 
+    def set_lowpan_context(self, cid, prefix):
+        self._lowpan_ip_header_factory.set_lowpan_context(cid, prefix)
+
     def decompress(self, data, message_info):
         ipv6_header = self._lowpan_ip_header_factory.parse(data, message_info)
 
@@ -1076,6 +1082,9 @@ class LowpanParser(object):
 
         return self._ipv6_packet_factory.parse(io.BytesIO(decompressed_data), message_info)
 
+    def set_lowpan_context(self, cid, prefix):
+        self._lowpan_decompressor.set_lowpan_context(cid, prefix)
+
     def parse(self, data, message_info):
 
         while data.tell() < len(data.getvalue()):

tests/scripts/thread-cert/message.py

@@ -443,6 +443,9 @@ class MessageFactory:
         mac_frame.parse(data)
         return mac_frame
 
+    def set_lowpan_context(self, cid, prefix):
+        self._lowpan_parser.set_lowpan_context(cid, prefix)
+
     def create(self, data):
         message = Message()
         message.channel = struct.unpack(">B", data.read(1))

tests/scripts/thread-cert/node.py

@@ -194,6 +194,9 @@ class Node:
     def get_addrs(self):
         return self.interface.get_addrs()
 
+    def get_addr(self, prefix):
+        return self.interface.get_addr(prefix)
+
     def add_service(self, enterpriseNumber, serviceData, serverData):
         self.interface.add_service(enterpriseNumber, serviceData, serverData)
 

tests/scripts/thread-cert/node_cli.py

@@ -32,6 +32,7 @@ import sys
 import time
 import pexpect
 import re
+import ipaddress
 
 import config
 
@@ -388,6 +389,17 @@ class otCli:
 
         return addrs
 
+    def get_addr(self, prefix):
+        network = ipaddress.ip_network(unicode(prefix))
+        addrs = self.get_addrs()
+
+        for addr in addrs:
+            ipv6_address = ipaddress.ip_address(addr.decode("utf-8"))
+            if ipv6_address in network:
+                return ipv6_address.exploded.encode('utf-8')
+
+        return None
+
     def add_service(self, enterpriseNumber, serviceData, serverData):
         cmd = 'service add ' + enterpriseNumber + ' ' + serviceData+ ' '  + serverData
         self.send_command(cmd)

tests/scripts/thread-cert/sniffer.py

@@ -116,6 +116,9 @@ class Sniffer:
         self._thread.join()
         self._thread = None
 
+    def set_lowpan_context(self, cid, prefix):
+        self._message_factory.set_lowpan_context(cid, prefix)
+
     def get_messages_sent_by(self, nodeid):
         """ Get sniffed messages.