espressif/esp-nimble
1
0
Fork 0
mirror of https://github.com/espressif/esp-nimble.git synced 2026-08-21 09:20:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

tools/hci_throughput: testing over encrypted link

Browse Source 
Added option enable_encryption to run throughput
tests over encrypted link.
This commit is contained in:
Jakub
2022-08-04 08:57:21 +02:00
committed by Szymon Janc
parent f9eb3c8d27
commit 82dd24d825
5 changed files with 135 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/hci_throughput/config.yaml.sample
+1
View File
@@ -43,6 +43,7 @@ adv:
peer_address: 00:00:00:00:00:00
advertising_channel_map: 7
advertising_filter_policy: 0
enable_encryption: true
conn:
le_scan_interval: 2400
le_scan_window: 2400
tools/hci_throughput/hci.py
+37
View File
@@ -35,6 +35,7 @@ HCI_EVENT_PACKET = 0x04
L2CAP_HDR_BYTES = 4
HCI_EV_CODE_DISCONN_CMP = 0x05
HCI_EV_CODE_ENCRYPTION_CHANGE = 0x08
HCI_EV_CODE_CMD_CMP = 0x0e
HCI_EV_CODE_CMD_STATUS = 0x0f
HCI_EV_CODE_LE_META_EVENT = 0x3e
@@ -42,6 +43,7 @@ HCI_SUBEV_CODE_LE_ENHANCED_CONN_CMP = 0x0a
HCI_SUBEV_CODE_LE_DATA_LEN_CHANGE = 0x07
HCI_SUBEV_CODE_LE_PHY_UPDATE_CMP = 0x0c
HCI_SUBEV_CODE_LE_CHAN_SEL_ALG = 0x14
HCI_SUBEV_CODE_LE_LONG_TERM_KEY_REQUEST = 0x05
HCI_EV_NUM_COMP_PKTS = 0x13
CONN_FAILED_TO_BE_ESTABLISHED = 0x3e
@@ -66,6 +68,8 @@ OCF_LE_SET_ADVERTISE_ENABLE = 0x000a
OCF_LE_SET_SCAN_PARAMETERS = 0x000b
OCF_LE_SET_SCAN_ENABLE = 0x000c
OCF_LE_CREATE_CONN = 0x000d
OCF_LE_ENABLE_ENCRYPTION = 0x0019
OCF_LE_LONG_TERM_KEY_REQUEST_REPLY = 0x001A
OCF_LE_SET_DATA_LEN = 0x0022
OCF_LE_READ_SUGGESTED_DFLT_DATA_LEN = 0x0023
OCF_LE_READ_MAX_DATA_LEN = 0x002f
@@ -106,6 +110,7 @@ num_of_completed_packets_cnt = 0
num_of_completed_packets_time = 0
read_local_commands = None
le_read_local_supported_features = None
ltk = None
############
# FUNCTIONS
@@ -303,6 +308,21 @@ class HCI_Ev_Cmd_Status:
self.opcode = opcode
@dataclass
class HCI_Ev_LE_Encryption_Change():
status: int
connection_handle: int
encryption_enabled: int
def __init__(self):
self.set()
def set(self, status=0, connection_handle=0, encryption_enabled=0):
self.status = status
self.connection_handle = connection_handle
self.encryption_enabled = encryption_enabled
@dataclass
class HCI_Ev_LE_Meta:
subevent_code: int
@@ -374,6 +394,23 @@ class HCI_Ev_LE_Data_Length_Change(HCI_Ev_LE_Meta):
self.triggered = triggered
@dataclass
class HCI_Ev_LE_Long_Term_Key_Request(HCI_Ev_LE_Meta):
conn_handle: int
random_number: int
encrypted_diversifier: int
def __init__(self):
self.set()
def set(self, subevent_code=0, conn_handle=0, random_number=0,
encrypted_diversifier=0):
super().set(subevent_code)
self.conn_handle = conn_handle
self.random_number = random_number
self.encrypted_diversifier = encrypted_diversifier
@dataclass
class HCI_Ev_LE_PHY_Update_Complete(HCI_Ev_LE_Meta):
status: int
tools/hci_throughput/hci_commands.py
+84
View File
@@ -47,6 +47,7 @@ class HCI_Commands():
self.async_sem_cmd = asyncio.Semaphore()
self.async_ev_cmd_end = asyncio.Event()
self.async_ev_connected = asyncio.Event()
self.async_ev_encryption_change = asyncio.Event()
self.async_ev_set_data_len = asyncio.Event()
self.async_ev_update_phy = asyncio.Event()
self.async_ev_num_cmp_pckts = asyncio.Event()
@@ -219,6 +220,41 @@ class HCI_Commands():
await self.async_ev_cmd_end.wait()
self.async_ev_cmd_end.clear()
async def cmd_le_enable_encryption(self, conn_handle: int, random_number: int, ediv: int, ltk: int):
async with self.async_sem_cmd:
hci.ltk = ltk
random_number_bytes = random_number.to_bytes(8, byteorder='little')
ltk_bytes = ltk.to_bytes(16, byteorder='little')
data_bytes = struct.pack("<H", conn_handle) + random_number_bytes + \
struct.pack("<H", ediv) + ltk_bytes
self.hci_send_cmd.set(
hci.OGF_LE_CTL,
hci.OCF_LE_ENABLE_ENCRYPTION,
data_bytes)
logging.debug("%s %s", self.cmd_le_enable_encryption.__name__,
self.hci_send_cmd)
await self.send(self.hci_send_cmd.ba_full_message)
await self.async_ev_cmd_end.wait()
self.async_ev_cmd_end.clear()
async def cmd_le_long_term_key_request_reply(self, conn_handle: int, ltk: int):
async with self.async_sem_cmd:
ltk_bytes = ltk.to_bytes(16, byteorder='little')
data_bytes = struct.pack('<H', conn_handle) + ltk_bytes
self.hci_send_cmd.set(
hci.OGF_LE_CTL,
hci.OCF_LE_LONG_TERM_KEY_REQUEST_REPLY,
data_bytes)
logging.debug(
"%s %s",
self.cmd_le_long_term_key_request_reply.__name__,
self.hci_send_cmd)
await self.send(self.hci_send_cmd.ba_full_message)
# Is run from another command,
# don't need to wait for cmd complete.
# await self.async_ev_cmd_end.wait()
# self.async_ev_cmd_end.clear()
async def cmd_le_set_data_len(self, conn_handle: int, tx_octets: int, tx_time: int):
""" conn_handle: Range 0x0000 to 0x0EFF
tx_octets: Range 0x001B to 0x00FB
@@ -358,6 +394,11 @@ class HCI_Commands():
ev_cmd_stat.set(*struct.unpack('<BBH', bytes(data[:4])))
return ev_cmd_stat
def parse_ev_encryption_change(self, data: bytes):
ev_encryption_change = hci.HCI_Ev_LE_Encryption_Change()
ev_encryption_change.set(*struct.unpack('<BHB', bytes(data[:4])))
return ev_encryption_change
def parse_ev_le_meta(self, data: bytes):
ev_le_meta = hci.HCI_Ev_LE_Meta()
ev_le_meta.set(data[0])
@@ -377,6 +418,12 @@ class HCI_Commands():
ev_le_data_len_change.set(*struct.unpack('<BHHHHH', bytes(data[:11])))
return ev_le_data_len_change
def parse_subev_le_long_term_key_request(self, data: bytes):
ev_le_long_term_key_request = hci.HCI_Ev_LE_Long_Term_Key_Request()
ev_le_long_term_key_request.set(
*struct.unpack('<BHQH', bytes(data[:13])))
return ev_le_long_term_key_request
def parse_subev_le_phy_update_cmp(self, data: bytes):
le_phy_update_cmp = hci.HCI_Ev_LE_PHY_Update_Complete()
le_phy_update_cmp.set(*struct.unpack('<BBHBB', data))
@@ -533,6 +580,15 @@ class HCI_Commands():
self.hci_recv_ev_packet.current_event))
return hci.HCI_SUBEV_CODE_LE_ENHANCED_CONN_CMP
elif subev_code == hci.HCI_SUBEV_CODE_LE_LONG_TERM_KEY_REQUEST:
self.hci_recv_ev_packet.current_event = \
self.parse_subev_le_long_term_key_request(
self.hci_recv_ev_packet.recv_data)
hci.events_list.append(
(hci.HCI_SUBEV_CODE_LE_LONG_TERM_KEY_REQUEST,
self.hci_recv_ev_packet.current_event))
return hci.HCI_SUBEV_CODE_LE_LONG_TERM_KEY_REQUEST
elif subev_code == hci.HCI_SUBEV_CODE_LE_DATA_LEN_CHANGE:
self.hci_recv_ev_packet.current_event = \
self.parse_subev_le_data_len_change(
@@ -584,6 +640,14 @@ class HCI_Commands():
self.hci_recv_ev_packet.current_event))
return hci.HCI_EV_CODE_CMD_STATUS
elif self.hci_recv_ev_packet.ev_code == hci.HCI_EV_CODE_ENCRYPTION_CHANGE:
self.hci_recv_ev_packet.current_event = \
self.parse_ev_encryption_change(
self.hci_recv_ev_packet.recv_data)
hci.events_list.append((hci.HCI_EV_CODE_ENCRYPTION_CHANGE,
self.hci_recv_ev_packet.current_event))
return hci.HCI_EV_CODE_ENCRYPTION_CHANGE
elif self.hci_recv_ev_packet.ev_code == hci.HCI_EV_CODE_LE_META_EVENT:
self.hci_recv_ev_packet.current_event = \
self.parse_ev_le_meta(self.hci_recv_ev_packet.recv_data)
@@ -647,6 +711,19 @@ class HCI_Commands():
logging.error("Status: %s for event: %s", status, curr_ev)
self.async_ev_cmd_end.set()
elif event_code == hci.HCI_EV_CODE_ENCRYPTION_CHANGE:
logging.debug(
"Received code: %s - HCI_EV_CODE_ENCRYPTION_CHANGE",
event_code)
status = curr_ev.status
encryption_enabled = curr_ev.encryption_enabled
if (status == 0 and encryption_enabled != 0):
self.async_ev_encryption_change.set()
else:
raise Exception(
"Encryption failed. Status: %d, encryption enabled: %d",
status, encryption_enabled)
elif event_code == hci.HCI_EV_CODE_LE_META_EVENT:
logging.debug(
"Received code: %s - HCI_EV_CODE_LE_META_EVENT", event_code)
@@ -678,6 +755,13 @@ class HCI_Commands():
"Received subev code: %s - HCI_SUBEV_CODE_LE_CHAN_SEL_ALG",
subev_code)
elif subev_code == hci.HCI_SUBEV_CODE_LE_LONG_TERM_KEY_REQUEST:
logging.debug(
"Received subev code: %s - HCI_SUBEV_CODE_LE_LONG_TERM_KEY_REQUEST",
subev_code)
await self.cmd_le_long_term_key_request_reply(
hci.conn_handle, hci.ltk)
elif subev_code < 0:
logging.warning(f"Unknown received subevent: {buffer}\n")
tools/hci_throughput/hci_device.py
+6 -1
View File
@@ -218,6 +218,10 @@ async def async_main_tx(bt_dev: hci_commands.HCI_Commands, ini: dict, cfg: dict)
hci.le_read_local_supported_features.le_features)
await hci_commands.wait_for_event(bt_dev.async_ev_update_phy, hci.WAIT_FOR_EVENT_TIMEOUT)
if cfg["enable_encryption"]:
await bt_dev.cmd_le_enable_encryption(hci.conn_handle, random_number=0, ediv=0, ltk=hci.ltk)
await hci_commands.wait_for_event(bt_dev.async_ev_encryption_change, 10)
############
# L2CAP SEND
############
@@ -289,9 +293,10 @@ def parse_cfg_files(args) -> dict:
with open(args.init_file, "r") as file:
init_file = yaml.safe_load(file)
ini = init_file[args.mode]
global test_dir, transport_directory
global test_dir, transport_directory, ltk
test_dir = init_file["test_dir"]
transport_directory = init_file["transport_directory"]
hci.ltk = int(init_file["ltk"], 16)
with open(args.config_file) as f:
cfg = yaml.safe_load(f)
tools/hci_throughput/main.py
+7 -1
View File
@@ -29,6 +29,7 @@ import csv
import util
import os
import math
import random
PROCESS_TIMEOUT = 500 # seconds, adjust if necessary
@@ -94,6 +95,10 @@ def change_config_var(filename: str, group: str, variable: str,
yaml.safe_dump(cfg, file, indent=1, sort_keys=False,
default_style=None, default_flow_style=False)
def generate_long_term_key():
rand_val = random.getrandbits(128)
return rand_val.to_bytes(16, byteorder='little')
def get_init_dict(filename: str, args_list: list, modes: list, dir: str,
transport_directory: str):
@@ -115,7 +120,8 @@ def get_init_dict(filename: str, args_list: list, modes: list, dir: str,
"peer_address": args_list[0][2]
},
"test_dir": dir,
"transport_directory": transport_directory
"transport_directory": transport_directory,
"ltk": hex(random.getrandbits(128))
}
with open(filename, 'w') as file: