espressif/openthread
1
0
Fork 0
mirror of https://github.com/espressif/openthread.git synced 2026-09-12 20:20:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

[style] apply google python style guide (#4501)

Browse Source 
This commit applies and enforces Google's python style for tests.
This commit is contained in:
Yakun Xu
2020-02-04 10:27:50 -08:00
committed by GitHub
parent 8368d440dd
commit 33808ebfba
501 changed files with 6674 additions and 5515 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/scripts/thread-cert/test_crypto.py
+155 -93
View File
@@ -36,12 +36,15 @@ 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
@@ -73,7 +76,8 @@ 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():
@@ -96,7 +100,8 @@ 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(
@@ -105,32 +110,38 @@ 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])
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))
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])
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):
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(
@@ -139,31 +150,37 @@ 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])
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))
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])
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):
def test_should_encrypt_and_decrypt_random_data_content_when_proper_methods_are_called(
self):
# GIVEN
data = any_data()
@@ -173,18 +190,22 @@ 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)
@@ -195,7 +216,6 @@ class TestCryptoEngine(unittest.TestCase):
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
@@ -203,7 +223,8 @@ 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
"""
@@ -217,12 +238,21 @@ class TestCryptoMaterialCreator(unittest.TestCase):
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]))
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):
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
"""
@@ -236,12 +266,21 @@ class TestCryptoMaterialCreator(unittest.TestCase):
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]))
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):
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
"""
@@ -255,10 +294,18 @@ class TestCryptoMaterialCreator(unittest.TestCase):
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]))
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):
@@ -272,7 +319,8 @@ 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)
@@ -281,7 +329,8 @@ 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.
@@ -290,64 +339,68 @@ 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)
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
)
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
)
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))
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
@@ -355,13 +408,15 @@ 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))
key_id_mode, security_level, any_frame_counter(),
any_key_id(key_id_mode))
# WHEN
actual_security_level = aux_sec_hdr_obj.security_level
@@ -369,13 +424,15 @@ 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))
key_id_mode, any_security_level(), frame_counter,
any_key_id(key_id_mode))
# WHEN
actual_frame_counter = aux_sec_hdr_obj.frame_counter
@@ -397,7 +454,8 @@ 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)
@@ -409,12 +467,14 @@ class TestAuxiliarySecurityHeader(unittest.TestCase):
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()
@@ -423,13 +483,15 @@ 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)