arno/trezor-firmware
1
0
Fork 0
mirror of https://github.com/trezor/trezor-firmware.git synced 2025-01-10 07:20:56 +00:00
Code Issues Releases Wiki Activity

fix: fix change failed test cause

Browse Source
This commit is contained in:
M1nd3r 2025-01-08 17:01:38 +01:00
parent a785f83af0
commit 62b4c4afca
1 changed files with 5 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
core/src/trezor/wire/thp/channel.py
View File

@ -408,7 +408,7 @@ class Channel:
session_id, session_id,
) )
self.set_channel_state(ChannelState.INVALIDATED) self.set_channel_state(ChannelState.INVALIDATED)
task = self._write_and_encrypt(self.buffer[:noise_payload_len], force) task = self._write_and_encrypt(buffer, noise_payload_len, force)
if task is not None: if task is not None:
await task await task
@ -425,15 +425,14 @@ class Channel:
) )
def _write_and_encrypt( def _write_and_encrypt(
self, payload: bytes, force: bool = False self, payload: bytes, noise_payload_len: int, force: bool = False
) -> Awaitable[None] | None: ) -> Awaitable[None] | None:
payload_length = len(payload)
buffer = memory_manager.get_existing_write_buffer(self.get_channel_id_int()) buffer = memory_manager.get_existing_write_buffer(self.get_channel_id_int())
# if buffer is WireBufferError: # if buffer is WireBufferError:
# pass # TODO handle deviceBUSY # pass # TODO handle deviceBUSY
self._encrypt(buffer, payload_length) self._encrypt(buffer, noise_payload_len)
payload_length = payload_length + TAG_LENGTH payload_length = noise_payload_len + TAG_LENGTH
if self.write_task_spawn is not None: if self.write_task_spawn is not None:
self.write_task_spawn.close() # UPS TODO might break something self.write_task_spawn.close() # UPS TODO might break something
@ -444,7 +443,7 @@ class Channel:
self._log("Writing FORCE message (without async or retransmission).") self._log("Writing FORCE message (without async or retransmission).")
return self._write_encrypted_payload_loop( return self._write_encrypted_payload_loop(
ENCRYPTED, memoryview(self.buffer[:payload_length]) ENCRYPTED, memoryview(buffer[:payload_length])
) )
self.write_task_spawn = loop.spawn( self.write_task_spawn = loop.spawn(
self._write_encrypted_payload_loop( self._write_encrypted_payload_loop(