| #!/usr/bin/env python3 |
| # |
| # Copyright (c) 2017-2018, 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 binascii |
| |
| import ipaddress |
| import ipv6 |
| import network_data |
| import network_layer |
| import common |
| import config |
| import mesh_cop |
| import mle |
| |
| from enum import IntEnum |
| |
| |
| class CheckType(IntEnum): |
| CONTAIN = 0 |
| NOT_CONTAIN = 1 |
| OPTIONAL = 2 |
| |
| |
| class NetworkDataCheckType: |
| PREFIX_CNT = 1 |
| PREFIX_CONTENT = 2 |
| |
| |
| def check_address_query(command_msg, source_node, destination_address): |
| """Verify source_node sent a properly formatted Address Query Request message to the destination_address. |
| """ |
| command_msg.assertCoapMessageContainsTlv(network_layer.TargetEid) |
| |
| 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. The source node's rloc is: " + |
| str(ipv6.ip_address(source_rloc)) + ", but the source_address in command msg is: " + |
| str(command_msg.ipv6_packet.ipv6_header.source_address)) |
| |
| if isinstance(destination_address, bytearray): |
| destination_address = bytes(destination_address) |
| |
| assert (ipv6.ip_address(destination_address) == command_msg.ipv6_packet.ipv6_header.destination_address |
| ), "Error: The IPv6 destination address is not expected." |
| |
| |
| 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) |
| |
| 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." |
| |
| |
| def check_address_error_notification(command_msg, source_node, destination_address): |
| """Verify source_node sent a properly formatted Address Error Notification command message to destination_address. |
| """ |
| command_msg.assertCoapMessageRequestUriPath('/a/ae') |
| command_msg.assertCoapMessageContainsTlv(network_layer.TargetEid) |
| command_msg.assertCoapMessageContainsTlv(network_layer.MlEid) |
| |
| 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. The source node's rloc is: " + |
| str(ipv6.ip_address(source_rloc)) + ", but the source_address in command msg is: " + |
| str(command_msg.ipv6_packet.ipv6_header.source_address)) |
| |
| if isinstance(destination_address, bytearray): |
| destination_address = bytes(destination_address) |
| |
| assert (ipv6.ip_address(destination_address) == command_msg.ipv6_packet.ipv6_header.destination_address), ( |
| "Error: The IPv6 destination address is not expected. The destination node's rloc is: " + |
| str(ipv6.ip_address(destination_address)) + ", but the destination_address in command msg is: " + |
| str(command_msg.ipv6_packet.ipv6_header.destination_address)) |
| |
| |
| def check_address_solicit(command_msg, was_router): |
| command_msg.assertCoapMessageRequestUriPath('/a/as') |
| command_msg.assertCoapMessageContainsTlv(network_layer.MacExtendedAddress) |
| command_msg.assertCoapMessageContainsTlv(network_layer.Status) |
| if was_router: |
| command_msg.assertCoapMessageContainsTlv(network_layer.Rloc16) |
| else: |
| command_msg.assertMleMessageDoesNotContainTlv(network_layer.Rloc16) |
| |
| |
| def check_address_release(command_msg, destination_node): |
| """Verify the message is a properly formatted address release destined to the given node. |
| """ |
| command_msg.assertCoapMessageRequestUriPath('/a/ar') |
| command_msg.assertCoapMessageContainsTlv(network_layer.Rloc16) |
| command_msg.assertCoapMessageContainsTlv(network_layer.MacExtendedAddress) |
| |
| 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 destination is not RLOC address" |
| |
| |
| def check_tlv_request_tlv(command_msg, check_type, tlv_id): |
| """Verify if TLV Request TLV contains specified TLV ID |
| """ |
| tlv_request_tlv = command_msg.get_mle_message_tlv(mle.TlvRequest) |
| |
| if check_type == CheckType.CONTAIN: |
| assert (tlv_request_tlv is not None), "Error: The msg doesn't contain TLV Request TLV" |
| assert any( |
| tlv_id == tlv |
| for tlv in tlv_request_tlv.tlvs), "Error: The msg doesn't contain TLV Request TLV ID: {}".format(tlv_id) |
| |
| elif check_type == CheckType.NOT_CONTAIN: |
| if tlv_request_tlv is not None: |
| assert (any(tlv_id == tlv for tlv in tlv_request_tlv.tlvs) is |
| False), "Error: The msg contains TLV Request TLV ID: {}".format(tlv_id) |
| |
| elif check_type == CheckType.OPTIONAL: |
| if tlv_request_tlv is not None: |
| if any(tlv_id == tlv for tlv in tlv_request_tlv.tlvs): |
| print("TLV Request TLV contains TLV ID: {}".format(tlv_id)) |
| else: |
| print("TLV Request TLV doesn't contain TLV ID: {}".format(tlv_id)) |
| else: |
| print("The msg doesn't contain TLV Request TLV") |
| |
| else: |
| raise ValueError("Invalid check type") |
| |
| |
| def check_link_request( |
| command_msg, |
| source_address=CheckType.OPTIONAL, |
| leader_data=CheckType.OPTIONAL, |
| tlv_request_address16=CheckType.OPTIONAL, |
| tlv_request_route64=CheckType.OPTIONAL, |
| tlv_request_link_margin=CheckType.OPTIONAL, |
| ): |
| """Verify a properly formatted Link Request command message. |
| """ |
| command_msg.assertMleMessageContainsTlv(mle.Challenge) |
| command_msg.assertMleMessageContainsTlv(mle.Version) |
| |
| check_mle_optional_tlv(command_msg, source_address, mle.SourceAddress) |
| check_mle_optional_tlv(command_msg, leader_data, mle.LeaderData) |
| |
| check_tlv_request_tlv(command_msg, tlv_request_address16, mle.TlvType.ADDRESS16) |
| check_tlv_request_tlv(command_msg, tlv_request_route64, mle.TlvType.ROUTE64) |
| check_tlv_request_tlv(command_msg, tlv_request_link_margin, mle.TlvType.LINK_MARGIN) |
| |
| |
| def check_link_accept( |
| command_msg, |
| destination_node, |
| leader_data=CheckType.OPTIONAL, |
| link_margin=CheckType.OPTIONAL, |
| mle_frame_counter=CheckType.OPTIONAL, |
| challenge=CheckType.OPTIONAL, |
| address16=CheckType.OPTIONAL, |
| route64=CheckType.OPTIONAL, |
| tlv_request_link_margin=CheckType.OPTIONAL, |
| ): |
| """verify a properly formatted link accept command message. |
| """ |
| command_msg.assertMleMessageContainsTlv(mle.LinkLayerFrameCounter) |
| command_msg.assertMleMessageContainsTlv(mle.SourceAddress) |
| command_msg.assertMleMessageContainsTlv(mle.Response) |
| command_msg.assertMleMessageContainsTlv(mle.Version) |
| |
| check_mle_optional_tlv(command_msg, leader_data, mle.LeaderData) |
| check_mle_optional_tlv(command_msg, link_margin, mle.LinkMargin) |
| check_mle_optional_tlv(command_msg, mle_frame_counter, mle.MleFrameCounter) |
| check_mle_optional_tlv(command_msg, challenge, mle.Challenge) |
| check_mle_optional_tlv(command_msg, address16, mle.Address16) |
| check_mle_optional_tlv(command_msg, route64, mle.Route64) |
| |
| check_tlv_request_tlv(command_msg, tlv_request_link_margin, mle.TlvType.LINK_MARGIN) |
| |
| destination_link_local = destination_node.get_ip6_address(config.ADDRESS_TYPE.LINK_LOCAL) |
| assert (ipv6.ip_address(destination_link_local) == command_msg.ipv6_packet.ipv6_header.destination_address |
| ), "Error: The destination is unexpected" |
| |
| |
| def check_icmp_path(sniffer, path, nodes, icmp_type=ipv6.ICMP_ECHO_REQUEST): |
| """Verify icmp message is forwarded along the path. |
| """ |
| len_path = len(path) |
| |
| # Verify icmp message is forwarded to the next node of the path. |
| for i in range(0, len_path): |
| node_msg = sniffer.get_messages_sent_by(path[i]) |
| node_icmp_msg = node_msg.get_icmp_message(icmp_type) |
| |
| if i < len_path - 1: |
| next_node = nodes[path[i + 1]] |
| next_node_rloc16 = next_node.get_addr16() |
| assert (next_node_rloc16 == node_icmp_msg.mac_header.dest_address.rloc), "Error: The path is unexpected." |
| else: |
| return True |
| |
| return False |
| |
| |
| def check_id_set(command_msg, router_id): |
| """Check the command_msg's Route64 tlv to verify router_id is an active router. |
| """ |
| tlv = command_msg.assertMleMessageContainsTlv(mle.Route64) |
| return (tlv.router_id_mask >> (63 - router_id)) & 1 |
| |
| |
| def get_routing_cost(command_msg, router_id): |
| """Check the command_msg's Route64 tlv to get the routing cost to router. |
| """ |
| tlv = command_msg.assertMleMessageContainsTlv(mle.Route64) |
| |
| # Get router's mask pos |
| # Turn the number into binary string. Need to consider the preceding 0 |
| # omitted during conversion. |
| router_id_mask_str = bin(tlv.router_id_mask).replace('0b', '') |
| prefix_len = 64 - len(router_id_mask_str) |
| routing_entry_pos = 0 |
| |
| for i in range(0, router_id - prefix_len): |
| if router_id_mask_str[i] == '1': |
| routing_entry_pos += 1 |
| |
| assert router_id_mask_str[router_id - prefix_len] == '1', \ |
| (("Error: The router isn't in the topology. \n", |
| "route64 tlv is: %s. \nrouter_id is: %s. \nrouting_entry_pos is: %s. \nrouter_id_mask_str is: %s.") % |
| (tlv, router_id, routing_entry_pos, router_id_mask_str)) |
| |
| return tlv.link_quality_and_route_data[routing_entry_pos].route |
| |
| |
| def check_mle_optional_tlv(command_msg, type, tlv): |
| if type == CheckType.CONTAIN: |
| command_msg.assertMleMessageContainsTlv(tlv) |
| elif type == CheckType.NOT_CONTAIN: |
| command_msg.assertMleMessageDoesNotContainTlv(tlv) |
| elif type == CheckType.OPTIONAL: |
| command_msg.assertMleMessageContainsOptionalTlv(tlv) |
| else: |
| raise ValueError("Invalid check type") |
| |
| |
| def check_mle_advertisement(command_msg): |
| command_msg.assertSentWithHopLimit(255) |
| command_msg.assertSentToDestinationAddress(config.LINK_LOCAL_ALL_NODES_ADDRESS) |
| command_msg.assertMleMessageContainsTlv(mle.SourceAddress) |
| command_msg.assertMleMessageContainsTlv(mle.LeaderData) |
| command_msg.assertMleMessageContainsTlv(mle.Route64) |
| |
| |
| def check_parent_request(command_msg, is_first_request): |
| """Verify a properly formatted Parent Request command message. |
| """ |
| if command_msg.mle.aux_sec_hdr.key_id_mode != 0x2: |
| raise ValueError("The Key Identifier Mode of the Security Control Field SHALL be set to 0x02") |
| |
| command_msg.assertSentWithHopLimit(255) |
| command_msg.assertSentToDestinationAddress(config.LINK_LOCAL_ALL_ROUTERS_ADDRESS) |
| command_msg.assertMleMessageContainsTlv(mle.Mode) |
| command_msg.assertMleMessageContainsTlv(mle.Challenge) |
| command_msg.assertMleMessageContainsTlv(mle.Version) |
| scan_mask = command_msg.assertMleMessageContainsTlv(mle.ScanMask) |
| if not scan_mask.router: |
| raise ValueError("Parent request without R bit set") |
| if is_first_request: |
| if scan_mask.end_device: |
| raise ValueError("First parent request with E bit set") |
| elif not scan_mask.end_device: |
| raise ValueError("Second parent request without E bit set") |
| |
| |
| def check_parent_response(command_msg, mle_frame_counter=CheckType.OPTIONAL): |
| """Verify a properly formatted Parent Response command message. |
| """ |
| command_msg.assertMleMessageContainsTlv(mle.Challenge) |
| command_msg.assertMleMessageContainsTlv(mle.Connectivity) |
| command_msg.assertMleMessageContainsTlv(mle.LeaderData) |
| command_msg.assertMleMessageContainsTlv(mle.LinkLayerFrameCounter) |
| command_msg.assertMleMessageContainsTlv(mle.LinkMargin) |
| command_msg.assertMleMessageContainsTlv(mle.Response) |
| command_msg.assertMleMessageContainsTlv(mle.SourceAddress) |
| command_msg.assertMleMessageContainsTlv(mle.Version) |
| |
| check_mle_optional_tlv(command_msg, mle_frame_counter, mle.MleFrameCounter) |
| |
| |
| def check_child_id_request( |
| command_msg, |
| tlv_request=CheckType.OPTIONAL, |
| mle_frame_counter=CheckType.OPTIONAL, |
| address_registration=CheckType.OPTIONAL, |
| active_timestamp=CheckType.OPTIONAL, |
| pending_timestamp=CheckType.OPTIONAL, |
| route64=CheckType.OPTIONAL, |
| ): |
| """Verify a properly formatted Child Id Request command message. |
| """ |
| if command_msg.mle.aux_sec_hdr.key_id_mode != 0x2: |
| raise ValueError("The Key Identifier Mode of the Security Control Field SHALL be set to 0x02") |
| |
| command_msg.assertMleMessageContainsTlv(mle.LinkLayerFrameCounter) |
| command_msg.assertMleMessageContainsTlv(mle.Mode) |
| command_msg.assertMleMessageContainsTlv(mle.Response) |
| command_msg.assertMleMessageContainsTlv(mle.Timeout) |
| command_msg.assertMleMessageContainsTlv(mle.Version) |
| |
| check_mle_optional_tlv(command_msg, tlv_request, mle.TlvRequest) |
| check_mle_optional_tlv(command_msg, mle_frame_counter, mle.MleFrameCounter) |
| check_mle_optional_tlv(command_msg, address_registration, mle.AddressRegistration) |
| check_mle_optional_tlv(command_msg, active_timestamp, mle.ActiveTimestamp) |
| check_mle_optional_tlv(command_msg, pending_timestamp, mle.PendingTimestamp) |
| check_mle_optional_tlv(command_msg, route64, mle.Route64) |
| |
| check_tlv_request_tlv(command_msg, CheckType.CONTAIN, mle.TlvType.ADDRESS16) |
| check_tlv_request_tlv(command_msg, CheckType.CONTAIN, mle.TlvType.NETWORK_DATA) |
| |
| |
| def check_child_id_response( |
| command_msg, |
| route64=CheckType.OPTIONAL, |
| network_data=CheckType.OPTIONAL, |
| address_registration=CheckType.OPTIONAL, |
| active_timestamp=CheckType.OPTIONAL, |
| pending_timestamp=CheckType.OPTIONAL, |
| active_operational_dataset=CheckType.OPTIONAL, |
| pending_operational_dataset=CheckType.OPTIONAL, |
| network_data_check=None, |
| ): |
| """Verify a properly formatted Child Id Response command message. |
| """ |
| command_msg.assertMleMessageContainsTlv(mle.SourceAddress) |
| command_msg.assertMleMessageContainsTlv(mle.LeaderData) |
| command_msg.assertMleMessageContainsTlv(mle.Address16) |
| |
| check_mle_optional_tlv(command_msg, route64, mle.Route64) |
| check_mle_optional_tlv(command_msg, network_data, mle.NetworkData) |
| check_mle_optional_tlv(command_msg, address_registration, mle.AddressRegistration) |
| check_mle_optional_tlv(command_msg, active_timestamp, mle.ActiveTimestamp) |
| check_mle_optional_tlv(command_msg, pending_timestamp, mle.PendingTimestamp) |
| check_mle_optional_tlv(command_msg, active_operational_dataset, mle.ActiveOperationalDataset) |
| check_mle_optional_tlv(command_msg, pending_operational_dataset, mle.PendingOperationalDataset) |
| |
| if network_data_check is not None: |
| network_data_tlv = command_msg.assertMleMessageContainsTlv(mle.NetworkData) |
| network_data_check.check(network_data_tlv) |
| |
| |
| def check_prefix(prefix): |
| """Verify if a prefix contains 6loWPAN sub-TLV and border router sub-TLV |
| """ |
| assert contains_tlv(prefix.sub_tlvs, network_data.BorderRouter), 'Prefix doesn\'t contain a border router sub-TLV!' |
| assert contains_tlv(prefix.sub_tlvs, network_data.LowpanId), 'Prefix doesn\'t contain a LowpanId sub-TLV!' |
| |
| |
| def check_child_update_request_from_child( |
| command_msg, |
| source_address=CheckType.OPTIONAL, |
| leader_data=CheckType.OPTIONAL, |
| challenge=CheckType.OPTIONAL, |
| time_out=CheckType.OPTIONAL, |
| address_registration=CheckType.OPTIONAL, |
| tlv_request_tlv=CheckType.OPTIONAL, |
| active_timestamp=CheckType.OPTIONAL, |
| CIDs=(), |
| ): |
| |
| command_msg.assertMleMessageContainsTlv(mle.Mode) |
| check_mle_optional_tlv(command_msg, source_address, mle.SourceAddress) |
| check_mle_optional_tlv(command_msg, leader_data, mle.LeaderData) |
| check_mle_optional_tlv(command_msg, challenge, mle.Challenge) |
| check_mle_optional_tlv(command_msg, time_out, mle.Timeout) |
| check_mle_optional_tlv(command_msg, address_registration, mle.AddressRegistration) |
| check_mle_optional_tlv(command_msg, tlv_request_tlv, mle.TlvRequest) |
| check_mle_optional_tlv(command_msg, active_timestamp, mle.ActiveTimestamp) |
| |
| if (address_registration == CheckType.CONTAIN) and len(CIDs) > 0: |
| _check_address_registration(command_msg, CIDs) |
| |
| |
| def check_coap_optional_tlv(coap_msg, type, tlv): |
| if type == CheckType.CONTAIN: |
| coap_msg.assertCoapMessageContainsTlv(tlv) |
| elif type == CheckType.NOT_CONTAIN: |
| coap_msg.assertCoapMessageDoesNotContainTlv(tlv) |
| elif type == CheckType.OPTIONAL: |
| coap_msg.assertCoapMessageContainsOptionalTlv(tlv) |
| else: |
| raise ValueError("Invalid check type") |
| |
| |
| def check_router_id_cached(node, router_id, cached=True): |
| """Verify if the node has cached any entries based on the router ID |
| """ |
| eidcaches = node.get_eidcaches() |
| if cached: |
| assert any(router_id == (int(rloc, 16) >> 10) for (_, rloc) in eidcaches) |
| else: |
| assert (any(router_id == (int(rloc, 16) >> 10) for (_, rloc) in eidcaches) is False) |
| |
| |
| def contains_tlv(sub_tlvs, tlv_type): |
| """Verify if a specific type of tlv is included in a sub-tlv list. |
| """ |
| return any(isinstance(sub_tlv, tlv_type) for sub_tlv in sub_tlvs) |
| |
| |
| def contains_tlvs(sub_tlvs, tlv_types): |
| """Verify if all types of tlv in a list are included in a sub-tlv list. |
| """ |
| return all((any(isinstance(sub_tlv, tlv_type) for sub_tlv in sub_tlvs)) for tlv_type in tlv_types) |
| |
| |
| def check_secure_mle_key_id_mode(command_msg, key_id_mode): |
| """Verify if the mle command message sets the right key id mode. |
| """ |
| assert isinstance(command_msg.mle, mle.MleMessageSecured) |
| assert command_msg.mle.aux_sec_hdr.key_id_mode == key_id_mode |
| |
| |
| def check_data_response(command_msg, network_data_check=None, active_timestamp=CheckType.OPTIONAL): |
| """Verify a properly formatted Data Response command message. |
| """ |
| check_secure_mle_key_id_mode(command_msg, 0x02) |
| command_msg.assertMleMessageContainsTlv(mle.SourceAddress) |
| command_msg.assertMleMessageContainsTlv(mle.LeaderData) |
| check_mle_optional_tlv(command_msg, active_timestamp, mle.ActiveTimestamp) |
| if network_data_check is not None: |
| network_data_tlv = command_msg.assertMleMessageContainsTlv(mle.NetworkData) |
| network_data_check.check(network_data_tlv) |
| |
| |
| def check_child_update_request_from_parent( |
| command_msg, |
| leader_data=CheckType.OPTIONAL, |
| network_data=CheckType.OPTIONAL, |
| challenge=CheckType.OPTIONAL, |
| tlv_request=CheckType.OPTIONAL, |
| active_timestamp=CheckType.OPTIONAL, |
| ): |
| """Verify a properly formatted Child Update Request(from parent) command message. |
| """ |
| check_secure_mle_key_id_mode(command_msg, 0x02) |
| |
| command_msg.assertMleMessageContainsTlv(mle.SourceAddress) |
| check_mle_optional_tlv(command_msg, leader_data, mle.LeaderData) |
| check_mle_optional_tlv(command_msg, network_data, mle.NetworkData) |
| check_mle_optional_tlv(command_msg, challenge, mle.Challenge) |
| check_mle_optional_tlv(command_msg, tlv_request, mle.TlvRequest) |
| check_mle_optional_tlv(command_msg, active_timestamp, mle.ActiveTimestamp) |
| |
| |
| def check_child_update_response( |
| command_msg, |
| timeout=CheckType.OPTIONAL, |
| address_registration=CheckType.OPTIONAL, |
| address16=CheckType.OPTIONAL, |
| leader_data=CheckType.OPTIONAL, |
| network_data=CheckType.OPTIONAL, |
| response=CheckType.OPTIONAL, |
| link_layer_frame_counter=CheckType.OPTIONAL, |
| mle_frame_counter=CheckType.OPTIONAL, |
| CIDs=(), |
| ): |
| """Verify a properly formatted Child Update Response from parent |
| """ |
| check_secure_mle_key_id_mode(command_msg, 0x02) |
| |
| command_msg.assertMleMessageContainsTlv(mle.SourceAddress) |
| command_msg.assertMleMessageContainsTlv(mle.Mode) |
| check_mle_optional_tlv(command_msg, timeout, mle.Timeout) |
| check_mle_optional_tlv(command_msg, address_registration, mle.AddressRegistration) |
| check_mle_optional_tlv(command_msg, address16, mle.Address16) |
| check_mle_optional_tlv(command_msg, leader_data, mle.LeaderData) |
| check_mle_optional_tlv(command_msg, network_data, mle.NetworkData) |
| check_mle_optional_tlv(command_msg, response, mle.Response) |
| check_mle_optional_tlv(command_msg, link_layer_frame_counter, mle.LinkLayerFrameCounter) |
| check_mle_optional_tlv(command_msg, mle_frame_counter, mle.MleFrameCounter) |
| |
| if (address_registration == CheckType.CONTAIN) and len(CIDs) > 0: |
| _check_address_registration(command_msg, CIDs) |
| |
| |
| def _check_address_registration(command_msg, CIDs=()): |
| addresses = command_msg.assertMleMessageContainsTlv(mle.AddressRegistration).addresses |
| for cid in CIDs: |
| found = False |
| for address in addresses: |
| if isinstance(address, mle.AddressCompressed): |
| if cid == address.cid: |
| found = True |
| break |
| assert found, "AddressRegistration TLV doesn't have CID {} ".format(cid) |
| |
| |
| def get_sub_tlv(tlvs, tlv_type): |
| for sub_tlv in tlvs: |
| if isinstance(sub_tlv, tlv_type): |
| return sub_tlv |
| |
| |
| def check_address_registration_tlv( |
| command_msg, |
| full_address, |
| ): |
| """Check whether or not a full IPv6 address in AddressRegistrationTlv. |
| """ |
| found = False |
| addr = ipaddress.ip_address(full_address) |
| addresses = command_msg.assertMleMessageContainsTlv(mle.AddressRegistration).addresses |
| |
| for item in addresses: |
| if isinstance(item, mle.AddressFull) and ipaddress.ip_address(item.ipv6_address) == addr: |
| found = True |
| break |
| |
| return found |
| |
| |
| def check_compressed_address_registration_tlv(command_msg, cid, iid, cid_present_once=False): |
| '''Check whether or not a compressed IPv6 address in AddressRegistrationTlv. |
| note: only compare the iid part of the address. |
| |
| Args: |
| command_msg (MleMessage) : The Mle message to check. |
| cid (int): The context id of the domain prefix. |
| iid (string): The Interface Identifier. |
| cid_present_once(boolean): True if cid entry should apprear only once in AR Tlv. |
| False otherwise. |
| ''' |
| found = False |
| cid_cnt = 0 |
| |
| addresses = command_msg.assertMleMessageContainsTlv(mle.AddressRegistration).addresses |
| |
| for item in addresses: |
| if isinstance(item, mle.AddressCompressed): |
| if cid == item.cid: |
| cid_cnt = cid_cnt + 1 |
| if iid == item.iid.hex(): |
| found = True |
| break |
| assert found, 'Error: Expected (cid, iid):({},{}) Not Found'.format(cid, iid) |
| |
| assert cid_present_once == (cid_cnt == 1), 'Error: Expected cid present {} but present {}'.format( |
| 'once' if cid_present_once else '', cid_cnt) |
| |
| |
| def assert_contains_tlv(tlvs, check_type, tlv_type): |
| """Assert a tlv list contains specific tlv and return the first qualified. |
| """ |
| tlvs = [tlv for tlv in tlvs if isinstance(tlv, tlv_type)] |
| if check_type is CheckType.CONTAIN: |
| assert tlvs |
| return tlvs[0] |
| elif check_type is CheckType.NOT_CONTAIN: |
| assert not tlvs |
| return None |
| elif check_type is CheckType.OPTIONAL: |
| return None |
| else: |
| raise ValueError("Invalid check type: {}".format(check_type)) |
| |
| |
| def check_discovery_request(command_msg, thread_version: str = None): |
| """Verify a properly formatted Thread Discovery Request command message. |
| """ |
| assert not isinstance(command_msg.mle, mle.MleMessageSecured) |
| tlvs = command_msg.assertMleMessageContainsTlv(mle.ThreadDiscovery).tlvs |
| request = assert_contains_tlv(tlvs, CheckType.CONTAIN, mesh_cop.DiscoveryRequest) |
| assert not thread_version or thread_version in ['1.1', '1.2'] |
| if thread_version == '1.1': |
| assert request.version == config.THREAD_VERSION_1_1 |
| elif thread_version == '1.2': |
| assert request.version == config.THREAD_VERSION_1_2 |
| |
| |
| def check_discovery_response(command_msg, |
| request_src_addr, |
| steering_data=CheckType.OPTIONAL, |
| thread_version: str = None): |
| """Verify a properly formatted Thread Discovery Response command message. |
| """ |
| assert not isinstance(command_msg.mle, mle.MleMessageSecured) |
| assert (command_msg.mac_header.src_address.type == common.MacAddressType.LONG) |
| assert command_msg.mac_header.dest_address == request_src_addr |
| |
| tlvs = command_msg.assertMleMessageContainsTlv(mle.ThreadDiscovery).tlvs |
| response = assert_contains_tlv(tlvs, CheckType.CONTAIN, mesh_cop.DiscoveryResponse) |
| assert not thread_version or thread_version in ['1.1', '1.2'] |
| if thread_version == '1.1': |
| assert response.version == config.THREAD_VERSION_1_1 |
| elif thread_version == '1.2': |
| assert response.version == config.THREAD_VERSION_1_2 |
| assert_contains_tlv(tlvs, CheckType.CONTAIN, mesh_cop.ExtendedPanid) |
| assert_contains_tlv(tlvs, CheckType.CONTAIN, mesh_cop.NetworkName) |
| assert_contains_tlv(tlvs, steering_data, mesh_cop.SteeringData) |
| assert_contains_tlv(tlvs, steering_data, mesh_cop.JoinerUdpPort) |
| |
| check_type = (CheckType.CONTAIN if response.native_flag else CheckType.OPTIONAL) |
| assert_contains_tlv(tlvs, check_type, mesh_cop.CommissionerUdpPort) |
| |
| |
| def get_joiner_udp_port_in_discovery_response(command_msg): |
| """Get the udp port specified in a DISCOVERY RESPONSE message |
| """ |
| tlvs = command_msg.assertMleMessageContainsTlv(mle.ThreadDiscovery).tlvs |
| udp_port_tlv = assert_contains_tlv(tlvs, CheckType.CONTAIN, mesh_cop.JoinerUdpPort) |
| return udp_port_tlv.udp_port |
| |
| |
| def check_joiner_commissioning_messages(commissioning_messages, url=''): |
| """Verify COAP messages sent by joiner while commissioning process. |
| """ |
| print(commissioning_messages) |
| assert len(commissioning_messages) >= 4 |
| join_fin_req = commissioning_messages[0] |
| assert join_fin_req.type == mesh_cop.MeshCopMessageType.JOIN_FIN_REQ |
| if url: |
| provisioning_url = assert_contains_tlv(join_fin_req.tlvs, CheckType.CONTAIN, mesh_cop.ProvisioningUrl) |
| assert url == provisioning_url.url |
| else: |
| assert_contains_tlv(join_fin_req.tlvs, CheckType.NOT_CONTAIN, mesh_cop.ProvisioningUrl) |
| |
| join_ent_rsp = commissioning_messages[3] |
| assert join_ent_rsp.type == mesh_cop.MeshCopMessageType.JOIN_ENT_RSP |
| |
| |
| def check_commissioner_commissioning_messages(commissioning_messages, state=mesh_cop.MeshCopState.ACCEPT): |
| """Verify COAP messages sent by commissioner while commissioning process. |
| """ |
| assert len(commissioning_messages) >= 2 |
| join_fin_rsq = commissioning_messages[1] |
| assert join_fin_rsq.type == mesh_cop.MeshCopMessageType.JOIN_FIN_RSP |
| rsq_state = assert_contains_tlv(join_fin_rsq.tlvs, CheckType.CONTAIN, mesh_cop.State) |
| assert rsq_state.state == state |
| |
| |
| def check_joiner_router_commissioning_messages(commissioning_messages): |
| """Verify COAP messages sent by joiner router while commissioning process. |
| """ |
| if len(commissioning_messages) >= 4: |
| join_ent_ntf = commissioning_messages[2] |
| else: |
| join_ent_ntf = commissioning_messages[0] |
| assert join_ent_ntf.type == mesh_cop.MeshCopMessageType.JOIN_ENT_NTF |
| return None |
| |
| |
| def check_payload_same(tp1, tp2): |
| """Verfiy two payloads are totally the same. |
| A payload is a tuple of tlvs. |
| """ |
| assert len(tp1) == len(tp2) |
| for tlv in tp2: |
| peer_tlv = get_sub_tlv(tp1, type(tlv)) |
| assert (peer_tlv is not None and |
| peer_tlv == tlv), 'peer_tlv:{}, tlv:{} type:{}'.format(peer_tlv, tlv, type(tlv)) |
| |
| |
| def check_coap_message(msg, payloads, dest_addrs=None): |
| if dest_addrs is not None: |
| found = False |
| for dest in dest_addrs: |
| if msg.ipv6_packet.ipv6_header.destination_address == dest: |
| found = True |
| break |
| assert found, 'Destination address incorrect' |
| check_payload_same(msg.coap.payload, payloads) |
| |
| |
| class SinglePrefixCheck: |
| |
| def __init__(self, prefix=None, border_router_16=None): |
| self._prefix = prefix |
| self._border_router_16 = border_router_16 |
| |
| def check(self, prefix_tlv): |
| border_router_tlv = assert_contains_tlv(prefix_tlv.sub_tlvs, CheckType.CONTAIN, network_data.BorderRouter) |
| assert_contains_tlv(prefix_tlv.sub_tlvs, CheckType.CONTAIN, network_data.LowpanId) |
| result = True |
| if self._prefix is not None: |
| result &= self._prefix == binascii.hexlify(prefix_tlv.prefix) |
| if self._border_router_16 is not None: |
| result &= (self._border_router_16 == border_router_tlv.border_router_16) |
| return result |
| |
| |
| class PrefixesCheck: |
| |
| def __init__(self, prefix_cnt=0, prefix_check_list=()): |
| self._prefix_cnt = prefix_cnt |
| self._prefix_check_list = prefix_check_list |
| |
| def check(self, prefix_tlvs): |
| # if prefix_cnt is given, then check count only |
| if self._prefix_cnt > 0: |
| assert (len(prefix_tlvs) >= self._prefix_cnt), 'prefix count is less than expected' |
| else: |
| for prefix_check in self._prefix_check_list: |
| found = False |
| for prefix_tlv in prefix_tlvs: |
| if prefix_check.check(prefix_tlv): |
| found = True |
| break |
| assert found, 'Some prefix is absent: {}'.format(prefix_check) |
| |
| |
| class CommissioningDataCheck: |
| |
| def __init__(self, stable=None, sub_tlv_type_list=()): |
| self._stable = stable |
| self._sub_tlv_type_list = sub_tlv_type_list |
| |
| def check(self, commissioning_data_tlv): |
| if self._stable is not None: |
| assert (self._stable == commissioning_data_tlv.stable), 'Commissioning Data stable flag is not correct' |
| assert contains_tlvs(commissioning_data_tlv.sub_tlvs, |
| self._sub_tlv_type_list), 'Some sub tlvs are missing in Commissioning Data' |
| |
| |
| class NetworkDataCheck: |
| |
| def __init__(self, prefixes_check=None, commissioning_data_check=None): |
| self._prefixes_check = prefixes_check |
| self._commissioning_data_check = commissioning_data_check |
| |
| def check(self, network_data_tlv): |
| if self._prefixes_check is not None: |
| prefix_tlvs = [tlv for tlv in network_data_tlv.tlvs if isinstance(tlv, network_data.Prefix)] |
| self._prefixes_check.check(prefix_tlvs) |
| if self._commissioning_data_check is not None: |
| commissioning_data_tlv = assert_contains_tlv( |
| network_data_tlv.tlvs, |
| CheckType.CONTAIN, |
| network_data.CommissioningData, |
| ) |
| self._commissioning_data_check.check(commissioning_data_tlv) |
