[tcat] improved debug info format using hexadecimal + ASCII dump (#11881)

This improves debuggability of TCAT client and server, by using one
unified format (hex + ASCII) to show transmitted and received TCAT
data within the TLS session, as well as showing size of the encrypted
(TLS) data. For encrypted data, only size is now shown to avoid
clutter.  Showing the hex + ASCII dump allows devs/testers to visually
read TCAT TLVs from screen and identify how all TCAT commands are
processed by the Thread device.
This commit is contained in:
Esko Dijk
2025-09-09 09:11:05 -07:00
committed by GitHub
parent db7f037f73
commit 3431162a09
6 changed files with 91 additions and 8 deletions
+1
View File
@@ -874,6 +874,7 @@ Error TcatAgent::VerifyHash(const Message &aIncomingMessage,
VerifyOrExit(mRandomChallenge != 0, error = kErrorSecurity);
CalculateHash(mRandomChallenge, reinterpret_cast<const char *>(aBuf), aBufLen, hash);
DumpDebg("Hash", &hash, sizeof(hash));
VerifyOrExit(aIncomingMessage.Compare(aOffset, hash), error = kErrorSecurity);
+11 -1
View File
@@ -420,6 +420,7 @@ void BleSecure::HandleTlsReceive(void *aContext, uint8_t *aBuf, uint16_t aLength
void BleSecure::HandleTlsReceive(uint8_t *aBuf, uint16_t aLength)
{
VerifyOrExit(mReceivedMessage != nullptr);
DumpDebg("Rx", aBuf, aLength);
if (!mTlvMode)
{
@@ -527,6 +528,11 @@ void BleSecure::HandleTransmit(void)
Error error = kErrorNone;
ot::Message *message = mTransmitQueue.GetHead();
#if OT_SHOULD_LOG_AT(OT_LOG_LEVEL_DEBG)
uint16_t len;
uint8_t buf[kTlsDataMaxSize];
#endif
VerifyOrExit(message != nullptr);
mTransmitQueue.Dequeue(*message);
@@ -536,7 +542,11 @@ void BleSecure::HandleTransmit(void)
}
SuccessOrExit(error = mTls.Send(*message));
LogDebg("Transmit");
#if OT_SHOULD_LOG_AT(OT_LOG_LEVEL_DEBG)
len = message->ReadBytes(message->GetOffset(), buf, sizeof(buf));
DumpDebg("Tx", buf, len);
#endif
exit:
FreeMessageOnError(message, error);
+5 -5
View File
@@ -57,7 +57,7 @@ class BleStream:
await self.client.disconnect()
def __handle_rx(self, _: BleakGATTCharacteristic, data: bytearray):
logger.debug(f'received {len(data)} bytes')
logger.debug(f'rx {len(data)} bytes')
self.__receive_buffer += data
self.__last_recv_time = time.time()
@@ -74,7 +74,7 @@ class BleStream:
return self
async def send(self, data):
logger.debug(f'sending {data}')
logger.debug(f'tx {len(data)} bytes')
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):
@@ -88,10 +88,10 @@ class BleStream:
while time.time() - self.__last_recv_time <= recv_timeout:
await sleep(0.1)
message = self.__receive_buffer[:bufsize]
data = self.__receive_buffer[:bufsize]
self.__receive_buffer = self.__receive_buffer[bufsize:]
logger.debug(f'retrieved {message}')
return message
logger.debug(f'rx {len(data)} bytes')
return data
async def disconnect(self):
if self.client.is_connected:
@@ -114,6 +114,7 @@ class BleStreamSecure:
return True
async def send(self, bytes):
logger.debug(f"tx {len(bytes)} bytes\n" + utils.hexdump_ot("Tx", bytes))
self.ssl_object.write(bytes)
encode = self.outgoing.read(4096)
await self.stream.send(encode)
@@ -140,6 +141,8 @@ class BleStreamSecure:
await asyncio.sleep(0.1)
more = await self.stream.recv(buffersize)
self.incoming.write(more)
logger.debug(f"rx {len(decode)} bytes\n" + utils.hexdump_ot("Rx", decode))
return decode
async def send_with_resp(self, bytes):
+2 -2
View File
@@ -45,14 +45,14 @@ class UdpStream:
self.address = (address, self.BASE_PORT + node_id)
async def send(self, data):
logger.debug(f'sending {len(data)} bytes: {data}')
logger.debug(f'tx {len(data)} bytes')
return self.socket.sendto(data, self.address)
async def recv(self, bufsize):
ready = select.select([self.socket], [], [], self.MAX_SERVER_TIMEOUT_SEC)
if ready[0]:
data = self.socket.recv(bufsize)
logger.debug(f'received {len(data)} bytes: {data}')
logger.debug(f'rx {len(data)} bytes')
return data
else:
raise Exception('simulation UdpStream recv timeout - likely, TCAT is stopped on TCAT Device')
+69
View File
@@ -27,6 +27,7 @@
"""
import base64
from tlv import advertised_tlv
@@ -74,3 +75,71 @@ def base64_string(bindata):
def load_cert_pem(fn):
with open(fn, 'r') as file:
return file.read()
def hexdump_ot(text: str, data: bytes, line_prefix: str = "") -> str:
"""
Formats a byte string into a hex dump format similar to OpenThread logs.
Args:
text: String that is printed in the header.
data: The byte array to format.
line_prefix: A string to prepend to each line of the output, such as a log timestamp.
Returns:
A multi-line string containing the formatted hex dump.
"""
lines = []
data_len = len(data)
# 1. Header
header = superimpose_centered_string("=" * 72, f"[{text} len={data_len:03d}]")
lines.append(line_prefix + header)
# 2. Process data in 16-byte chunks
chunk_size = 16
for i in range(0, data_len, chunk_size):
chunk = data[i:i + chunk_size]
# Split into two 8-byte hex groups
hex_part1 = ' '.join(f'{b:02X}' for b in chunk[0:8])
hex_part2 = ' '.join(f'{b:02X}' for b in chunk[8:16])
# Create the ASCII representation (replace non-printables with '.')
ascii_part = ''.join(chr(b) if 32 <= b <= 126 else '.' for b in chunk)
# Format the complete line with fixed-width padding
# Width of each hex half: 8 bytes * 2 chars/byte + 7 spaces = 23 characters
line = f"| {hex_part1:<23} | {hex_part2:<23} | {ascii_part:<16} |"
lines.append(line_prefix + line)
# 3. Create the footer line
footer = "-" * 72
lines.append(line_prefix + footer)
return "\n".join(lines)
def superimpose_centered_string(background: str, foreground: str) -> str:
"""
Superimposes a foreground string onto the center of a background string.
Args:
background: The string to use as the background.
foreground: The string to place on top of the background.
Returns:
A new string with the foreground centered within the background.
If the foreground is longer than or equal to the background's length,
the foreground string is returned on its own.
"""
len_bg = len(background)
len_fg = len(foreground)
if len_fg >= len_bg:
return foreground
start_index = (len_bg - len_fg) // 2
end_index = start_index + len_fg
return background[:start_index] + foreground + background[end_index:]