From 2636df195279ddd9f7f1d9b5e199cf5afa696ae9 Mon Sep 17 00:00:00 2001 From: Zhanglong Xia Date: Tue, 23 Jan 2018 00:48:00 +0800 Subject: [PATCH] [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. --- .../Cert_5_3_08_ChildAddressSet.py | 177 +++++++++++++----- tests/scripts/thread-cert/Makefile.am | 1 + tests/scripts/thread-cert/config.py | 5 +- tests/scripts/thread-cert/lowpan.py | 9 + tests/scripts/thread-cert/message.py | 3 + tests/scripts/thread-cert/node.py | 3 + tests/scripts/thread-cert/node_cli.py | 12 ++ tests/scripts/thread-cert/sniffer.py | 3 + 8 files changed, 161 insertions(+), 52 deletions(-) diff --git a/tests/scripts/thread-cert/Cert_5_3_08_ChildAddressSet.py b/tests/scripts/thread-cert/Cert_5_3_08_ChildAddressSet.py index 8011f1a6f..1b36db570 100755 --- a/tests/scripts/thread-cert/Cert_5_3_08_ChildAddressSet.py +++ b/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() diff --git a/tests/scripts/thread-cert/Makefile.am b/tests/scripts/thread-cert/Makefile.am index c2c3f0b73..6cbe91f66 100644 --- a/tests/scripts/thread-cert/Makefile.am +++ b/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 \ diff --git a/tests/scripts/thread-cert/config.py b/tests/scripts/thread-cert/config.py index 356c2015b..eadf5c117 100644 --- a/tests/scripts/thread-cert/config.py +++ b/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 } diff --git a/tests/scripts/thread-cert/lowpan.py b/tests/scripts/thread-cert/lowpan.py index 7a883076c..945b44c4f 100644 --- a/tests/scripts/thread-cert/lowpan.py +++ b/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()): diff --git a/tests/scripts/thread-cert/message.py b/tests/scripts/thread-cert/message.py index 5e80e6760..c1604e615 100644 --- a/tests/scripts/thread-cert/message.py +++ b/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)) diff --git a/tests/scripts/thread-cert/node.py b/tests/scripts/thread-cert/node.py index 4e1be862c..cd5caa9d0 100755 --- a/tests/scripts/thread-cert/node.py +++ b/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) diff --git a/tests/scripts/thread-cert/node_cli.py b/tests/scripts/thread-cert/node_cli.py index da135fb2b..dfa61edf8 100644 --- a/tests/scripts/thread-cert/node_cli.py +++ b/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) diff --git a/tests/scripts/thread-cert/sniffer.py b/tests/scripts/thread-cert/sniffer.py index 1e695a80b..6249166b2 100644 --- a/tests/scripts/thread-cert/sniffer.py +++ b/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.