From b289c9412217f82c8351bbe110fcfc2848bcfcb4 Mon Sep 17 00:00:00 2001 From: Li Cao Date: Tue, 16 Jul 2019 23:47:38 +0800 Subject: [PATCH] [tests] fix mac header ie parsing problem (#3976) --- tests/scripts/thread-cert/common.py | 2 +- tests/scripts/thread-cert/mac802154.py | 214 +++++++++++++------- tests/scripts/thread-cert/net_crypto.py | 8 +- tests/scripts/thread-cert/test_mac802154.py | 2 +- 4 files changed, 146 insertions(+), 80 deletions(-) diff --git a/tests/scripts/thread-cert/common.py b/tests/scripts/thread-cert/common.py index b3bc5a470..4e460af2f 100644 --- a/tests/scripts/thread-cert/common.py +++ b/tests/scripts/thread-cert/common.py @@ -51,7 +51,7 @@ class MessageInfo(object): self.aux_sec_hdr_bytes = None self.mhr_bytes = None - self.nonpayload_fields = None + self.extra_open_fields = None self.source_mac_address = None self.destination_mac_address = None diff --git a/tests/scripts/thread-cert/mac802154.py b/tests/scripts/thread-cert/mac802154.py index 955973f4b..09a677306 100644 --- a/tests/scripts/thread-cert/mac802154.py +++ b/tests/scripts/thread-cert/mac802154.py @@ -66,6 +66,27 @@ class DeviceDescriptors: return short_address +class InformationElement: + + """Class representing 802.15.4 MAC Information Element.""" + def __init__(self, id, length, content): + self._id = id + self._length = length + self._content = content + + @property + def id(self): + return self._id + + @property + def length(self): + return self._length + + @property + def content(self): + return self._content + + class MacHeader: """Class representing 802.15.4 MAC header.""" @@ -138,7 +159,33 @@ class MacFrame: IEEE802154_HEADER_IE_HT1 = 0x7E IEEE802154_HEADER_IE_HT2 = 0x7F + IEEE802154_VERSION_2015 = 0x02 + def parse(self, data): + """Parse a MAC 802.15.4 frame + + Format of MAC 802.15.4 Frame: + | Frame Header | ASH | IE | Frame Payload | MIC | FCS | + + Frame Header: Frame Control | Sequence Number | Addressing Field + ASH: Auxiliary Security Header (optional) + IE: Header IE | Payload IE (optional) + Frame Payload: Data payload (optional) + MIC: To authenticate the frame (optional) + + In the parsing process, Frame Header is parsed first. Then it can be + determined if security is enabled for this frame. If it's enabled, + then ASH should be parsed. + After that, go directly to the end to parse FCS first instead + of parsing IE or Frame Payload. This is because sometimes, IE doesn't + have a termination IE and parsing needs to know the start of MIC to determine + where's the end of IEs(In such cases there are no payload). + After parsing FCS, MIC is parsed if security is enabled. + Then the end of IE(or payload) is known. And at last IEs and payload + are parsed. + And after parsing everything, build MacHeader and MacPayload with + the things parsed. + """ mhr_start = data.tell() fc, seq = struct.unpack("> 7 - header_ie_length = ( - header_ie & MacFrame.IEEE802154_HEADER_IE_LENGTH_MASK - ) - if header_ie_length: - data.read(header_ie_length) - - if header_ie_id in [ - MacFrame.IEEE802154_HEADER_IE_HT1, - MacFrame.IEEE802154_HEADER_IE_HT2, - ]: - break - header_ie_end = data.tell() - - # Check end of MAC frame - if frame_type == MacHeader.FrameType.COMMAND: - command_type = ord(data.read(1)) - else: - command_type = None - - payload_pos = data.tell() + # Record the beginning position of header IE. + # Or the position of payload if there is no header IE. + cur_pos = data.tell() + header_ie_start = cur_pos data.seek(-2, io.SEEK_END) fcs_start = data.tell() + fcs = self._parse_fcs(data, fcs_start) + data.seek(fcs_start) if aux_sec_header and aux_sec_header.security_level: mic, payload_end = self._parse_mic( data, aux_sec_header.security_level ) else: - payload_end = data.tell() + payload_end = fcs_start mic = None - fcs = self._parse_fcs(data, fcs_start) + # There may be no termination IE. In such case, there is no payload, get + # IE until the beginning position of MIC or a termination IE. + header_ie_list = [] + data.seek(cur_pos) + while ie_present and cur_pos + 2 < payload_end: + header_ie = struct.unpack("> 7 + # Currently, payload IE doesn't exist in the code. So only HT2 is required. + # TODO: support HT1 when there are Payload IEs in our code + assert id != MacFrame.IEEE802154_HEADER_IE_HT1, \ + 'Currently there should be no HT1!' + header_ie_length = ( + header_ie & MacFrame.IEEE802154_HEADER_IE_LENGTH_MASK + ) + assert cur_pos + 2 + header_ie_length <= payload_end, \ + 'Parsing Header IE error, IE id:{} length:{}'.format(id, header_ie_length) + header_ie_content = data.read(header_ie_length) + header_ie_list.append( + InformationElement(id, header_ie_length, header_ie_content) + ) + cur_pos += 2 + header_ie_length + if id == MacFrame.IEEE802154_HEADER_IE_HT2: + break + header_ie_end = cur_pos + + payload_pos = cur_pos + payload_len = payload_end - payload_pos + data.seek(payload_pos) + payload = data.read(payload_len) + + if security_enabled: + mhr_len = mhr_end - mhr_start + data.seek(mhr_start) + mhr_bytes = data.read(mhr_len) + + aux_sec_header_len = aux_sec_header_end - mhr_end + aux_sec_hdr_bytes = data.read(aux_sec_header_len) + + extra_open_fields = bytearray([]) + + if ie_present: + data.seek(header_ie_start) + extra_open_fields += data.read( + header_ie_end - header_ie_start + ) + + message_info = MessageInfo() + message_info.aux_sec_hdr = aux_sec_header + message_info.aux_sec_hdr_bytes = aux_sec_hdr_bytes + message_info.extra_open_fields = extra_open_fields + message_info.mhr_bytes = mhr_bytes + + open_payload = [] + private_payload = payload + + # Check end of MAC frame + if frame_type == MacHeader.FrameType.COMMAND: + if frame_version < MacFrame.IEEE802154_VERSION_2015: + extra_open_fields.append(payload[0]) + open_payload = payload[0:1] + private_payload = payload[1:] + message_info.open_payload_length = 1 + + if src_address.type == MacAddressType.SHORT: + message_info.source_mac_address = DeviceDescriptors.get_extended( + src_address).mac_address + else: + message_info.source_mac_address = src_address.mac_address + + sec_obj = CryptoEngine( + MacCryptoMaterialCreator(config.DEFAULT_MASTER_KEY) + ) + self.payload = MacPayload( + bytearray(open_payload) + + sec_obj.decrypt(private_payload, mic, message_info) + ) + + else: + self.payload = MacPayload(payload) + + if frame_type == MacHeader.FrameType.COMMAND: + command_type = self.payload.data[0] + else: + command_type = None # Create Header object self.header = MacHeader( @@ -259,51 +370,6 @@ class MacFrame: fcs, ) - # Create Payload object - payload_len = payload_end - payload_pos - data.seek(payload_pos) - - payload = data.read(payload_len) - - if security_enabled: - mhr_len = mhr_end - mhr_start - data.seek(mhr_start) - mhr_bytes = data.read(mhr_len) - - aux_sec_header_len = aux_sec_header_end - mhr_end - aux_sec_hdr_bytes = data.read(aux_sec_header_len) - - non_payload_fields = bytearray([]) - - if ie_present: - data.seek(header_ie_start) - non_payload_fields += data.read( - header_ie_end - header_ie_start - ) - if command_type is not None: - non_payload_fields.append(command_type) - - message_info = MessageInfo() - message_info.aux_sec_hdr = aux_sec_header - message_info.aux_sec_hdr_bytes = aux_sec_hdr_bytes - message_info.nonpayload_fields = non_payload_fields - message_info.mhr_bytes = mhr_bytes - if src_address.type == MacAddressType.SHORT: - message_info.source_mac_address = DeviceDescriptors.get_extended( - src_address).mac_address - else: - message_info.source_mac_address = src_address.mac_address - - sec_obj = CryptoEngine( - MacCryptoMaterialCreator(config.DEFAULT_MASTER_KEY) - ) - self.payload = MacPayload( - sec_obj.decrypt(payload, mic, message_info) - ) - - else: - self.payload = MacPayload(payload) - def _parse_address(self, data, mode): if mode == MacHeader.AddressMode.SHORT: return MacAddress( diff --git a/tests/scripts/thread-cert/net_crypto.py b/tests/scripts/thread-cert/net_crypto.py index 6bdd263a7..a2f7abaa7 100644 --- a/tests/scripts/thread-cert/net_crypto.py +++ b/tests/scripts/thread-cert/net_crypto.py @@ -159,7 +159,7 @@ class MacCryptoMaterialCreator(CryptoMaterialCreator): return bytes(eui64 + struct.pack(">LB", frame_counter, security_level)) def _create_authenticated_data( - self, mhr, auxiliary_security_header, nonpayload_fields + self, mhr, auxiliary_security_header, extra_open_fields ): """ Create Authenticated Data @@ -168,13 +168,13 @@ class MacCryptoMaterialCreator(CryptoMaterialCreator): Args: mhr (bytes) auxiliary_security_header (bytes) - nonpayload_fields (bytes) + extra_open_fields (bytes) Returns: bytes: Authenticated Data """ - return bytes(mhr + auxiliary_security_header + nonpayload_fields) + return bytes(mhr + auxiliary_security_header + extra_open_fields) def create_key_and_nonce_and_authenticated_data(self, message_info): _, mac_key = self._generate_keys( @@ -190,7 +190,7 @@ class MacCryptoMaterialCreator(CryptoMaterialCreator): auth_data = self._create_authenticated_data( message_info.mhr_bytes, message_info.aux_sec_hdr_bytes, - message_info.nonpayload_fields, + message_info.extra_open_fields, ) return mac_key, nonce, auth_data diff --git a/tests/scripts/thread-cert/test_mac802154.py b/tests/scripts/thread-cert/test_mac802154.py index 7f2794cdd..2bd10a827 100755 --- a/tests/scripts/thread-cert/test_mac802154.py +++ b/tests/scripts/thread-cert/test_mac802154.py @@ -198,7 +198,7 @@ class TestMacParser(unittest.TestCase): self.assertEqual(0xface, frame.header.src_pan_id) self.assertEqual(0xf00d, frame.header.src_address.rloc) - self.assertEqual(bytearray([]), frame.payload.data) + self.assertEqual(bytearray([0x4]), frame.payload.data) def test_should_decrypt_data_frame(self):