espressif/openthread
1
0
Fork 0
mirror of https://github.com/espressif/openthread.git synced 2026-08-23 18:59:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

[style] change python yapf column_limit to 119 (#5339)

Browse Source
This commit is contained in:
Simon Lin
2020-08-06 21:40:24 -07:00
committed by GitHub
parent e810357adb
commit d7343c877b
150 changed files with 2910 additions and 5676 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/scripts/thread-cert/test_crypto.py
+82 -142
View File
@@ -36,15 +36,12 @@ import ipaddress
import common
import net_crypto
master_key = bytearray([
0x00, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, 0x88, 0x99, 0xaa, 0xbb,
0xcc, 0xdd, 0xee, 0xff
])
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 = 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
@@ -76,8 +73,7 @@ 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)
return net_crypto.AuxiliarySecurityHeader(key_id_mode, any_security_level(), any_frame_counter(), key_id)
def any_frame_counter():
@@ -100,8 +96,7 @@ def any_master_key():
class TestCryptoEngine(unittest.TestCase):
def test_should_decrypt_bytearray_to_mle_message_when_decrypt_method_is_called(
self):
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(
@@ -110,38 +105,32 @@ class TestCryptoEngine(unittest.TestCase):
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)
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
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))
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
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):
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(
@@ -150,37 +139,31 @@ class TestCryptoEngine(unittest.TestCase):
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)
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
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))
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
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):
def test_should_encrypt_and_decrypt_random_data_content_when_proper_methods_are_called(self):
# GIVEN
data = any_data()
@@ -190,22 +173,18 @@ class TestCryptoEngine(unittest.TestCase):
security_level = 5
message_info = common.MessageInfo()
message_info.source_mac_address = common.MacAddress.from_eui64(
any_eui64())
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)
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))
net_crypto_engine = net_crypto.CryptoEngine(net_crypto.MleCryptoMaterialCreator(master_key))
# WHEN
enc_data, mic = net_crypto_engine.encrypt(data, message_info)
@@ -223,8 +202,7 @@ class TestCryptoMaterialCreator(unittest.TestCase):
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):
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
"""
@@ -240,19 +218,14 @@ class TestCryptoMaterialCreator(unittest.TestCase):
# THEN
self.assertEqual(
mle_key,
bytearray([
0x54, 0x45, 0xf4, 0x15, 0x8f, 0xd7, 0x59, 0x12, 0x17, 0x58,
0x09, 0xf8, 0xb5, 0x7a, 0x66, 0xa4
]))
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
]))
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):
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
"""
@@ -268,19 +241,14 @@ class TestCryptoMaterialCreator(unittest.TestCase):
# THEN
self.assertEqual(
mle_key,
bytearray([
0x8f, 0x4c, 0xd1, 0xa2, 0x7d, 0x95, 0xc0, 0x7d, 0x12, 0xdb,
0x89, 0x74, 0xbd, 0x61, 0x5c, 0x13
]))
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
]))
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):
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
"""
@@ -296,16 +264,12 @@ class TestCryptoMaterialCreator(unittest.TestCase):
# THEN
self.assertEqual(
mle_key,
bytearray([
0x01, 0x6e, 0x2a, 0xb8, 0xec, 0x88, 0x87, 0x96, 0x87, 0xa7,
0x2e, 0x0a, 0x35, 0x7e, 0xcf, 0x2a
]))
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
]))
bytearray([0x56, 0x41, 0x09, 0xe9, 0xd2, 0xaa, 0xd7, 0xf7, 0x23, 0xec, 0x3b, 0x96, 0x11, 0x0e, 0xef,
0xa3]))
class TestMleCryptoMaterialCreator(unittest.TestCase):
@@ -319,8 +283,7 @@ class TestMleCryptoMaterialCreator(unittest.TestCase):
creator = net_crypto.MleCryptoMaterialCreator(master_key)
# WHEN
nonce = creator._create_nonce(source_eui64, frame_counter,
security_level)
nonce = creator._create_nonce(source_eui64, frame_counter, security_level)
# THEN
nonce_bytes = io.BytesIO(nonce)
@@ -329,8 +292,7 @@ class TestMleCryptoMaterialCreator(unittest.TestCase):
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):
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.
@@ -339,68 +301,55 @@ class TestMleCryptoMaterialCreator(unittest.TestCase):
# 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())
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)
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))
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):
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_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)
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)
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)
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)
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):
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))
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
@@ -408,15 +357,13 @@ class TestAuxiliarySecurityHeader(unittest.TestCase):
# THEN
self.assertEqual(key_id_mode, actual_key_id_mode)
def test_should_return_security_level_value_when_security_level_property_is_called(
self):
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))
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
@@ -424,15 +371,13 @@ class TestAuxiliarySecurityHeader(unittest.TestCase):
# THEN
self.assertEqual(security_level, actual_security_level)
def test_should_return_frame_counter_value_when_frame_counter_property_is_called(
self):
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))
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
@@ -445,8 +390,8 @@ class TestAuxiliarySecurityHeader(unittest.TestCase):
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)
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
@@ -454,27 +399,24 @@ class TestAuxiliarySecurityHeader(unittest.TestCase):
# THEN
self.assertEqual(key_id, actual_key_id)
def test_should_return_sequence_counter_value_when_sequence_counter_property_is_called(
self):
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)
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)
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):
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()
@@ -483,15 +425,13 @@ class TestAuxiliarySecurityHeaderFactory(unittest.TestCase):
factory = net_crypto.AuxiliarySecurityHeaderFactory()
data = bytearray([sec_lvl | key_id_mode << 3]) + struct.pack(
"<I", frame_counter) + key_id
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.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)