xref: /aosp_15_r20/external/openthread/tests/scripts/thread-cert/test_crypto.py (revision cfb92d1480a9e65faed56933e9c12405f45898b4)

#!/usr/bin/env python3
#
#  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

network_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_network_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(network_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(network_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()

        network_key = any_network_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(network_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 generation 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(network_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(network_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(network_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(network_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(network_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_auth_data_when_create_key_and_nonce_and_auth_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(network_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()