mirror of
https://github.com/espressif/openthread.git
synced 2026-11-05 14:03:14 +00:00
[tcat_ble_client][ble] handle sudden BLE disconnection in TCAT Commisioner (#13262)
An issue was seen where a sudden BLE disconnect would not be recognized by the TCAT Commissioner. Reason: its connection state was based on the TLS session only, and TLS stays active in that case. This PR adds detection of the BLE link state: if down, the user is informed that the connection is lost and internal state is reset to 'disconnected'. To be able to test this behavior also in simulation, the simulation platform is extended with a 'BLE connected' state. If BLE is switched off, this results in a simulated disconnect signal sent over UDP so that the TCAT Commissioner BLE-simulation also detects the BLE disconnection.
This commit is contained in:
@@ -48,7 +48,7 @@ static uint8_t sBleBuffer[PLAT_BLE_MSG_DATA_MAX];
|
||||
|
||||
static int sFd = -1;
|
||||
static bool sIsConnected = false;
|
||||
static bool sIsDisconnecting = false;
|
||||
static bool sIsDisconnecting = false; // value only relevant when sIsConnected is true
|
||||
static bool sIsAdvertising = false;
|
||||
static bool sIsEnabled = false;
|
||||
|
||||
@@ -97,6 +97,19 @@ static void deinitFds(void)
|
||||
}
|
||||
}
|
||||
|
||||
// Notify the simulated BLE peer (TCAT Commissioner) that the link is being dropped, by
|
||||
// sending a 0-length UDP datagram.
|
||||
static void sendDisconnectSignal(void)
|
||||
{
|
||||
if (sFd != -1 && sIsConnected)
|
||||
{
|
||||
if (sendto(sFd, sBleBuffer, 0, 0, (struct sockaddr *)&sSockaddr, sizeof(sSockaddr)) == -1)
|
||||
{
|
||||
perror("BLE simulation sendDisconnectSignal() sendto failed.");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
otError otPlatBleGetAdvertisementBuffer(otInstance *aInstance, uint8_t **aAdvertisementBuffer)
|
||||
{
|
||||
OT_UNUSED_VARIABLE(aInstance);
|
||||
@@ -113,10 +126,9 @@ otError otPlatBleEnable(otInstance *aInstance)
|
||||
if (!sIsEnabled)
|
||||
{
|
||||
initFds();
|
||||
sIsEnabled = true;
|
||||
sIsConnected = false;
|
||||
sIsDisconnecting = false;
|
||||
sIsAdvertising = false;
|
||||
sIsEnabled = true;
|
||||
sIsConnected = false;
|
||||
sIsAdvertising = false;
|
||||
}
|
||||
return OT_ERROR_NONE;
|
||||
}
|
||||
@@ -126,6 +138,11 @@ otError otPlatBleDisable(otInstance *aInstance)
|
||||
OT_UNUSED_VARIABLE(aInstance);
|
||||
if (sIsEnabled)
|
||||
{
|
||||
if (sIsConnected)
|
||||
{
|
||||
sendDisconnectSignal();
|
||||
sIsConnected = false;
|
||||
}
|
||||
deinitFds();
|
||||
sIsEnabled = false;
|
||||
}
|
||||
@@ -185,7 +202,7 @@ otError otPlatBleGattServerIndicate(otInstance *aInstance, uint16_t aHandle, con
|
||||
ssize_t rval;
|
||||
otError error = OT_ERROR_NONE;
|
||||
|
||||
otEXPECT_ACTION(sFd != -1, error = OT_ERROR_INVALID_STATE);
|
||||
otEXPECT_ACTION(sFd != -1 && sIsConnected, error = OT_ERROR_INVALID_STATE);
|
||||
rval = sendto(sFd, (const char *)aPacket->mValue, aPacket->mLength, 0, (struct sockaddr *)&sSockaddr,
|
||||
sizeof(sSockaddr));
|
||||
if (rval == -1)
|
||||
@@ -214,7 +231,7 @@ void platformBleUpdateFdSet(fd_set *aReadFdSet, fd_set *aWriteFdSet, struct time
|
||||
}
|
||||
|
||||
// A pending disconnection must be delivered promptly; ensure the main loop does not block.
|
||||
if (sIsDisconnecting && aTimeout != NULL)
|
||||
if (sFd != -1 && sIsConnected && sIsDisconnecting && aTimeout != NULL)
|
||||
{
|
||||
aTimeout->tv_sec = 0;
|
||||
aTimeout->tv_usec = 0;
|
||||
@@ -227,24 +244,25 @@ void platformBleProcess(otInstance *aInstance, const fd_set *aReadFdSet, const f
|
||||
|
||||
otEXPECT(sFd != -1);
|
||||
|
||||
// Deliver a pending disconnection (requested earlier via otPlatBleGapDisconnect).
|
||||
if (sIsDisconnecting)
|
||||
// Deliver a pending disconnection (requested earlier via otPlatBleGapDisconnect)
|
||||
if (sIsDisconnecting && sIsConnected)
|
||||
{
|
||||
// Drain any remaining data and drop it: prevent stale data from triggering a new connection later on.
|
||||
while (recvfrom(sFd, sBleBuffer, sizeof(sBleBuffer), MSG_DONTWAIT, NULL, NULL) >= 0)
|
||||
{
|
||||
}
|
||||
|
||||
sIsConnected = false;
|
||||
sIsDisconnecting = false;
|
||||
sendDisconnectSignal();
|
||||
sIsConnected = false;
|
||||
otPlatBleGapOnDisconnected(aInstance, 0);
|
||||
}
|
||||
else if (FD_ISSET(sFd, aReadFdSet))
|
||||
{
|
||||
socklen_t len = sizeof(sSockaddr);
|
||||
ssize_t rval;
|
||||
memset(&sSockaddr, 0, sizeof(sSockaddr));
|
||||
rval = recvfrom(sFd, sBleBuffer, sizeof(sBleBuffer), 0, (struct sockaddr *)&sSockaddr, &len);
|
||||
struct sockaddr_in sockaddr;
|
||||
socklen_t len = sizeof(sockaddr);
|
||||
ssize_t rval;
|
||||
|
||||
memset(&sockaddr, 0, sizeof(sockaddr));
|
||||
rval = recvfrom(sFd, sBleBuffer, sizeof(sBleBuffer), 0, (struct sockaddr *)&sockaddr, &len);
|
||||
if (rval > 0)
|
||||
{
|
||||
otBleRadioPacket myPacket;
|
||||
@@ -255,8 +273,10 @@ void platformBleProcess(otInstance *aInstance, const fd_set *aReadFdSet, const f
|
||||
// closing TCAT session until the device is ready again for a new session.
|
||||
otEXPECT(sIsAdvertising);
|
||||
|
||||
sIsConnected = true;
|
||||
sIsAdvertising = false; // per API contract, advertising stops once a client connects
|
||||
sSockaddr = sockaddr;
|
||||
sIsConnected = true;
|
||||
sIsDisconnecting = false;
|
||||
sIsAdvertising = false; // per API contract, advertising stops once a client connects
|
||||
otLogDebgPlat("BLE client connected");
|
||||
otPlatBleGapOnConnected(aInstance, 0);
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
"""
|
||||
Copyright (c) 2024-2025, The OpenThread Authors.
|
||||
Copyright (c) 2024-2026, The OpenThread Authors.
|
||||
All rights reserved.
|
||||
|
||||
Redistribution and use in source and binary forms, with or without
|
||||
@@ -32,10 +32,12 @@ import logging
|
||||
|
||||
from bleak import BLEDevice
|
||||
|
||||
from ble.ble_stream import BleConnectionClosed
|
||||
from ble.ble_stream_secure import BleStreamSecure
|
||||
from ble.udp_stream import UdpStream
|
||||
from ble import ble_scanner
|
||||
from cli.cli import CLI
|
||||
from cli.base_commands import connection_closed_helper, disconnect_helper
|
||||
from dataset.dataset import ThreadDataset
|
||||
from cli.command import CommandResult
|
||||
from tlv.tcat_tlv import TcatTLVType
|
||||
@@ -92,6 +94,8 @@ async def main():
|
||||
try:
|
||||
result: CommandResult = await cli.evaluate_input(user_input)
|
||||
result.pretty_print()
|
||||
except BleConnectionClosed:
|
||||
await connection_closed_helper(cli.context)
|
||||
except Exception as e:
|
||||
logger.error(e)
|
||||
logger.debug(e, exc_info=True)
|
||||
@@ -112,12 +116,31 @@ async def receive_loop(cli_context: dict):
|
||||
while True:
|
||||
bless: BleStreamSecure = cli_context['ble_sstream']
|
||||
if bless is not None:
|
||||
data = await bless.recv_unsolicited_event()
|
||||
try:
|
||||
data = await bless.recv_unsolicited_event()
|
||||
except BleConnectionClosed:
|
||||
await connection_closed_helper(cli_context)
|
||||
continue
|
||||
|
||||
if data:
|
||||
logger.info('Received event data from TCAT Device:\n' + hexdump_ot("Event", data))
|
||||
tlv = TLV.from_bytes(data)
|
||||
validate_unsolicited_tlv(tlv)
|
||||
continue
|
||||
|
||||
# The connection can also end without BleConnectionClosed being raised: either the peer
|
||||
# closed TLS gracefully (close-notify), or the BLE link itself dropped.
|
||||
if not bless.is_connected:
|
||||
# Ignore a stream that was already torn down locally (e.g. by the 'disconnect' command)
|
||||
if cli_context['ble_sstream'] is not bless:
|
||||
continue
|
||||
if bless.close_notify_received:
|
||||
print('TCAT Device disconnected: the TLS connection was closed by the device.')
|
||||
await disconnect_helper(cli_context)
|
||||
else:
|
||||
await connection_closed_helper(cli_context)
|
||||
continue
|
||||
|
||||
await asyncio.sleep(0.100)
|
||||
|
||||
|
||||
|
||||
@@ -38,6 +38,10 @@ from bleak.backends.characteristic import BleakGATTCharacteristic
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class BleConnectionClosed(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class BleStream:
|
||||
|
||||
def __init__(self, client, service_uuid, tx_char_uuid, rx_char_uuid):
|
||||
@@ -77,6 +81,8 @@ class BleStream:
|
||||
|
||||
async def send(self, data):
|
||||
logger.debug(f'tx {len(data)} bytes')
|
||||
if not self.client.is_connected:
|
||||
raise BleConnectionClosed('BLE connection was closed')
|
||||
services = self.client.services.get_service(self.service_uuid)
|
||||
rx_char = services.get_characteristic(self.rx_char_uuid)
|
||||
for s in BleStream.__sliced(data, rx_char.max_write_without_response_size):
|
||||
@@ -84,7 +90,10 @@ class BleStream:
|
||||
return len(data)
|
||||
|
||||
async def recv(self, bufsize, recv_timeout=0.200):
|
||||
# check to drain bytes already received before reporting the closed connection
|
||||
if not self.__receive_buffer:
|
||||
if not self.client.is_connected:
|
||||
raise BleConnectionClosed('BLE connection was closed')
|
||||
return b''
|
||||
|
||||
while asyncio.get_running_loop().time() - self.__last_recv_time <= recv_timeout:
|
||||
@@ -98,3 +107,7 @@ class BleStream:
|
||||
async def disconnect(self):
|
||||
if self.client.is_connected:
|
||||
await self.client.disconnect()
|
||||
|
||||
@property
|
||||
def is_connected(self):
|
||||
return self.client.is_connected or len(self.__receive_buffer) > 0
|
||||
|
||||
@@ -337,8 +337,19 @@ class BleStreamSecure:
|
||||
|
||||
@property
|
||||
def is_connected(self):
|
||||
# Also consult the underlying link (self.stream.is_connected): the TLS flags
|
||||
# below are only updated reactively when an inbound close-notify is processed,
|
||||
# so they can still report "connected" after the peer has already dropped the
|
||||
# BLE link. Checking the link makes the pre-command connection check reflect
|
||||
# reality instead of failing later at the actual write.
|
||||
return not self._close_notify_sent and not self._close_notify_received and \
|
||||
self._peer_public_key is not None and self.ssl_object is not None
|
||||
self._peer_public_key is not None and self.ssl_object is not None and \
|
||||
self.stream.is_connected
|
||||
|
||||
@property
|
||||
def close_notify_received(self):
|
||||
"""True once the peer's close-notify alert has been received, whichever side started the close."""
|
||||
return self._close_notify_received
|
||||
|
||||
@property
|
||||
def peer_public_key(self):
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
"""
|
||||
Copyright (c) 2024-2025, The OpenThread Authors.
|
||||
Copyright (c) 2024-2026, The OpenThread Authors.
|
||||
All rights reserved.
|
||||
|
||||
Redistribution and use in source and binary forms, with or without
|
||||
@@ -30,6 +30,8 @@ import logging
|
||||
import socket
|
||||
import select
|
||||
|
||||
from ble.ble_stream import BleConnectionClosed
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@@ -40,6 +42,7 @@ class UdpStream:
|
||||
def __init__(self, address, node_id):
|
||||
self.__receive_buffer = b''
|
||||
self.__last_recv_time = None
|
||||
self.__connected = True
|
||||
self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
self.socket.setblocking(False)
|
||||
self.address = (address, self.BASE_PORT + node_id)
|
||||
@@ -49,12 +52,21 @@ class UdpStream:
|
||||
|
||||
async def send(self, data):
|
||||
logger.debug(f'tx {len(data)} bytes')
|
||||
if not self.__connected:
|
||||
raise BleConnectionClosed('BLE connection (simulation) was closed')
|
||||
return self.socket.sendto(data, self.address)
|
||||
|
||||
async def recv(self, bufsize):
|
||||
if not self.__connected:
|
||||
raise BleConnectionClosed('BLE connection (simulation) was closed')
|
||||
ready = select.select([self.socket], [], [], self.MAX_SERVER_TIMEOUT_SEC)
|
||||
if ready[0]:
|
||||
data = self.socket.recv(bufsize)
|
||||
# A received 0-byte datagram simulates the peer dropping the BLE link
|
||||
if len(data) == 0:
|
||||
logger.debug('rx: BLE link disconnection was simulated (0-byte UDP packet)')
|
||||
self.__connected = False
|
||||
raise BleConnectionClosed('BLE connection (simulation) was closed')
|
||||
logger.debug(f'rx {len(data)} bytes')
|
||||
return data
|
||||
else:
|
||||
@@ -67,4 +79,9 @@ class UdpStream:
|
||||
self.socket.sendto(b'', self.address)
|
||||
|
||||
async def disconnect(self):
|
||||
self.__connected = False
|
||||
self.socket.close()
|
||||
|
||||
@property
|
||||
def is_connected(self):
|
||||
return self.__connected
|
||||
|
||||
@@ -472,15 +472,17 @@ async def connect_helper(device: BLEDevice | UdpStream,
|
||||
else:
|
||||
ble_stream = device
|
||||
ble_sstream = BleStreamSecure(ble_stream)
|
||||
context['ble_sstream'] = ble_sstream
|
||||
context['ble_stream'] = ble_stream
|
||||
|
||||
cert_path = context['cmd_args'].cert_path if context['cmd_args'] else 'auth'
|
||||
ble_sstream.load_cert(
|
||||
certfile=path.join(cert_path, 'commissioner_cert.pem'),
|
||||
keyfile=path.join(cert_path, 'commissioner_key.pem'),
|
||||
cafile=path.join(cert_path, 'ca_cert.pem'),
|
||||
)
|
||||
try:
|
||||
ble_sstream.load_cert(
|
||||
certfile=path.join(cert_path, 'commissioner_cert.pem'),
|
||||
keyfile=path.join(cert_path, 'commissioner_key.pem'),
|
||||
cafile=path.join(cert_path, 'ca_cert.pem'),
|
||||
)
|
||||
except Exception:
|
||||
await ble_stream.disconnect()
|
||||
raise
|
||||
logger.info(f"Certificates and key loaded from '{cert_path}'")
|
||||
|
||||
print('Setting up secure channel...')
|
||||
@@ -495,6 +497,10 @@ async def connect_helper(device: BLEDevice | UdpStream,
|
||||
logger.error(e)
|
||||
|
||||
if ok:
|
||||
# Store the connection only once established: the background receive loop treats a
|
||||
# non-connected stream in the context as a closed link, so it must not see it during the handshake.
|
||||
context['ble_sstream'] = ble_sstream
|
||||
context['ble_stream'] = ble_stream
|
||||
print('Done')
|
||||
return True
|
||||
else:
|
||||
@@ -508,23 +514,47 @@ async def connect_helper(device: BLEDevice | UdpStream,
|
||||
async def disconnect_helper(context: dict) -> None:
|
||||
"""Helper function for CLI and commands to disconnect from a TCAT device."""
|
||||
bless: BleStreamSecure = context['ble_sstream']
|
||||
bles = context['ble_stream']
|
||||
context['ble_sstream'] = None
|
||||
context['ble_stream'] = None
|
||||
|
||||
doing_disconn = False
|
||||
if bless is not None and bless.is_connected:
|
||||
print('Disconnecting...')
|
||||
doing_disconn = True
|
||||
logger.debug('Closing TLS connection.')
|
||||
await bless.close(timeout=5.0)
|
||||
context['ble_sstream'] = None
|
||||
|
||||
bles = context['ble_stream']
|
||||
if bles is not None:
|
||||
logger.debug('Closing BLE connection.')
|
||||
await bles.disconnect()
|
||||
context['ble_stream'] = None
|
||||
if doing_disconn:
|
||||
print('Done')
|
||||
|
||||
|
||||
async def connection_closed_helper(context: dict) -> bool:
|
||||
"""Formally tear down a TCAT link that was lost unexpectedly (e.g. the peer dropped BLE).
|
||||
|
||||
Unlike `disconnect_helper`, this sends no Disconnect TLV and performs no TLS close-notify,
|
||||
because the link is already gone. Returns True if a connection was actually torn down.
|
||||
"""
|
||||
if context['ble_sstream'] is None and context['ble_stream'] is None:
|
||||
return False # nothing to tear down (already disconnected)
|
||||
|
||||
print('TCAT Device disconnected: the BLE connection was closed unexpectedly.')
|
||||
|
||||
# Clear references first so this stays correct if both the receive loop and a command
|
||||
# detect the closed link concurrently, then drop the (already dead) link without any
|
||||
# over-the-link traffic.
|
||||
bles = context['ble_stream']
|
||||
context['ble_sstream'] = None
|
||||
context['ble_stream'] = None
|
||||
if bles is not None:
|
||||
logger.debug('Closing BLE connection.')
|
||||
await bles.disconnect()
|
||||
return True
|
||||
|
||||
|
||||
class ScanCommand(Command):
|
||||
|
||||
def get_help_string(self) -> str:
|
||||
|
||||
Reference in New Issue
Block a user