diff --git a/examples/platforms/simulation/ble.c b/examples/platforms/simulation/ble.c index 7a57c5284..aeaa1d7ef 100644 --- a/examples/platforms/simulation/ble.c +++ b/examples/platforms/simulation/ble.c @@ -256,8 +256,13 @@ void platformBleProcess(otInstance *aInstance, const fd_set *aReadFdSet, const f } else if (rval == 0) { - // socket is closed, which should not happen - assert(false); + // A zero-length datagram: simulates Commissioner suddenly disconnected the BLE link. + if (sIsConnected) + { + sIsConnected = false; + otLogDebgPlat("BLE client link disconnected"); + otPlatBleGapOnDisconnected(aInstance, 0); + } } else if (errno != EINTR && errno != EAGAIN) { diff --git a/tools/tcat_ble_client/ble/udp_stream.py b/tools/tcat_ble_client/ble/udp_stream.py index 2d3b980a0..7f1edf742 100644 --- a/tools/tcat_ble_client/ble/udp_stream.py +++ b/tools/tcat_ble_client/ble/udp_stream.py @@ -60,6 +60,11 @@ class UdpStream: else: return b'' + async def simulation_ble_disconnect(self): + # Simulate a BLE link break (e.g. peer out of range) by sending a zero-length UDP + # datagram. Unlike `disconnect`, this does not send a Disconnect TLV and does not + # perform a clean TLS shutdown. + self.socket.sendto(b'', self.address) + async def disconnect(self): - if self.socket is not None: - self.socket.close() + self.socket.close() diff --git a/tools/tcat_ble_client/cli/base_commands.py b/tools/tcat_ble_client/cli/base_commands.py index c466d7a64..1bdad9cdd 100644 --- a/tools/tcat_ble_client/cli/base_commands.py +++ b/tools/tcat_ble_client/cli/base_commands.py @@ -244,6 +244,27 @@ class DisconnectCommand(Command): return CommandResultNone() +class SimulationBleDisconnectCommand(Command): + + def get_help_string(self) -> str: + return 'Simulate a BLE link break to a simulated TCAT device (no Disconnect TLV, no TLS shutdown).' + + async def execute_default(self, args, context) -> CommandResult: + ble_stream = context['ble_stream'] + if not isinstance(ble_stream, UdpStream): + return CommandResultError('only available for a simulation connection (use \'simulation \' first).') + + print('Disconnecting simulated BLE link...') + # Signal the abrupt link break to the device, then drop local stream state without a TLS + # shutdown (no close-notify), mirroring a real abrupt disconnect on the client side too. + await ble_stream.simulation_ble_disconnect() + await ble_stream.disconnect() + context['ble_stream'] = None + context['ble_sstream'] = None + print('Done') + return CommandResultNone() + + class ExtractDatasetCommand(BleCommand): def get_log_string(self) -> str: diff --git a/tools/tcat_ble_client/cli/cli.py b/tools/tcat_ble_client/cli/cli.py index 25fc6a0a3..dfc1c8354 100644 --- a/tools/tcat_ble_client/cli/cli.py +++ b/tools/tcat_ble_client/cli/cli.py @@ -38,7 +38,7 @@ from cli.base_commands import (DisconnectCommand, HelpCommand, HelloCommand, Com GetRandomNumberChallenge, ThreadStateCommand, ScanCommand, PresentHash, DiagnosticTlvsCommand, GetApplicationLayersCommand, SendVendorData, SendApplicationData1, SendApplicationData2, SendApplicationData3, SendApplicationData4, - SimulationCommand, connect_helper, disconnect_helper) + SimulationCommand, SimulationBleDisconnectCommand, connect_helper, disconnect_helper) from .command import CommandResultNone, CommandResult from .tlv_commands import TlvCommand from cli.dataset_commands import (DatasetCommand) @@ -72,6 +72,7 @@ class CLI: 'thread': ThreadStateCommand(), 'scan': ScanCommand(), 'simulation': SimulationCommand(), + 'simulation_ble_disconnect': SimulationBleDisconnectCommand(), 'random_challenge': GetRandomNumberChallenge(), 'present_hash': PresentHash(), 'tlv': TlvCommand(),