Fix minor and formatting issues

M1nd3r/thp5
M1nd3r 3 months ago committed by M1nd3r
parent d8c5678960
commit 3abb67f47e

@ -1,7 +1,6 @@
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
import storage.cache as storage_cache import storage.cache as storage_cache
import storage.cache_thp as storage_thp_cache
import storage.device as storage_device import storage.device as storage_device
from trezor import TR, config, utils, wire, workflow from trezor import TR, config, utils, wire, workflow
from trezor.enums import HomescreenFormat, MessageType from trezor.enums import HomescreenFormat, MessageType

@ -9,7 +9,7 @@ from storage.cache_common import InvalidSessionError, SessionlessCache
SESSIONLESS_FLAG = const(128) SESSIONLESS_FLAG = const(128)
if TYPE_CHECKING: if TYPE_CHECKING:
from typing import Sequence, TypeVar, overload from typing import TypeVar, overload
T = TypeVar("T") T = TypeVar("T")
@ -102,7 +102,8 @@ def delete(key: int) -> None:
if TYPE_CHECKING: if TYPE_CHECKING:
@overload @overload
def get(key: int) -> bytes | None: ... def get(key: int) -> bytes | None:
...
@overload @overload
def get(key: int, default: T) -> bytes | T: # noqa: F811 def get(key: int, default: T) -> bytes | T: # noqa: F811

@ -2,10 +2,10 @@ import builtins
from micropython import const from micropython import const
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from trezor import utils from trezor import utils
from storage.cache_common import DataCache, InvalidSessionError from storage.cache_common import DataCache
if TYPE_CHECKING: if TYPE_CHECKING:
from typing import Sequence, TypeVar, overload from typing import TypeVar
T = TypeVar("T") T = TypeVar("T")

@ -27,7 +27,8 @@ class DataCache:
if TYPE_CHECKING: if TYPE_CHECKING:
@overload @overload
def get(self, key: int) -> bytes | None: ... def get(self, key: int) -> bytes | None: # noqa: F811
...
@overload @overload
def get(self, key: int, default: T) -> bytes | T: # noqa: F811 def get(self, key: int, default: T) -> bytes | T: # noqa: F811

@ -7,7 +7,7 @@ from trezor import utils
if TYPE_CHECKING: if TYPE_CHECKING:
from typing import Sequence, TypeVar, overload from typing import TypeVar
T = TypeVar("T") T = TypeVar("T")
@ -218,6 +218,9 @@ def start_session(session_id: bytes | None) -> bytes: # TODO incomplete
def start_existing_session(session_id: bytes) -> bytes: def start_existing_session(session_id: bytes) -> bytes:
global _active_session_idx
global _is_active_session_authenticated
if session_id is None: if session_id is None:
raise ValueError("session_id cannot be None") raise ValueError("session_id cannot be None")
if get_active_session_id() == session_id: if get_active_session_id() == session_id:

@ -148,19 +148,24 @@ def chunks(items: Chunkable, size: int) -> Iterator[Chunkable]:
if TYPE_CHECKING: if TYPE_CHECKING:
class HashContext(Protocol): class HashContext(Protocol):
def update(self, __buf: bytes) -> None: ... def update(self, __buf: bytes) -> None:
...
def digest(self) -> bytes: ... def digest(self) -> bytes:
...
class HashContextInitable(HashContext, Protocol): class HashContextInitable(HashContext, Protocol):
def __init__( # pylint: disable=super-init-not-called def __init__( # pylint: disable=super-init-not-called
self, __data: bytes | None = None self, __data: bytes | None = None
) -> None: ... ) -> None:
...
class Writer(Protocol): class Writer(Protocol):
def append(self, __b: int) -> None: ... def append(self, __b: int) -> None:
...
def extend(self, __buf: bytes) -> None: ... def extend(self, __buf: bytes) -> None:
...
if False: # noqa if False: # noqa

@ -23,17 +23,15 @@ reads the message's header. When the message type is known the first handler is
""" """
from apps import workflow_handlers
from micropython import const from micropython import const
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from storage.cache_codec import InvalidSessionError from storage.cache_common import InvalidSessionError
from trezor import log, loop, protobuf, utils, workflow from trezor import log, loop, protobuf, utils, workflow
from trezor.enums import FailureType from trezor.enums import FailureType
from trezor.messages import Failure from trezor.messages import Failure
from trezor.wire import codec_v1, context, protocol_common from trezor.wire import codec_v1, context, protocol_common
from trezor.wire.errors import ActionCancelled, DataError, Error from trezor.wire.errors import ActionCancelled, DataError, Error
import trezor.enums.MessageType as MT
# Import all errors into namespace, so that `wire.Error` is available from # Import all errors into namespace, so that `wire.Error` is available from
# other packages. # other packages.

@ -80,12 +80,14 @@ class Context:
@overload @overload
async def read( async def read(
self, expected_types: Container[int] self, expected_types: Container[int]
) -> protobuf.MessageType: ... ) -> protobuf.MessageType:
...
@overload @overload
async def read( async def read(
self, expected_types: Container[int], expected_type: type[LoadedMessageType] self, expected_types: Container[int], expected_type: type[LoadedMessageType]
) -> LoadedMessageType: ... ) -> LoadedMessageType:
...
async def read( async def read(
self, self,

@ -1,10 +1,8 @@
import ustruct import ustruct
from storage import cache_thp as storage_thp_cache from storage import cache_thp as storage_thp_cache
from storage.cache_thp import SessionThpCache, BROADCAST_CHANNEL_ID from storage.cache_thp import SessionThpCache
from trezor import io
from trezor.wire.protocol_common import WireError from trezor.wire.protocol_common import WireError
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from ubinascii import hexlify
if TYPE_CHECKING: if TYPE_CHECKING:
from trezorio import WireInterface from trezorio import WireInterface

@ -1,17 +1,15 @@
import ustruct import ustruct
from micropython import const from micropython import const
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from storage.cache_thp import SessionThpCache from storage.cache_thp import SessionThpCache, BROADCAST_CHANNEL_ID
from trezor import io, loop, utils from trezor import io, loop, utils
from trezor.crypto import crc from trezor.crypto import crc
from trezor.wire.protocol_common import Message from trezor.wire.protocol_common import Message
import trezor.wire.thp_session as THP import trezor.wire.thp_session as THP
from trezor.wire.thp_session import ( from trezor.wire.thp_session import (
ThpError, ThpError,
SessionState, SessionState
BROADCAST_CHANNEL_ID,
) )
from ubinascii import hexlify
if TYPE_CHECKING: if TYPE_CHECKING:
from trezorio import WireInterface from trezorio import WireInterface

Loading…
Cancel
Save