1
0
mirror of https://github.com/trezor/trezor-firmware.git synced 2025-07-23 23:18:16 +00:00

feat(core): send address & public key response before showing StatusScreen

Co-authored-by: matejcik <ja@matejcik.cz>
This commit is contained in:
Roman Zeyde 2025-04-03 16:20:46 +03:00
parent 3f520b4ce2
commit 946fc89449
37 changed files with 1650 additions and 1464 deletions

View File

@ -0,0 +1 @@
[T3T1] Show confirmation layout after sending response to host.

View File

@ -709,7 +709,6 @@ static void _librust_qstrs(void) {
MP_QSTR_time_ms;
MP_QSTR_timer;
MP_QSTR_title;
MP_QSTR_title_success;
MP_QSTR_total_fee_new;
MP_QSTR_total_len;
MP_QSTR_touch_event;

View File

@ -543,7 +543,6 @@ extern "C" fn new_flow_get_address(n_args: usize, args: *const Obj, kwargs: *mut
let account: Option<TString> = kwargs.get(Qstr::MP_QSTR_account)?.try_into_option()?;
let path: Option<TString> = kwargs.get(Qstr::MP_QSTR_path)?.try_into_option()?;
let xpubs: Obj = kwargs.get(Qstr::MP_QSTR_xpubs)?;
let title_success: TString = kwargs.get(Qstr::MP_QSTR_title_success)?.try_into()?;
let br_code: u16 = kwargs.get(Qstr::MP_QSTR_br_code)?.try_into()?;
let br_name: TString = kwargs.get(Qstr::MP_QSTR_br_name)?.try_into()?;
@ -558,7 +557,6 @@ extern "C" fn new_flow_get_address(n_args: usize, args: *const Obj, kwargs: *mut
account,
path,
xpubs,
title_success,
br_code,
br_name,
)?;
@ -1414,7 +1412,6 @@ pub static mp_module_trezorui_api: Module = obj_module! {
/// account: str | None,
/// path: str | None,
/// xpubs: list[tuple[str, str]],
/// title_success: str,
/// br_code: ButtonRequestType,
/// br_name: str,
/// ) -> LayoutObj[UiResult]:

View File

@ -602,7 +602,6 @@ impl FirmwareUI for UIBolt {
_account: Option<TString<'static>>,
_path: Option<TString<'static>>,
_xpubs: Obj,
_title_success: TString<'static>,
_br_code: u16,
_br_name: TString<'static>,
) -> Result<impl LayoutMaybeTrace, Error> {

View File

@ -736,7 +736,6 @@ impl FirmwareUI for UICaesar {
_account: Option<TString<'static>>,
_path: Option<TString<'static>>,
_xpubs: Obj,
_title_success: TString<'static>,
_br_code: u16,
_br_name: TString<'static>,
) -> Result<impl LayoutMaybeTrace, Error> {

View File

@ -172,12 +172,12 @@ impl StatusScreen {
)
}
pub fn new_success_timeout(msg: TString<'static>) -> Self {
pub fn new_success_timeout(msg: TString<'static>, time_ms: u32) -> Self {
Self::new(
theme::ICON_SIMPLE_CHECKMARK30,
theme::GREEN_LIME,
theme::GREEN_LIGHT,
DismissType::Timeout(Timeout::new(TIMEOUT_MS)),
DismissType::Timeout(Timeout::new(time_ms)),
msg,
)
}

View File

@ -20,7 +20,7 @@ use crate::{
};
use super::super::{
component::{AddressDetails, Frame, PromptScreen, StatusScreen, SwipeContent, VerticalMenu},
component::{AddressDetails, Frame, PromptScreen, SwipeContent, VerticalMenu},
theme,
};
@ -30,7 +30,6 @@ const QR_BORDER: i16 = 4;
pub enum GetAddress {
Address,
Tap,
Confirmed,
Menu,
QrCode,
AccountInfo,
@ -64,9 +63,8 @@ impl FlowController for GetAddress {
fn handle_event(&'static self, msg: FlowMsg) -> Decision {
match (self, msg) {
(Self::Address, FlowMsg::Info) => Self::Menu.goto(),
(Self::Tap, FlowMsg::Confirmed) => Self::Confirmed.swipe_up(),
(Self::Tap, FlowMsg::Confirmed) => self.return_msg(FlowMsg::Confirmed),
(Self::Tap, FlowMsg::Info) => Self::Menu.swipe_left(),
(Self::Confirmed, _) => self.return_msg(FlowMsg::Confirmed),
(Self::Menu, FlowMsg::Choice(0)) => Self::QrCode.swipe_left(),
(Self::Menu, FlowMsg::Choice(1)) => Self::AccountInfo.swipe_left(),
(Self::Menu, FlowMsg::Choice(2)) => Self::Cancel.swipe_left(),
@ -93,7 +91,6 @@ pub fn new_get_address(
account: Option<TString<'static>>,
path: Option<TString<'static>>,
xpubs: Obj, // TODO: get rid of Obj
title_success: TString<'static>,
br_code: u16,
br_name: TString<'static>,
) -> Result<SwipeFlow, error::Error> {
@ -131,14 +128,6 @@ pub fn new_get_address(
.with_swipe(Direction::Left, SwipeSettings::default())
.map(super::util::map_to_confirm);
let content_confirmed = Frame::left_aligned(
TR::words__title_success.into(),
StatusScreen::new_success_timeout(title_success),
)
.with_footer(TR::instructions__continue_in_app.into(), None)
.with_result_icon(theme::ICON_BULLET_CHECKMARK, theme::GREEN_LIGHT)
.map(|_| Some(FlowMsg::Confirmed));
// Menu
let content_menu = Frame::left_aligned(
"".into(),
@ -200,7 +189,6 @@ pub fn new_get_address(
let mut res = SwipeFlow::new(&GetAddress::Address)?;
res.add_page(&GetAddress::Address, content_address)?
.add_page(&GetAddress::Tap, content_tap)?
.add_page(&GetAddress::Confirmed, content_confirmed)?
.add_page(&GetAddress::Menu, content_menu)?
.add_page(&GetAddress::QrCode, content_qr)?
.add_page(&GetAddress::AccountInfo, content_account)?

View File

@ -9,6 +9,7 @@ use crate::{
ui::{
component::{
connect::Connect,
swipe_detect::SwipeSettings,
text::{
op::OpTextLayout,
paragraphs::{
@ -19,7 +20,7 @@ use crate::{
},
Border, CachedJpeg, ComponentExt, Empty, FormattedText, Never, Timeout,
},
geometry::{self, Offset},
geometry::{self, Direction, Offset},
layout::{
obj::{LayoutMaybeTrace, LayoutObj, RootComponent},
util::{PropsList, RecoveryType},
@ -630,7 +631,6 @@ impl FirmwareUI for UIDelizia {
account: Option<TString<'static>>,
path: Option<TString<'static>>,
xpubs: Obj,
title_success: TString<'static>,
br_code: u16,
br_name: TString<'static>,
) -> Result<impl LayoutMaybeTrace, Error> {
@ -645,7 +645,6 @@ impl FirmwareUI for UIDelizia {
account,
path,
xpubs,
title_success,
br_code,
br_name,
)?;
@ -1045,24 +1044,34 @@ impl FirmwareUI for UIDelizia {
fn show_success(
title: TString<'static>,
_button: TString<'static>,
button: TString<'static>,
description: TString<'static>,
_allow_cancel: bool,
_time_ms: u32,
time_ms: u32,
) -> Result<Gc<LayoutObj>, Error> {
let instruction = if button.is_empty() {
TR::instructions__tap_to_continue.into()
} else {
button
};
// description used in the Footer
let description = if description.is_empty() {
None
} else {
Some(description)
};
let content = StatusScreen::new_success(title);
let content = if time_ms > 0 {
StatusScreen::new_success_timeout(title, time_ms)
} else {
StatusScreen::new_success(title)
};
let layout = LayoutObj::new(SwipeUpScreen::new(
Frame::left_aligned(
TR::words__title_success.into(),
SwipeContent::new(content).with_no_attach_anim(),
)
.with_swipeup_footer(description)
.with_footer(instruction, description)
.with_swipe(Direction::Up, SwipeSettings::default())
.with_result_icon(theme::ICON_BULLET_CHECKMARK, theme::GREEN_LIGHT),
))?;
Ok(layout)

View File

@ -203,7 +203,6 @@ pub trait FirmwareUI {
account: Option<TString<'static>>,
path: Option<TString<'static>>,
xpubs: Obj, // TODO: replace Obj
title_success: TString<'static>,
br_code: u16,
br_name: TString<'static>,
) -> Result<impl LayoutMaybeTrace, Error>;

View File

@ -362,7 +362,6 @@ def flow_get_address(
account: str | None,
path: str | None,
xpubs: list[tuple[str, str]],
title_success: str,
br_code: ButtonRequestType,
br_name: str,
) -> LayoutObj[UiResult]:

View File

@ -9,6 +9,7 @@ from .keychain import with_keychain
if TYPE_CHECKING:
from trezor.messages import Address, GetAddress, HDNodeType
from trezor.wire import MaybeEarlyResponse
from apps.common.coininfo import CoinInfo
from apps.common.keychain import Keychain
@ -35,14 +36,19 @@ def _get_xpubs(
@with_keychain
async def get_address(msg: GetAddress, keychain: Keychain, coin: CoinInfo) -> Address:
async def get_address(
msg: GetAddress, keychain: Keychain, coin: CoinInfo
) -> MaybeEarlyResponse[Address]:
from trezor import TR
from trezor.enums import InputScriptType
from trezor.messages import Address
from trezor.ui.layouts import (
confirm_multisig_different_paths_warning,
confirm_multisig_warning,
show_address,
show_continue_in_app,
)
from trezor.wire import early_response
from apps.common.address_mac import get_address_mac
from apps.common.paths import address_n_to_str, validate_path
@ -103,6 +109,8 @@ async def get_address(msg: GetAddress, keychain: Keychain, coin: CoinInfo) -> Ad
):
mac = get_address_mac(address, coin.slip44, keychain)
response = Address(address=address, mac=mac)
if msg.show_display:
path = address_n_to_str(address_n)
if multisig:
@ -156,4 +164,8 @@ async def get_address(msg: GetAddress, keychain: Keychain, coin: CoinInfo) -> Ad
chunkify=bool(msg.chunkify),
)
return Address(address=address, mac=mac)
return await early_response(
response, show_continue_in_app(TR.address__confirmed)
)
else:
return response

View File

@ -5,6 +5,7 @@ if TYPE_CHECKING:
from trezor.enums import InputScriptType
from trezor.messages import GetPublicKey, PublicKey
from trezor.protobuf import MessageType
from trezor.wire import MaybeEarlyResponse
from apps.common.keychain import Keychain
@ -13,10 +14,11 @@ async def get_public_key(
msg: GetPublicKey,
auth_msg: MessageType | None = None,
keychain: Keychain | None = None,
) -> PublicKey:
) -> MaybeEarlyResponse[PublicKey]:
from trezor import TR, wire
from trezor.enums import InputScriptType
from trezor.messages import HDNodeType, PublicKey, UnlockPath
from trezor.wire import early_response
from apps.common import coininfo, paths
from apps.common.keychain import FORBIDDEN_KEY_PATH, get_keychain
@ -86,9 +88,19 @@ async def get_public_key(
descriptor = _xpub_descriptor(
node, xpub_magic, address_n, script_type, keychain.root_fingerprint()
)
response = PublicKey(
node=node_type,
xpub=node_xpub,
root_fingerprint=keychain.root_fingerprint(),
descriptor=descriptor,
)
if msg.show_display:
from trezor.ui.layouts import confirm_path_warning, show_pubkey
from trezor.ui.layouts import (
confirm_path_warning,
show_continue_in_app,
show_pubkey,
)
from apps.common.paths import address_n_to_str
@ -116,13 +128,11 @@ async def get_public_key(
mismatch_title=TR.addr_mismatch__xpub_mismatch,
br_name="show_xpub",
)
return PublicKey(
node=node_type,
xpub=node_xpub,
root_fingerprint=keychain.root_fingerprint(),
descriptor=descriptor,
)
return await early_response(
response, show_continue_in_app(TR.address__public_key_confirmed)
)
else:
return response
def _xpub_descriptor(

View File

@ -21,6 +21,7 @@ if TYPE_CHECKING:
VerifyMessage,
)
from trezor.protobuf import MessageType
from trezor.wire import MaybeEarlyResponse
from typing_extensions import Protocol
from apps.common import coininfo
@ -43,7 +44,7 @@ if TYPE_CHECKING:
script_type: InputScriptType
MsgIn = TypeVar("MsgIn", bound=BitcoinMessage)
HandlerWithCoinInfo = Callable[..., Awaitable[MsgOut]]
HandlerWithCoinInfo = Callable[..., Awaitable[MaybeEarlyResponse[MsgOut]]]
# BIP-45 for multisig: https://github.com/bitcoin/bips/blob/master/bip-0045.mediawiki
PATTERN_BIP45 = "m/45'/[0-100]/change/address_index"
@ -316,7 +317,7 @@ def with_keychain(func: HandlerWithCoinInfo[MsgOut]) -> Handler[MsgIn, MsgOut]:
async def wrapper(
msg: BitcoinMessage,
auth_msg: MessageType | None = None,
) -> MsgOut:
) -> MaybeEarlyResponse[MsgOut]:
coin = _get_coin_by_name(msg.coin_name)
unlock_schemas = _get_unlock_schemas(msg, auth_msg, coin)
keychain = await _get_keychain_for_coin(coin, unlock_schemas)

View File

@ -4,14 +4,17 @@ from . import seed
if TYPE_CHECKING:
from trezor.messages import CardanoAddress, CardanoGetAddress
from trezor.wire import MaybeEarlyResponse
@seed.with_keychain
async def get_address(
msg: CardanoGetAddress, keychain: seed.Keychain
) -> CardanoAddress:
) -> MaybeEarlyResponse[CardanoAddress]:
from trezor import log, wire
from trezor.messages import CardanoAddress
from trezor.ui.layouts import show_continue_in_app
from trezor.wire import early_response
from . import addresses
from .helpers.credential import Credential, should_show_credentials
@ -32,7 +35,11 @@ async def get_address(
log.exception(__name__, e)
raise wire.ProcessError("Deriving address failed")
response = CardanoAddress(address=address)
if msg.show_display:
from trezor import TR
# _display_address
if should_show_credentials(address_parameters):
await show_credentials(
@ -42,5 +49,8 @@ async def get_address(
await show_cardano_address(
address_parameters, address, msg.protocol_magic, chunkify=bool(msg.chunkify)
)
return await early_response(
response, show_continue_in_app(TR.address__confirmed)
)
return CardanoAddress(address=address)
return response

View File

@ -7,12 +7,13 @@ from . import seed
if TYPE_CHECKING:
from trezor.messages import CardanoGetPublicKey, CardanoPublicKey
from trezor.wire import MaybeEarlyResponse
@seed.with_keychain
async def get_public_key(
msg: CardanoGetPublicKey, keychain: seed.Keychain
) -> CardanoPublicKey:
) -> MaybeEarlyResponse[CardanoPublicKey]:
from trezor import log, wire
from trezor.ui.layouts import show_pubkey
@ -37,10 +38,17 @@ async def get_public_key(
raise wire.ProcessError("Deriving public key failed")
if msg.show_display:
from trezor.ui.layouts import show_continue_in_app
from trezor.wire import early_response
from apps.common.paths import address_n_to_str
path = address_n_to_str(address_n)
await show_pubkey(key.xpub, TR.address__public_key, path=path)
return await early_response(
key, show_continue_in_app(TR.address__public_key_confirmed)
)
return key

View File

@ -21,6 +21,7 @@ if TYPE_CHECKING:
from trezor import messages
from trezor.crypto import bip32
from trezor.enums import CardanoDerivationType
from trezor.wire import MaybeEarlyResponse
from apps.common.keychain import Handler, MsgOut
from apps.common.paths import Bip32Path
@ -33,7 +34,9 @@ if TYPE_CHECKING:
)
MsgIn = TypeVar("MsgIn", bound=CardanoMessages)
HandlerWithKeychain = Callable[[MsgIn, "Keychain"], Awaitable[MsgOut]]
HandlerWithKeychain = Callable[
[MsgIn, "Keychain"], Awaitable[MaybeEarlyResponse[MsgOut]]
]
class Keychain:
@ -183,7 +186,7 @@ async def _get_keychain(derivation_type: CardanoDerivationType) -> Keychain:
def with_keychain(func: HandlerWithKeychain[MsgIn, MsgOut]) -> Handler[MsgIn, MsgOut]:
async def wrapper(msg: MsgIn) -> MsgOut:
async def wrapper(msg: MsgIn) -> MaybeEarlyResponse[MsgOut]:
keychain = await _get_keychain(msg.derivation_type)
return await func(msg, keychain)

View File

@ -9,6 +9,7 @@ if TYPE_CHECKING:
from typing import Any, Awaitable, Callable, Iterable, TypeVar
from trezor.protobuf import MessageType
from trezor.wire import MaybeEarlyResponse
from typing_extensions import Protocol
from .seed import Slip21Node
@ -27,8 +28,10 @@ if TYPE_CHECKING:
MsgIn = TypeVar("MsgIn", bound=MessageType)
MsgOut = TypeVar("MsgOut", bound=MessageType)
Handler = Callable[[MsgIn], Awaitable[MsgOut]]
HandlerWithKeychain = Callable[[MsgIn, "Keychain"], Awaitable[MsgOut]]
Handler = Callable[[MsgIn], Awaitable[MaybeEarlyResponse[MsgOut]]]
HandlerWithKeychain = Callable[
[MsgIn, "Keychain"], Awaitable[MaybeEarlyResponse[MsgOut]]
]
class Deletable(Protocol):
def __del__(self) -> None: ...
@ -194,7 +197,7 @@ def with_slip44_keychain(
schemas = [s.copy() for s in schemas]
def decorator(func: HandlerWithKeychain[MsgIn, MsgOut]) -> Handler[MsgIn, MsgOut]:
async def wrapper(msg: MsgIn) -> MsgOut:
async def wrapper(msg: MsgIn) -> MaybeEarlyResponse[MsgOut]:
keychain = await get_keychain(curve, schemas)
with keychain:
return await func(msg, keychain)

View File

@ -4,14 +4,19 @@ from apps.common.keychain import auto_keychain
if TYPE_CHECKING:
from trezor.messages import EosGetPublicKey, EosPublicKey
from trezor.wire import MaybeEarlyResponse
from apps.common.keychain import Keychain
@auto_keychain(__name__)
async def get_public_key(msg: EosGetPublicKey, keychain: Keychain) -> EosPublicKey:
async def get_public_key(
msg: EosGetPublicKey, keychain: Keychain
) -> MaybeEarlyResponse[EosPublicKey]:
from trezor.crypto.curve import secp256k1
from trezor.messages import EosPublicKey
from trezor.ui.layouts import show_continue_in_app
from trezor.wire import early_response
from apps.common import paths
@ -24,11 +29,18 @@ async def get_public_key(msg: EosGetPublicKey, keychain: Keychain) -> EosPublicK
public_key = secp256k1.publickey(node.private_key(), True)
wif = public_key_to_wif(public_key)
response = EosPublicKey(wif_public_key=wif, raw_public_key=public_key)
if msg.show_display:
from trezor import TR
from . import PATTERN, SLIP44_ID
path = paths.address_n_to_str(msg.address_n)
account = paths.get_account_name("EOS", msg.address_n, PATTERN, SLIP44_ID)
await require_get_public_key(wif, path, account)
return EosPublicKey(wif_public_key=wif, raw_public_key=public_key)
return await early_response(
response, show_continue_in_app(TR.address__public_key_confirmed)
)
return response

View File

@ -4,6 +4,7 @@ from .keychain import PATTERNS_ADDRESS, with_keychain_from_path
if TYPE_CHECKING:
from trezor.messages import EthereumAddress, EthereumGetAddress
from trezor.wire import MaybeEarlyResponse
from apps.common.keychain import Keychain
@ -15,9 +16,10 @@ async def get_address(
msg: EthereumGetAddress,
keychain: Keychain,
defs: Definitions,
) -> EthereumAddress:
) -> MaybeEarlyResponse[EthereumAddress]:
from trezor.messages import EthereumAddress
from trezor.ui.layouts import show_address
from trezor.ui.layouts import show_address, show_continue_in_app
from trezor.wire import early_response
from apps.common import paths
@ -30,8 +32,11 @@ async def get_address(
node = keychain.derive(address_n)
address = address_from_bytes(node.ethereum_pubkeyhash(), defs.network)
response = EthereumAddress(address=address)
if msg.show_display:
from trezor import TR
slip44_id = address_n[1] # it depends on the network (ETH vs ETC...)
await show_address(
address,
@ -41,5 +46,8 @@ async def get_address(
),
chunkify=bool(msg.chunkify),
)
return await early_response(
response, show_continue_in_app(TR.address__confirmed)
)
return EthereumAddress(address=address)
return response

View File

@ -2,21 +2,32 @@ from typing import TYPE_CHECKING
if TYPE_CHECKING:
from trezor.messages import EthereumGetPublicKey, EthereumPublicKey
from trezor.wire import MaybeEarlyResponse
async def get_public_key(msg: EthereumGetPublicKey) -> EthereumPublicKey:
async def get_public_key(
msg: EthereumGetPublicKey,
) -> MaybeEarlyResponse[EthereumPublicKey]:
from ubinascii import hexlify
from trezor.messages import EthereumPublicKey, GetPublicKey
from trezor.ui.layouts import show_pubkey
from trezor.messages import EthereumPublicKey, GetPublicKey, PublicKey
from trezor.ui.layouts import show_continue_in_app, show_pubkey
from trezor.wire import early_response
from apps.bitcoin import get_public_key as bitcoin_get_public_key
# we use the Bitcoin format for Ethereum xpubs
btc_pubkey_msg = GetPublicKey(address_n=msg.address_n)
resp = await bitcoin_get_public_key.get_public_key(btc_pubkey_msg)
btc_pubkey_msg = GetPublicKey(address_n=msg.address_n, show_display=False)
btc_resp = await bitcoin_get_public_key.get_public_key(btc_pubkey_msg)
assert PublicKey.is_type_of(btc_resp)
response = EthereumPublicKey(node=btc_resp.node, xpub=btc_resp.xpub)
if msg.show_display:
await show_pubkey(hexlify(resp.node.public_key).decode())
from trezor import TR
return EthereumPublicKey(node=resp.node, xpub=resp.xpub)
await show_pubkey(hexlify(response.node.public_key).decode())
return await early_response(
response, show_continue_in_app(TR.address__public_key_confirmed)
)
return response

View File

@ -17,6 +17,7 @@ if TYPE_CHECKING:
EthereumSignTxEIP1559,
EthereumSignTypedData,
)
from trezor.wire import MaybeEarlyResponse
from apps.common.keychain import Handler, Keychain, MsgOut
@ -30,7 +31,7 @@ if TYPE_CHECKING:
HandlerAddressN = Callable[
[MsgInAddressN, Keychain, definitions.Definitions],
Awaitable[MsgOut],
Awaitable[MaybeEarlyResponse[MsgOut]],
]
# messages for "with_keychain_and_defs_from_chain_id" decorator
@ -42,7 +43,7 @@ if TYPE_CHECKING:
HandlerChainId = Callable[
[MsgInSignTx, Keychain, definitions.Definitions],
Awaitable[MsgOut],
Awaitable[MaybeEarlyResponse[MsgOut]],
]
@ -120,7 +121,7 @@ def with_keychain_from_path(
def decorator(
func: HandlerAddressN[MsgInAddressN, MsgOut]
) -> Handler[MsgInAddressN, MsgOut]:
async def wrapper(msg: MsgInAddressN) -> MsgOut:
async def wrapper(msg: MsgInAddressN) -> MaybeEarlyResponse[MsgOut]:
slip44 = _slip44_from_address_n(msg.address_n)
defs = _defs_from_message(msg, slip44=slip44)
schemas = _schemas_from_network(patterns, defs.network)
@ -137,7 +138,7 @@ def with_keychain_from_chain_id(
func: HandlerChainId[MsgInSignTx, MsgOut]
) -> Handler[MsgInSignTx, MsgOut]:
# this is only for SignTx, and only PATTERN_ADDRESS is allowed
async def wrapper(msg: MsgInSignTx) -> MsgOut:
async def wrapper(msg: MsgInSignTx) -> MaybeEarlyResponse[MsgOut]:
defs = _defs_from_message(msg, chain_id=msg.chain_id)
schemas = _schemas_from_network(PATTERNS_ADDRESS, defs.network)
keychain = await get_keychain(CURVE, schemas)

View File

@ -100,7 +100,7 @@ async def show_dry_run_result(result: bool, is_slip39: bool) -> None:
text = TR.recovery__dry_run_slip39_valid_match
else:
text = TR.recovery__dry_run_bip39_valid_match
await show_success("success_dry_recovery", text, button=TR.buttons__continue)
await show_success("success_dry_recovery", text)
else:
if is_slip39:
text = TR.recovery__dry_run_slip39_valid_mismatch

View File

@ -144,7 +144,12 @@ async def reset_device(msg: ResetDevice) -> Success:
async def _entropy_check(secret: bytes) -> bool:
"""Returns True to indicate that entropy check loop should end."""
from trezor.messages import EntropyCheckContinue, EntropyCheckReady, GetPublicKey
from trezor.messages import (
EntropyCheckContinue,
EntropyCheckReady,
GetPublicKey,
PublicKey,
)
from trezor.wire.context import call_any
from apps.bitcoin.get_public_key import get_public_key
@ -171,6 +176,7 @@ async def _entropy_check(secret: bytes) -> bool:
curve_name = req.ecdsa_curve_name or coininfo.by_name(req.coin_name).curve_name
keychain = Keychain(seed, curve_name, [paths.AlwaysMatchingSchema])
msg = await get_public_key(req, keychain=keychain)
assert PublicKey.is_type_of(msg)
async def _backup_bip39(mnemonic: str) -> None:

View File

@ -4,15 +4,19 @@ from apps.common.keychain import auto_keychain
if TYPE_CHECKING:
from trezor.messages import MoneroAddress, MoneroGetAddress
from trezor.wire import MaybeEarlyResponse
from apps.common.keychain import Keychain
@auto_keychain(__name__)
async def get_address(msg: MoneroGetAddress, keychain: Keychain) -> MoneroAddress:
async def get_address(
msg: MoneroGetAddress, keychain: Keychain
) -> MaybeEarlyResponse[MoneroAddress]:
from trezor import wire
from trezor.messages import MoneroAddress
from trezor.ui.layouts import show_address
from trezor.ui.layouts import show_address, show_continue_in_app
from trezor.wire import early_response
from apps.common import paths
from apps.monero import misc
@ -65,7 +69,11 @@ async def get_address(msg: MoneroGetAddress, keychain: Keychain) -> MoneroAddres
crypto_helpers.encodepoint(pub_view),
)
response = MoneroAddress(address=addr.encode())
if msg.show_display:
from trezor import TR
from . import PATTERN, SLIP44_ID
await show_address(
@ -75,5 +83,8 @@ async def get_address(msg: MoneroGetAddress, keychain: Keychain) -> MoneroAddres
account=paths.get_account_name("XMR", msg.address_n, PATTERN, SLIP44_ID),
chunkify=bool(msg.chunkify),
)
return await early_response(
response, show_continue_in_app(TR.address__confirmed)
)
return MoneroAddress(address=addr.encode())
return response

View File

@ -6,14 +6,18 @@ from . import CURVE, PATTERNS, SLIP44_ID
if TYPE_CHECKING:
from trezor.messages import NEMAddress, NEMGetAddress
from trezor.wire import MaybeEarlyResponse
from apps.common.keychain import Keychain
@with_slip44_keychain(*PATTERNS, slip44_id=SLIP44_ID, curve=CURVE)
async def get_address(msg: NEMGetAddress, keychain: Keychain) -> NEMAddress:
async def get_address(
msg: NEMGetAddress, keychain: Keychain
) -> MaybeEarlyResponse[NEMAddress]:
from trezor.messages import NEMAddress
from trezor.ui.layouts import show_address
from trezor.ui.layouts import show_address, show_continue_in_app
from trezor.wire import early_response
from apps.common import paths
@ -28,8 +32,11 @@ async def get_address(msg: NEMGetAddress, keychain: Keychain) -> NEMAddress:
node = keychain.derive(address_n)
address = node.nem_address(network)
response = NEMAddress(address=address)
if msg.show_display:
from trezor import TR
from . import PATTERNS, SLIP44_ID
await show_address(
@ -40,5 +47,8 @@ async def get_address(msg: NEMGetAddress, keychain: Keychain) -> NEMAddress:
network=get_network_str(network),
chunkify=bool(msg.chunkify),
)
return await early_response(
response, show_continue_in_app(TR.address__confirmed)
)
return NEMAddress(address=address)
return response

View File

@ -4,15 +4,19 @@ from apps.common.keychain import auto_keychain
if TYPE_CHECKING:
from trezor.messages import RippleAddress, RippleGetAddress
from trezor.wire import MaybeEarlyResponse
from apps.common.keychain import Keychain
@auto_keychain(__name__)
async def get_address(msg: RippleGetAddress, keychain: Keychain) -> RippleAddress:
async def get_address(
msg: RippleGetAddress, keychain: Keychain
) -> MaybeEarlyResponse[RippleAddress]:
# NOTE: local imports here saves 20 bytes
from trezor.messages import RippleAddress
from trezor.ui.layouts import show_address
from trezor.ui.layouts import show_address, show_continue_in_app
from trezor.wire import early_response
from apps.common import paths
@ -25,8 +29,11 @@ async def get_address(msg: RippleGetAddress, keychain: Keychain) -> RippleAddres
node = keychain.derive(address_n)
pubkey = node.public_key()
address = address_from_public_key(pubkey)
response = RippleAddress(address=address)
if msg.show_display:
from trezor import TR
from . import PATTERN, SLIP44_ID
await show_address(
@ -35,5 +42,8 @@ async def get_address(msg: RippleGetAddress, keychain: Keychain) -> RippleAddres
account=paths.get_account_name("XRP", msg.address_n, PATTERN, SLIP44_ID),
chunkify=bool(msg.chunkify),
)
return await early_response(
response, show_continue_in_app(TR.address__confirmed)
)
return RippleAddress(address=address)
return response

View File

@ -8,6 +8,7 @@ from . import CURVE, PATTERNS, SLIP44_ID
if TYPE_CHECKING:
from trezor.messages import SolanaAddress, SolanaGetAddress
from trezor.wire import MaybeEarlyResponse
from apps.common.keychain import Keychain
@ -16,9 +17,10 @@ if TYPE_CHECKING:
async def get_address(
msg: SolanaGetAddress,
keychain: Keychain,
) -> SolanaAddress:
) -> MaybeEarlyResponse[SolanaAddress]:
from trezor.messages import SolanaAddress
from trezor.ui.layouts import show_address
from trezor.ui.layouts import show_address, show_continue_in_app
from trezor.wire import early_response
from apps.common import paths
@ -26,12 +28,18 @@ async def get_address(
public_key = derive_public_key(keychain, msg.address_n)
address = base58.encode(public_key)
response = SolanaAddress(address=address)
if msg.show_display:
from trezor import TR
await show_address(
address,
path=paths.address_n_to_str(msg.address_n),
chunkify=bool(msg.chunkify),
)
return await early_response(
response, show_continue_in_app(TR.address__confirmed)
)
return SolanaAddress(address=address)
return response

View File

@ -9,6 +9,7 @@ from . import CURVE, PATTERNS, SLIP44_ID
if TYPE_CHECKING:
from trezor.messages import SolanaGetPublicKey, SolanaPublicKey
from trezor.wire import MaybeEarlyResponse
from apps.common.keychain import Keychain
@ -16,19 +17,26 @@ if TYPE_CHECKING:
@with_slip44_keychain(*PATTERNS, slip44_id=SLIP44_ID, curve=CURVE)
async def get_public_key(
msg: SolanaGetPublicKey, keychain: Keychain
) -> SolanaPublicKey:
) -> MaybeEarlyResponse[SolanaPublicKey]:
from trezor.messages import SolanaPublicKey
from trezor.ui.layouts import show_pubkey
from trezor.ui.layouts import show_continue_in_app, show_pubkey
from trezor.wire import early_response
public_key = derive_public_key(keychain, msg.address_n)
response = SolanaPublicKey(public_key=public_key)
if msg.show_display:
from trezor import TR
from apps.common.paths import address_n_to_str
path = address_n_to_str(msg.address_n)
await show_pubkey(base58.encode(public_key), path=path)
return await early_response(
response, show_continue_in_app(TR.address__public_key_confirmed)
)
return SolanaPublicKey(public_key=public_key)
return response
def derive_public_key(keychain: Keychain, address_n: list[int]) -> bytes:

View File

@ -4,14 +4,18 @@ from apps.common.keychain import auto_keychain
if TYPE_CHECKING:
from trezor.messages import StellarAddress, StellarGetAddress
from trezor.wire import MaybeEarlyResponse
from apps.common.keychain import Keychain
@auto_keychain(__name__)
async def get_address(msg: StellarGetAddress, keychain: Keychain) -> StellarAddress:
async def get_address(
msg: StellarGetAddress, keychain: Keychain
) -> MaybeEarlyResponse[StellarAddress]:
from trezor.messages import StellarAddress
from trezor.ui.layouts import show_address
from trezor.ui.layouts import show_address, show_continue_in_app
from trezor.wire import early_response
from apps.common import paths, seed
@ -24,8 +28,11 @@ async def get_address(msg: StellarGetAddress, keychain: Keychain) -> StellarAddr
node = keychain.derive(address_n)
pubkey = seed.remove_ed25519_prefix(node.public_key())
address = helpers.address_from_public_key(pubkey)
response = StellarAddress(address=address)
if msg.show_display:
from trezor import TR
from . import PATTERN, SLIP44_ID
await show_address(
@ -35,5 +42,8 @@ async def get_address(msg: StellarGetAddress, keychain: Keychain) -> StellarAddr
account=paths.get_account_name("XLM", msg.address_n, PATTERN, SLIP44_ID),
chunkify=bool(msg.chunkify),
)
return await early_response(
response, show_continue_in_app(TR.address__confirmed)
)
return StellarAddress(address=address)
return response

View File

@ -6,15 +6,19 @@ from . import CURVE, PATTERNS, SLIP44_ID
if TYPE_CHECKING:
from trezor.messages import TezosAddress, TezosGetAddress
from trezor.wire import MaybeEarlyResponse
from apps.common.keychain import Keychain
@with_slip44_keychain(*PATTERNS, slip44_id=SLIP44_ID, curve=CURVE)
async def get_address(msg: TezosGetAddress, keychain: Keychain) -> TezosAddress:
async def get_address(
msg: TezosGetAddress, keychain: Keychain
) -> MaybeEarlyResponse[TezosAddress]:
from trezor.crypto import hashlib
from trezor.messages import TezosAddress
from trezor.ui.layouts import show_address
from trezor.ui.layouts import show_address, show_continue_in_app
from trezor.wire import early_response
from apps.common import paths, seed
@ -29,8 +33,11 @@ async def get_address(msg: TezosGetAddress, keychain: Keychain) -> TezosAddress:
pk = seed.remove_ed25519_prefix(node.public_key())
pkh = hashlib.blake2b(pk, outlen=helpers.PUBLIC_KEY_HASH_SIZE).digest()
address = helpers.base58_encode_check(pkh, helpers.TEZOS_ED25519_ADDRESS_PREFIX)
response = TezosAddress(address=address)
if msg.show_display:
from trezor import TR
from . import PATTERNS, SLIP44_ID
await show_address(
@ -39,5 +46,8 @@ async def get_address(msg: TezosGetAddress, keychain: Keychain) -> TezosAddress:
account=paths.get_account_name("XTZ", address_n, PATTERNS, SLIP44_ID),
chunkify=bool(msg.chunkify),
)
return await early_response(
response, show_continue_in_app(TR.address__confirmed)
)
return TezosAddress(address=address)
return response

View File

@ -6,14 +6,18 @@ from . import CURVE, PATTERNS, SLIP44_ID
if TYPE_CHECKING:
from trezor.messages import TezosGetPublicKey, TezosPublicKey
from trezor.wire import MaybeEarlyResponse
from apps.common.keychain import Keychain
@with_slip44_keychain(*PATTERNS, slip44_id=SLIP44_ID, curve=CURVE)
async def get_public_key(msg: TezosGetPublicKey, keychain: Keychain) -> TezosPublicKey:
async def get_public_key(
msg: TezosGetPublicKey, keychain: Keychain
) -> MaybeEarlyResponse[TezosPublicKey]:
from trezor.messages import TezosPublicKey
from trezor.ui.layouts import show_pubkey
from trezor.ui.layouts import show_continue_in_app, show_pubkey
from trezor.wire import early_response
from apps.common import paths, seed
@ -24,12 +28,18 @@ async def get_public_key(msg: TezosGetPublicKey, keychain: Keychain) -> TezosPub
node = keychain.derive(msg.address_n)
pk = seed.remove_ed25519_prefix(node.public_key())
pk_prefixed = helpers.base58_encode_check(pk, helpers.TEZOS_PUBLICKEY_PREFIX)
response = TezosPublicKey(public_key=pk_prefixed)
if msg.show_display:
from trezor import TR
from . import PATTERNS, SLIP44_ID
account = paths.get_account_name("XTZ", msg.address_n, PATTERNS, SLIP44_ID)
path = paths.address_n_to_str(msg.address_n)
await show_pubkey(pk_prefixed, account=account, path=path)
return await early_response(
response, show_continue_in_app(TR.address__public_key_confirmed)
)
return TezosPublicKey(public_key=pk_prefixed)
return response

View File

@ -401,6 +401,10 @@ def show_success(
)
async def show_continue_in_app(content: str) -> None:
return
async def confirm_output(
address: str,
amount: str,

View File

@ -464,6 +464,10 @@ def show_success(
)
async def show_continue_in_app(content: str) -> None:
return
async def confirm_output(
address: str,
amount: str,

View File

@ -235,12 +235,6 @@ async def show_address(
)
return result
title_success = (
TR.address__public_key_confirmed
if title in ("XPUB", TR.address__public_key)
else TR.address__confirmed
)
await raise_if_not_confirmed(
trezorui_api.flow_get_address(
address=address,
@ -253,7 +247,6 @@ async def show_address(
account=account,
path=path,
xpubs=[(xpub_title(i), xpub) for i, xpub in enumerate(xpubs)],
title_success=title_success,
br_name=br_name,
br_code=br_code,
),
@ -349,22 +342,33 @@ def show_danger(
def show_success(
br_name: str,
br_name: str | None,
content: str,
subheader: str | None = None,
button: str | None = None,
time_ms: int = 0,
) -> Awaitable[None]:
return raise_if_not_confirmed(
trezorui_api.show_success(
title=content,
button="",
description=subheader if subheader else "",
button=button or "",
description=subheader or "",
time_ms=time_ms,
),
br_name,
ButtonRequestType.Success,
)
def show_continue_in_app(content: str) -> Awaitable[None]:
return show_success(
content=content,
button=TR.instructions__continue_in_app,
time_ms=3200,
br_name=None,
)
async def confirm_output(
address: str,
amount: str | None = None,

View File

@ -41,7 +41,7 @@ if __debug__:
if TYPE_CHECKING:
from trezorio import WireInterface
from typing import Any, Callable, Coroutine, TypeVar
from typing import Any, Awaitable, Callable, Coroutine, Generic, TypeAlias, TypeVar
Msg = TypeVar("Msg", bound=protobuf.MessageType)
HandlerTask = Coroutine[Any, Any, protobuf.MessageType]
@ -49,6 +49,28 @@ if TYPE_CHECKING:
LoadedMessageType = TypeVar("LoadedMessageType", bound=protobuf.MessageType)
class EarlyResponse(Generic[Msg]):
"""Marker type (when the response is sent before the last layout is shown)."""
MaybeEarlyResponse: TypeAlias = Msg | EarlyResponse[Msg]
else:
EarlyResponse = object
_EARLY_RESPONSE = EarlyResponse()
async def early_response(response: Msg, layout: Awaitable[None]) -> EarlyResponse[Msg]:
from .context import get_context
# first, send the response back to the client
await get_context().write(response)
# then, show the success layout
await layout
# return a special marker object
return _EARLY_RESPONSE
class BufferProvider:
def __init__(self, size: int) -> None:

View File

@ -63,6 +63,8 @@ async def handle_single_message(ctx: Context, msg: Message) -> bool:
the type of message is supposed to be optimized and not disrupt the running state,
this function will return `True`.
"""
from . import _EARLY_RESPONSE
if __debug__:
try:
msg_type = protobuf.type_for_wire(msg.type).MESSAGE_NAME
@ -75,8 +77,6 @@ async def handle_single_message(ctx: Context, msg: Message) -> bool:
msg_type,
)
res_msg: protobuf.MessageType | None = None
# We need to find a handler for this message type.
try:
handler: Handler = find_handler(ctx.iface, msg.type)
@ -159,7 +159,7 @@ async def handle_single_message(ctx: Context, msg: Message) -> bool:
wire_log.exception(__name__, ctx.iface, exc)
res_msg = failure(exc)
if res_msg is not None:
if res_msg is not _EARLY_RESPONSE:
# perform the write outside the big try-except block, so that usb write
# problem bubbles up
await ctx.write(res_msg)

File diff suppressed because it is too large Load Diff