From 1ae6b68eaf0e9ac1e8e0ec57122eea3b3bf65657 Mon Sep 17 00:00:00 2001 From: David Misiak Date: Wed, 18 May 2022 17:05:46 +0200 Subject: [PATCH] refactor(cardano): use module imports for messages, layout --- core/src/apps/cardano/address.py | 41 ++-- core/src/apps/cardano/auxiliary_data.py | 27 +-- core/src/apps/cardano/certificates.py | 35 ++- core/src/apps/cardano/get_address.py | 19 +- .../apps/cardano/get_native_script_hash.py | 21 +- core/src/apps/cardano/get_public_key.py | 17 +- core/src/apps/cardano/helpers/credential.py | 15 +- core/src/apps/cardano/layout.py | 45 ++-- core/src/apps/cardano/native_script.py | 14 +- core/src/apps/cardano/seed.py | 15 +- core/src/apps/cardano/sign_tx/__init__.py | 5 +- .../apps/cardano/sign_tx/multisig_signer.py | 25 +- .../apps/cardano/sign_tx/ordinary_signer.py | 22 +- .../src/apps/cardano/sign_tx/plutus_signer.py | 49 ++-- .../apps/cardano/sign_tx/pool_owner_signer.py | 26 +- core/src/apps/cardano/sign_tx/signer.py | 222 ++++++++---------- 16 files changed, 249 insertions(+), 349 deletions(-) diff --git a/core/src/apps/cardano/address.py b/core/src/apps/cardano/address.py index e86489991..401fc1647 100644 --- a/core/src/apps/cardano/address.py +++ b/core/src/apps/cardano/address.py @@ -1,6 +1,6 @@ -from typing import TYPE_CHECKING +from typing import Any -from trezor import wire +from trezor import messages, wire from trezor.crypto import base58 from trezor.enums import CardanoAddressType @@ -10,15 +10,6 @@ from .helpers.paths import SCHEMA_STAKING_ANY_ACCOUNT from .helpers.utils import get_public_key_hash, variable_length_encode from .seed import is_byron_path, is_shelley_path -if TYPE_CHECKING: - from typing import Any - - from trezor.messages import ( - CardanoAddressParametersType, - CardanoBlockchainPointerType, - ) - from . import seed - ADDRESS_TYPES_SHELLEY = ( CardanoAddressType.BASE, CardanoAddressType.BASE_SCRIPT_KEY, @@ -48,7 +39,9 @@ def assert_address_params_cond(condition: bool) -> None: raise wire.ProcessError("Invalid address parameters") -def validate_address_parameters(parameters: CardanoAddressParametersType) -> None: +def validate_address_parameters( + parameters: messages.CardanoAddressParametersType, +) -> None: _validate_address_parameters_structure(parameters) if parameters.address_type == CardanoAddressType.BYRON: @@ -102,7 +95,7 @@ def validate_address_parameters(parameters: CardanoAddressParametersType) -> Non def _validate_address_parameters_structure( - parameters: CardanoAddressParametersType, + parameters: messages.CardanoAddressParametersType, ) -> None: address_n = parameters.address_n address_n_staking = parameters.address_n_staking @@ -206,7 +199,7 @@ def _validate_script_hash(script_hash: bytes | None) -> None: def validate_output_address_parameters( - parameters: CardanoAddressParametersType, + parameters: messages.CardanoAddressParametersType, ) -> None: validate_address_parameters(parameters) @@ -338,7 +331,7 @@ def _get_address_network_id(address: bytes) -> int: def derive_human_readable_address( keychain: seed.Keychain, - parameters: CardanoAddressParametersType, + parameters: messages.CardanoAddressParametersType, protocol_magic: int, network_id: int, ) -> str: @@ -364,7 +357,7 @@ def encode_human_readable_address(address_bytes: bytes) -> str: def derive_address_bytes( keychain: seed.Keychain, - parameters: CardanoAddressParametersType, + parameters: messages.CardanoAddressParametersType, protocol_magic: int, network_id: int, ) -> bytes: @@ -379,7 +372,9 @@ def derive_address_bytes( def _derive_shelley_address( - keychain: seed.Keychain, parameters: CardanoAddressParametersType, network_id: int + keychain: seed.Keychain, + parameters: messages.CardanoAddressParametersType, + network_id: int, ) -> bytes: header = _create_address_header(parameters.address_type, network_id) @@ -394,8 +389,8 @@ def _create_address_header(address_type: CardanoAddressType, network_id: int) -> return header.to_bytes(1, "little") -def _get_address_payment_part( - keychain: seed.Keychain, parameters: CardanoAddressParametersType +def _get_payment_part( + keychain: seed.Keychain, parameters: messages.CardanoAddressParametersType ) -> bytes: if parameters.address_n: return get_public_key_hash(keychain, parameters.address_n) @@ -405,8 +400,8 @@ def _get_address_payment_part( return bytes() -def _get_address_staking_part( - keychain: seed.Keychain, parameters: CardanoAddressParametersType +def _get_staking_part( + keychain: seed.Keychain, parameters: messages.CardanoAddressParametersType ) -> bytes: if parameters.staking_key_hash: return parameters.staking_key_hash @@ -420,7 +415,9 @@ def _get_address_staking_part( return bytes() -def _encode_certificate_pointer(pointer: CardanoBlockchainPointerType) -> bytes: +def _encode_certificate_pointer( + pointer: messages.CardanoBlockchainPointerType, +) -> bytes: block_index_encoded = variable_length_encode(pointer.block_index) tx_index_encoded = variable_length_encode(pointer.tx_index) certificate_index_encoded = variable_length_encode(pointer.certificate_index) diff --git a/core/src/apps/cardano/auxiliary_data.py b/core/src/apps/cardano/auxiliary_data.py index e6e729b71..ba08f177a 100644 --- a/core/src/apps/cardano/auxiliary_data.py +++ b/core/src/apps/cardano/auxiliary_data.py @@ -1,10 +1,9 @@ from typing import TYPE_CHECKING -from trezor import wire +from trezor import messages, wire from trezor.crypto import hashlib from trezor.crypto.curve import ed25519 from trezor.enums import CardanoAddressType, CardanoTxAuxiliaryDataSupplementType -from trezor.messages import CardanoTxAuxiliaryDataSupplement from apps.common import cbor @@ -20,11 +19,6 @@ from .helpers.utils import derive_public_key from .layout import confirm_catalyst_registration, show_auxiliary_data_hash if TYPE_CHECKING: - from trezor.messages import ( - CardanoCatalystRegistrationParametersType, - CardanoTxAuxiliaryData, - ) - CatalystRegistrationPayload = dict[int, bytes | int] SignedCatalystRegistrationPayload = tuple[CatalystRegistrationPayload, bytes] CatalystRegistrationSignature = dict[int, bytes] @@ -42,7 +36,7 @@ METADATA_KEY_CATALYST_REGISTRATION = 61284 METADATA_KEY_CATALYST_REGISTRATION_SIGNATURE = 61285 -def validate_auxiliary_data(auxiliary_data: CardanoTxAuxiliaryData) -> None: +def validate(auxiliary_data: messages.CardanoTxAuxiliaryData) -> None: fields_provided = 0 if auxiliary_data.hash: fields_provided += 1 @@ -63,7 +57,7 @@ def _validate_auxiliary_data_hash(auxiliary_data_hash: bytes) -> None: def _validate_catalyst_registration_parameters( - catalyst_registration_parameters: CardanoCatalystRegistrationParametersType, + catalyst_registration_parameters: messages.CardanoCatalystRegistrationParametersType, ) -> None: if ( len(catalyst_registration_parameters.voting_public_key) @@ -87,7 +81,8 @@ async def show_auxiliary_data( ctx: wire.Context, keychain: seed.Keychain, auxiliary_data_hash: bytes, - catalyst_registration_parameters: CardanoCatalystRegistrationParametersType | None, + catalyst_registration_parameters: messages.CardanoCatalystRegistrationParametersType + | None, protocol_magic: int, network_id: int, ) -> None: @@ -106,7 +101,7 @@ async def show_auxiliary_data( async def _show_catalyst_registration( ctx: wire.Context, keychain: seed.Keychain, - catalyst_registration_parameters: CardanoCatalystRegistrationParametersType, + catalyst_registration_parameters: messages.CardanoCatalystRegistrationParametersType, protocol_magic: int, network_id: int, ) -> None: @@ -128,10 +123,10 @@ async def _show_catalyst_registration( def get_auxiliary_data_hash_and_supplement( keychain: seed.Keychain, - auxiliary_data: CardanoTxAuxiliaryData, + auxiliary_data: messages.CardanoTxAuxiliaryData, protocol_magic: int, network_id: int, -) -> tuple[bytes, CardanoTxAuxiliaryDataSupplement]: +) -> tuple[bytes, messages.CardanoTxAuxiliaryDataSupplement]: if parameters := auxiliary_data.catalyst_registration_parameters: ( catalyst_registration_payload, @@ -142,7 +137,7 @@ def get_auxiliary_data_hash_and_supplement( auxiliary_data_hash = _get_catalyst_registration_auxiliary_data_hash( catalyst_registration_payload, catalyst_signature ) - auxiliary_data_supplement = CardanoTxAuxiliaryDataSupplement( + auxiliary_data_supplement = messages.CardanoTxAuxiliaryDataSupplement( type=CardanoTxAuxiliaryDataSupplementType.CATALYST_REGISTRATION_SIGNATURE, auxiliary_data_hash=auxiliary_data_hash, catalyst_signature=catalyst_signature, @@ -150,7 +145,7 @@ def get_auxiliary_data_hash_and_supplement( return auxiliary_data_hash, auxiliary_data_supplement else: assert auxiliary_data.hash is not None # validate_auxiliary_data - return auxiliary_data.hash, CardanoTxAuxiliaryDataSupplement( + return auxiliary_data.hash, messages.CardanoTxAuxiliaryDataSupplement( type=CardanoTxAuxiliaryDataSupplementType.NONE ) @@ -182,7 +177,7 @@ def _cborize_catalyst_registration( def _get_signed_catalyst_registration_payload( keychain: seed.Keychain, - catalyst_registration_parameters: CardanoCatalystRegistrationParametersType, + catalyst_registration_parameters: messages.CardanoCatalystRegistrationParametersType, protocol_magic: int, network_id: int, ) -> SignedCatalystRegistrationPayload: diff --git a/core/src/apps/cardano/certificates.py b/core/src/apps/cardano/certificates.py index 746680907..6c572b81c 100644 --- a/core/src/apps/cardano/certificates.py +++ b/core/src/apps/cardano/certificates.py @@ -13,14 +13,7 @@ from .helpers.utils import get_public_key_hash, validate_stake_credential if TYPE_CHECKING: from typing import Any - from trezor.messages import ( - CardanoPoolMetadataType, - CardanoPoolOwner, - CardanoPoolParametersType, - CardanoPoolRelayParameters, - CardanoTxCertificate, - ) - + from trezor import messages from apps.common.cbor import CborSequence from . import seed @@ -37,7 +30,7 @@ MAX_PORT_NUMBER = 65535 def validate_certificate( - certificate: CardanoTxCertificate, + certificate: messages.CardanoTxCertificate, protocol_magic: int, network_id: int, account_path_checker: AccountPathChecker, @@ -70,7 +63,7 @@ def validate_certificate( account_path_checker.add_certificate(certificate) -def _validate_certificate_structure(certificate: CardanoTxCertificate) -> None: +def _validate_certificate_structure(certificate: messages.CardanoTxCertificate) -> None: pool = certificate.pool pool_parameters = certificate.pool_parameters @@ -93,7 +86,7 @@ def _validate_certificate_structure(certificate: CardanoTxCertificate) -> None: def cborize_certificate( - keychain: seed.Keychain, certificate: CardanoTxCertificate + keychain: seed.Keychain, certificate: messages.CardanoTxCertificate ) -> CborSequence: if certificate.type in ( CardanoCertificateType.STAKE_REGISTRATION, @@ -140,7 +133,7 @@ def cborize_certificate_stake_credential( def cborize_initial_pool_registration_certificate_fields( - certificate: CardanoTxCertificate, + certificate: messages.CardanoTxCertificate, ) -> CborSequence: assert certificate.type == CardanoCertificateType.STAKE_POOL_REGISTRATION @@ -172,7 +165,9 @@ def assert_certificate_cond(condition: bool) -> None: def _validate_pool_parameters( - pool_parameters: CardanoPoolParametersType, protocol_magic: int, network_id: int + pool_parameters: messages.CardanoPoolParametersType, + protocol_magic: int, + network_id: int, ) -> None: assert_certificate_cond(len(pool_parameters.pool_id) == POOL_HASH_SIZE) assert_certificate_cond(len(pool_parameters.vrf_key_hash) == VRF_KEY_HASH_SIZE) @@ -192,7 +187,7 @@ def _validate_pool_parameters( def validate_pool_owner( - owner: CardanoPoolOwner, account_path_checker: AccountPathChecker + owner: messages.CardanoPoolOwner, account_path_checker: AccountPathChecker ) -> None: assert_certificate_cond( owner.staking_key_hash is not None or owner.staking_key_path is not None @@ -207,7 +202,7 @@ def validate_pool_owner( account_path_checker.add_pool_owner(owner) -def validate_pool_relay(pool_relay: CardanoPoolRelayParameters) -> None: +def validate_pool_relay(pool_relay: messages.CardanoPoolRelayParameters) -> None: if pool_relay.type == CardanoPoolRelayType.SINGLE_HOST_IP: assert_certificate_cond( pool_relay.ipv4_address is not None or pool_relay.ipv6_address is not None @@ -236,13 +231,15 @@ def validate_pool_relay(pool_relay: CardanoPoolRelayParameters) -> None: raise RuntimeError # should be unreachable -def _validate_pool_metadata(pool_metadata: CardanoPoolMetadataType) -> None: +def _validate_pool_metadata(pool_metadata: messages.CardanoPoolMetadataType) -> None: assert_certificate_cond(len(pool_metadata.url) <= MAX_URL_LENGTH) assert_certificate_cond(len(pool_metadata.hash) == POOL_METADATA_HASH_SIZE) assert_certificate_cond(all((32 <= ord(c) < 127) for c in pool_metadata.url)) -def cborize_pool_owner(keychain: seed.Keychain, pool_owner: CardanoPoolOwner) -> bytes: +def cborize_pool_owner( + keychain: seed.Keychain, pool_owner: messages.CardanoPoolOwner +) -> bytes: if pool_owner.staking_key_path: return get_public_key_hash(keychain, pool_owner.staking_key_path) elif pool_owner.staking_key_hash: @@ -266,7 +263,7 @@ def _cborize_ipv6_address(ipv6_address: bytes | None) -> bytes | None: def cborize_pool_relay( - pool_relay: CardanoPoolRelayParameters, + pool_relay: messages.CardanoPoolRelayParameters, ) -> CborSequence: if pool_relay.type == CardanoPoolRelayType.SINGLE_HOST_IP: return ( @@ -291,7 +288,7 @@ def cborize_pool_relay( def cborize_pool_metadata( - pool_metadata: CardanoPoolMetadataType | None, + pool_metadata: messages.CardanoPoolMetadataType | None, ) -> CborSequence | None: if not pool_metadata: return None diff --git a/core/src/apps/cardano/get_address.py b/core/src/apps/cardano/get_address.py index 184a4e94b..6c1d4c732 100644 --- a/core/src/apps/cardano/get_address.py +++ b/core/src/apps/cardano/get_address.py @@ -1,7 +1,4 @@ -from typing import TYPE_CHECKING - -from trezor import log, wire -from trezor.messages import CardanoAddress +from trezor import log, messages, wire from . import seed from .address import derive_human_readable_address, validate_address_parameters @@ -9,17 +6,11 @@ from .helpers.credential import Credential, should_show_address_credentials from .helpers.utils import validate_network_info from .layout import show_address_credentials, show_cardano_address -if TYPE_CHECKING: - from trezor.messages import ( - CardanoAddressParametersType, - CardanoGetAddress, - ) - @seed.with_keychain async def get_address( - ctx: wire.Context, msg: CardanoGetAddress, keychain: seed.Keychain -) -> CardanoAddress: + ctx: wire.Context, msg: messages.CardanoGetAddress, keychain: seed.Keychain +) -> messages.CardanoAddress: address_parameters = msg.address_parameters validate_network_info(msg.network_id, msg.protocol_magic) @@ -37,12 +28,12 @@ async def get_address( if msg.show_display: await _display_address(ctx, address_parameters, address, msg.protocol_magic) - return CardanoAddress(address=address) + return messages.CardanoAddress(address=address) async def _display_address( ctx: wire.Context, - address_parameters: CardanoAddressParametersType, + address_parameters: messages.CardanoAddressParametersType, address: str, protocol_magic: int, ) -> None: diff --git a/core/src/apps/cardano/get_native_script_hash.py b/core/src/apps/cardano/get_native_script_hash.py index 7ebb5298a..e5f413067 100644 --- a/core/src/apps/cardano/get_native_script_hash.py +++ b/core/src/apps/cardano/get_native_script_hash.py @@ -1,26 +1,19 @@ -from typing import TYPE_CHECKING - -from trezor import wire +from trezor import messages, wire from trezor.enums import CardanoNativeScriptHashDisplayFormat -from trezor.messages import CardanoNativeScriptHash - -from . import native_script, seed -from .layout import show_native_script, show_script_hash -if TYPE_CHECKING: - from trezor.messages import CardanoGetNativeScriptHash +from . import layout, native_script, seed @seed.with_keychain async def get_native_script_hash( - ctx: wire.Context, msg: CardanoGetNativeScriptHash, keychain: seed.Keychain -) -> CardanoNativeScriptHash: + ctx: wire.Context, msg: messages.CardanoGetNativeScriptHash, keychain: seed.Keychain +) -> messages.CardanoNativeScriptHash: native_script.validate_native_script(msg.script) script_hash = native_script.get_native_script_hash(keychain, msg.script) if msg.display_format != CardanoNativeScriptHashDisplayFormat.HIDE: - await show_native_script(ctx, msg.script) - await show_script_hash(ctx, script_hash, msg.display_format) + await layout.show_native_script(ctx, msg.script) + await layout.show_script_hash(ctx, script_hash, msg.display_format) - return CardanoNativeScriptHash(script_hash=script_hash) + return messages.CardanoNativeScriptHash(script_hash=script_hash) diff --git a/core/src/apps/cardano/get_public_key.py b/core/src/apps/cardano/get_public_key.py index 6f8983b52..547c83523 100644 --- a/core/src/apps/cardano/get_public_key.py +++ b/core/src/apps/cardano/get_public_key.py @@ -1,8 +1,6 @@ -from typing import TYPE_CHECKING from ubinascii import hexlify -from trezor import log, wire -from trezor.messages import CardanoPublicKey, HDNodeType +from trezor import log, messages, wire from trezor.ui.layouts import show_pubkey from apps.common import paths @@ -11,14 +9,11 @@ from . import seed from .helpers.paths import SCHEMA_MINT, SCHEMA_PUBKEY from .helpers.utils import derive_public_key -if TYPE_CHECKING: - from trezor.messages import CardanoGetPublicKey - @seed.with_keychain async def get_public_key( - ctx: wire.Context, msg: CardanoGetPublicKey, keychain: seed.Keychain -) -> CardanoPublicKey: + ctx: wire.Context, msg: messages.CardanoGetPublicKey, keychain: seed.Keychain +) -> messages.CardanoPublicKey: await paths.validate_path( ctx, keychain, @@ -41,14 +36,14 @@ async def get_public_key( def _get_public_key( keychain: seed.Keychain, derivation_path: list[int] -) -> CardanoPublicKey: +) -> messages.CardanoPublicKey: node = keychain.derive(derivation_path) public_key = hexlify(derive_public_key(keychain, derivation_path)).decode() chain_code = hexlify(node.chain_code()).decode() xpub_key = public_key + chain_code - node_type = HDNodeType( + node_type = messages.HDNodeType( depth=node.depth(), child_num=node.child_num(), fingerprint=node.fingerprint(), @@ -56,4 +51,4 @@ def _get_public_key( public_key=derive_public_key(keychain, derivation_path), ) - return CardanoPublicKey(node=node_type, xpub=xpub_key) + return messages.CardanoPublicKey(node=node_type, xpub=xpub_key) diff --git a/core/src/apps/cardano/helpers/credential.py b/core/src/apps/cardano/helpers/credential.py index 47c5220fb..7557c7b50 100644 --- a/core/src/apps/cardano/helpers/credential.py +++ b/core/src/apps/cardano/helpers/credential.py @@ -7,10 +7,7 @@ from .paths import CHAIN_STAKING_KEY, SCHEMA_PAYMENT, SCHEMA_STAKING from .utils import bech32, to_account_path if TYPE_CHECKING: - from trezor.messages import ( - CardanoBlockchainPointerType, - CardanoAddressParametersType, - ) + from trezor import messages from trezor.ui.layouts import PropertyType CREDENTIAL_TYPE_PAYMENT: str = "payment" @@ -30,7 +27,7 @@ class Credential: path: list[int] key_hash: bytes | None script_hash: bytes | None - pointer: CardanoBlockchainPointerType | None + pointer: messages.CardanoBlockchainPointerType | None is_reward: bool = False is_no_staking: bool = False @@ -45,7 +42,7 @@ class Credential: path: list[int], key_hash: bytes | None, script_hash: bytes | None, - pointer: CardanoBlockchainPointerType | None, + pointer: messages.CardanoBlockchainPointerType | None, ): self.type_name = type_name self.address_type = address_type @@ -56,7 +53,7 @@ class Credential: @classmethod def payment_credential( - cls, address_params: CardanoAddressParametersType + cls, address_params: messages.CardanoAddressParametersType ) -> "Credential": address_type = address_params.address_type credential = cls( @@ -99,7 +96,7 @@ class Credential: @classmethod def stake_credential( - cls, address_params: CardanoAddressParametersType + cls, address_params: messages.CardanoAddressParametersType ) -> "Credential": address_type = address_params.address_type credential = cls( @@ -207,7 +204,7 @@ class Credential: def should_show_address_credentials( - address_parameters: CardanoAddressParametersType, + address_parameters: messages.CardanoAddressParametersType, ) -> bool: return not ( address_parameters.address_type == CardanoAddressType.BASE diff --git a/core/src/apps/cardano/layout.py b/core/src/apps/cardano/layout.py index b168c2ef2..c6f0c0a8e 100644 --- a/core/src/apps/cardano/layout.py +++ b/core/src/apps/cardano/layout.py @@ -1,6 +1,6 @@ from typing import TYPE_CHECKING -from trezor import ui +from trezor import messages, ui from trezor.enums import ( ButtonRequestType, CardanoAddressType, @@ -8,7 +8,6 @@ from trezor.enums import ( CardanoNativeScriptHashDisplayFormat, CardanoNativeScriptType, ) -from trezor.messages import CardanoAddressParametersType from trezor.strings import format_amount from trezor.ui.layouts import ( confirm_blob, @@ -36,18 +35,6 @@ from .seed import is_minting_path, is_multisig_path if TYPE_CHECKING: from trezor import wire - from trezor.messages import ( - CardanoNativeScript, - CardanoTxCertificate, - CardanoTxCollateralInput, - CardanoTxInput, - CardanoTxRequiredSigner, - CardanoTxWithdrawal, - CardanoPoolParametersType, - CardanoPoolOwner, - CardanoPoolMetadataType, - CardanoToken, - ) from trezor.ui.layouts import PropertyType from .helpers.credential import Credential @@ -95,7 +82,7 @@ def is_printable_ascii_bytestring(bytestr: bytes) -> bool: async def show_native_script( ctx: wire.Context, - script: CardanoNativeScript, + script: messages.CardanoNativeScript, indices: list[int] | None = None, ) -> None: script_heading = "Script" @@ -213,7 +200,7 @@ async def show_plutus_transaction(ctx: wire.Context) -> None: ) -async def confirm_input(ctx: wire.Context, input: CardanoTxInput) -> None: +async def confirm_input(ctx: wire.Context, input: messages.CardanoTxInput) -> None: await confirm_properties( ctx, "confirm_input", @@ -249,7 +236,7 @@ async def confirm_sending( async def confirm_sending_token( - ctx: wire.Context, policy_id: bytes, token: CardanoToken + ctx: wire.Context, policy_id: bytes, token: messages.CardanoToken ) -> None: assert token.amount is not None # _validate_token @@ -490,7 +477,7 @@ async def confirm_transaction( async def confirm_certificate( - ctx: wire.Context, certificate: CardanoTxCertificate + ctx: wire.Context, certificate: messages.CardanoTxCertificate ) -> None: # stake pool registration requires custom confirmation logic not covered # in this call @@ -517,7 +504,9 @@ async def confirm_certificate( async def confirm_stake_pool_parameters( - ctx: wire.Context, pool_parameters: CardanoPoolParametersType, network_id: int + ctx: wire.Context, + pool_parameters: messages.CardanoPoolParametersType, + network_id: int, ) -> None: margin_percentage = ( 100.0 * pool_parameters.margin_numerator / pool_parameters.margin_denominator @@ -547,7 +536,7 @@ async def confirm_stake_pool_parameters( async def confirm_stake_pool_owner( ctx: wire.Context, keychain: seed.Keychain, - owner: CardanoPoolOwner, + owner: messages.CardanoPoolOwner, protocol_magic: int, network_id: int, ) -> None: @@ -558,7 +547,7 @@ async def confirm_stake_pool_owner( ( derive_human_readable_address( keychain, - CardanoAddressParametersType( + messages.CardanoAddressParametersType( address_type=CardanoAddressType.REWARD, address_n=owner.staking_key_path, ), @@ -575,7 +564,7 @@ async def confirm_stake_pool_owner( "Pool owner:", derive_human_readable_address( keychain, - CardanoAddressParametersType( + messages.CardanoAddressParametersType( address_type=CardanoAddressType.REWARD, staking_key_hash=owner.staking_key_hash, ), @@ -596,7 +585,7 @@ async def confirm_stake_pool_owner( async def confirm_stake_pool_metadata( ctx: wire.Context, - metadata: CardanoPoolMetadataType | None, + metadata: messages.CardanoPoolMetadataType | None, ) -> None: if metadata is None: await confirm_properties( @@ -643,7 +632,7 @@ async def confirm_stake_pool_registration_final( async def confirm_withdrawal( ctx: wire.Context, - withdrawal: CardanoTxWithdrawal, + withdrawal: messages.CardanoTxWithdrawal, reward_address_bytes: bytes, network_id: int, ) -> None: @@ -726,7 +715,7 @@ async def show_auxiliary_data_hash( async def confirm_token_minting( - ctx: wire.Context, policy_id: bytes, token: CardanoToken + ctx: wire.Context, policy_id: bytes, token: messages.CardanoToken ) -> None: assert token.mint_amount is not None # _validate_token await confirm_properties( @@ -777,7 +766,7 @@ async def confirm_script_data_hash(ctx: wire.Context, script_data_hash: bytes) - async def confirm_collateral_input( - ctx: wire.Context, collateral_input: CardanoTxCollateralInput + ctx: wire.Context, collateral_input: messages.CardanoTxCollateralInput ) -> None: await confirm_properties( ctx, @@ -792,7 +781,7 @@ async def confirm_collateral_input( async def confirm_required_signer( - ctx: wire.Context, required_signer: CardanoTxRequiredSigner + ctx: wire.Context, required_signer: messages.CardanoTxRequiredSigner ) -> None: assert ( required_signer.key_hash is not None or required_signer.key_path @@ -814,7 +803,7 @@ async def confirm_required_signer( async def show_cardano_address( ctx: wire.Context, - address_parameters: CardanoAddressParametersType, + address_parameters: messages.CardanoAddressParametersType, address: str, protocol_magic: int, ) -> None: diff --git a/core/src/apps/cardano/native_script.py b/core/src/apps/cardano/native_script.py index 1f55a23e5..4c1a1058d 100644 --- a/core/src/apps/cardano/native_script.py +++ b/core/src/apps/cardano/native_script.py @@ -1,6 +1,6 @@ from typing import TYPE_CHECKING -from trezor import wire +from trezor import messages, wire from trezor.crypto import hashlib from trezor.enums import CardanoNativeScriptType @@ -14,12 +14,10 @@ from .seed import Keychain, is_multisig_path if TYPE_CHECKING: from typing import Any - from trezor.messages import CardanoNativeScript - from apps.common.cbor import CborSequence -def validate_native_script(script: CardanoNativeScript | None) -> None: +def validate_native_script(script: messages.CardanoNativeScript | None) -> None: INVALID_NATIVE_SCRIPT = wire.ProcessError("Invalid native script") if not script: @@ -61,7 +59,7 @@ def validate_native_script(script: CardanoNativeScript | None) -> None: raise INVALID_NATIVE_SCRIPT -def _validate_native_script_structure(script: CardanoNativeScript) -> None: +def _validate_native_script_structure(script: messages.CardanoNativeScript) -> None: key_hash = script.key_hash key_path = script.key_path scripts = script.scripts @@ -114,14 +112,16 @@ def _validate_native_script_structure(script: CardanoNativeScript) -> None: raise wire.ProcessError("Invalid native script") -def get_native_script_hash(keychain: Keychain, script: CardanoNativeScript) -> bytes: +def get_native_script_hash( + keychain: seed.Keychain, script: messages.CardanoNativeScript +) -> bytes: script_cbor = cbor.encode(cborize_native_script(keychain, script)) prefixed_script_cbor = b"\00" + script_cbor return hashlib.blake2b(data=prefixed_script_cbor, outlen=SCRIPT_HASH_SIZE).digest() def cborize_native_script( - keychain: Keychain, script: CardanoNativeScript + keychain: seed.Keychain, script: messages.CardanoNativeScript ) -> CborSequence: script_content: CborSequence if script.type == CardanoNativeScriptType.PUB_KEY: diff --git a/core/src/apps/cardano/seed.py b/core/src/apps/cardano/seed.py index 852e67db2..daf585684 100644 --- a/core/src/apps/cardano/seed.py +++ b/core/src/apps/cardano/seed.py @@ -16,18 +16,13 @@ if TYPE_CHECKING: from apps.common.paths import Bip32Path from apps.common.keychain import MsgOut, Handler - from trezor.messages import ( - CardanoGetAddress, - CardanoGetPublicKey, - CardanoGetNativeScriptHash, - CardanoSignTxInit, - ) + from trezor import messages CardanoMessages = ( - CardanoGetAddress - | CardanoGetPublicKey - | CardanoGetNativeScriptHash - | CardanoSignTxInit + messages.CardanoGetAddress + | messages.CardanoGetPublicKey + | messages.CardanoGetNativeScriptHash + | messages.CardanoSignTxInit ) MsgIn = TypeVar("MsgIn", bound=CardanoMessages) diff --git a/core/src/apps/cardano/sign_tx/__init__.py b/core/src/apps/cardano/sign_tx/__init__.py index cf4eda083..366e3e6c9 100644 --- a/core/src/apps/cardano/sign_tx/__init__.py +++ b/core/src/apps/cardano/sign_tx/__init__.py @@ -1,8 +1,7 @@ from typing import Type -from trezor import log, wire +from trezor import log, messages, wire from trezor.enums import CardanoTxSigningMode -from trezor.messages import CardanoSignTxFinished, CardanoSignTxInit from .. import seed from .signer import Signer @@ -41,4 +40,4 @@ async def sign_tx( log.exception(__name__, e) raise wire.ProcessError("Signing failed") - return CardanoSignTxFinished() + return messages.CardanoSignTxFinished() diff --git a/core/src/apps/cardano/sign_tx/multisig_signer.py b/core/src/apps/cardano/sign_tx/multisig_signer.py index 43a77e650..f2c9c44d6 100644 --- a/core/src/apps/cardano/sign_tx/multisig_signer.py +++ b/core/src/apps/cardano/sign_tx/multisig_signer.py @@ -1,16 +1,8 @@ -from trezor import wire +from trezor import messages, wire from trezor.enums import CardanoCertificateType -from trezor.messages import ( - CardanoSignTxInit, - CardanoTxCertificate, - CardanoTxOutput, - CardanoTxWithdrawal, - CardanoTxWitnessRequest, -) -from .. import seed +from .. import layout, seed from ..helpers.paths import SCHEMA_MINT -from ..layout import show_multisig_transaction from ..seed import is_multisig_path from .signer import Signer @@ -21,7 +13,10 @@ class MultisigSigner(Signer): """ def __init__( - self, ctx: wire.Context, msg: CardanoSignTxInit, keychain: seed.Keychain + self, + ctx: wire.Context, + msg: messages.CardanoSignTxInit, + keychain: seed.Keychain, ) -> None: super().__init__(ctx, msg, keychain) @@ -34,7 +29,7 @@ class MultisigSigner(Signer): raise wire.ProcessError("Invalid tx signing request") async def _show_tx_signing_request(self) -> None: - await show_multisig_transaction(self.ctx) + await layout.show_multisig_transaction(self.ctx) await super()._show_tx_signing_request() async def _confirm_tx(self, tx_hash: bytes) -> None: @@ -56,20 +51,20 @@ class MultisigSigner(Signer): if output.address_parameters is not None: raise wire.ProcessError("Invalid output") - def _validate_certificate(self, certificate: CardanoTxCertificate) -> None: + def _validate_certificate(self, certificate: messages.CardanoTxCertificate) -> None: super()._validate_certificate(certificate) if certificate.type == CardanoCertificateType.STAKE_POOL_REGISTRATION: raise wire.ProcessError("Invalid certificate") if certificate.path or certificate.key_hash: raise wire.ProcessError("Invalid certificate") - def _validate_withdrawal(self, withdrawal: CardanoTxWithdrawal) -> None: + def _validate_withdrawal(self, withdrawal: messages.CardanoTxWithdrawal) -> None: super()._validate_withdrawal(withdrawal) if withdrawal.path or withdrawal.key_hash: raise wire.ProcessError("Invalid withdrawal") def _validate_witness_request( - self, witness_request: CardanoTxWitnessRequest + self, witness_request: messages.CardanoTxWitnessRequest ) -> None: super()._validate_witness_request(witness_request) is_minting = SCHEMA_MINT.match(witness_request.path) diff --git a/core/src/apps/cardano/sign_tx/ordinary_signer.py b/core/src/apps/cardano/sign_tx/ordinary_signer.py index 2ea66db95..17e0ad25b 100644 --- a/core/src/apps/cardano/sign_tx/ordinary_signer.py +++ b/core/src/apps/cardano/sign_tx/ordinary_signer.py @@ -1,20 +1,13 @@ -from trezor import wire +from trezor import messages, wire from trezor.enums import CardanoCertificateType -from trezor.messages import ( - CardanoSignTxInit, - CardanoTxCertificate, - CardanoTxWithdrawal, - CardanoTxWitnessRequest, -) -from .. import seed +from .. import layout, seed from ..helpers.paths import ( SCHEMA_MINT, SCHEMA_PAYMENT, SCHEMA_STAKING, WITNESS_PATH_NAME, ) -from ..layout import confirm_witness_request from ..seed import is_byron_path, is_shelley_path from .signer import Signer @@ -26,7 +19,10 @@ class OrdinarySigner(Signer): """ def __init__( - self, ctx: wire.Context, msg: CardanoSignTxInit, keychain: seed.Keychain + self, + ctx: wire.Context, + msg: messages.CardanoSignTxInit, + keychain: seed.Keychain, ) -> None: super().__init__(ctx, msg, keychain) @@ -59,13 +55,13 @@ class OrdinarySigner(Signer): if certificate.script_hash or certificate.key_hash: raise wire.ProcessError("Invalid certificate") - def _validate_withdrawal(self, withdrawal: CardanoTxWithdrawal) -> None: + def _validate_withdrawal(self, withdrawal: messages.CardanoTxWithdrawal) -> None: super()._validate_withdrawal(withdrawal) if withdrawal.script_hash or withdrawal.key_hash: raise wire.ProcessError("Invalid withdrawal") def _validate_witness_request( - self, witness_request: CardanoTxWitnessRequest + self, witness_request: messages.CardanoTxWitnessRequest ) -> None: super()._validate_witness_request(witness_request) is_minting = SCHEMA_MINT.match(witness_request.path) @@ -89,6 +85,6 @@ class OrdinarySigner(Signer): is_minting = SCHEMA_MINT.match(witness_path) if is_minting: - await confirm_witness_request(self.ctx, witness_path) + await layout.confirm_witness_request(self.ctx, witness_path) elif not is_payment and not is_staking: await self._fail_or_warn_path(witness_path, WITNESS_PATH_NAME) diff --git a/core/src/apps/cardano/sign_tx/plutus_signer.py b/core/src/apps/cardano/sign_tx/plutus_signer.py index a3a371d3b..b3bf22c77 100644 --- a/core/src/apps/cardano/sign_tx/plutus_signer.py +++ b/core/src/apps/cardano/sign_tx/plutus_signer.py @@ -1,25 +1,9 @@ -from trezor import wire +from trezor import messages, wire from trezor.enums import CardanoCertificateType -from trezor.messages import ( - CardanoAddressParametersType, - CardanoSignTxInit, - CardanoTxCertificate, - CardanoTxInput, - CardanoTxOutput, - CardanoTxWitnessRequest, -) -from .. import seed +from .. import layout, seed from ..helpers.credential import Credential, should_show_address_credentials from ..helpers.paths import SCHEMA_MINT -from ..layout import ( - confirm_input, - confirm_transaction, - show_device_owned_output_credentials, - show_plutus_transaction, - show_warning_no_collateral_inputs, - show_warning_no_script_data_hash, -) from ..seed import is_multisig_path, is_shelley_path from .signer import Signer @@ -31,18 +15,21 @@ class PlutusSigner(Signer): """ def __init__( - self, ctx: wire.Context, msg: CardanoSignTxInit, keychain: seed.Keychain + self, + ctx: wire.Context, + msg: messages.CardanoSignTxInit, + keychain: seed.Keychain, ) -> None: super().__init__(ctx, msg, keychain) async def _show_tx_signing_request(self) -> None: - await show_plutus_transaction(self.ctx) + await layout.show_plutus_transaction(self.ctx) await super()._show_tx_signing_request() # These items should be present if a Plutus script is to be executed. if self.msg.script_data_hash is None: - await show_warning_no_script_data_hash(self.ctx) + await layout.show_warning_no_script_data_hash(self.ctx) if self.msg.collateral_inputs_count == 0: - await show_warning_no_collateral_inputs(self.ctx) + await layout.show_warning_no_collateral_inputs(self.ctx) async def _confirm_transaction(self, tx_hash: bytes) -> None: # super() omitted intentionally @@ -50,7 +37,7 @@ class PlutusSigner(Signer): # computed by a trusted device (in case the tx contains many items which are # tedious to check one by one on the Trezor screen). is_network_id_verifiable = self._is_network_id_verifiable() - await confirm_transaction( + await layout.confirm_transaction( self.ctx, self.msg.fee, self.msg.network_id, @@ -61,43 +48,43 @@ class PlutusSigner(Signer): tx_hash, ) - async def _show_input(self, input: CardanoTxInput) -> None: + async def _show_input(self, input: messages.CardanoTxInput) -> None: # super() omitted intentionally # The inputs are not interchangeable (because of datums), so we must show them. - await confirm_input(self.ctx, input) + await layout.confirm_input(self.ctx, input) async def _show_output_credentials( - self, address_parameters: CardanoAddressParametersType + self, address_parameters: messages.CardanoAddressParametersType ) -> None: # In ordinary txs, change outputs with matching payment and staking paths can be # hidden, but we need to show them in Plutus txs because of the script # evaluation. We at least hide the staking path if it matches the payment path. show_both_credentials = should_show_address_credentials(address_parameters) - await show_device_owned_output_credentials( + await layout.show_device_owned_output_credentials( self.ctx, Credential.payment_credential(address_parameters), Credential.stake_credential(address_parameters), show_both_credentials, ) - def _should_show_output(self, output: CardanoTxOutput) -> bool: + def _should_show_output(self, output: messages.CardanoTxOutput) -> bool: # super() omitted intentionally # All outputs need to be shown (even device-owned), because they might influence # the script evaluation. return True - def _is_change_output(self, output: CardanoTxOutput) -> bool: + def _is_change_output(self, output: messages.CardanoTxOutput) -> bool: # super() omitted intentionally # In Plutus txs, we don't call device-owned outputs "change" outputs. return False - def _validate_certificate(self, certificate: CardanoTxCertificate) -> None: + def _validate_certificate(self, certificate: messages.CardanoTxCertificate) -> None: super()._validate_certificate(certificate) if certificate.type == CardanoCertificateType.STAKE_POOL_REGISTRATION: raise wire.ProcessError("Invalid certificate") def _validate_witness_request( - self, witness_request: CardanoTxWitnessRequest + self, witness_request: messages.CardanoTxWitnessRequest ) -> None: super()._validate_witness_request(witness_request) is_minting = SCHEMA_MINT.match(witness_request.path) diff --git a/core/src/apps/cardano/sign_tx/pool_owner_signer.py b/core/src/apps/cardano/sign_tx/pool_owner_signer.py index a0a8dc406..ea0b98ce1 100644 --- a/core/src/apps/cardano/sign_tx/pool_owner_signer.py +++ b/core/src/apps/cardano/sign_tx/pool_owner_signer.py @@ -1,15 +1,8 @@ -from trezor import wire +from trezor import messages, wire from trezor.enums import CardanoCertificateType -from trezor.messages import ( - CardanoSignTxInit, - CardanoTxCertificate, - CardanoTxOutput, - CardanoTxWitnessRequest, -) -from .. import seed +from .. import layout, seed from ..helpers.paths import SCHEMA_STAKING_ANY_ACCOUNT -from ..layout import confirm_stake_pool_registration_final from .signer import Signer @@ -27,7 +20,10 @@ class PoolOwnerSigner(Signer): """ def __init__( - self, ctx: wire.Context, msg: CardanoSignTxInit, keychain: seed.Keychain + self, + ctx: wire.Context, + msg: messages.CardanoSignTxInit, + keychain: seed.Keychain, ) -> None: super().__init__(ctx, msg, keychain) @@ -51,30 +47,30 @@ class PoolOwnerSigner(Signer): async def _confirm_transaction(self, tx_hash: bytes) -> None: # super() omitted intentionally - await confirm_stake_pool_registration_final( + await layout.confirm_stake_pool_registration_final( self.ctx, self.msg.protocol_magic, self.msg.ttl, self.msg.validity_interval_start, ) - def _validate_output(self, output: CardanoTxOutput) -> None: + def _validate_output(self, output: messages.CardanoTxOutput) -> None: super()._validate_output(output) if output.address_parameters is not None or output.datum_hash is not None: raise wire.ProcessError("Invalid output") - def _should_show_output(self, output: CardanoTxOutput) -> bool: + def _should_show_output(self, output: messages.CardanoTxOutput) -> bool: # super() omitted intentionally # There are no spending witnesses, it is thus safe to hide outputs. return False - def _validate_certificate(self, certificate: CardanoTxCertificate) -> None: + def _validate_certificate(self, certificate: messages.CardanoTxCertificate) -> None: super()._validate_certificate(certificate) if certificate.type != CardanoCertificateType.STAKE_POOL_REGISTRATION: raise wire.ProcessError("Invalid certificate") def _validate_witness_request( - self, witness_request: CardanoTxWitnessRequest + self, witness_request: messages.CardanoTxWitnessRequest ) -> None: super()._validate_witness_request(witness_request) if not SCHEMA_STAKING_ANY_ACCOUNT.match(witness_request.path): diff --git a/core/src/apps/cardano/sign_tx/signer.py b/core/src/apps/cardano/sign_tx/signer.py index b35969215..dfc9dbcf1 100644 --- a/core/src/apps/cardano/sign_tx/signer.py +++ b/core/src/apps/cardano/sign_tx/signer.py @@ -1,7 +1,7 @@ from micropython import const from typing import TYPE_CHECKING -from trezor import wire +from trezor import messages, wire from trezor.crypto import hashlib from trezor.crypto.curve import ed25519 from trezor.enums import ( @@ -9,31 +9,10 @@ from trezor.enums import ( CardanoCertificateType, CardanoTxWitnessType, ) -from trezor.messages import ( - CardanoAddressParametersType, - CardanoAssetGroup, - CardanoPoolOwner, - CardanoPoolRelayParameters, - CardanoSignTxInit, - CardanoToken, - CardanoTxAuxiliaryData, - CardanoTxBodyHash, - CardanoTxCertificate, - CardanoTxCollateralInput, - CardanoTxHostAck, - CardanoTxInput, - CardanoTxItemAck, - CardanoTxMint, - CardanoTxOutput, - CardanoTxRequiredSigner, - CardanoTxWithdrawal, - CardanoTxWitnessRequest, - CardanoTxWitnessResponse, -) from apps.common import cbor, safety_checks -from .. import seed +from .. import layout from ..address import ( ADDRESS_TYPES_PAYMENT_SCRIPT, derive_address_bytes, @@ -82,35 +61,17 @@ from ..helpers.utils import ( validate_network_info, validate_stake_credential, ) -from ..layout import ( - confirm_certificate, - confirm_collateral_input, - confirm_required_signer, - confirm_script_data_hash, - confirm_sending, - confirm_sending_token, - confirm_stake_pool_metadata, - confirm_stake_pool_owner, - confirm_stake_pool_parameters, - confirm_token_minting, - confirm_transaction, - confirm_withdrawal, - confirm_witness_request, - show_change_output_credentials, - show_warning_path, - show_warning_tx_contains_mint, - show_warning_tx_network_unverifiable, - show_warning_tx_output_contains_datum_hash, - show_warning_tx_output_contains_tokens, - show_warning_tx_output_no_datum_hash, -) from ..seed import is_byron_path, is_minting_path, is_multisig_path, is_shelley_path if TYPE_CHECKING: from typing import Any from apps.common.paths import PathSchema - CardanoTxResponseType = CardanoTxItemAck | CardanoTxWitnessResponse + from .. import seed + + CardanoTxResponseType = ( + messages.CardanoTxItemAck | messages.CardanoTxWitnessResponse + ) MINTING_POLICY_ID_LENGTH = 28 MAX_ASSET_NAME_LENGTH = 32 @@ -142,7 +103,10 @@ class Signer: """ def __init__( - self, ctx: wire.Context, msg: CardanoSignTxInit, keychain: seed.Keychain + self, + ctx: wire.Context, + msg: messages.CardanoSignTxInit, + keychain: seed.Keychain, ) -> None: self.ctx = ctx self.msg = msg @@ -179,8 +143,10 @@ class Signer: await self._confirm_transaction(tx_hash) response_after_witness_requests = await self._process_witness_requests(tx_hash) - await self.ctx.call(response_after_witness_requests, CardanoTxHostAck) - await self.ctx.call(CardanoTxBodyHash(tx_hash=tx_hash), CardanoTxHostAck) + await self.ctx.call(response_after_witness_requests, messages.CardanoTxHostAck) + await self.ctx.call( + messages.CardanoTxBodyHash(tx_hash=tx_hash), messages.CardanoTxHostAck + ) # signing request @@ -262,7 +228,7 @@ class Signer: async def _show_tx_signing_request(self) -> None: if not self._is_network_id_verifiable(): - await show_warning_tx_network_unverifiable(self.ctx) + await layout.show_warning_tx_network_unverifiable(self.ctx) async def _confirm_tx(self, tx_hash: bytes) -> None: # Final signing confirmation is handled separately in each signing mode. @@ -274,18 +240,18 @@ class Signer: self, inputs_list: HashBuilderList[tuple[bytes, int]] ) -> None: for _ in range(self.msg.inputs_count): - input: CardanoTxInput = await self.ctx.call( - CardanoTxItemAck(), CardanoTxInput + input: messages.CardanoTxInput = await self.ctx.call( + messages.CardanoTxItemAck(), messages.CardanoTxInput ) self._validate_input(input) await self._show_input(input) inputs_list.append((input.prev_hash, input.prev_index)) - def _validate_input(self, input: CardanoTxInput) -> None: + def _validate_input(self, input: messages.CardanoTxInput) -> None: if len(input.prev_hash) != INPUT_PREV_HASH_SIZE: raise wire.ProcessError("Invalid input") - async def _show_input(self, input: CardanoTxInput) -> None: + async def _show_input(self, input: messages.CardanoTxInput) -> None: # We never show the inputs, except for Plutus txs. pass @@ -294,8 +260,8 @@ class Signer: async def _process_outputs(self, outputs_list: HashBuilderList) -> None: total_amount = 0 for _ in range(self.msg.outputs_count): - output: CardanoTxOutput = await self.ctx.call( - CardanoTxItemAck(), CardanoTxOutput + output: messages.CardanoTxOutput = await self.ctx.call( + messages.CardanoTxItemAck(), messages.CardanoTxOutput ) self._validate_output(output) await self._show_output(output) @@ -334,7 +300,7 @@ class Signer: if total_amount > LOVELACE_MAX_SUPPLY: raise wire.ProcessError("Total transaction amount is out of range!") - def _validate_output(self, output: CardanoTxOutput) -> None: + def _validate_output(self, output: messages.CardanoTxOutput) -> None: if output.address_parameters is not None and output.address is not None: raise wire.ProcessError("Invalid output") @@ -357,21 +323,21 @@ class Signer: self.account_path_checker.add_output(output) - async def _show_output(self, output: CardanoTxOutput) -> None: + async def _show_output(self, output: messages.CardanoTxOutput) -> None: if not self._should_show_output(output): return if output.datum_hash is not None: - await show_warning_tx_output_contains_datum_hash( + await layout.show_warning_tx_output_contains_datum_hash( self.ctx, output.datum_hash ) address_type = self._get_output_address_type(output) if output.datum_hash is None and address_type in ADDRESS_TYPES_PAYMENT_SCRIPT: - await show_warning_tx_output_no_datum_hash(self.ctx) + await layout.show_warning_tx_output_no_datum_hash(self.ctx) if output.asset_groups_count > 0: - await show_warning_tx_output_contains_tokens(self.ctx) + await layout.show_warning_tx_output_contains_tokens(self.ctx) if output.address_parameters is not None: address = derive_human_readable_address( @@ -385,7 +351,7 @@ class Signer: assert output.address is not None # _validate_output address = output.address - await confirm_sending( + await layout.confirm_sending( self.ctx, output.amount, address, @@ -394,15 +360,15 @@ class Signer: ) async def _show_output_credentials( - self, address_parameters: CardanoAddressParametersType + self, address_parameters: messages.CardanoAddressParametersType ) -> None: - await show_change_output_credentials( + await layout.show_change_output_credentials( self.ctx, Credential.payment_credential(address_parameters), Credential.stake_credential(address_parameters), ) - def _should_show_output(self, output: CardanoTxOutput) -> bool: + def _should_show_output(self, output: messages.CardanoTxOutput) -> bool: """ Determines whether the output should be shown. Extracted from _show_output because of readability and because the same decision is made when displaying output tokens. @@ -424,7 +390,7 @@ class Signer: return True - def _is_change_output(self, output: CardanoTxOutput) -> bool: + def _is_change_output(self, output: messages.CardanoTxOutput) -> bool: """Used only to determine what message to show to the user when confirming sending.""" return output.address_parameters is not None @@ -437,8 +403,8 @@ class Signer: should_show_tokens: bool, ) -> None: for _ in range(asset_groups_count): - asset_group: CardanoAssetGroup = await self.ctx.call( - CardanoTxItemAck(), CardanoAssetGroup + asset_group: messages.CardanoAssetGroup = await self.ctx.call( + messages.CardanoTxItemAck(), messages.CardanoAssetGroup ) self._validate_asset_group(asset_group) @@ -455,7 +421,7 @@ class Signer: ) def _validate_asset_group( - self, asset_group: CardanoAssetGroup, is_mint: bool = False + self, asset_group: messages.CardanoAssetGroup, is_mint: bool = False ) -> None: INVALID_TOKEN_BUNDLE = ( wire.ProcessError("Invalid mint token bundle") @@ -478,15 +444,19 @@ class Signer: should_show_tokens: bool, ) -> None: for _ in range(tokens_count): - token: CardanoToken = await self.ctx.call(CardanoTxItemAck(), CardanoToken) + token: messages.CardanoToken = await self.ctx.call( + messages.CardanoTxItemAck(), messages.CardanoToken + ) self._validate_token(token) if should_show_tokens: - await confirm_sending_token(self.ctx, policy_id, token) + await layout.confirm_sending_token(self.ctx, policy_id, token) assert token.amount is not None # _validate_token tokens_dict.add(token.asset_name_bytes, token.amount) - def _validate_token(self, token: CardanoToken, is_mint: bool = False) -> None: + def _validate_token( + self, token: messages.CardanoToken, is_mint: bool = False + ) -> None: INVALID_TOKEN_BUNDLE = ( wire.ProcessError("Invalid mint token bundle") if is_mint @@ -507,8 +477,8 @@ class Signer: async def _process_certificates(self, certificates_list: HashBuilderList) -> None: for _ in range(self.msg.certificates_count): - certificate: CardanoTxCertificate = await self.ctx.call( - CardanoTxItemAck(), CardanoTxCertificate + certificate: messages.CardanoTxCertificate = await self.ctx.call( + messages.CardanoTxItemAck(), messages.CardanoTxCertificate ) self._validate_certificate(certificate) await self._show_certificate(certificate) @@ -550,7 +520,7 @@ class Signer: cborize_certificate(self.keychain, certificate) ) - def _validate_certificate(self, certificate: CardanoTxCertificate) -> None: + def _validate_certificate(self, certificate: messages.CardanoTxCertificate) -> None: validate_certificate( certificate, self.msg.protocol_magic, @@ -558,7 +528,9 @@ class Signer: self.account_path_checker, ) - async def _show_certificate(self, certificate: CardanoTxCertificate) -> None: + async def _show_certificate( + self, certificate: messages.CardanoTxCertificate + ) -> None: if certificate.path: await self._fail_or_warn_if_invalid_path( SCHEMA_STAKING, certificate.path, CERTIFICATE_PATH_NAME @@ -566,14 +538,14 @@ class Signer: if certificate.type == CardanoCertificateType.STAKE_POOL_REGISTRATION: assert certificate.pool_parameters is not None - await confirm_stake_pool_parameters( + await layout.confirm_stake_pool_parameters( self.ctx, certificate.pool_parameters, self.msg.network_id ) - await confirm_stake_pool_metadata( + await layout.confirm_stake_pool_metadata( self.ctx, certificate.pool_parameters.metadata ) else: - await confirm_certificate(self.ctx, certificate) + await layout.confirm_certificate(self.ctx, certificate) # pool owners @@ -582,8 +554,8 @@ class Signer: ) -> None: owners_as_path_count = 0 for _ in range(owners_count): - owner: CardanoPoolOwner = await self.ctx.call( - CardanoTxItemAck(), CardanoPoolOwner + owner: messages.CardanoPoolOwner = await self.ctx.call( + messages.CardanoTxItemAck(), messages.CardanoPoolOwner ) validate_pool_owner(owner, self.account_path_checker) await self._show_pool_owner(owner) @@ -594,13 +566,13 @@ class Signer: assert_certificate_cond(owners_as_path_count == 1) - async def _show_pool_owner(self, owner: CardanoPoolOwner) -> None: + async def _show_pool_owner(self, owner: messages.CardanoPoolOwner) -> None: if owner.staking_key_path: await self._fail_or_warn_if_invalid_path( SCHEMA_STAKING, owner.staking_key_path, POOL_OWNER_STAKING_PATH_NAME ) - await confirm_stake_pool_owner( + await layout.confirm_stake_pool_owner( self.ctx, self.keychain, owner, self.msg.protocol_magic, self.msg.network_id ) @@ -612,8 +584,8 @@ class Signer: relays_count: int, ) -> None: for _ in range(relays_count): - relay: CardanoPoolRelayParameters = await self.ctx.call( - CardanoTxItemAck(), CardanoPoolRelayParameters + relay: messages.CardanoPoolRelayParameters = await self.ctx.call( + messages.CardanoTxItemAck(), messages.CardanoPoolRelayParameters ) validate_pool_relay(relay) relays_list.append(cborize_pool_relay(relay)) @@ -624,19 +596,19 @@ class Signer: self, withdrawals_dict: HashBuilderDict[bytes, int] ) -> None: for _ in range(self.msg.withdrawals_count): - withdrawal: CardanoTxWithdrawal = await self.ctx.call( - CardanoTxItemAck(), CardanoTxWithdrawal + withdrawal: messages.CardanoTxWithdrawal = await self.ctx.call( + messages.CardanoTxItemAck(), messages.CardanoTxWithdrawal ) self._validate_withdrawal(withdrawal) reward_address_bytes = self._derive_withdrawal_reward_address_bytes( withdrawal ) - await confirm_withdrawal( + await layout.confirm_withdrawal( self.ctx, withdrawal, reward_address_bytes, self.msg.network_id ) withdrawals_dict.add(reward_address_bytes, withdrawal.amount) - def _validate_withdrawal(self, withdrawal: CardanoTxWithdrawal) -> None: + def _validate_withdrawal(self, withdrawal: messages.CardanoTxWithdrawal) -> None: validate_stake_credential( withdrawal.path, withdrawal.script_hash, @@ -652,8 +624,8 @@ class Signer: # auxiliary data async def _process_auxiliary_data(self) -> None: - auxiliary_data: CardanoTxAuxiliaryData = await self.ctx.call( - CardanoTxItemAck(), CardanoTxAuxiliaryData + auxiliary_data: messages.CardanoTxAuxiliaryData = await self.ctx.call( + messages.CardanoTxItemAck(), messages.CardanoTxAuxiliaryData ) validate_auxiliary_data(auxiliary_data) @@ -673,22 +645,22 @@ class Signer: ) self.tx_dict.add(TX_BODY_KEY_AUXILIARY_DATA, auxiliary_data_hash) - await self.ctx.call(auxiliary_data_supplement, CardanoTxHostAck) + await self.ctx.call(auxiliary_data_supplement, messages.CardanoTxHostAck) # minting async def _process_minting( self, minting_dict: HashBuilderDict[bytes, HashBuilderDict] ) -> None: - token_minting: CardanoTxMint = await self.ctx.call( - CardanoTxItemAck(), CardanoTxMint + token_minting: messages.CardanoTxMint = await self.ctx.call( + messages.CardanoTxItemAck(), messages.CardanoTxMint ) - await show_warning_tx_contains_mint(self.ctx) + await layout.show_warning_tx_contains_mint(self.ctx) for _ in range(token_minting.asset_groups_count): - asset_group: CardanoAssetGroup = await self.ctx.call( - CardanoTxItemAck(), CardanoAssetGroup + asset_group: messages.CardanoAssetGroup = await self.ctx.call( + messages.CardanoTxItemAck(), messages.CardanoAssetGroup ) self._validate_asset_group(asset_group, is_mint=True) @@ -711,9 +683,11 @@ class Signer: tokens_count: int, ) -> None: for _ in range(tokens_count): - token: CardanoToken = await self.ctx.call(CardanoTxItemAck(), CardanoToken) + token: messages.CardanoToken = await self.ctx.call( + messages.CardanoTxItemAck(), messages.CardanoToken + ) self._validate_token(token, is_mint=True) - await confirm_token_minting(self.ctx, policy_id, token) + await layout.confirm_token_minting(self.ctx, policy_id, token) assert token.mint_amount is not None # _validate_token tokens.add(token.asset_name_bytes, token.mint_amount) @@ -723,7 +697,7 @@ class Signer: async def _process_script_data_hash(self) -> None: assert self.msg.script_data_hash is not None self._validate_script_data_hash() - await confirm_script_data_hash(self.ctx, self.msg.script_data_hash) + await layout.confirm_script_data_hash(self.ctx, self.msg.script_data_hash) self.tx_dict.add(TX_BODY_KEY_SCRIPT_DATA_HASH, self.msg.script_data_hash) def _validate_script_data_hash(self) -> None: @@ -737,17 +711,17 @@ class Signer: self, collateral_inputs_list: HashBuilderList[tuple[bytes, int]] ) -> None: for _ in range(self.msg.collateral_inputs_count): - collateral_input: CardanoTxCollateralInput = await self.ctx.call( - CardanoTxItemAck(), CardanoTxCollateralInput + collateral_input: messages.CardanoTxCollateralInput = await self.ctx.call( + messages.CardanoTxItemAck(), messages.CardanoTxCollateralInput ) self._validate_collateral_input(collateral_input) - await confirm_collateral_input(self.ctx, collateral_input) + await layout.confirm_collateral_input(self.ctx, collateral_input) collateral_inputs_list.append( (collateral_input.prev_hash, collateral_input.prev_index) ) def _validate_collateral_input( - self, collateral_input: CardanoTxCollateralInput + self, collateral_input: messages.CardanoTxCollateralInput ) -> None: if len(collateral_input.prev_hash) != INPUT_PREV_HASH_SIZE: raise wire.ProcessError("Invalid collateral input") @@ -758,11 +732,11 @@ class Signer: self, required_signers_list: HashBuilderList[bytes] ) -> None: for _ in range(self.msg.required_signers_count): - required_signer: CardanoTxRequiredSigner = await self.ctx.call( - CardanoTxItemAck(), CardanoTxRequiredSigner + required_signer: messages.CardanoTxRequiredSigner = await self.ctx.call( + messages.CardanoTxItemAck(), messages.CardanoTxRequiredSigner ) self._validate_required_signer(required_signer) - await confirm_required_signer(self.ctx, required_signer) + await layout.confirm_required_signer(self.ctx, required_signer) key_hash = required_signer.key_hash or get_public_key_hash( self.keychain, required_signer.key_path @@ -770,7 +744,7 @@ class Signer: required_signers_list.append(key_hash) def _validate_required_signer( - self, required_signer: CardanoTxRequiredSigner + self, required_signer: messages.CardanoTxRequiredSigner ) -> None: INVALID_REQUIRED_SIGNER = wire.ProcessError("Invalid required signer") @@ -793,10 +767,12 @@ class Signer: # witness requests async def _process_witness_requests(self, tx_hash: bytes) -> CardanoTxResponseType: - response: CardanoTxResponseType = CardanoTxItemAck() + response: CardanoTxResponseType = messages.CardanoTxItemAck() for _ in range(self.msg.witness_requests_count): - witness_request = await self.ctx.call(response, CardanoTxWitnessRequest) + witness_request = await self.ctx.call( + response, messages.CardanoTxWitnessRequest + ) self._validate_witness_request(witness_request) path = witness_request.path await self._show_witness_request(path) @@ -808,7 +784,7 @@ class Signer: return response def _validate_witness_request( - self, witness_request: CardanoTxWitnessRequest + self, witness_request: messages.CardanoTxWitnessRequest ) -> None: self.account_path_checker.add_witness_request(witness_request) @@ -816,7 +792,7 @@ class Signer: self, witness_path: list[int], ) -> None: - await confirm_witness_request(self.ctx, witness_path) + await layout.confirm_witness_request(self.ctx, witness_path) # helpers @@ -836,7 +812,7 @@ class Signer: or self.msg.withdrawals_count != 0 ) - def _get_output_address(self, output: CardanoTxOutput) -> bytes: + def _get_output_address(self, output: messages.CardanoTxOutput) -> bytes: if output.address_parameters: return derive_address_bytes( self.keychain, @@ -848,14 +824,16 @@ class Signer: assert output.address is not None # _validate_output return get_address_bytes_unsafe(output.address) - def _get_output_address_type(self, output: CardanoTxOutput) -> CardanoAddressType: + def _get_output_address_type( + self, output: messages.CardanoTxOutput + ) -> CardanoAddressType: if output.address_parameters: return output.address_parameters.address_type assert output.address is not None # _validate_output return get_address_type(get_address_bytes_unsafe(output.address)) def _derive_withdrawal_reward_address_bytes( - self, withdrawal: CardanoTxWithdrawal + self, withdrawal: messages.CardanoTxWithdrawal ) -> bytes: reward_address_type = ( CardanoAddressType.REWARD @@ -864,7 +842,7 @@ class Signer: ) return derive_address_bytes( self.keychain, - CardanoAddressParametersType( + messages.CardanoAddressParametersType( address_type=reward_address_type, address_n_staking=withdrawal.path, staking_key_hash=withdrawal.key_hash, @@ -876,9 +854,9 @@ class Signer: def _get_byron_witness( self, path: list[int], tx_hash: bytes - ) -> CardanoTxWitnessResponse: + ) -> messages.CardanoTxWitnessResponse: node = self.keychain.derive(path) - return CardanoTxWitnessResponse( + return messages.CardanoTxWitnessResponse( type=CardanoTxWitnessType.BYRON_WITNESS, pub_key=derive_public_key(self.keychain, path), signature=self._sign_tx_hash(tx_hash, path), @@ -887,8 +865,8 @@ class Signer: def _get_shelley_witness( self, path: list[int], tx_hash: bytes - ) -> CardanoTxWitnessResponse: - return CardanoTxWitnessResponse( + ) -> messages.CardanoTxWitnessResponse: + return messages.CardanoTxWitnessResponse( type=CardanoTxWitnessType.SHELLEY_WITNESS, pub_key=derive_public_key(self.keychain, path), signature=self._sign_tx_hash(tx_hash, path), @@ -910,10 +888,10 @@ class Signer: if safety_checks.is_strict(): raise wire.DataError(f"Invalid {path_name.lower()}") else: - await show_warning_path(self.ctx, path, path_name) + await layout.show_warning_path(self.ctx, path, path_name) def _fail_if_strict_and_unusual( - self, address_parameters: CardanoAddressParametersType + self, address_parameters: messages.CardanoAddressParametersType ) -> None: if not safety_checks.is_strict(): return