arno
/
trezor-firmware
mirror of https://github.com/trezor/trezor-firmware.git
1
0
Fork 0
Code Issues Releases Wiki Activity

refactor(cardano): inline validation errors

Browse Source
pull/2374/head
David Misiak 2 years ago committed by matejcik
parent 2724d29968
commit 0836e91faa
12 changed files with 161 additions and 220 deletions
Show Stats Download Patch File Download Diff File

130
core/src/apps/cardano/address.py
Unescape View File

@ -1,18 +1,11 @@
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from trezor import wire
from trezor.crypto import base58 from trezor.crypto import base58
from trezor.enums import CardanoAddressType from trezor.enums import CardanoAddressType
from .byron_address import derive_byron_address, validate_byron_address from .byron_address import derive_byron_address, validate_byron_address
from .helpers import ( from .helpers import ADDRESS_KEY_HASH_SIZE, SCRIPT_HASH_SIZE, bech32, network_ids
ADDRESS_KEY_HASH_SIZE,
INVALID_ADDRESS,
INVALID_ADDRESS_PARAMETERS,
NETWORK_MISMATCH,
SCRIPT_HASH_SIZE,
bech32,
network_ids,
)
from .helpers.paths import SCHEMA_STAKING_ANY_ACCOUNT from .helpers.paths import SCHEMA_STAKING_ANY_ACCOUNT
from .helpers.utils import get_public_key_hash, variable_length_encode from .helpers.utils import get_public_key_hash, variable_length_encode
from .seed import is_byron_path, is_shelley_path from .seed import is_byron_path, is_shelley_path
@ -50,16 +43,19 @@ MIN_ADDRESS_BYTES_LENGTH = 29
MAX_ADDRESS_BYTES_LENGTH = 65 MAX_ADDRESS_BYTES_LENGTH = 65
def assert_address_params_cond(condition: bool) -> None:
if not condition:
raise wire.ProcessError("Invalid address parameters")
def validate_address_parameters(parameters: CardanoAddressParametersType) -> None: def validate_address_parameters(parameters: CardanoAddressParametersType) -> None:
_validate_address_parameters_structure(parameters) _validate_address_parameters_structure(parameters)
if parameters.address_type == CardanoAddressType.BYRON: if parameters.address_type == CardanoAddressType.BYRON:
if not is_byron_path(parameters.address_n): assert_address_params_cond(is_byron_path(parameters.address_n))
raise INVALID_ADDRESS_PARAMETERS
elif parameters.address_type == CardanoAddressType.BASE: elif parameters.address_type == CardanoAddressType.BASE:
if not is_shelley_path(parameters.address_n): assert_address_params_cond(is_shelley_path(parameters.address_n))
raise INVALID_ADDRESS_PARAMETERS
_validate_base_address_staking_info( _validate_base_address_staking_info(
parameters.address_n_staking, parameters.staking_key_hash parameters.address_n_staking, parameters.staking_key_hash
) )
@ -71,8 +67,7 @@ def validate_address_parameters(parameters: CardanoAddressParametersType) -> Non
) )
elif parameters.address_type == CardanoAddressType.BASE_KEY_SCRIPT: elif parameters.address_type == CardanoAddressType.BASE_KEY_SCRIPT:
if not is_shelley_path(parameters.address_n): assert_address_params_cond(is_shelley_path(parameters.address_n))
raise INVALID_ADDRESS_PARAMETERS
_validate_script_hash(parameters.script_staking_hash) _validate_script_hash(parameters.script_staking_hash)
elif parameters.address_type == CardanoAddressType.BASE_SCRIPT_SCRIPT: elif parameters.address_type == CardanoAddressType.BASE_SCRIPT_SCRIPT:
@ -80,34 +75,30 @@ def validate_address_parameters(parameters: CardanoAddressParametersType) -> Non
_validate_script_hash(parameters.script_staking_hash) _validate_script_hash(parameters.script_staking_hash)
elif parameters.address_type == CardanoAddressType.POINTER: elif parameters.address_type == CardanoAddressType.POINTER:
if not is_shelley_path(parameters.address_n): assert_address_params_cond(is_shelley_path(parameters.address_n))
raise INVALID_ADDRESS_PARAMETERS assert_address_params_cond(parameters.certificate_pointer is not None)
if parameters.certificate_pointer is None:
raise INVALID_ADDRESS_PARAMETERS
elif parameters.address_type == CardanoAddressType.POINTER_SCRIPT: elif parameters.address_type == CardanoAddressType.POINTER_SCRIPT:
_validate_script_hash(parameters.script_payment_hash) _validate_script_hash(parameters.script_payment_hash)
if parameters.certificate_pointer is None: assert_address_params_cond(parameters.certificate_pointer is not None)
raise INVALID_ADDRESS_PARAMETERS
elif parameters.address_type == CardanoAddressType.ENTERPRISE: elif parameters.address_type == CardanoAddressType.ENTERPRISE:
if not is_shelley_path(parameters.address_n): assert_address_params_cond(is_shelley_path(parameters.address_n))
raise INVALID_ADDRESS_PARAMETERS
elif parameters.address_type == CardanoAddressType.ENTERPRISE_SCRIPT: elif parameters.address_type == CardanoAddressType.ENTERPRISE_SCRIPT:
_validate_script_hash(parameters.script_payment_hash) _validate_script_hash(parameters.script_payment_hash)
elif parameters.address_type == CardanoAddressType.REWARD: elif parameters.address_type == CardanoAddressType.REWARD:
if not is_shelley_path(parameters.address_n_staking): assert_address_params_cond(is_shelley_path(parameters.address_n_staking))
raise INVALID_ADDRESS_PARAMETERS assert_address_params_cond(
if not SCHEMA_STAKING_ANY_ACCOUNT.match(parameters.address_n_staking): SCHEMA_STAKING_ANY_ACCOUNT.match(parameters.address_n_staking)
raise INVALID_ADDRESS_PARAMETERS )
elif parameters.address_type == CardanoAddressType.REWARD_SCRIPT: elif parameters.address_type == CardanoAddressType.REWARD_SCRIPT:
_validate_script_hash(parameters.script_staking_hash) _validate_script_hash(parameters.script_staking_hash)
else: else:
raise INVALID_ADDRESS_PARAMETERS raise RuntimeError # should be unreachable
def _validate_address_parameters_structure( def _validate_address_parameters_structure(
@ -190,32 +181,28 @@ def _validate_address_parameters_structure(
), ),
} }
if parameters.address_type not in fields_to_be_empty or any( assert_address_params_cond(parameters.address_type in fields_to_be_empty)
fields_to_be_empty[parameters.address_type] assert_address_params_cond(not any(fields_to_be_empty[parameters.address_type]))
):
raise INVALID_ADDRESS_PARAMETERS
def _validate_base_address_staking_info( def _validate_base_address_staking_info(
staking_path: list[int], staking_path: list[int],
staking_key_hash: bytes | None, staking_key_hash: bytes | None,
) -> None: ) -> None:
if staking_key_hash and staking_path: assert_address_params_cond(not (staking_key_hash and staking_path))
raise INVALID_ADDRESS_PARAMETERS
if staking_key_hash: if staking_key_hash:
if len(staking_key_hash) != ADDRESS_KEY_HASH_SIZE: assert_address_params_cond(len(staking_key_hash) == ADDRESS_KEY_HASH_SIZE)
raise INVALID_ADDRESS_PARAMETERS
elif staking_path: elif staking_path:
if not SCHEMA_STAKING_ANY_ACCOUNT.match(staking_path): assert_address_params_cond(SCHEMA_STAKING_ANY_ACCOUNT.match(staking_path))
raise INVALID_ADDRESS_PARAMETERS
else: else:
raise INVALID_ADDRESS_PARAMETERS raise wire.ProcessError("Invalid address parameters")
def _validate_script_hash(script_hash: bytes | None) -> None: def _validate_script_hash(script_hash: bytes | None) -> None:
if not script_hash or len(script_hash) != SCRIPT_HASH_SIZE: assert_address_params_cond(
raise INVALID_ADDRESS_PARAMETERS script_hash is not None and len(script_hash) == SCRIPT_HASH_SIZE
)
def validate_output_address_parameters( def validate_output_address_parameters(
@ -223,16 +210,23 @@ def validate_output_address_parameters(
) -> None: ) -> None:
validate_address_parameters(parameters) validate_address_parameters(parameters)
if parameters.address_type not in ( # Change outputs with script payment part are forbidden.
CardanoAddressType.BASE, # Reward addresses are forbidden as outputs in general, see also validate_output_address
CardanoAddressType.BASE_KEY_SCRIPT, assert_address_params_cond(
CardanoAddressType.POINTER, parameters.address_type
CardanoAddressType.ENTERPRISE, in (
CardanoAddressType.BYRON, CardanoAddressType.BASE,
): CardanoAddressType.BASE_KEY_SCRIPT,
# Change outputs with script payment part are forbidden. CardanoAddressType.POINTER,
# Reward addresses are forbidden as outputs in general, see also validate_output_address CardanoAddressType.ENTERPRISE,
raise INVALID_ADDRESS_PARAMETERS CardanoAddressType.BYRON,
)
)
def assert_address_cond(condition: bool) -> None:
if not condition:
raise wire.ProcessError("Invalid address")
def _validate_address_and_get_type( def _validate_address_and_get_type(
@ -242,8 +236,8 @@ def _validate_address_and_get_type(
Validates Cardano address and returns its type Validates Cardano address and returns its type
for the convenience of outward-facing functions. for the convenience of outward-facing functions.
""" """
if address is None or len(address) == 0: assert_address_cond(address is not None)
raise INVALID_ADDRESS assert_address_cond(len(address) > 0)
address_bytes = get_address_bytes_unsafe(address) address_bytes = get_address_bytes_unsafe(address)
address_type = get_address_type(address_bytes) address_type = get_address_type(address_bytes)
@ -253,26 +247,24 @@ def _validate_address_and_get_type(
elif address_type in ADDRESS_TYPES_SHELLEY: elif address_type in ADDRESS_TYPES_SHELLEY:
_validate_shelley_address(address, address_bytes, network_id) _validate_shelley_address(address, address_bytes, network_id)
else: else:
raise INVALID_ADDRESS raise wire.ProcessError("Invalid address")
return address_type return address_type
def validate_output_address(address: str, protocol_magic: int, network_id: int) -> None: def validate_output_address(address: str, protocol_magic: int, network_id: int) -> None:
address_type = _validate_address_and_get_type(address, protocol_magic, network_id) address_type = _validate_address_and_get_type(address, protocol_magic, network_id)
assert_address_cond(
if address_type in (CardanoAddressType.REWARD, CardanoAddressType.REWARD_SCRIPT): address_type
raise INVALID_ADDRESS not in (CardanoAddressType.REWARD, CardanoAddressType.REWARD_SCRIPT)
)
def validate_reward_address(address: str, protocol_magic: int, network_id: int) -> None: def validate_reward_address(address: str, protocol_magic: int, network_id: int) -> None:
address_type = _validate_address_and_get_type(address, protocol_magic, network_id) address_type = _validate_address_and_get_type(address, protocol_magic, network_id)
assert_address_cond(
if address_type not in ( address_type in (CardanoAddressType.REWARD, CardanoAddressType.REWARD_SCRIPT)
CardanoAddressType.REWARD, )
CardanoAddressType.REWARD_SCRIPT,
):
raise INVALID_ADDRESS
def get_address_bytes_unsafe(address: str) -> bytes: def get_address_bytes_unsafe(address: str) -> bytes:
@ -282,7 +274,7 @@ def get_address_bytes_unsafe(address: str) -> bytes:
try: try:
address_bytes = base58.decode(address) address_bytes = base58.decode(address)
except ValueError: except ValueError:
raise INVALID_ADDRESS raise wire.ProcessError("Invalid address")
return address_bytes return address_bytes
@ -302,8 +294,9 @@ def _validate_shelley_address(
def _validate_address_size(address_bytes: bytes) -> None: def _validate_address_size(address_bytes: bytes) -> None:
if not MIN_ADDRESS_BYTES_LENGTH <= len(address_bytes) <= MAX_ADDRESS_BYTES_LENGTH: assert_address_cond(
raise INVALID_ADDRESS MIN_ADDRESS_BYTES_LENGTH <= len(address_bytes) <= MAX_ADDRESS_BYTES_LENGTH
)
def _validate_address_bech32_hrp( def _validate_address_bech32_hrp(
@ -312,8 +305,7 @@ def _validate_address_bech32_hrp(
valid_hrp = _get_bech32_hrp_for_address(address_type, network_id) valid_hrp = _get_bech32_hrp_for_address(address_type, network_id)
bech32_hrp = bech32.get_hrp(address_str) bech32_hrp = bech32.get_hrp(address_str)
if valid_hrp != bech32_hrp: assert_address_cond(valid_hrp == bech32_hrp)
raise INVALID_ADDRESS
def _get_bech32_hrp_for_address( def _get_bech32_hrp_for_address(
@ -337,7 +329,7 @@ def _get_bech32_hrp_for_address(
def _validate_address_network_id(address: bytes, network_id: int) -> None: def _validate_address_network_id(address: bytes, network_id: int) -> None:
if _get_address_network_id(address) != network_id: if _get_address_network_id(address) != network_id:
raise NETWORK_MISMATCH raise wire.ProcessError("Output address network mismatch")
def _get_address_network_id(address: bytes) -> int: def _get_address_network_id(address: bytes) -> int:

15
core/src/apps/cardano/auxiliary_data.py
Unescape View File

@ -1,5 +1,6 @@
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from trezor import wire
from trezor.crypto import hashlib from trezor.crypto import hashlib
from trezor.crypto.curve import ed25519 from trezor.crypto.curve import ed25519
from trezor.enums import CardanoAddressType, CardanoTxAuxiliaryDataSupplementType from trezor.enums import CardanoAddressType, CardanoTxAuxiliaryDataSupplementType
@ -12,15 +13,13 @@ from .address import (
derive_human_readable_address, derive_human_readable_address,
validate_address_parameters, validate_address_parameters,
) )
from .helpers import INVALID_AUXILIARY_DATA, bech32 from .helpers import bech32
from .helpers.bech32 import HRP_JORMUN_PUBLIC_KEY from .helpers.bech32 import HRP_JORMUN_PUBLIC_KEY
from .helpers.paths import SCHEMA_STAKING_ANY_ACCOUNT from .helpers.paths import SCHEMA_STAKING_ANY_ACCOUNT
from .helpers.utils import derive_public_key from .helpers.utils import derive_public_key
from .layout import confirm_catalyst_registration, show_auxiliary_data_hash from .layout import confirm_catalyst_registration, show_auxiliary_data_hash
if TYPE_CHECKING: if TYPE_CHECKING:
from trezor import wire
from trezor.messages import ( from trezor.messages import (
CardanoCatalystRegistrationParametersType, CardanoCatalystRegistrationParametersType,
CardanoTxAuxiliaryData, CardanoTxAuxiliaryData,
@ -55,12 +54,12 @@ def validate_auxiliary_data(auxiliary_data: CardanoTxAuxiliaryData) -> None:
) )
if fields_provided != 1: if fields_provided != 1:
raise INVALID_AUXILIARY_DATA raise wire.ProcessError("Invalid auxiliary data")
def _validate_auxiliary_data_hash(auxiliary_data_hash: bytes) -> None: def _validate_auxiliary_data_hash(auxiliary_data_hash: bytes) -> None:
if len(auxiliary_data_hash) != AUXILIARY_DATA_HASH_SIZE: if len(auxiliary_data_hash) != AUXILIARY_DATA_HASH_SIZE:
raise INVALID_AUXILIARY_DATA raise wire.ProcessError("Invalid auxiliary data")
def _validate_catalyst_registration_parameters( def _validate_catalyst_registration_parameters(
@ -70,16 +69,16 @@ def _validate_catalyst_registration_parameters(
len(catalyst_registration_parameters.voting_public_key) len(catalyst_registration_parameters.voting_public_key)
!= CATALYST_VOTING_PUBLIC_KEY_LENGTH != CATALYST_VOTING_PUBLIC_KEY_LENGTH
): ):
raise INVALID_AUXILIARY_DATA raise wire.ProcessError("Invalid auxiliary data")
if not SCHEMA_STAKING_ANY_ACCOUNT.match( if not SCHEMA_STAKING_ANY_ACCOUNT.match(
catalyst_registration_parameters.staking_path catalyst_registration_parameters.staking_path
): ):
raise INVALID_AUXILIARY_DATA raise wire.ProcessError("Invalid auxiliary data")
address_parameters = catalyst_registration_parameters.reward_address_parameters address_parameters = catalyst_registration_parameters.reward_address_parameters
if address_parameters.address_type == CardanoAddressType.BYRON: if address_parameters.address_type == CardanoAddressType.BYRON:
raise INVALID_AUXILIARY_DATA raise wire.ProcessError("Invalid auxiliary data")
validate_address_parameters(address_parameters) validate_address_parameters(address_parameters)

24
core/src/apps/cardano/byron_address.py
Unescape View File

@ -1,11 +1,11 @@
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from trezor import log from trezor import log, wire
from trezor.crypto import crc, hashlib from trezor.crypto import crc, hashlib
from apps.common import cbor from apps.common import cbor
from .helpers import INVALID_ADDRESS, NETWORK_MISMATCH, protocol_magics from .helpers import protocol_magics
from .helpers.utils import derive_public_key from .helpers.utils import derive_public_key
if TYPE_CHECKING: if TYPE_CHECKING:
@ -62,21 +62,21 @@ def _decode_address_raw(address: bytes) -> bytes:
except ValueError as e: except ValueError as e:
if __debug__: if __debug__:
log.exception(__name__, e) log.exception(__name__, e)
raise INVALID_ADDRESS raise wire.ProcessError("Invalid address")
if not isinstance(address_unpacked, list) or len(address_unpacked) != 2: if not isinstance(address_unpacked, list) or len(address_unpacked) != 2:
raise INVALID_ADDRESS raise wire.ProcessError("Invalid address")
address_data_encoded = address_unpacked[0] address_data_encoded = address_unpacked[0]
if not isinstance(address_data_encoded, bytes): if not isinstance(address_data_encoded, bytes):
raise INVALID_ADDRESS raise wire.ProcessError("Invalid address")
address_crc = address_unpacked[1] address_crc = address_unpacked[1]
if not isinstance(address_crc, int): if not isinstance(address_crc, int):
raise INVALID_ADDRESS raise wire.ProcessError("Invalid address")
if address_crc != crc.crc32(address_data_encoded): if address_crc != crc.crc32(address_data_encoded):
raise INVALID_ADDRESS raise wire.ProcessError("Invalid address")
return address_data_encoded return address_data_encoded
@ -91,24 +91,24 @@ def _validate_address_data_protocol_magic(
""" """
address_data = cbor.decode(address_data_encoded) address_data = cbor.decode(address_data_encoded)
if not isinstance(address_data, list) or len(address_data) < 2: if not isinstance(address_data, list) or len(address_data) < 2:
raise INVALID_ADDRESS raise wire.ProcessError("Invalid address")
attributes = address_data[1] attributes = address_data[1]
if protocol_magics.is_mainnet(protocol_magic): if protocol_magics.is_mainnet(protocol_magic):
if PROTOCOL_MAGIC_KEY in attributes: if PROTOCOL_MAGIC_KEY in attributes:
raise NETWORK_MISMATCH raise wire.ProcessError("Output address network mismatch")
else: # testnet else: # testnet
if len(attributes) == 0 or PROTOCOL_MAGIC_KEY not in attributes: if len(attributes) == 0 or PROTOCOL_MAGIC_KEY not in attributes:
raise NETWORK_MISMATCH raise wire.ProcessError("Output address network mismatch")
protocol_magic_cbor = attributes[PROTOCOL_MAGIC_KEY] protocol_magic_cbor = attributes[PROTOCOL_MAGIC_KEY]
address_protocol_magic = cbor.decode(protocol_magic_cbor) address_protocol_magic = cbor.decode(protocol_magic_cbor)
if not isinstance(address_protocol_magic, int): if not isinstance(address_protocol_magic, int):
raise INVALID_ADDRESS raise wire.ProcessError("Invalid address")
if address_protocol_magic != protocol_magic: if address_protocol_magic != protocol_magic:
raise NETWORK_MISMATCH raise wire.ProcessError("Output address network mismatch")
def _address_hash(data: list) -> bytes: def _address_hash(data: list) -> bytes:

21
core/src/apps/cardano/certificates.py
Unescape View File

@ -1,11 +1,12 @@
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from trezor import wire
from trezor.enums import CardanoCertificateType, CardanoPoolRelayType from trezor.enums import CardanoCertificateType, CardanoPoolRelayType
from apps.common import cbor from apps.common import cbor
from .address import get_address_bytes_unsafe, validate_reward_address from .address import get_address_bytes_unsafe, validate_reward_address
from .helpers import ADDRESS_KEY_HASH_SIZE, INVALID_CERTIFICATE, LOVELACE_MAX_SUPPLY from .helpers import ADDRESS_KEY_HASH_SIZE, LOVELACE_MAX_SUPPLY
from .helpers.paths import SCHEMA_STAKING_ANY_ACCOUNT from .helpers.paths import SCHEMA_STAKING_ANY_ACCOUNT
from .helpers.utils import get_public_key_hash, validate_stake_credential from .helpers.utils import get_public_key_hash, validate_stake_credential
@ -52,16 +53,16 @@ def validate_certificate(
certificate.path, certificate.path,
certificate.script_hash, certificate.script_hash,
certificate.key_hash, certificate.key_hash,
INVALID_CERTIFICATE, wire.ProcessError("Invalid certificate"),
) )
if certificate.type == CardanoCertificateType.STAKE_DELEGATION: if certificate.type == CardanoCertificateType.STAKE_DELEGATION:
if not certificate.pool or len(certificate.pool) != POOL_HASH_SIZE: if not certificate.pool or len(certificate.pool) != POOL_HASH_SIZE:
raise INVALID_CERTIFICATE raise wire.ProcessError("Invalid certificate")
if certificate.type == CardanoCertificateType.STAKE_POOL_REGISTRATION: if certificate.type == CardanoCertificateType.STAKE_POOL_REGISTRATION:
if certificate.pool_parameters is None: if certificate.pool_parameters is None:
raise INVALID_CERTIFICATE raise wire.ProcessError("Invalid certificate")
_validate_pool_parameters( _validate_pool_parameters(
certificate.pool_parameters, protocol_magic, network_id certificate.pool_parameters, protocol_magic, network_id
) )
@ -88,7 +89,7 @@ def _validate_certificate_structure(certificate: CardanoTxCertificate) -> None:
if certificate.type not in fields_to_be_empty or any( if certificate.type not in fields_to_be_empty or any(
fields_to_be_empty[certificate.type] fields_to_be_empty[certificate.type]
): ):
raise INVALID_CERTIFICATE raise wire.ProcessError("Invalid certificate")
def cborize_certificate( def cborize_certificate(
@ -119,7 +120,7 @@ def cborize_certificate(
certificate.pool, certificate.pool,
) )
else: else:
raise INVALID_CERTIFICATE raise RuntimeError # should be unreachable
def cborize_certificate_stake_credential( def cborize_certificate_stake_credential(
@ -135,7 +136,7 @@ def cborize_certificate_stake_credential(
return 1, script_hash return 1, script_hash
# should be unreachable unless there's a bug in validation # should be unreachable unless there's a bug in validation
raise INVALID_CERTIFICATE raise RuntimeError
def cborize_initial_pool_registration_certificate_fields( def cborize_initial_pool_registration_certificate_fields(
@ -167,7 +168,7 @@ def cborize_initial_pool_registration_certificate_fields(
def assert_certificate_cond(condition: bool) -> None: def assert_certificate_cond(condition: bool) -> None:
if not condition: if not condition:
raise INVALID_CERTIFICATE raise wire.ProcessError("Invalid certificate")
def _validate_pool_parameters( def _validate_pool_parameters(
@ -232,7 +233,7 @@ def validate_pool_relay(pool_relay: CardanoPoolRelayParameters) -> None:
and len(pool_relay.host_name) <= MAX_URL_LENGTH and len(pool_relay.host_name) <= MAX_URL_LENGTH
) )
else: else:
raise INVALID_CERTIFICATE raise RuntimeError # should be unreachable
def _validate_pool_metadata(pool_metadata: CardanoPoolMetadataType) -> None: def _validate_pool_metadata(pool_metadata: CardanoPoolMetadataType) -> None:
@ -286,7 +287,7 @@ def cborize_pool_relay(
pool_relay.host_name, pool_relay.host_name,
) )
else: else:
raise INVALID_CERTIFICATE raise RuntimeError # should be unreachable
def cborize_pool_metadata( def cborize_pool_metadata(

26
core/src/apps/cardano/helpers/__init__.py
Unescape View File

@ -1,29 +1,3 @@
from trezor import wire
INVALID_ADDRESS = wire.ProcessError("Invalid address")
INVALID_ADDRESS_PARAMETERS = wire.ProcessError("Invalid address parameters")
NETWORK_MISMATCH = wire.ProcessError("Output address network mismatch")
INVALID_TX_SIGNING_REQUEST = wire.ProcessError("Invalid tx signing request")
INVALID_INPUT = wire.ProcessError("Invalid input")
INVALID_OUTPUT = wire.ProcessError("Invalid output")
INVALID_CERTIFICATE = wire.ProcessError("Invalid certificate")
INVALID_WITHDRAWAL = wire.ProcessError("Invalid withdrawal")
INVALID_TOKEN_BUNDLE_OUTPUT = wire.ProcessError("Invalid token bundle in output")
INVALID_AUXILIARY_DATA = wire.ProcessError("Invalid auxiliary data")
INVALID_STAKE_POOL_REGISTRATION_TX_STRUCTURE = wire.ProcessError(
"Stakepool registration transaction cannot contain other certificates, withdrawals or minting"
)
INVALID_STAKEPOOL_REGISTRATION_TX_WITNESSES = wire.ProcessError(
"Stakepool registration transaction can only contain staking witnesses"
)
INVALID_WITNESS_REQUEST = wire.ProcessError("Invalid witness request")
INVALID_NATIVE_SCRIPT = wire.ProcessError("Invalid native script")
INVALID_TOKEN_BUNDLE_MINT = wire.ProcessError("Invalid mint token bundle")
INVALID_OUTPUT_DATUM_HASH = wire.ProcessError("Invalid output datum hash")
INVALID_SCRIPT_DATA_HASH = wire.ProcessError("Invalid script data hash")
INVALID_COLLATERAL_INPUT = wire.ProcessError("Invalid collateral input")
INVALID_REQUIRED_SIGNER = wire.ProcessError("Invalid required signer")
LOVELACE_MAX_SUPPLY = 45_000_000_000 * 1_000_000 LOVELACE_MAX_SUPPLY = 45_000_000_000 * 1_000_000
INPUT_PREV_HASH_SIZE = 32 INPUT_PREV_HASH_SIZE = 32
ADDRESS_KEY_HASH_SIZE = 28 ADDRESS_KEY_HASH_SIZE = 28

21
core/src/apps/cardano/helpers/account_path_check.py
Unescape View File

@ -1,18 +1,13 @@
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from trezor import wire
from ...common.paths import HARDENED from ...common.paths import HARDENED
from ..seed import is_byron_path, is_minting_path, is_multisig_path, is_shelley_path from ..seed import is_byron_path, is_minting_path, is_multisig_path, is_shelley_path
from . import (
INVALID_CERTIFICATE,
INVALID_OUTPUT,
INVALID_WITHDRAWAL,
INVALID_WITNESS_REQUEST,
)
from .paths import ACCOUNT_PATH_INDEX, ACCOUNT_PATH_LENGTH from .paths import ACCOUNT_PATH_INDEX, ACCOUNT_PATH_LENGTH
from .utils import to_account_path from .utils import to_account_path
if TYPE_CHECKING: if TYPE_CHECKING:
from trezor import wire
from trezor.messages import ( from trezor.messages import (
CardanoPoolOwner, CardanoPoolOwner,
CardanoTxCertificate, CardanoTxCertificate,
@ -81,25 +76,27 @@ class AccountPathChecker:
if not output.address_parameters.address_n: if not output.address_parameters.address_n:
return return
self._add(output.address_parameters.address_n, INVALID_OUTPUT) self._add(
output.address_parameters.address_n, wire.ProcessError("Invalid output")
)
def add_certificate(self, certificate: CardanoTxCertificate) -> None: def add_certificate(self, certificate: CardanoTxCertificate) -> None:
if not certificate.path: if not certificate.path:
return return
self._add(certificate.path, INVALID_CERTIFICATE) self._add(certificate.path, wire.ProcessError("Invalid certificate"))
def add_pool_owner(self, pool_owner: CardanoPoolOwner) -> None: def add_pool_owner(self, pool_owner: CardanoPoolOwner) -> None:
if not pool_owner.staking_key_path: if not pool_owner.staking_key_path:
return return
self._add(pool_owner.staking_key_path, INVALID_CERTIFICATE) self._add(pool_owner.staking_key_path, wire.ProcessError("Invalid certificate"))
def add_withdrawal(self, withdrawal: CardanoTxWithdrawal) -> None: def add_withdrawal(self, withdrawal: CardanoTxWithdrawal) -> None:
if not withdrawal.path: if not withdrawal.path:
return return
self._add(withdrawal.path, INVALID_WITHDRAWAL) self._add(withdrawal.path, wire.ProcessError("Invalid withdrawal"))
def add_witness_request(self, witness_request: CardanoTxWitnessRequest) -> None: def add_witness_request(self, witness_request: CardanoTxWitnessRequest) -> None:
self._add(witness_request.path, INVALID_WITNESS_REQUEST) self._add(witness_request.path, wire.ProcessError("Invalid witness request"))

15
core/src/apps/cardano/native_script.py
Unescape View File

@ -1,13 +1,10 @@
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from trezor import wire
from trezor.crypto import hashlib from trezor.crypto import hashlib
from trezor.enums import CardanoNativeScriptType from trezor.enums import CardanoNativeScriptType
from apps.cardano.helpers import ( from apps.cardano.helpers import ADDRESS_KEY_HASH_SIZE, SCRIPT_HASH_SIZE
ADDRESS_KEY_HASH_SIZE,
INVALID_NATIVE_SCRIPT,
SCRIPT_HASH_SIZE,
)
from apps.common import cbor from apps.common import cbor
from .helpers.paths import SCHEMA_MINT from .helpers.paths import SCHEMA_MINT
@ -23,6 +20,8 @@ if TYPE_CHECKING:
def validate_native_script(script: CardanoNativeScript | None) -> None: def validate_native_script(script: CardanoNativeScript | None) -> None:
INVALID_NATIVE_SCRIPT = wire.ProcessError("Invalid native script")
if not script: if not script:
raise INVALID_NATIVE_SCRIPT raise INVALID_NATIVE_SCRIPT
@ -112,7 +111,7 @@ def _validate_native_script_structure(script: CardanoNativeScript) -> None:
} }
if script.type not in fields_to_be_empty or any(fields_to_be_empty[script.type]): if script.type not in fields_to_be_empty or any(fields_to_be_empty[script.type]):
raise INVALID_NATIVE_SCRIPT raise wire.ProcessError("Invalid native script")
def get_native_script_hash(keychain: Keychain, script: CardanoNativeScript) -> bytes: def get_native_script_hash(keychain: Keychain, script: CardanoNativeScript) -> bytes:
@ -131,7 +130,7 @@ def cborize_native_script(
elif script.key_path: elif script.key_path:
script_content = (get_public_key_hash(keychain, script.key_path),) script_content = (get_public_key_hash(keychain, script.key_path),)
else: else:
raise INVALID_NATIVE_SCRIPT raise wire.ProcessError("Invalid native script")
elif script.type == CardanoNativeScriptType.ALL: elif script.type == CardanoNativeScriptType.ALL:
script_content = ( script_content = (
tuple( tuple(
@ -159,6 +158,6 @@ def cborize_native_script(
elif script.type == CardanoNativeScriptType.INVALID_HEREAFTER: elif script.type == CardanoNativeScriptType.INVALID_HEREAFTER:
script_content = (script.invalid_hereafter,) script_content = (script.invalid_hereafter,)
else: else:
raise INVALID_NATIVE_SCRIPT raise RuntimeError # should be unreachable
return (script.type,) + script_content return (script.type,) + script_content

26
core/src/apps/cardano/sign_tx/multisig_signer.py
Unescape View File

@ -9,13 +9,6 @@ from trezor.messages import (
) )
from .. import seed from .. import seed
from ..helpers import (
INVALID_CERTIFICATE,
INVALID_OUTPUT,
INVALID_TX_SIGNING_REQUEST,
INVALID_WITHDRAWAL,
INVALID_WITNESS_REQUEST,
)
from ..helpers.paths import SCHEMA_MINT from ..helpers.paths import SCHEMA_MINT
from ..layout import show_multisig_transaction from ..layout import show_multisig_transaction
from ..seed import is_multisig_path from ..seed import is_multisig_path
@ -38,7 +31,7 @@ class MultisigSigner(Signer):
self.msg.collateral_inputs_count != 0 self.msg.collateral_inputs_count != 0
or self.msg.required_signers_count != 0 or self.msg.required_signers_count != 0
): ):
raise INVALID_TX_SIGNING_REQUEST raise wire.ProcessError("Invalid tx signing request")
async def _show_tx_signing_request(self) -> None: async def _show_tx_signing_request(self) -> None:
await show_multisig_transaction(self.ctx) await show_multisig_transaction(self.ctx)
@ -61,19 +54,19 @@ class MultisigSigner(Signer):
def _validate_output(self, output: messages.CardanoTxOutput) -> None: def _validate_output(self, output: messages.CardanoTxOutput) -> None:
super()._validate_output(output) super()._validate_output(output)
if output.address_parameters is not None: if output.address_parameters is not None:
raise INVALID_OUTPUT raise wire.ProcessError("Invalid output")
def _validate_certificate(self, certificate: CardanoTxCertificate) -> None: def _validate_certificate(self, certificate: CardanoTxCertificate) -> None:
super()._validate_certificate(certificate) super()._validate_certificate(certificate)
if certificate.type == CardanoCertificateType.STAKE_POOL_REGISTRATION: if certificate.type == CardanoCertificateType.STAKE_POOL_REGISTRATION:
raise INVALID_CERTIFICATE raise wire.ProcessError("Invalid certificate")
if certificate.path or certificate.key_hash: if certificate.path or certificate.key_hash:
raise INVALID_CERTIFICATE raise wire.ProcessError("Invalid certificate")
def _validate_withdrawal(self, withdrawal: CardanoTxWithdrawal) -> None: def _validate_withdrawal(self, withdrawal: CardanoTxWithdrawal) -> None:
super()._validate_withdrawal(withdrawal) super()._validate_withdrawal(withdrawal)
if withdrawal.path or withdrawal.key_hash: if withdrawal.path or withdrawal.key_hash:
raise INVALID_WITHDRAWAL raise wire.ProcessError("Invalid withdrawal")
def _validate_witness_request( def _validate_witness_request(
self, witness_request: CardanoTxWitnessRequest self, witness_request: CardanoTxWitnessRequest
@ -82,7 +75,8 @@ class MultisigSigner(Signer):
is_minting = SCHEMA_MINT.match(witness_request.path) is_minting = SCHEMA_MINT.match(witness_request.path)
transaction_has_token_minting = self.msg.minting_asset_groups_count > 0 transaction_has_token_minting = self.msg.minting_asset_groups_count > 0
if not is_multisig_path(witness_request.path) and not is_minting: if not (
raise INVALID_WITNESS_REQUEST is_multisig_path(witness_request.path)
if is_minting and not transaction_has_token_minting: or (is_minting and transaction_has_token_minting)
raise INVALID_WITNESS_REQUEST ):
raise wire.ProcessError("Invalid witness request")

20
core/src/apps/cardano/sign_tx/ordinary_signer.py
Unescape View File

@ -8,12 +8,6 @@ from trezor.messages import (
) )
from .. import seed from .. import seed
from ..helpers import (
INVALID_CERTIFICATE,
INVALID_TX_SIGNING_REQUEST,
INVALID_WITHDRAWAL,
INVALID_WITNESS_REQUEST,
)
from ..helpers.paths import ( from ..helpers.paths import (
SCHEMA_MINT, SCHEMA_MINT,
SCHEMA_PAYMENT, SCHEMA_PAYMENT,
@ -42,7 +36,7 @@ class OrdinarySigner(Signer):
self.msg.collateral_inputs_count != 0 self.msg.collateral_inputs_count != 0
or self.msg.required_signers_count != 0 or self.msg.required_signers_count != 0
): ):
raise INVALID_TX_SIGNING_REQUEST raise wire.ProcessError("Invalid tx signing request")
async def _confirm_tx(self, tx_hash: bytes) -> None: async def _confirm_tx(self, tx_hash: bytes) -> None:
# super() omitted intentionally # super() omitted intentionally
@ -61,14 +55,14 @@ class OrdinarySigner(Signer):
def _validate_certificate(self, certificate: messages.CardanoTxCertificate) -> None: def _validate_certificate(self, certificate: messages.CardanoTxCertificate) -> None:
super()._validate_certificate(certificate) super()._validate_certificate(certificate)
if certificate.type == CardanoCertificateType.STAKE_POOL_REGISTRATION: if certificate.type == CardanoCertificateType.STAKE_POOL_REGISTRATION:
raise INVALID_CERTIFICATE raise wire.ProcessError("Invalid certificate")
if certificate.script_hash or certificate.key_hash: if certificate.script_hash or certificate.key_hash:
raise INVALID_CERTIFICATE raise wire.ProcessError("Invalid certificate")
def _validate_withdrawal(self, withdrawal: CardanoTxWithdrawal) -> None: def _validate_withdrawal(self, withdrawal: CardanoTxWithdrawal) -> None:
super()._validate_withdrawal(withdrawal) super()._validate_withdrawal(withdrawal)
if withdrawal.script_hash or withdrawal.key_hash: if withdrawal.script_hash or withdrawal.key_hash:
raise INVALID_WITHDRAWAL raise wire.ProcessError("Invalid withdrawal")
def _validate_witness_request( def _validate_witness_request(
self, witness_request: CardanoTxWitnessRequest self, witness_request: CardanoTxWitnessRequest
@ -80,11 +74,9 @@ class OrdinarySigner(Signer):
if not ( if not (
is_byron_path(witness_request.path) is_byron_path(witness_request.path)
or is_shelley_path(witness_request.path) or is_shelley_path(witness_request.path)
or is_minting or (is_minting and transaction_has_token_minting)
): ):
raise INVALID_WITNESS_REQUEST raise wire.ProcessError("Invalid witness request")
if is_minting and not transaction_has_token_minting:
raise INVALID_WITNESS_REQUEST
async def _show_witness_request(self, witness_path: list[int]) -> None: async def _show_witness_request(self, witness_path: list[int]) -> None:
# super() omitted intentionally # super() omitted intentionally

5
core/src/apps/cardano/sign_tx/plutus_signer.py
Unescape View File

@ -10,7 +10,6 @@ from trezor.messages import (
) )
from .. import seed from .. import seed
from ..helpers import INVALID_CERTIFICATE, INVALID_WITNESS_REQUEST
from ..helpers.credential import Credential, should_show_address_credentials from ..helpers.credential import Credential, should_show_address_credentials
from ..helpers.paths import SCHEMA_MINT from ..helpers.paths import SCHEMA_MINT
from ..layout import ( from ..layout import (
@ -95,7 +94,7 @@ class PlutusSigner(Signer):
def _validate_certificate(self, certificate: CardanoTxCertificate) -> None: def _validate_certificate(self, certificate: CardanoTxCertificate) -> None:
super()._validate_certificate(certificate) super()._validate_certificate(certificate)
if certificate.type == CardanoCertificateType.STAKE_POOL_REGISTRATION: if certificate.type == CardanoCertificateType.STAKE_POOL_REGISTRATION:
raise INVALID_CERTIFICATE raise wire.ProcessError("Invalid certificate")
def _validate_witness_request( def _validate_witness_request(
self, witness_request: CardanoTxWitnessRequest self, witness_request: CardanoTxWitnessRequest
@ -109,4 +108,4 @@ class PlutusSigner(Signer):
or is_multisig_path(witness_request.path) or is_multisig_path(witness_request.path)
or is_minting or is_minting
): ):
raise INVALID_WITNESS_REQUEST raise wire.ProcessError("Invalid witness request")

25
core/src/apps/cardano/sign_tx/pool_owner_signer.py
Unescape View File

@ -8,13 +8,6 @@ from trezor.messages import (
) )
from .. import seed from .. import seed
from ..helpers import (
INVALID_CERTIFICATE,
INVALID_OUTPUT,
INVALID_STAKE_POOL_REGISTRATION_TX_STRUCTURE,
INVALID_STAKEPOOL_REGISTRATION_TX_WITNESSES,
INVALID_TX_SIGNING_REQUEST,
)
from ..helpers.paths import SCHEMA_STAKING_ANY_ACCOUNT from ..helpers.paths import SCHEMA_STAKING_ANY_ACCOUNT
from ..layout import confirm_stake_pool_registration_final from ..layout import confirm_stake_pool_registration_final
from .signer import Signer from .signer import Signer
@ -45,14 +38,16 @@ class PoolOwnerSigner(Signer):
or self.msg.withdrawals_count != 0 or self.msg.withdrawals_count != 0
or self.msg.minting_asset_groups_count != 0 or self.msg.minting_asset_groups_count != 0
): ):
raise INVALID_STAKE_POOL_REGISTRATION_TX_STRUCTURE raise wire.ProcessError(
"Stakepool registration transaction cannot contain other certificates, withdrawals or minting"
)
if ( if (
self.msg.script_data_hash is not None self.msg.script_data_hash is not None
or self.msg.collateral_inputs_count != 0 or self.msg.collateral_inputs_count != 0
or self.msg.required_signers_count != 0 or self.msg.required_signers_count != 0
): ):
raise INVALID_TX_SIGNING_REQUEST raise wire.ProcessError("Invalid tx signing request")
async def _confirm_transaction(self, tx_hash: bytes) -> None: async def _confirm_transaction(self, tx_hash: bytes) -> None:
# super() omitted intentionally # super() omitted intentionally
@ -65,10 +60,8 @@ class PoolOwnerSigner(Signer):
def _validate_output(self, output: CardanoTxOutput) -> None: def _validate_output(self, output: CardanoTxOutput) -> None:
super()._validate_output(output) super()._validate_output(output)
if output.address_parameters is not None: if output.address_parameters is not None or output.datum_hash is not None:
raise INVALID_OUTPUT raise wire.ProcessError("Invalid output")
if output.datum_hash is not None:
raise INVALID_OUTPUT
def _should_show_output(self, output: CardanoTxOutput) -> bool: def _should_show_output(self, output: CardanoTxOutput) -> bool:
# super() omitted intentionally # super() omitted intentionally
@ -78,14 +71,16 @@ class PoolOwnerSigner(Signer):
def _validate_certificate(self, certificate: CardanoTxCertificate) -> None: def _validate_certificate(self, certificate: CardanoTxCertificate) -> None:
super()._validate_certificate(certificate) super()._validate_certificate(certificate)
if certificate.type != CardanoCertificateType.STAKE_POOL_REGISTRATION: if certificate.type != CardanoCertificateType.STAKE_POOL_REGISTRATION:
raise INVALID_CERTIFICATE raise wire.ProcessError("Invalid certificate")
def _validate_witness_request( def _validate_witness_request(
self, witness_request: CardanoTxWitnessRequest self, witness_request: CardanoTxWitnessRequest
) -> None: ) -> None:
super()._validate_witness_request(witness_request) super()._validate_witness_request(witness_request)
if not SCHEMA_STAKING_ANY_ACCOUNT.match(witness_request.path): if not SCHEMA_STAKING_ANY_ACCOUNT.match(witness_request.path):
raise INVALID_STAKEPOOL_REGISTRATION_TX_WITNESSES raise wire.ProcessError(
"Stakepool registration transaction can only contain staking witnesses"
)
def _is_network_id_verifiable(self) -> bool: def _is_network_id_verifiable(self) -> bool:
# super() omitted intentionally # super() omitted intentionally

53
core/src/apps/cardano/sign_tx/signer.py
Unescape View File

@ -62,16 +62,6 @@ from ..certificates import (
from ..helpers import ( from ..helpers import (
ADDRESS_KEY_HASH_SIZE, ADDRESS_KEY_HASH_SIZE,
INPUT_PREV_HASH_SIZE, INPUT_PREV_HASH_SIZE,
INVALID_COLLATERAL_INPUT,
INVALID_INPUT,
INVALID_OUTPUT,
INVALID_OUTPUT_DATUM_HASH,
INVALID_REQUIRED_SIGNER,
INVALID_SCRIPT_DATA_HASH,
INVALID_TOKEN_BUNDLE_MINT,
INVALID_TOKEN_BUNDLE_OUTPUT,
INVALID_TX_SIGNING_REQUEST,
INVALID_WITHDRAWAL,
LOVELACE_MAX_SUPPLY, LOVELACE_MAX_SUPPLY,
OUTPUT_DATUM_HASH_SIZE, OUTPUT_DATUM_HASH_SIZE,
SCRIPT_DATA_HASH_SIZE, SCRIPT_DATA_HASH_SIZE,
@ -176,7 +166,7 @@ class Signer:
) )
) )
self.tx_dict: HashBuilderDict[int, Any] = HashBuilderDict( self.tx_dict: HashBuilderDict[int, Any] = HashBuilderDict(
tx_body_map_item_count, INVALID_TX_SIGNING_REQUEST tx_body_map_item_count, wire.ProcessError("Invalid tx signing request")
) )
async def sign(self) -> None: async def sign(self) -> None:
@ -222,7 +212,7 @@ class Signer:
if self.msg.withdrawals_count > 0: if self.msg.withdrawals_count > 0:
withdrawals_dict: HashBuilderDict[bytes, int] = HashBuilderDict( withdrawals_dict: HashBuilderDict[bytes, int] = HashBuilderDict(
self.msg.withdrawals_count, INVALID_WITHDRAWAL self.msg.withdrawals_count, wire.ProcessError("Invalid withdrawal")
) )
with self.tx_dict.add(TX_BODY_KEY_WITHDRAWALS, withdrawals_dict): with self.tx_dict.add(TX_BODY_KEY_WITHDRAWALS, withdrawals_dict):
await self._process_withdrawals(withdrawals_dict) await self._process_withdrawals(withdrawals_dict)
@ -237,7 +227,8 @@ class Signer:
if self.msg.minting_asset_groups_count > 0: if self.msg.minting_asset_groups_count > 0:
minting_dict: HashBuilderDict[bytes, HashBuilderDict] = HashBuilderDict( minting_dict: HashBuilderDict[bytes, HashBuilderDict] = HashBuilderDict(
self.msg.minting_asset_groups_count, INVALID_TOKEN_BUNDLE_MINT self.msg.minting_asset_groups_count,
wire.ProcessError("Invalid mint token bundle"),
) )
with self.tx_dict.add(TX_BODY_KEY_MINT, minting_dict): with self.tx_dict.add(TX_BODY_KEY_MINT, minting_dict):
await self._process_minting(minting_dict) await self._process_minting(minting_dict)
@ -292,7 +283,7 @@ class Signer:
def _validate_input(self, input: CardanoTxInput) -> None: def _validate_input(self, input: CardanoTxInput) -> None:
if len(input.prev_hash) != INPUT_PREV_HASH_SIZE: if len(input.prev_hash) != INPUT_PREV_HASH_SIZE:
raise INVALID_INPUT raise wire.ProcessError("Invalid input")
async def _show_input(self, input: CardanoTxInput) -> None: async def _show_input(self, input: CardanoTxInput) -> None:
# We never show the inputs, except for Plutus txs. # We never show the inputs, except for Plutus txs.
@ -326,7 +317,8 @@ class Signer:
asset_groups_dict: HashBuilderDict[ asset_groups_dict: HashBuilderDict[
bytes, HashBuilderDict[bytes, int] bytes, HashBuilderDict[bytes, int]
] = HashBuilderDict( ] = HashBuilderDict(
output.asset_groups_count, INVALID_TOKEN_BUNDLE_OUTPUT output.asset_groups_count,
wire.ProcessError("Invalid token bundle in output"),
) )
with output_value_list.append(asset_groups_dict): with output_value_list.append(asset_groups_dict):
await self._process_asset_groups( await self._process_asset_groups(
@ -344,7 +336,7 @@ class Signer:
def _validate_output(self, output: CardanoTxOutput) -> None: def _validate_output(self, output: CardanoTxOutput) -> None:
if output.address_parameters is not None and output.address is not None: if output.address_parameters is not None and output.address is not None:
raise INVALID_OUTPUT raise wire.ProcessError("Invalid output")
if output.address_parameters is not None: if output.address_parameters is not None:
validate_output_address_parameters(output.address_parameters) validate_output_address_parameters(output.address_parameters)
@ -354,14 +346,14 @@ class Signer:
output.address, self.msg.protocol_magic, self.msg.network_id output.address, self.msg.protocol_magic, self.msg.network_id
) )
else: else:
raise INVALID_OUTPUT raise wire.ProcessError("Invalid output")
if output.datum_hash is not None: if output.datum_hash is not None:
if len(output.datum_hash) != OUTPUT_DATUM_HASH_SIZE: if len(output.datum_hash) != OUTPUT_DATUM_HASH_SIZE:
raise INVALID_OUTPUT_DATUM_HASH raise wire.ProcessError("Invalid output datum hash")
address_type = self._get_output_address_type(output) address_type = self._get_output_address_type(output)
if address_type not in ADDRESS_TYPES_PAYMENT_SCRIPT: if address_type not in ADDRESS_TYPES_PAYMENT_SCRIPT:
raise INVALID_OUTPUT raise wire.ProcessError("Invalid output")
self.account_path_checker.add_output(output) self.account_path_checker.add_output(output)
@ -451,7 +443,8 @@ class Signer:
self._validate_asset_group(asset_group) self._validate_asset_group(asset_group)
tokens: HashBuilderDict[bytes, int] = HashBuilderDict( tokens: HashBuilderDict[bytes, int] = HashBuilderDict(
asset_group.tokens_count, INVALID_TOKEN_BUNDLE_OUTPUT asset_group.tokens_count,
wire.ProcessError("Invalid token bundle in output"),
) )
with asset_groups_dict.add(asset_group.policy_id, tokens): with asset_groups_dict.add(asset_group.policy_id, tokens):
await self._process_tokens( await self._process_tokens(
@ -465,7 +458,9 @@ class Signer:
self, asset_group: CardanoAssetGroup, is_mint: bool = False self, asset_group: CardanoAssetGroup, is_mint: bool = False
) -> None: ) -> None:
INVALID_TOKEN_BUNDLE = ( INVALID_TOKEN_BUNDLE = (
INVALID_TOKEN_BUNDLE_MINT if is_mint else INVALID_TOKEN_BUNDLE_OUTPUT wire.ProcessError("Invalid mint token bundle")
if is_mint
else wire.ProcessError("Invalid token bundle in output")
) )
if len(asset_group.policy_id) != MINTING_POLICY_ID_LENGTH: if len(asset_group.policy_id) != MINTING_POLICY_ID_LENGTH:
@ -493,7 +488,9 @@ class Signer:
def _validate_token(self, token: CardanoToken, is_mint: bool = False) -> None: def _validate_token(self, token: CardanoToken, is_mint: bool = False) -> None:
INVALID_TOKEN_BUNDLE = ( INVALID_TOKEN_BUNDLE = (
INVALID_TOKEN_BUNDLE_MINT if is_mint else INVALID_TOKEN_BUNDLE_OUTPUT wire.ProcessError("Invalid mint token bundle")
if is_mint
else wire.ProcessError("Invalid token bundle in output")
) )
if is_mint: if is_mint:
@ -644,11 +641,11 @@ class Signer:
withdrawal.path, withdrawal.path,
withdrawal.script_hash, withdrawal.script_hash,
withdrawal.key_hash, withdrawal.key_hash,
INVALID_WITHDRAWAL, wire.ProcessError("Invalid withdrawal"),
) )
if not 0 <= withdrawal.amount < LOVELACE_MAX_SUPPLY: if not 0 <= withdrawal.amount < LOVELACE_MAX_SUPPLY:
raise INVALID_WITHDRAWAL raise wire.ProcessError("Invalid withdrawal")
self.account_path_checker.add_withdrawal(withdrawal) self.account_path_checker.add_withdrawal(withdrawal)
@ -696,7 +693,7 @@ class Signer:
self._validate_asset_group(asset_group, is_mint=True) self._validate_asset_group(asset_group, is_mint=True)
tokens: HashBuilderDict[bytes, int] = HashBuilderDict( tokens: HashBuilderDict[bytes, int] = HashBuilderDict(
asset_group.tokens_count, INVALID_TOKEN_BUNDLE_MINT asset_group.tokens_count, wire.ProcessError("Invalid mint token bundle")
) )
with minting_dict.add(asset_group.policy_id, tokens): with minting_dict.add(asset_group.policy_id, tokens):
await self._process_minting_tokens( await self._process_minting_tokens(
@ -732,7 +729,7 @@ class Signer:
def _validate_script_data_hash(self) -> None: def _validate_script_data_hash(self) -> None:
assert self.msg.script_data_hash is not None assert self.msg.script_data_hash is not None
if len(self.msg.script_data_hash) != SCRIPT_DATA_HASH_SIZE: if len(self.msg.script_data_hash) != SCRIPT_DATA_HASH_SIZE:
raise INVALID_SCRIPT_DATA_HASH raise wire.ProcessError("Invalid script data hash")
# collateral inputs # collateral inputs
@ -753,7 +750,7 @@ class Signer:
self, collateral_input: CardanoTxCollateralInput self, collateral_input: CardanoTxCollateralInput
) -> None: ) -> None:
if len(collateral_input.prev_hash) != INPUT_PREV_HASH_SIZE: if len(collateral_input.prev_hash) != INPUT_PREV_HASH_SIZE:
raise INVALID_COLLATERAL_INPUT raise wire.ProcessError("Invalid collateral input")
# required signers # required signers
@ -775,6 +772,8 @@ class Signer:
def _validate_required_signer( def _validate_required_signer(
self, required_signer: CardanoTxRequiredSigner self, required_signer: CardanoTxRequiredSigner
) -> None: ) -> None:
INVALID_REQUIRED_SIGNER = wire.ProcessError("Invalid required signer")
if required_signer.key_hash and required_signer.key_path: if required_signer.key_hash and required_signer.key_path:
raise INVALID_REQUIRED_SIGNER raise INVALID_REQUIRED_SIGNER

Write Preview
Loading…
Cancel
Save