| #!/usr/bin/env python |
| # |
| # Copyright (c) 2016, 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 io |
| import random |
| import struct |
| import unittest |
| import ipaddress |
| |
| import common |
| import net_crypto |
| import mle |
| |
| master_key = bytearray([0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, |
| 0x88, 0x99, 0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff]) |
| |
| |
| def convert_aux_sec_hdr_to_bytearray(aux_sec_hdr): |
| data = bytearray([aux_sec_hdr.security_level | ((aux_sec_hdr.key_id_mode & 0x03) << 3)]) |
| data += struct.pack("<L", aux_sec_hdr.frame_counter) |
| data += aux_sec_hdr.key_id |
| return data |
| |
| |
| def any_eui64(): |
| return bytearray([random.getrandbits(8) for _ in range(8)]) |
| |
| |
| def any_security_level(): |
| return random.getrandbits(3) |
| |
| |
| def any_key_id_mode(): |
| """ |
| Only key id mode 2. |
| """ |
| return 2 |
| |
| |
| def any_key_id(key_id_mode): |
| if key_id_mode == 2: |
| length = 5 |
| |
| return bytearray([random.getrandbits(8) for _ in range(length)]) |
| |
| |
| def any_auxiliary_security_header(): |
| key_id_mode = any_key_id_mode() |
| key_id = any_key_id(key_id_mode) |
| |
| return net_crypto.AuxiliarySecurityHeader(key_id_mode, any_security_level(), any_frame_counter(), key_id) |
| |
| |
| def any_frame_counter(): |
| return random.getrandbits(32) |
| |
| |
| def any_ip_address(): |
| ip_address_bytes = bytearray([random.getrandbits(8) for _ in range(16)]) |
| return ipaddress.ip_address(bytes(ip_address_bytes)) |
| |
| |
| def any_data(length=None): |
| length = length if length is not None else random.randint(0, 128) |
| return bytearray([random.getrandbits(8) for _ in range(length)]) |
| |
| |
| def any_master_key(): |
| return bytearray([random.getrandbits(8) for _ in range(16)]) |
| |
| |
| class TestCryptoEngine(unittest.TestCase): |
| |
| def test_should_decrypt_bytearray_to_mle_message_when_decrypt_method_is_called(self): |
| # GIVEN |
| message_info = common.MessageInfo() |
| message_info.source_mac_address = common.MacAddress.from_eui64( |
| bytearray([0x00, 0x35, 0xcc, 0x94, 0xd7, 0x7a, 0x07, 0xe8])) |
| |
| message_info.source_ipv6 = "fe80::235:cc94:d77a:07e8" |
| message_info.destination_ipv6 = "ff02::2" |
| |
| message_info.aux_sec_hdr = net_crypto.AuxiliarySecurityHeader(key_id_mode=2, |
| security_level=5, |
| frame_counter=262165, |
| key_id=bytearray([0x00, 0x00, 0x00, 0x00, 0x01])) |
| message_info.aux_sec_hdr_bytes = convert_aux_sec_hdr_to_bytearray(message_info.aux_sec_hdr) |
| |
| data = bytearray([0x9a, 0x5a, 0x9a, 0x5b, 0xba, 0x25, 0x9c, 0x5e, |
| 0x58, 0xa2, 0x7e, 0x75, 0x74, 0xef, 0x79, 0xbc, |
| 0x4f, 0xa3, 0xf9, 0xae, 0xa8, 0x34, 0xf6, 0xf2, |
| 0x37, 0x21, 0x93, 0x60]) |
| |
| mic = bytearray([0xe1, 0xb5, 0xa2, 0x53]) |
| |
| net_crypto_engine = net_crypto.CryptoEngine(net_crypto.MleCryptoMaterialCreator(master_key)) |
| |
| # WHEN |
| mle_msg = net_crypto_engine.decrypt(data, mic, message_info) |
| |
| # THEN |
| expected_mle_msg = bytearray([0x04, 0x00, 0x02, 0x00, 0x00, 0x09, 0x0b, 0x8f, |
| 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x40, 0x00, |
| 0x01, 0xf1, 0x0b, 0x08, 0x65, 0x5e, 0x0f, 0x83, |
| 0x40, 0xc7, 0x83, 0x31]) |
| self.assertEqual(expected_mle_msg, mle_msg) |
| |
| def test_should_encrypt_mle_message_to_bytearray_when_encrypt_method_is_called(self): |
| # GIVEN |
| message_info = common.MessageInfo() |
| message_info.source_mac_address = common.MacAddress.from_eui64( |
| bytearray([0x00, 0x35, 0xcc, 0x94, 0xd7, 0x7a, 0x07, 0xe8])) |
| |
| message_info.source_ipv6 = "fe80::235:cc94:d77a:07e8" |
| message_info.destination_ipv6 = "ff02::2" |
| |
| message_info.aux_sec_hdr = net_crypto.AuxiliarySecurityHeader(key_id_mode=2, |
| security_level=5, |
| frame_counter=262165, |
| key_id=bytearray([0x00, 0x00, 0x00, 0x00, 0x01])) |
| message_info.aux_sec_hdr_bytes = convert_aux_sec_hdr_to_bytearray(message_info.aux_sec_hdr) |
| |
| mle_msg = bytearray([0x04, 0x00, 0x02, 0x00, 0x00, 0x09, 0x0b, 0x8f, |
| 0x80, 0x00, 0x00, 0x00, 0x00, 0x00, 0x40, 0x00, |
| 0x01, 0xf1, 0x0b, 0x08, 0x65, 0x5e, 0x0f, 0x83, |
| 0x40, 0xc7, 0x83, 0x31]) |
| |
| net_crypto_engine = net_crypto.CryptoEngine(net_crypto.MleCryptoMaterialCreator(master_key)) |
| |
| # WHEN |
| encrypted_data, mic = net_crypto_engine.encrypt(mle_msg, message_info) |
| |
| # THEN |
| expected_encrypted_data = bytearray([0x9a, 0x5a, 0x9a, 0x5b, 0xba, 0x25, 0x9c, 0x5e, |
| 0x58, 0xa2, 0x7e, 0x75, 0x74, 0xef, 0x79, 0xbc, |
| 0x4f, 0xa3, 0xf9, 0xae, 0xa8, 0x34, 0xf6, 0xf2, |
| 0x37, 0x21, 0x93, 0x60, 0xe1, 0xb5, 0xa2, 0x53]) |
| |
| self.assertEqual(expected_encrypted_data, encrypted_data + mic) |
| |
| def test_should_encrypt_and_decrypt_random_data_content_when_proper_methods_are_called(self): |
| # GIVEN |
| data = any_data() |
| |
| master_key = any_master_key() |
| |
| key_id_mode = 2 |
| security_level = 5 |
| |
| message_info = common.MessageInfo() |
| message_info.source_mac_address = common.MacAddress.from_eui64(any_eui64()) |
| |
| message_info.source_ipv6 = any_ip_address() |
| message_info.destination_ipv6 = any_ip_address() |
| |
| message_info.aux_sec_hdr = net_crypto.AuxiliarySecurityHeader(key_id_mode=key_id_mode, |
| security_level=security_level, |
| frame_counter=any_frame_counter(), |
| key_id=any_key_id(key_id_mode)) |
| message_info.aux_sec_hdr_bytes = convert_aux_sec_hdr_to_bytearray(message_info.aux_sec_hdr) |
| |
| net_crypto_engine = net_crypto.CryptoEngine(net_crypto.MleCryptoMaterialCreator(master_key)) |
| |
| # WHEN |
| enc_data, mic = net_crypto_engine.encrypt(data, message_info) |
| dec_data = net_crypto_engine.decrypt(enc_data, mic, message_info) |
| |
| # THEN |
| self.assertEqual(data, dec_data) |
| |
| |
| class TestCryptoMaterialCreator(unittest.TestCase): |
| |
| """ Key generaion was described in Thread specification. |
| |
| Read more: Thread 1.1.0 Specification Candidate Final - 7.1.4 Key Generation |
| |
| Test vectors was taken from thread specification. |
| """ |
| |
| def test_should_generate_mle_and_mac_key_when_generate_keys_method_is_called_with_sequence_counter_equal_0(self): |
| """ |
| 7.1.4.1 Test Vector 1 |
| """ |
| |
| # GIVEN |
| sequence_counter = 0 |
| |
| creator = net_crypto.CryptoMaterialCreator(master_key) |
| |
| # WHEN |
| mle_key, mac_key = creator._generate_keys(sequence_counter) |
| |
| # THEN |
| self.assertEqual(mle_key, bytearray([0x54, 0x45, 0xf4, 0x15, 0x8f, 0xd7, 0x59, 0x12, |
| 0x17, 0x58, 0x09, 0xf8, 0xb5, 0x7a, 0x66, 0xa4])) |
| self.assertEqual(mac_key, bytearray([0xde, 0x89, 0xc5, 0x3a, 0xf3, 0x82, 0xb4, 0x21, |
| 0xe0, 0xfd, 0xe5, 0xa9, 0xba, 0xe3, 0xbe, 0xf0])) |
| |
| def test_should_generate_mle_and_mac_key_when_generate_keys_method_is_called_with_sequence_counter_equal_1(self): |
| """ |
| 7.1.4.2 Test Vector 2 |
| """ |
| |
| # GIVEN |
| sequence_counter = 1 |
| |
| creator = net_crypto.CryptoMaterialCreator(master_key) |
| |
| # WHEN |
| mle_key, mac_key = creator._generate_keys(sequence_counter) |
| |
| # THEN |
| self.assertEqual(mle_key, bytearray([0x8f, 0x4c, 0xd1, 0xa2, 0x7d, 0x95, 0xc0, 0x7d, |
| 0x12, 0xdb, 0x89, 0x74, 0xbd, 0x61, 0x5c, 0x13])) |
| self.assertEqual(mac_key, bytearray([0x9b, 0xe0, 0xd1, 0xaf, 0x7b, 0xd8, 0x73, 0x50, |
| 0xde, 0xab, 0xcd, 0xd0, 0x7f, 0xeb, 0xb9, 0xd5])) |
| |
| def test_should_generate_mle_and_mac_key_when_generate_keys_method_is_called_with_sequence_counter_equal_2(self): |
| """ |
| 7.1.4.3 Test Vector 3 |
| """ |
| |
| # GIVEN |
| sequence_counter = 2 |
| |
| creator = net_crypto.CryptoMaterialCreator(master_key) |
| |
| # WHEN |
| mle_key, mac_key = creator._generate_keys(sequence_counter) |
| |
| # THEN |
| self.assertEqual(mle_key, bytearray([0x01, 0x6e, 0x2a, 0xb8, 0xec, 0x88, 0x87, 0x96, |
| 0x87, 0xa7, 0x2e, 0x0a, 0x35, 0x7e, 0xcf, 0x2a])) |
| self.assertEqual(mac_key, bytearray([0x56, 0x41, 0x09, 0xe9, 0xd2, 0xaa, 0xd7, 0xf7, |
| 0x23, 0xec, 0x3b, 0x96, 0x11, 0x0e, 0xef, 0xa3])) |
| |
| |
| class TestMleCryptoMaterialCreator(unittest.TestCase): |
| |
| def test_should_create_nonce_when_create_nonce_method_is_called(self): |
| # GIVEN |
| source_eui64 = any_eui64() |
| frame_counter = any_frame_counter() |
| security_level = any_security_level() |
| |
| creator = net_crypto.MleCryptoMaterialCreator(master_key) |
| |
| # WHEN |
| nonce = creator._create_nonce(source_eui64, frame_counter, security_level) |
| |
| # THEN |
| nonce_bytes = io.BytesIO(nonce) |
| |
| self.assertEqual(source_eui64, nonce_bytes.read(8)) |
| self.assertEqual(struct.pack(">L", frame_counter), nonce_bytes.read(4)) |
| self.assertEqual(security_level, ord(nonce_bytes.read(1))) |
| |
| def test_should_create_authenticated_data_when_create_authenticated_data_method_is_called(self): |
| """ |
| Only Key id mode 2. |
| Length of the Auxiliary Security Header is constantly equal 10. |
| """ |
| |
| # GIVEN |
| source_address = any_ip_address() |
| destination_address = any_ip_address() |
| auxiliary_security_header_bytes = convert_aux_sec_hdr_to_bytearray(any_auxiliary_security_header()) |
| |
| creator = net_crypto.MleCryptoMaterialCreator(master_key) |
| |
| # WHEN |
| authenticated_data = creator._create_authenticated_data( |
| source_address, destination_address, auxiliary_security_header_bytes) |
| |
| # THEN |
| authenticated_data_bytes = io.BytesIO(authenticated_data) |
| |
| self.assertEqual(source_address.packed, authenticated_data_bytes.read(16)) |
| self.assertEqual(destination_address.packed, authenticated_data_bytes.read(16)) |
| self.assertEqual(auxiliary_security_header_bytes, authenticated_data_bytes.read(10)) |
| |
| def test_should_create_key_and_nonce_and_authenticated_data_when_create_key_and_nonce_and_authenticated_data_is_called(self): |
| # GIVEN |
| message_info = common.MessageInfo() |
| message_info.source_mac_address = common.MacAddress.from_eui64(any_eui64()) |
| |
| message_info.source_ipv6 = any_ip_address() |
| message_info.destination_ipv6 = any_ip_address() |
| |
| message_info.aux_sec_hdr = any_auxiliary_security_header() |
| message_info.aux_sec_hdr_bytes = convert_aux_sec_hdr_to_bytearray(message_info.aux_sec_hdr) |
| |
| creator = net_crypto.MleCryptoMaterialCreator(master_key) |
| |
| # WHEN |
| key, nonce, auth_data = creator.create_key_and_nonce_and_authenticated_data(message_info) |
| |
| # THEN |
| self.assertEqual(message_info.source_mac_address.mac_address + |
| struct.pack(">LB", |
| message_info.aux_sec_hdr.frame_counter, |
| message_info.aux_sec_hdr.security_level), nonce) |
| |
| self.assertEqual(message_info.source_ipv6.packed + |
| message_info.destination_ipv6.packed + |
| message_info.aux_sec_hdr_bytes, auth_data) |
| |
| |
| class TestAuxiliarySecurityHeader(unittest.TestCase): |
| |
| def test_should_return_key_id_mode_value_when_key_id_mode_property_is_called(self): |
| # GIVEN |
| key_id_mode = any_key_id_mode() |
| |
| aux_sec_hdr_obj = net_crypto.AuxiliarySecurityHeader( |
| key_id_mode, any_security_level(), any_frame_counter(), any_key_id(key_id_mode)) |
| |
| # WHEN |
| actual_key_id_mode = aux_sec_hdr_obj.key_id_mode |
| |
| # THEN |
| self.assertEqual(key_id_mode, actual_key_id_mode) |
| |
| def test_should_return_security_level_value_when_security_level_property_is_called(self): |
| # GIVEN |
| security_level = any_security_level() |
| key_id_mode = any_key_id_mode() |
| |
| aux_sec_hdr_obj = net_crypto.AuxiliarySecurityHeader( |
| key_id_mode, security_level, any_frame_counter(), any_key_id(key_id_mode)) |
| |
| # WHEN |
| actual_security_level = aux_sec_hdr_obj.security_level |
| |
| # THEN |
| self.assertEqual(security_level, actual_security_level) |
| |
| def test_should_return_frame_counter_value_when_frame_counter_property_is_called(self): |
| # GIVEN |
| frame_counter = any_frame_counter() |
| key_id_mode = any_key_id_mode() |
| |
| aux_sec_hdr_obj = net_crypto.AuxiliarySecurityHeader( |
| key_id_mode, any_security_level(), frame_counter, any_key_id(key_id_mode)) |
| |
| # WHEN |
| actual_frame_counter = aux_sec_hdr_obj.frame_counter |
| |
| # THEN |
| self.assertEqual(frame_counter, actual_frame_counter) |
| |
| def test_should_return_key_id_value_when_key_id_property_is_called(self): |
| # GIVEN |
| key_id_mode = any_key_id_mode() |
| key_id = any_key_id(key_id_mode) |
| |
| aux_sec_hdr_obj = net_crypto.AuxiliarySecurityHeader( |
| key_id_mode, any_security_level(), any_frame_counter(), key_id) |
| |
| # WHEN |
| actual_key_id = aux_sec_hdr_obj.key_id |
| |
| # THEN |
| self.assertEqual(key_id, actual_key_id) |
| |
| def test_should_return_sequence_counter_value_when_sequence_counter_property_is_called(self): |
| # GIVEN |
| key_id_mode = 2 |
| key_id = any_key_id(key_id_mode) |
| |
| aux_sec_hdr_obj = net_crypto.AuxiliarySecurityHeader( |
| key_id_mode, any_security_level(), any_frame_counter(), key_id) |
| |
| # WHEN |
| actual_sequence_counter = aux_sec_hdr_obj.sequence_counter |
| |
| # THEN |
| self.assertEqual(struct.unpack(">I", key_id[:4])[0], actual_sequence_counter) |
| |
| |
| class TestAuxiliarySecurityHeaderFactory(unittest.TestCase): |
| |
| def test_should_create_AuxiliarySecurityHeader_from_bytearray_when_parse_method_is_called(self): |
| # GIVEN |
| key_id_mode = any_key_id_mode() |
| sec_lvl = any_security_level() |
| frame_counter = any_frame_counter() |
| key_id = any_key_id(key_id_mode) |
| |
| factory = net_crypto.AuxiliarySecurityHeaderFactory() |
| |
| data = bytearray([sec_lvl | key_id_mode << 3]) + struct.pack("<I", frame_counter) + key_id |
| |
| # WHEN |
| aux_sec_hdr = factory.parse(io.BytesIO(data), common.MessageInfo()) |
| |
| # THEN |
| self.assertTrue(isinstance(aux_sec_hdr, net_crypto.AuxiliarySecurityHeader)) |
| self.assertEqual(key_id_mode, aux_sec_hdr.key_id_mode) |
| self.assertEqual(sec_lvl, aux_sec_hdr.security_level) |
| self.assertEqual(frame_counter, aux_sec_hdr.frame_counter) |
| |
| if __name__ == "__main__": |
| unittest.main() |
