From d6a35190363f19b68bf793bae9ff0924c8f449a7 Mon Sep 17 00:00:00 2001 From: hjian2017 <30794429+hjian2017@users.noreply.github.com> Date: Wed, 13 Sep 2017 16:44:56 +0800 Subject: [PATCH] [tests] add traffic analysis for Cert_5_2_05_AddressQuery.py (#2173) Add a fuction in node.py node_cli.py to node's address. Add a module command to verify commands. Spinel-cli.py doesn't request address for dhcp prefix now, so put Cert_5_2_05_AddressQuery.py in XFAIL on ncp. --- .../thread-cert/Cert_5_2_05_AddressQuery.py | 151 ++++++++++-------- tests/scripts/thread-cert/Makefile.am | 2 +- tests/scripts/thread-cert/command.py | 19 +++ tests/scripts/thread-cert/config.py | 18 ++- tests/scripts/thread-cert/node.py | 6 +- tests/scripts/thread-cert/node_cli.py | 70 +++++++- 6 files changed, 191 insertions(+), 75 deletions(-) create mode 100644 tests/scripts/thread-cert/command.py diff --git a/tests/scripts/thread-cert/Cert_5_2_05_AddressQuery.py b/tests/scripts/thread-cert/Cert_5_2_05_AddressQuery.py index 83774a5f5..f1ba6d6a1 100755 --- a/tests/scripts/thread-cert/Cert_5_2_05_AddressQuery.py +++ b/tests/scripts/thread-cert/Cert_5_2_05_AddressQuery.py @@ -31,111 +31,126 @@ import time import unittest import node +import config +import command -ED1 = 1 -BR1 = 2 -LEADER = 3 -ROUTER2 = 4 -REED = 5 -ED2 = 6 -ED3 = 7 - -MTDS = [ED1, ED2, ED3] +LEADER = 1 +ROUTER1 = 2 +BR = 3 +ED1 = 17 +DUT_REED = 18 +ROUTER_SELECTION_JITTER = 1 class Cert_5_2_5_AddressQuery(unittest.TestCase): def setUp(self): self.nodes = {} - for i in range(1,8): - self.nodes[i] = node.Node(i, (i in MTDS)) + for i in range(1, 19): + self.nodes[i] = node.Node(i, (i == ED1)) - self.nodes[LEADER].set_panid(0xface) + self.nodes[LEADER].set_panid() self.nodes[LEADER].set_mode('rsdn') - self.nodes[LEADER].add_whitelist(self.nodes[BR1].get_addr64()) - self.nodes[LEADER].add_whitelist(self.nodes[ROUTER2].get_addr64()) - self.nodes[LEADER].add_whitelist(self.nodes[REED].get_addr64()) self.nodes[LEADER].enable_whitelist() - self.nodes[BR1].set_panid(0xface) - self.nodes[BR1].set_mode('rsdn') - self.nodes[BR1].add_whitelist(self.nodes[LEADER].get_addr64()) - self.nodes[BR1].add_whitelist(self.nodes[ED1].get_addr64()) - self.nodes[BR1].enable_whitelist() - self.nodes[BR1].set_router_selection_jitter(1) + for i in range(2, 17): + self.nodes[i].set_panid() + self.nodes[i].set_mode('rsdn') + self.nodes[i].add_whitelist(self.nodes[LEADER].get_addr64()) + self.nodes[LEADER].add_whitelist(self.nodes[i].get_addr64()) + self.nodes[i].enable_whitelist() + self.nodes[i].set_router_selection_jitter(1) - self.nodes[ED1].set_panid(0xface) + self.nodes[ED1].set_panid() self.nodes[ED1].set_mode('rsn') - self.nodes[ED1].add_whitelist(self.nodes[BR1].get_addr64()) + self.nodes[ED1].add_whitelist(self.nodes[LEADER].get_addr64()) + self.nodes[LEADER].add_whitelist(self.nodes[ED1].get_addr64()) self.nodes[ED1].enable_whitelist() - self.nodes[REED].set_panid(0xface) - self.nodes[REED].set_mode('rsdn') - self.nodes[REED].add_whitelist(self.nodes[LEADER].get_addr64()) - self.nodes[REED].add_whitelist(self.nodes[ROUTER2].get_addr64()) - self.nodes[REED].set_router_upgrade_threshold(0) - self.nodes[REED].enable_whitelist() + self.nodes[DUT_REED].set_panid() + self.nodes[DUT_REED].set_mode('rsdn') + self.nodes[DUT_REED].add_whitelist(self.nodes[ROUTER1].get_addr64()) + self.nodes[ROUTER1].add_whitelist(self.nodes[DUT_REED].get_addr64()) + self.nodes[DUT_REED].enable_whitelist() + self.nodes[DUT_REED].set_router_selection_jitter(1) - self.nodes[ROUTER2].set_panid(0xface) - self.nodes[ROUTER2].set_mode('rsdn') - self.nodes[ROUTER2].add_whitelist(self.nodes[LEADER].get_addr64()) - self.nodes[ROUTER2].add_whitelist(self.nodes[REED].get_addr64()) - self.nodes[ROUTER2].add_whitelist(self.nodes[ED2].get_addr64()) - self.nodes[ROUTER2].add_whitelist(self.nodes[ED3].get_addr64()) - self.nodes[ROUTER2].enable_whitelist() - self.nodes[ROUTER2].set_router_selection_jitter(1) - - self.nodes[ED2].set_panid(0xface) - self.nodes[ED2].set_mode('rsn') - self.nodes[ED2].add_whitelist(self.nodes[ROUTER2].get_addr64()) - self.nodes[ED2].enable_whitelist() - - self.nodes[ED3].set_panid(0xface) - self.nodes[ED3].set_mode('rsn') - self.nodes[ED3].add_whitelist(self.nodes[ROUTER2].get_addr64()) - self.nodes[ED3].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): + # 1. LEADER: DHCPv6 Server for prefix 2001::/64. self.nodes[LEADER].start() self.nodes[LEADER].set_state('leader') self.assertEqual(self.nodes[LEADER].get_state(), 'leader') + self.nodes[LEADER].add_prefix('2001::/64', 'pdros') + self.nodes[LEADER].register_netdata() - self.nodes[BR1].start() + # 2. BR: SLAAC Server for prefix 2002::/64. + self.nodes[BR].start() time.sleep(5) - self.assertEqual(self.nodes[BR1].get_state(), 'router') + self.assertEqual(self.nodes[BR].get_state(), 'router') + self.nodes[BR].add_prefix('2002::/64', 'paros') + self.nodes[BR].register_netdata() - self.nodes[BR1].add_prefix('2001:2:0:3::/64', 'paros') - self.nodes[BR1].add_prefix('2001:2:0:4::/64', 'paros') - self.nodes[BR1].register_netdata() + # 3. Bring up remaining devices except DUT_REED. + for i in range(2, 17): + if i == BR: + continue + self.nodes[i].start() + time.sleep(5) + self.assertEqual(self.nodes[i].get_state(), 'router') self.nodes[ED1].start() time.sleep(5) self.assertEqual(self.nodes[ED1].get_state(), 'child') - self.nodes[REED].start() + # 4. Bring up DUT_REED. + self.nodes[DUT_REED].start() time.sleep(5) - self.assertEqual(self.nodes[REED].get_state(), 'child') + time.sleep(ROUTER_SELECTION_JITTER) - self.nodes[ROUTER2].start() - time.sleep(5) - self.assertEqual(self.nodes[ROUTER2].get_state(), 'router') + reed_messages = self.sniffer.get_messages_sent_by(DUT_REED) - self.nodes[ED2].start() - time.sleep(5) - self.assertEqual(self.nodes[ED2].get_state(), 'child') + # Verify DUT_REED doesn't try to become router. + msg = reed_messages.does_not_contain_coap_message() + assert msg is True, "Error: The REED sent an Address Solicit Request" - self.nodes[ED3].start() - time.sleep(5) - self.assertEqual(self.nodes[ED3].get_state(), 'child') + # 5. Enable a link between the DUT and BR to create a one-way link. + self.nodes[DUT_REED].add_whitelist(self.nodes[BR].get_addr64()) + self.nodes[BR].add_whitelist(self.nodes[DUT_REED].get_addr64()) - addrs = self.nodes[REED].get_addrs() - for addr in addrs: - if addr[0:4] != 'fe80': - self.assertTrue(self.nodes[ED2].ping(addr)) - time.sleep(1) + # 6. Verify DUT_REED would send Address Notification when ping to its ML-EID. + mleid = self.nodes[DUT_REED].get_ip6_address(config.ADDRESS_TYPE.ML_EID) + self.assertTrue(self.nodes[ED1].ping(mleid)) + + reed_messages = self.sniffer.get_messages_sent_by(DUT_REED) + msg = reed_messages.next_coap_message('0.02') + command.check_address_notification(msg, self.nodes[DUT_REED], self.nodes[LEADER]) + + # 7 & 8. Verify DUT_REED would send Address Notification when ping to its 2001::EID and 2002::EID. + flag2001 = 0 + flag2002 = 0 + for global_address in self.nodes[DUT_REED].get_ip6_address(config.ADDRESS_TYPE.GLOBAL): + if global_address[0:4] == '2001': + flag2001 += 1 + elif global_address[0:4] == '2002': + flag2002 += 1 + else: + raise "Error: Address is unexpected." + self.assertTrue(self.nodes[ED1].ping(global_address)) + + reed_messages = self.sniffer.get_messages_sent_by(DUT_REED) + msg = reed_messages.next_coap_message('0.02') + command.check_address_notification(msg, self.nodes[DUT_REED], self.nodes[LEADER]) + + assert flag2001 == 1 , "Error: Expecting address 2001::EID not appear." + assert flag2002 == 1 , "Error: Expecting address 2002::EID not appear." if __name__ == '__main__': unittest.main() diff --git a/tests/scripts/thread-cert/Makefile.am b/tests/scripts/thread-cert/Makefile.am index 86ba18dec..039e0878f 100644 --- a/tests/scripts/thread-cert/Makefile.am +++ b/tests/scripts/thread-cert/Makefile.am @@ -174,7 +174,6 @@ check_SCRIPTS = \ Cert_5_1_12_NewRouterNeighborSync.py \ Cert_5_1_13_RouterReset.py \ Cert_5_2_01_BecomeActiveRouter.py \ - Cert_5_2_05_AddressQuery.py \ Cert_5_2_06_RouterDowngrade.py \ Cert_5_2_07_REEDSynchronization.py \ Cert_5_2_04_REEDUpgrade.py \ @@ -262,6 +261,7 @@ TESTS = \ $(NULL) XFAIL_NCP_TESTS = \ + Cert_5_2_05_AddressQuery.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/command.py b/tests/scripts/thread-cert/command.py new file mode 100644 index 000000000..7ba9e54e6 --- /dev/null +++ b/tests/scripts/thread-cert/command.py @@ -0,0 +1,19 @@ +import ipv6 + +import network_layer +import config + +def check_address_notification(command_msg, source_node, destination_node): + """Verify source_node sent a properly formatted Address Notification command message to destination_node. + """ + command_msg.assertCoapMessageRequestUriPath('/a/an') + command_msg.assertCoapMessageContainsTlv(network_layer.TargetEid) + command_msg.assertCoapMessageContainsTlv(network_layer.Rloc16) + command_msg.assertCoapMessageContainsTlv(network_layer.MlEid) + + #pdb.set_trace() + source_rloc = source_node.get_ip6_address(config.ADDRESS_TYPE.RLOC) + assert ipv6.ip_address(source_rloc) == command_msg.ipv6_packet.ipv6_header.source_address, "Error: The IPv6 Source address is not the RLOC of the originator." + + destination_rloc = destination_node.get_ip6_address(config.ADDRESS_TYPE.RLOC) + assert ipv6.ip_address(destination_rloc) == command_msg.ipv6_packet.ipv6_header.destination_address, "Error: The IPv6 Destination address is not the RLOC of the destination." diff --git a/tests/scripts/thread-cert/config.py b/tests/scripts/thread-cert/config.py index c4278ab37..01054d797 100644 --- a/tests/scripts/thread-cert/config.py +++ b/tests/scripts/thread-cert/config.py @@ -26,6 +26,8 @@ # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE # POSSIBILITY OF SUCH DAMAGE. # +import os +from enum import Enum import coap import ipv6 @@ -37,9 +39,21 @@ import network_data import network_layer import sniffer +MESH_LOCAL_PREFIX = 'fdde:ad00:beef::/64' +MESH_LOCAL_PREFIX_REGEX_PATTERN = '^fdde:ad00:beef:(0){0,4}:' +ROUTING_LOCATOR = '64/:0:ff:fe00:/16' +ROUTING_LOCATOR_REGEX_PATTERN = '.*:(0)?:0{0,2}ff:fe00:\w{1,4}$' +LINK_LOCAL = 'fe80:/112' +LINK_LOCAL_REGEX_PATTERN = '^fe80:.*' +ALOC_FLAG_REGEX_PATTERN = '.*:fc..$' + DEFAULT_MASTER_KEY = bytearray([0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff]) +ADDRESS_TYPE = Enum('ADDRESS_TYPE', ('LINK_LOCAL', 'GLOBAL', 'RLOC', 'ALOC', 'ML_EID')) + +SNIFFER_ID = int(os.getenv('SNIFFER_ID', 34)) +PANID = 0xface def create_default_network_data_prefix_sub_tlvs_factories(): return { @@ -279,7 +293,7 @@ def create_default_lowpan_decompressor(context_manager): def create_default_thread_context_manager(): context_manager = lowpan.ContextManager() - context_manager[0] = lowpan.Context("fd00:0db8::/64") + context_manager[0] = lowpan.Context(MESH_LOCAL_PREFIX) return context_manager @@ -300,5 +314,5 @@ def create_default_thread_message_factory(master_key=DEFAULT_MASTER_KEY): return message.MessageFactory(lowpan_parser=lowpan_parser) -def create_default_thread_sniffer(nodeid): +def create_default_thread_sniffer(nodeid=SNIFFER_ID): return sniffer.Sniffer(nodeid, create_default_thread_message_factory()) diff --git a/tests/scripts/thread-cert/node.py b/tests/scripts/thread-cert/node.py index 76cbbca11..f9d20cbfb 100755 --- a/tests/scripts/thread-cert/node.py +++ b/tests/scripts/thread-cert/node.py @@ -35,6 +35,7 @@ if sys.platform != 'win32': else: import node_api import unittest +import config class Node: def __init__(self, nodeid, is_mtd=False): @@ -142,7 +143,7 @@ class Node: def get_panid(self): return self.interface.get_panid() - def set_panid(self, panid): + def set_panid(self, panid = config.PANID): self.interface.set_panid(panid) def get_partition_id(self): @@ -187,6 +188,9 @@ class Node: def get_addrs(self): return self.interface.get_addrs() + def get_ip6_address(self, address_type): + return self.interface.get_ip6_address(address_type) + def get_context_reuse_delay(self): return self.interface.get_context_reuse_delay() diff --git a/tests/scripts/thread-cert/node_cli.py b/tests/scripts/thread-cert/node_cli.py index 6f5ad360c..80691cc6b 100644 --- a/tests/scripts/thread-cert/node_cli.py +++ b/tests/scripts/thread-cert/node_cli.py @@ -31,6 +31,9 @@ import os import sys import time import pexpect +import re + +import config class otCli: def __init__(self, nodeid, is_mtd): @@ -83,7 +86,7 @@ class otCli: time.sleep(0.2) self.pexpect.expect('spinel-cli >') self.debug(int(os.getenv('DEBUG', '0'))) - + def __init_soc(self, nodeid): """ Initialize a System-on-a-chip node connected via UART. """ import fdpexpect @@ -98,7 +101,7 @@ class otCli: self.pexpect.close(force=True) def send_command(self, cmd): - print ("%d: %s" % (self.nodeid, cmd)) + print("%d: %s" % (self.nodeid, cmd)) self.pexpect.sendline(cmd) def get_commands(self): @@ -136,7 +139,7 @@ class otCli: def thread_stop(self): self.send_command('thread stop') self.pexpect.expect('Done') - + def commissioner_start(self): cmd = 'commissioner start' self.send_command(cmd) @@ -372,6 +375,67 @@ class otCli: return addrs + def __getLinkLocalAddress(self): + for ip6Addr in self.get_addrs(): + if re.match(config.LINK_LOCAL_REGEX_PATTERN, ip6Addr, re.I): + return ip6Addr + + return None + + def __getGlobalAddress(self): + global_address = [] + for ip6Addr in self.get_addrs(): + if (not re.match(config.LINK_LOCAL_REGEX_PATTERN, ip6Addr, re.I)) and \ + (not re.match(config.MESH_LOCAL_PREFIX_REGEX_PATTERN, ip6Addr, re.I)) and \ + (not re.match(config.ROUTING_LOCATOR_REGEX_PATTERN, ip6Addr, re.I)): + global_address.append(ip6Addr) + + return global_address + + def __getRloc(self): + for ip6Addr in self.get_addrs(): + if re.match(config.MESH_LOCAL_PREFIX_REGEX_PATTERN, ip6Addr, re.I) and \ + re.match(config.ROUTING_LOCATOR_REGEX_PATTERN, ip6Addr, re.I) and \ + not(re.match(config.ALOC_FLAG_REGEX_PATTERN, ip6Addr, re.I)): + return ip6Addr + return None + + def __getAloc(self): + aloc = [] + for ip6Addr in self.get_addrs(): + if re.match(config.MESH_LOCAL_PREFIX_REGEX_PATTERN, ip6Addr, re.I) and \ + re.match(config.ROUTING_LOCATOR_REGEX_PATTERN, ip6Addr, re.I) and \ + re.match(config.ALOC_FLAG_REGEX_PATTERN, ip6Addr, re.I): + aloc.append(ip6Addr) + + return aloc + + def __getMleid(self): + for ip6Addr in self.get_addrs(): + if re.match(config.MESH_LOCAL_PREFIX_REGEX_PATTERN, ip6Addr, re.I) and \ + not(re.match(config.ROUTING_LOCATOR_REGEX_PATTERN, ip6Addr, re.I)): + return ip6Addr + + return None + + def get_ip6_address(self, address_type): + """Get specific type of IPv6 address configured on thread device. + + Args: + address_type: the config.ADDRESS_TYPE type of IPv6 address. + + Returns: + IPv6 address string. + """ + switcher = { + config.ADDRESS_TYPE.LINK_LOCAL: self.__getLinkLocalAddress(), + config.ADDRESS_TYPE.GLOBAL: self.__getGlobalAddress(), + config.ADDRESS_TYPE.RLOC: self.__getRloc(), + config.ADDRESS_TYPE.ALOC: self.__getAloc(), + config.ADDRESS_TYPE.ML_EID: self.__getMleid() + } + return switcher.get(address_type, None) + def get_context_reuse_delay(self): self.send_command('contextreusedelay') i = self.pexpect.expect('(\d+)\r\n')