chore(core): decrease stellar size by 1400 bytes

pull/2633/head
grdddj 2 years ago committed by matejcik
parent f25119e44d
commit 9c0c3852f5

@ -59,21 +59,6 @@ op_codes: dict[int, int] = {
MessageType.StellarSetOptionsOp: 5,
}
op_wire_types = [
MessageType.StellarAccountMergeOp,
MessageType.StellarAllowTrustOp,
MessageType.StellarBumpSequenceOp,
MessageType.StellarChangeTrustOp,
MessageType.StellarCreateAccountOp,
MessageType.StellarCreatePassiveSellOfferOp,
MessageType.StellarManageDataOp,
MessageType.StellarManageBuyOfferOp,
MessageType.StellarManageSellOfferOp,
MessageType.StellarPathPaymentStrictReceiveOp,
MessageType.StellarPathPaymentStrictSendOp,
MessageType.StellarPaymentOp,
MessageType.StellarSetOptionsOp,
]
# https://www.stellar.org/developers/guides/concepts/accounts.html#balance
# https://github.com/stellar/go/blob/3d2c1defe73dbfed00146ebe0e8d7e07ce4bb1b6/amount/main.go#L23

@ -1,14 +1,9 @@
from typing import TYPE_CHECKING
from trezor.messages import StellarAddress, StellarGetAddress
from trezor.ui.layouts import show_address
from apps.common import paths, seed
from apps.common.keychain import auto_keychain
from . import helpers
if TYPE_CHECKING:
from trezor.messages import StellarGetAddress, StellarAddress
from trezor.wire import Context
from apps.common.keychain import Keychain
@ -17,6 +12,11 @@ if TYPE_CHECKING:
async def get_address(
ctx: Context, msg: StellarGetAddress, keychain: Keychain
) -> StellarAddress:
from apps.common import paths, seed
from trezor.messages import StellarAddress
from trezor.ui.layouts import show_address
from . import helpers
await paths.validate_path(ctx, keychain, msg.address_n)
node = keychain.derive(msg.address_n)
@ -25,6 +25,6 @@ async def get_address(
if msg.show_display:
title = paths.address_n_to_str(msg.address_n)
await show_address(ctx, address=address, case_sensitive=False, title=title)
await show_address(ctx, address, case_sensitive=False, title=title)
return StellarAddress(address=address)

@ -1,7 +1,4 @@
import ustruct
from trezor.crypto import base32
from trezor.wire import ProcessError
def public_key_from_address(address: str) -> bytes:
@ -9,8 +6,12 @@ def public_key_from_address(address: str) -> bytes:
Stellar address is in format:
<1-byte version> <32-bytes ed25519 public key> <2-bytes CRC-16 checksum>
"""
from trezor.wire import ProcessError
b = base32.decode(address)
_crc16_checksum_verify(b[:-2], b[-2:])
# verify checksum - function deleted as it saved 50 bytes from the binary
if _crc16_checksum(b[:-2]) != b[-2:]:
raise ProcessError("Invalid address checksum")
return b[1:-2]
@ -24,11 +25,6 @@ def address_from_public_key(pubkey: bytes) -> str:
return base32.encode(address)
def _crc16_checksum_verify(data: bytes, checksum: bytes) -> None:
if _crc16_checksum(data) != checksum:
raise ProcessError("Invalid address checksum")
def _crc16_checksum(data: bytes) -> bytes:
"""Returns the CRC-16 checksum of bytearray bytes
@ -36,6 +32,8 @@ def _crc16_checksum(data: bytes) -> bytes:
Initial value changed to 0x0000 to match Stellar configuration.
"""
import ustruct
crc = 0x0000
polynomial = 0x1021

@ -1,20 +1,14 @@
from typing import TYPE_CHECKING
import trezor.ui.layouts as layouts
from trezor import strings, ui
from trezor.enums import ButtonRequestType, StellarAssetType, StellarMemoType
from trezor.ui.layouts import (
confirm_action,
confirm_address,
confirm_blob,
confirm_metadata,
confirm_properties,
)
from trezor.wire import DataError
from trezor.enums import ButtonRequestType
from . import consts
if TYPE_CHECKING:
from trezor.wire import Context
from trezor.enums import StellarMemoType
from trezor.messages import StellarAsset
@ -25,39 +19,43 @@ async def require_confirm_init(
network_passphrase: str,
accounts_match: bool,
) -> None:
if accounts_match:
description = "Initialize signing with your account"
else:
description = "Initialize signing with"
await confirm_address(
description = "Initialize signing with" + " your account" if accounts_match else ""
await layouts.confirm_address(
ctx,
title="Confirm Stellar",
address=address,
br_type="confirm_init",
description=description,
"Confirm Stellar",
address,
description,
"confirm_init",
icon=ui.ICON_SEND,
)
network = get_network_warning(network_passphrase)
# get_network_warning
if network_passphrase == consts.NETWORK_PASSPHRASE_PUBLIC:
network = None
elif network_passphrase == consts.NETWORK_PASSPHRASE_TESTNET:
network = "testnet network"
else:
network = "private network"
if network:
await confirm_metadata(
await layouts.confirm_metadata(
ctx,
"confirm_init_network",
title="Confirm network",
content="Transaction is on {}",
param=network,
"Confirm network",
"Transaction is on {}",
network,
ButtonRequestType.ConfirmOutput,
icon=ui.ICON_CONFIRM,
br_code=ButtonRequestType.ConfirmOutput,
hide_continue=True,
)
async def require_confirm_timebounds(ctx: Context, start: int, end: int) -> None:
await confirm_properties(
await layouts.confirm_properties(
ctx,
"confirm_timebounds",
title="Confirm timebounds",
props=(
"Confirm timebounds",
(
(
"Valid from (UTC)",
strings.format_timestamp(start) if start > 0 else "[no restriction]",
@ -73,6 +71,8 @@ async def require_confirm_timebounds(ctx: Context, start: int, end: int) -> None
async def require_confirm_memo(
ctx: Context, memo_type: StellarMemoType, memo_text: str
) -> None:
from trezor.enums import StellarMemoType
if memo_type == StellarMemoType.TEXT:
description = "Memo (TEXT)"
elif memo_type == StellarMemoType.ID:
@ -82,40 +82,43 @@ async def require_confirm_memo(
elif memo_type == StellarMemoType.RETURN:
description = "Memo (RETURN)"
else:
return await confirm_action(
return await layouts.confirm_action(
ctx,
"confirm_memo",
title="Confirm memo",
action="No memo set!",
description="Important: Many exchanges require a memo when depositing",
"Confirm memo",
"No memo set!",
"Important: Many exchanges require a memo when depositing",
icon=ui.ICON_CONFIRM,
icon_color=ui.GREEN,
br_code=ButtonRequestType.ConfirmOutput,
)
await confirm_blob(
await layouts.confirm_blob(
ctx,
"confirm_memo",
title="Confirm memo",
description=description,
data=memo_text,
"Confirm memo",
memo_text,
description,
)
async def require_confirm_final(ctx: Context, fee: int, num_operations: int) -> None:
op_str = strings.format_plural("{count} {plural}", num_operations, "operation")
await confirm_metadata(
await layouts.confirm_metadata(
ctx,
"confirm_final",
title="Final confirm",
content="Sign this transaction made up of " + op_str + " and pay {}\nfor fee?",
param=format_amount(fee),
"Final confirm",
"Sign this transaction made up of " + op_str + " and pay {}\nfor fee?",
format_amount(fee),
hide_continue=True,
hold=True,
)
def format_asset(asset: StellarAsset | None) -> str:
from trezor.enums import StellarAssetType
from trezor.wire import DataError
if asset is None or asset.type == StellarAssetType.NATIVE:
return "XLM"
else:
@ -130,11 +133,3 @@ def format_amount(amount: int, asset: StellarAsset | None = None) -> str:
+ " "
+ format_asset(asset)
)
def get_network_warning(network_passphrase: str) -> str | None:
if network_passphrase == consts.NETWORK_PASSPHRASE_PUBLIC:
return None
if network_passphrase == consts.NETWORK_PASSPHRASE_TESTNET:
return "testnet network"
return "private network"

@ -1,57 +1,63 @@
from typing import TYPE_CHECKING
from .. import consts, writers
from . import layout, serialize
if TYPE_CHECKING:
from trezor.utils import Writer
from trezor.wire import Context
from consts import StellarMessageType
async def process_operation(ctx: Context, w: Writer, op: StellarMessageType) -> None:
# Importing the stuff inside (only) function saves around 100 bytes here
# (probably because the local lookup is more efficient than a global lookup)
# Saves about 75 bytes here, to have just one import instead of 13
import trezor.messages as messages
from .. import consts, writers
from . import layout, serialize
async def process_operation(
ctx: Context, w: Writer, op: consts.StellarMessageType
) -> None:
if op.source_account:
await layout.confirm_source_account(ctx, op.source_account)
serialize.write_account(w, op.source_account)
writers.write_uint32(w, consts.get_op_code(op))
if serialize.StellarAccountMergeOp.is_type_of(op):
# NOTE: each branch below has 45 bytes (26 the actions, 19 the condition)
if messages.StellarAccountMergeOp.is_type_of(op):
await layout.confirm_account_merge_op(ctx, op)
serialize.write_account_merge_op(w, op)
elif serialize.StellarAllowTrustOp.is_type_of(op):
elif messages.StellarAllowTrustOp.is_type_of(op):
await layout.confirm_allow_trust_op(ctx, op)
serialize.write_allow_trust_op(w, op)
elif serialize.StellarBumpSequenceOp.is_type_of(op):
elif messages.StellarBumpSequenceOp.is_type_of(op):
await layout.confirm_bump_sequence_op(ctx, op)
serialize.write_bump_sequence_op(w, op)
elif serialize.StellarChangeTrustOp.is_type_of(op):
elif messages.StellarChangeTrustOp.is_type_of(op):
await layout.confirm_change_trust_op(ctx, op)
serialize.write_change_trust_op(w, op)
elif serialize.StellarCreateAccountOp.is_type_of(op):
elif messages.StellarCreateAccountOp.is_type_of(op):
await layout.confirm_create_account_op(ctx, op)
serialize.write_create_account_op(w, op)
elif serialize.StellarCreatePassiveSellOfferOp.is_type_of(op):
elif messages.StellarCreatePassiveSellOfferOp.is_type_of(op):
await layout.confirm_create_passive_sell_offer_op(ctx, op)
serialize.write_create_passive_sell_offer_op(w, op)
elif serialize.StellarManageDataOp.is_type_of(op):
elif messages.StellarManageDataOp.is_type_of(op):
await layout.confirm_manage_data_op(ctx, op)
serialize.write_manage_data_op(w, op)
elif serialize.StellarManageBuyOfferOp.is_type_of(op):
elif messages.StellarManageBuyOfferOp.is_type_of(op):
await layout.confirm_manage_buy_offer_op(ctx, op)
serialize.write_manage_buy_offer_op(w, op)
elif serialize.StellarManageSellOfferOp.is_type_of(op):
elif messages.StellarManageSellOfferOp.is_type_of(op):
await layout.confirm_manage_sell_offer_op(ctx, op)
serialize.write_manage_sell_offer_op(w, op)
elif serialize.StellarPathPaymentStrictReceiveOp.is_type_of(op):
elif messages.StellarPathPaymentStrictReceiveOp.is_type_of(op):
await layout.confirm_path_payment_strict_receive_op(ctx, op)
serialize.write_path_payment_strict_receive_op(w, op)
elif serialize.StellarPathPaymentStrictSendOp.is_type_of(op):
elif messages.StellarPathPaymentStrictSendOp.is_type_of(op):
await layout.confirm_path_payment_strict_send_op(ctx, op)
serialize.write_path_payment_strict_send_op(w, op)
elif serialize.StellarPaymentOp.is_type_of(op):
elif messages.StellarPaymentOp.is_type_of(op):
await layout.confirm_payment_op(ctx, op)
serialize.write_payment_op(w, op)
elif serialize.StellarSetOptionsOp.is_type_of(op):
elif messages.StellarSetOptionsOp.is_type_of(op):
await layout.confirm_set_options_op(ctx, op)
serialize.write_set_options_op(w, op)
else:

@ -1,47 +1,44 @@
from typing import TYPE_CHECKING
from trezor.enums import StellarAssetType, StellarSignerType
from trezor.messages import (
StellarAccountMergeOp,
StellarAllowTrustOp,
StellarAsset,
StellarBumpSequenceOp,
StellarChangeTrustOp,
StellarCreateAccountOp,
StellarCreatePassiveSellOfferOp,
StellarManageBuyOfferOp,
StellarManageDataOp,
StellarManageSellOfferOp,
StellarPathPaymentStrictReceiveOp,
StellarPathPaymentStrictSendOp,
StellarPaymentOp,
StellarSetOptionsOp,
)
from trezor.ui.layouts import (
confirm_address,
confirm_amount,
confirm_blob,
confirm_metadata,
confirm_output,
confirm_properties,
confirm_text,
)
from trezor.wire import DataError, ProcessError
from .. import consts, helpers
from ..layout import format_amount, format_asset
from ..layout import format_amount
if TYPE_CHECKING:
from trezor.wire import Context
from trezor.messages import (
StellarAccountMergeOp,
StellarAllowTrustOp,
StellarAsset,
StellarBumpSequenceOp,
StellarChangeTrustOp,
StellarCreateAccountOp,
StellarCreatePassiveSellOfferOp,
StellarManageDataOp,
StellarManageSellOfferOp,
StellarPathPaymentStrictReceiveOp,
StellarPathPaymentStrictSendOp,
StellarPaymentOp,
StellarSetOptionsOp,
StellarManageBuyOfferOp,
)
async def confirm_source_account(ctx: Context, source_account: str) -> None:
await confirm_address(
ctx,
"Confirm operation",
source_account,
description="Source account:",
br_type="op_source_account",
"Source account:",
"op_source_account",
)
@ -49,8 +46,8 @@ async def confirm_allow_trust_op(ctx: Context, op: StellarAllowTrustOp) -> None:
await confirm_properties(
ctx,
"op_allow_trust",
title="Allow trust" if op.is_authorized else "Revoke trust",
props=(
"Allow trust" if op.is_authorized else "Revoke trust",
(
("Asset", op.asset_code),
("Trusted Account", op.trusted_account),
),
@ -62,8 +59,8 @@ async def confirm_account_merge_op(ctx: Context, op: StellarAccountMergeOp) -> N
ctx,
"Account Merge",
op.destination_account,
description="All XLM will be sent to:",
br_type="op_account_merge",
"All XLM will be sent to:",
"op_account_merge",
)
@ -72,18 +69,18 @@ async def confirm_bump_sequence_op(ctx: Context, op: StellarBumpSequenceOp) -> N
ctx,
"op_bump",
"Bump Sequence",
content="Set sequence to {}?",
param=str(op.bump_to),
"Set sequence to {}?",
str(op.bump_to),
)
async def confirm_change_trust_op(ctx: Context, op: StellarChangeTrustOp) -> None:
await confirm_amount(
ctx,
title="Delete trust" if op.limit == 0 else "Add trust",
amount=format_amount(op.limit, op.asset),
description="Limit:",
br_type="op_change_trust",
"Delete trust" if op.limit == 0 else "Add trust",
format_amount(op.limit, op.asset),
"Limit:",
"op_change_trust",
)
await confirm_asset_issuer(ctx, op.asset)
@ -93,7 +90,7 @@ async def confirm_create_account_op(ctx: Context, op: StellarCreateAccountOp) ->
ctx,
"op_create_account",
"Create Account",
props=(
(
("Account", op.new_account),
("Initial Balance", format_amount(op.starting_balance)),
),
@ -103,10 +100,7 @@ async def confirm_create_account_op(ctx: Context, op: StellarCreateAccountOp) ->
async def confirm_create_passive_sell_offer_op(
ctx: Context, op: StellarCreatePassiveSellOfferOp
) -> None:
if op.amount == 0:
text = "Delete Passive Offer"
else:
text = "New Passive Offer"
text = "Delete Passive Offer" if op.amount == 0 else "New Passive Offer"
await _confirm_offer(ctx, text, op)
@ -128,11 +122,7 @@ async def _confirm_manage_offer_op_common(
if op.offer_id == 0:
text = "New Offer"
else:
if op.amount == 0:
text = "Delete"
else:
text = "Update"
text += f" #{op.offer_id}"
text = f"{'Delete' if op.amount == 0 else 'Update'} #{op.offer_id}"
await _confirm_offer(ctx, text, op)
@ -143,35 +133,41 @@ async def _confirm_offer(
| StellarManageSellOfferOp
| StellarManageBuyOfferOp,
) -> None:
from trezor.messages import StellarManageBuyOfferOp
from ..layout import format_asset
buying_asset = op.buying_asset # local_cache_attribute
selling_asset = op.selling_asset # local_cache_attribute
if StellarManageBuyOfferOp.is_type_of(op):
buying = ("Buying:", format_amount(op.amount, op.buying_asset))
selling = ("Selling:", format_asset(op.selling_asset))
buying = ("Buying:", format_amount(op.amount, buying_asset))
selling = ("Selling:", format_asset(selling_asset))
price = (
f"Price per {format_asset(op.selling_asset)}:",
f"Price per {format_asset(selling_asset)}:",
str(op.price_n / op.price_d),
)
await confirm_properties(
ctx,
"op_offer",
title=title,
props=(buying, selling, price),
title,
(buying, selling, price),
)
else:
selling = ("Selling:", format_amount(op.amount, op.selling_asset))
buying = ("Buying:", format_asset(op.buying_asset))
selling = ("Selling:", format_amount(op.amount, selling_asset))
buying = ("Buying:", format_asset(buying_asset))
price = (
f"Price per {format_asset(op.buying_asset)}:",
f"Price per {format_asset(buying_asset)}:",
str(op.price_n / op.price_d),
)
await confirm_properties(
ctx,
"op_offer",
title=title,
props=(selling, buying, price),
title,
(selling, buying, price),
)
await confirm_asset_issuer(ctx, op.selling_asset)
await confirm_asset_issuer(ctx, op.buying_asset)
await confirm_asset_issuer(ctx, selling_asset)
await confirm_asset_issuer(ctx, buying_asset)
async def confirm_manage_data_op(ctx: Context, op: StellarManageDataOp) -> None:
@ -183,7 +179,7 @@ async def confirm_manage_data_op(ctx: Context, op: StellarManageDataOp) -> None:
ctx,
"op_data",
"Set data",
props=(("Key:", op.key), ("Value (SHA-256):", digest)),
(("Key:", op.key), ("Value (SHA-256):", digest)),
)
else:
await confirm_metadata(
@ -191,7 +187,7 @@ async def confirm_manage_data_op(ctx: Context, op: StellarManageDataOp) -> None:
"op_data",
"Clear data",
"Do you want to clear value key {}?",
param=op.key,
op.key,
)
@ -200,18 +196,18 @@ async def confirm_path_payment_strict_receive_op(
) -> None:
await confirm_output(
ctx,
address=op.destination_account,
amount=format_amount(op.destination_amount, op.destination_asset),
op.destination_account,
format_amount(op.destination_amount, op.destination_asset),
title="Path Pay",
)
await confirm_asset_issuer(ctx, op.destination_asset)
# confirm what the sender is using to pay
await confirm_amount(
ctx,
title="Debited amount",
amount=format_amount(op.send_max, op.send_asset),
description="Pay at most:",
br_type="op_path_payment_strict_receive",
"Debited amount",
format_amount(op.send_max, op.send_asset),
"Pay at most:",
"op_path_payment_strict_receive",
)
await confirm_asset_issuer(ctx, op.send_asset)
@ -221,18 +217,18 @@ async def confirm_path_payment_strict_send_op(
) -> None:
await confirm_output(
ctx,
address=op.destination_account,
amount=format_amount(op.destination_min, op.destination_asset),
op.destination_account,
format_amount(op.destination_min, op.destination_asset),
title="Path Pay at least",
)
await confirm_asset_issuer(ctx, op.destination_asset)
# confirm what the sender is using to pay
await confirm_amount(
ctx,
title="Debited amount",
amount=format_amount(op.send_amount, op.send_asset),
description="Pay:",
br_type="op_path_payment_strict_send",
"Debited amount",
format_amount(op.send_amount, op.send_asset),
"Pay:",
"op_path_payment_strict_send",
)
await confirm_asset_issuer(ctx, op.send_asset)
@ -240,20 +236,24 @@ async def confirm_path_payment_strict_send_op(
async def confirm_payment_op(ctx: Context, op: StellarPaymentOp) -> None:
await confirm_output(
ctx,
address=op.destination_account,
amount=format_amount(op.amount, op.asset),
op.destination_account,
format_amount(op.amount, op.asset),
)
await confirm_asset_issuer(ctx, op.asset)
async def confirm_set_options_op(ctx: Context, op: StellarSetOptionsOp) -> None:
from trezor.enums import StellarSignerType
from trezor.ui.layouts import confirm_blob, confirm_text
from .. import helpers
if op.inflation_destination_account:
await confirm_address(
ctx,
"Inflation",
op.inflation_destination_account,
description="Destination:",
br_type="op_inflation",
"Destination:",
"op_inflation",
)
if op.clear_flags:
@ -264,17 +264,27 @@ async def confirm_set_options_op(ctx: Context, op: StellarSetOptionsOp) -> None:
t = _format_flags(op.set_flags)
await confirm_text(ctx, "op_set_options", "Set flags", data=t)
thresholds = _format_thresholds(op)
thresholds: list[tuple[str, str]] = []
append = thresholds.append # local_cache_attribute
if op.master_weight is not None:
append(("Master Weight:", str(op.master_weight)))
if op.low_threshold is not None:
append(("Low:", str(op.low_threshold)))
if op.medium_threshold is not None:
append(("Medium:", str(op.medium_threshold)))
if op.high_threshold is not None:
append(("High:", str(op.high_threshold)))
if thresholds:
await confirm_properties(
ctx, "op_thresholds", "Account Thresholds", props=thresholds
)
await confirm_properties(ctx, "op_thresholds", "Account Thresholds", thresholds)
if op.home_domain:
await confirm_text(ctx, "op_home_domain", "Home Domain", op.home_domain)
signer_type = op.signer_type # local_cache_attribute
signer_key = op.signer_key # local_cache_attribute
if op.signer_type is not None:
if op.signer_key is None or op.signer_weight is None:
if signer_type is not None:
if signer_key is None or op.signer_weight is None:
raise DataError("Stellar: invalid signer option data.")
if op.signer_weight > 0:
@ -282,15 +292,15 @@ async def confirm_set_options_op(ctx: Context, op: StellarSetOptionsOp) -> None:
else:
title = "Remove Signer"
data: str | bytes = ""
if op.signer_type == StellarSignerType.ACCOUNT:
if signer_type == StellarSignerType.ACCOUNT:
description = "Account:"
data = helpers.address_from_public_key(op.signer_key)
elif op.signer_type == StellarSignerType.PRE_AUTH:
data = helpers.address_from_public_key(signer_key)
elif signer_type == StellarSignerType.PRE_AUTH:
description = "Pre-auth transaction:"
data = op.signer_key
elif op.signer_type == StellarSignerType.HASH:
data = signer_key
elif signer_type == StellarSignerType.HASH:
description = "Hash:"
data = op.signer_key
data = signer_key
else:
raise ProcessError("Stellar: invalid signer type")
@ -303,20 +313,9 @@ async def confirm_set_options_op(ctx: Context, op: StellarSetOptionsOp) -> None:
)
def _format_thresholds(op: StellarSetOptionsOp) -> list[tuple[str, str]]:
props = []
if op.master_weight is not None:
props.append(("Master Weight:", str(op.master_weight)))
if op.low_threshold is not None:
props.append(("Low:", str(op.low_threshold)))
if op.medium_threshold is not None:
props.append(("Medium:", str(op.medium_threshold)))
if op.high_threshold is not None:
props.append(("High:", str(op.high_threshold)))
return props
def _format_flags(flags: int) -> str:
from .. import consts
if flags > consts.FLAGS_MAX_SIZE:
raise ProcessError("Stellar: invalid flags")
flags_set = []
@ -330,6 +329,8 @@ def _format_flags(flags: int) -> str:
async def confirm_asset_issuer(ctx: Context, asset: StellarAsset) -> None:
from trezor.enums import StellarAssetType
if asset.type == StellarAssetType.NATIVE:
return
if asset.issuer is None or asset.code is None:
@ -338,6 +339,6 @@ async def confirm_asset_issuer(ctx: Context, asset: StellarAsset) -> None:
ctx,
"Confirm Issuer",
asset.issuer,
description=f"{asset.code} issuer:",
br_type="confirm_asset_issuer",
f"{asset.code} issuer:",
"confirm_asset_issuer",
)

@ -1,55 +1,63 @@
from typing import TYPE_CHECKING
from trezor.enums import StellarAssetType
from trezor.messages import (
StellarAccountMergeOp,
StellarAllowTrustOp,
StellarAsset,
StellarBumpSequenceOp,
StellarChangeTrustOp,
StellarCreateAccountOp,
StellarCreatePassiveSellOfferOp,
StellarManageBuyOfferOp,
StellarManageDataOp,
StellarManageSellOfferOp,
StellarPathPaymentStrictReceiveOp,
StellarPathPaymentStrictSendOp,
StellarPaymentOp,
StellarSetOptionsOp,
)
from trezor.wire import DataError, ProcessError
from .. import writers
from ..writers import (
write_bool,
write_bytes_fixed,
write_pubkey,
write_string,
write_uint32,
write_uint64,
)
if TYPE_CHECKING:
from trezor.utils import Writer
from trezor.messages import (
StellarAccountMergeOp,
StellarAllowTrustOp,
StellarAsset,
StellarBumpSequenceOp,
StellarChangeTrustOp,
StellarCreateAccountOp,
StellarCreatePassiveSellOfferOp,
StellarManageBuyOfferOp,
StellarManageDataOp,
StellarManageSellOfferOp,
StellarPathPaymentStrictReceiveOp,
StellarPathPaymentStrictSendOp,
StellarPaymentOp,
StellarSetOptionsOp,
)
def write_account_merge_op(w: Writer, msg: StellarAccountMergeOp) -> None:
writers.write_pubkey(w, msg.destination_account)
write_pubkey(w, msg.destination_account)
def write_allow_trust_op(w: Writer, msg: StellarAllowTrustOp) -> None:
# trustor account (the account being allowed to access the asset)
writers.write_pubkey(w, msg.trusted_account)
writers.write_uint32(w, msg.asset_type)
write_pubkey(w, msg.trusted_account)
write_uint32(w, msg.asset_type)
_write_asset_code(w, msg.asset_type, msg.asset_code)
writers.write_bool(w, msg.is_authorized)
write_bool(w, msg.is_authorized)
def write_bump_sequence_op(w: Writer, msg: StellarBumpSequenceOp) -> None:
writers.write_uint64(w, msg.bump_to)
write_uint64(w, msg.bump_to)
def write_change_trust_op(w: Writer, msg: StellarChangeTrustOp) -> None:
_write_asset(w, msg.asset)
writers.write_uint64(w, msg.limit)
write_uint64(w, msg.limit)
def write_create_account_op(w: Writer, msg: StellarCreateAccountOp) -> None:
writers.write_pubkey(w, msg.new_account)
writers.write_uint64(w, msg.starting_balance)
write_pubkey(w, msg.new_account)
write_uint64(w, msg.starting_balance)
def write_create_passive_sell_offer_op(
@ -57,18 +65,18 @@ def write_create_passive_sell_offer_op(
) -> None:
_write_asset(w, msg.selling_asset)
_write_asset(w, msg.buying_asset)
writers.write_uint64(w, msg.amount)
writers.write_uint32(w, msg.price_n)
writers.write_uint32(w, msg.price_d)
write_uint64(w, msg.amount)
write_uint32(w, msg.price_n)
write_uint32(w, msg.price_d)
def write_manage_data_op(w: Writer, msg: StellarManageDataOp) -> None:
if len(msg.key) > 64:
raise ProcessError("Stellar: max length of a key is 64 bytes")
writers.write_string(w, msg.key)
writers.write_bool(w, bool(msg.value))
write_string(w, msg.key)
write_bool(w, bool(msg.value))
if msg.value:
writers.write_string(w, msg.value)
write_string(w, msg.value)
def write_manage_buy_offer_op(w: Writer, msg: StellarManageBuyOfferOp) -> None:
@ -84,22 +92,22 @@ def _write_manage_offer_op_common(
) -> None:
_write_asset(w, msg.selling_asset)
_write_asset(w, msg.buying_asset)
writers.write_uint64(w, msg.amount) # amount to sell / buy
writers.write_uint32(w, msg.price_n) # numerator
writers.write_uint32(w, msg.price_d) # denominator
writers.write_uint64(w, msg.offer_id)
write_uint64(w, msg.amount) # amount to sell / buy
write_uint32(w, msg.price_n) # numerator
write_uint32(w, msg.price_d) # denominator
write_uint64(w, msg.offer_id)
def write_path_payment_strict_receive_op(
w: Writer, msg: StellarPathPaymentStrictReceiveOp
) -> None:
_write_asset(w, msg.send_asset)
writers.write_uint64(w, msg.send_max)
writers.write_pubkey(w, msg.destination_account)
write_uint64(w, msg.send_max)
write_pubkey(w, msg.destination_account)
_write_asset(w, msg.destination_asset)
writers.write_uint64(w, msg.destination_amount)
writers.write_uint32(w, len(msg.paths))
write_uint64(w, msg.destination_amount)
write_uint32(w, len(msg.paths))
for p in msg.paths:
_write_asset(w, p)
@ -108,77 +116,77 @@ def write_path_payment_strict_send_op(
w: Writer, msg: StellarPathPaymentStrictSendOp
) -> None:
_write_asset(w, msg.send_asset)
writers.write_uint64(w, msg.send_amount)
writers.write_pubkey(w, msg.destination_account)
write_uint64(w, msg.send_amount)
write_pubkey(w, msg.destination_account)
_write_asset(w, msg.destination_asset)
writers.write_uint64(w, msg.destination_min)
writers.write_uint32(w, len(msg.paths))
write_uint64(w, msg.destination_min)
write_uint32(w, len(msg.paths))
for p in msg.paths:
_write_asset(w, p)
def write_payment_op(w: Writer, msg: StellarPaymentOp) -> None:
writers.write_pubkey(w, msg.destination_account)
write_pubkey(w, msg.destination_account)
_write_asset(w, msg.asset)
writers.write_uint64(w, msg.amount)
write_uint64(w, msg.amount)
def write_set_options_op(w: Writer, msg: StellarSetOptionsOp) -> None:
# inflation destination
if msg.inflation_destination_account is None:
writers.write_bool(w, False)
write_bool(w, False)
else:
writers.write_bool(w, True)
writers.write_pubkey(w, msg.inflation_destination_account)
# clear flags
_write_set_options_int(w, msg.clear_flags)
# set flags
_write_set_options_int(w, msg.set_flags)
# account thresholds
_write_set_options_int(w, msg.master_weight)
_write_set_options_int(w, msg.low_threshold)
_write_set_options_int(w, msg.medium_threshold)
_write_set_options_int(w, msg.high_threshold)
write_bool(w, True)
write_pubkey(w, msg.inflation_destination_account)
# NOTE: saves 21 bytes compared to hardcoding the operations
for option in (
# clear flags
msg.clear_flags,
# set flags
msg.set_flags,
# account thresholds
msg.master_weight,
msg.low_threshold,
msg.medium_threshold,
msg.high_threshold,
):
if option is None:
write_bool(w, False)
else:
write_bool(w, True)
write_uint32(w, option)
# home domain
if msg.home_domain is None:
writers.write_bool(w, False)
write_bool(w, False)
else:
writers.write_bool(w, True)
write_bool(w, True)
if len(msg.home_domain) > 32:
raise ProcessError("Stellar: max length of a home domain is 32 bytes")
writers.write_string(w, msg.home_domain)
write_string(w, msg.home_domain)
# signer
if msg.signer_type is None:
writers.write_bool(w, False)
write_bool(w, False)
else:
if msg.signer_key is None or msg.signer_weight is None:
raise DataError(
"Stellar: signer_type, signer_key, signer_weight must be set together"
)
writers.write_bool(w, True)
writers.write_uint32(w, msg.signer_type)
writers.write_bytes_fixed(w, msg.signer_key, 32)
writers.write_uint32(w, msg.signer_weight)
def _write_set_options_int(w: Writer, value: int | None) -> None:
if value is None:
writers.write_bool(w, False)
else:
writers.write_bool(w, True)
writers.write_uint32(w, value)
write_bool(w, True)
write_uint32(w, msg.signer_type)
write_bytes_fixed(w, msg.signer_key, 32)
write_uint32(w, msg.signer_weight)
def write_account(w: Writer, source_account: str | None) -> None:
if source_account is None:
writers.write_bool(w, False)
write_bool(w, False)
else:
writers.write_bool(w, True)
writers.write_pubkey(w, source_account)
write_bool(w, True)
write_pubkey(w, source_account)
def _write_asset_code(
@ -195,22 +203,22 @@ def _write_asset_code(
if len(code) > 4:
raise DataError("Stellar: asset code too long for ALPHANUM4")
# pad with zeros to 4 chars
writers.write_bytes_fixed(w, code + bytes([0] * (4 - len(code))), 4)
write_bytes_fixed(w, code + bytes([0] * (4 - len(code))), 4)
elif asset_type == StellarAssetType.ALPHANUM12:
if len(code) > 12:
raise DataError("Stellar: asset code too long for ALPHANUM12")
# pad with zeros to 12 chars
writers.write_bytes_fixed(w, code + bytes([0] * (12 - len(code))), 12)
write_bytes_fixed(w, code + bytes([0] * (12 - len(code))), 12)
else:
raise ProcessError("Stellar: invalid asset type")
def _write_asset(w: Writer, asset: StellarAsset) -> None:
if asset.type == StellarAssetType.NATIVE:
writers.write_uint32(w, 0)
write_uint32(w, 0)
return
if asset.code is None or asset.issuer is None:
raise DataError("Stellar: invalid asset")
writers.write_uint32(w, asset.type)
write_uint32(w, asset.type)
_write_asset_code(w, asset.type, asset.code)
writers.write_pubkey(w, asset.issuer)
write_pubkey(w, asset.issuer)

@ -1,21 +1,10 @@
from typing import TYPE_CHECKING
from ubinascii import hexlify
from trezor.crypto.curve import ed25519
from trezor.crypto.hashlib import sha256
from trezor.enums import StellarMemoType
from trezor.messages import StellarSignedTx, StellarSignTx, StellarTxOpRequest
from trezor.wire import DataError, ProcessError
from apps.common import paths, seed
from apps.common.keychain import auto_keychain
from . import consts, helpers, layout, writers
from .operations import process_operation
if TYPE_CHECKING:
from trezor.messages import StellarSignTx, StellarSignedTx
from trezor.wire import Context
from trezor.utils import Writer
from apps.common.keychain import Keychain
@ -24,37 +13,32 @@ if TYPE_CHECKING:
async def sign_tx(
ctx: Context, msg: StellarSignTx, keychain: Keychain
) -> StellarSignedTx:
from ubinascii import hexlify
from trezor.messages import StellarSignedTx
from trezor.crypto.curve import ed25519
from apps.common import paths, seed
from trezor.enums import StellarMemoType
from trezor.crypto.hashlib import sha256
from trezor.wire import DataError, ProcessError
from trezor.messages import StellarTxOpRequest
from .operations import process_operation
from . import helpers
from . import consts, layout, writers
await paths.validate_path(ctx, keychain, msg.address_n)
node = keychain.derive(msg.address_n)
pubkey = seed.remove_ed25519_prefix(node.public_key())
num_operations = msg.num_operations # local_cache_attribute
if msg.num_operations == 0:
if num_operations == 0:
raise ProcessError("Stellar: At least one operation is required")
w = bytearray()
await _init(ctx, w, pubkey, msg)
await _timebounds(ctx, w, msg.timebounds_start, msg.timebounds_end)
await _memo(ctx, w, msg)
await _operations(ctx, w, msg.num_operations)
await _final(ctx, w, msg)
# sign
digest = sha256(w).digest()
signature = ed25519.sign(node.private_key(), digest)
# Add the public key for verification that the right account was used for signing
return StellarSignedTx(public_key=pubkey, signature=signature)
async def _final(ctx: Context, w: Writer, msg: StellarSignTx) -> None:
# 4 null bytes representing a (currently unused) empty union
writers.write_uint32(w, 0)
# final confirm
await layout.require_confirm_final(ctx, msg.fee, msg.num_operations)
async def _init(ctx: Context, w: Writer, pubkey: bytes, msg: StellarSignTx) -> None:
# ---------------------------------
# INIT
# ---------------------------------
network_passphrase_hash = sha256(msg.network_passphrase.encode()).digest()
writers.write_bytes_fixed(w, network_passphrase_hash, 32)
writers.write_bytes_fixed(w, consts.TX_TYPE, 4)
@ -71,44 +55,40 @@ async def _init(ctx: Context, w: Writer, pubkey: bytes, msg: StellarSignTx) -> N
ctx, msg.source_account, msg.network_passphrase, accounts_match
)
async def _timebounds(ctx: Context, w: Writer, start: int, end: int) -> None:
# ---------------------------------
# TIMEBOUNDS
# ---------------------------------
# confirm dialog
await layout.require_confirm_timebounds(ctx, start, end)
await layout.require_confirm_timebounds(
ctx, msg.timebounds_start, msg.timebounds_end
)
# timebounds are sent as uint32s since that's all we can display, but they must be hashed as 64bit
writers.write_bool(w, True)
writers.write_uint64(w, start)
writers.write_uint64(w, end)
writers.write_uint64(w, msg.timebounds_start)
writers.write_uint64(w, msg.timebounds_end)
memo_type = msg.memo_type # local_cache_attribute
memo_text = msg.memo_text # local_cache_attribute
async def _operations(ctx: Context, w: Writer, num_operations: int) -> None:
writers.write_uint32(w, num_operations)
for _ in range(num_operations):
op = await ctx.call_any(StellarTxOpRequest(), *consts.op_wire_types)
await process_operation(ctx, w, op) # type: ignore [Argument of type "MessageType" cannot be assigned to parameter "op" of type "StellarMessageType" in function "process_operation"]
async def _memo(ctx: Context, w: Writer, msg: StellarSignTx) -> None:
writers.write_uint32(w, msg.memo_type)
if msg.memo_type == StellarMemoType.NONE:
writers.write_uint32(w, memo_type)
if memo_type == StellarMemoType.NONE:
# nothing is serialized
memo_confirm_text = ""
elif msg.memo_type == StellarMemoType.TEXT:
elif memo_type == StellarMemoType.TEXT:
# Text: 4 bytes (size) + up to 28 bytes
if msg.memo_text is None:
if memo_text is None:
raise DataError("Stellar: Missing memo text")
if len(msg.memo_text) > 28:
if len(memo_text) > 28:
raise ProcessError("Stellar: max length of a memo text is 28 bytes")
writers.write_string(w, msg.memo_text)
memo_confirm_text = msg.memo_text
elif msg.memo_type == StellarMemoType.ID:
writers.write_string(w, memo_text)
memo_confirm_text = memo_text
elif memo_type == StellarMemoType.ID:
# ID: 64 bit unsigned integer
if msg.memo_id is None:
raise DataError("Stellar: Missing memo id")
writers.write_uint64(w, msg.memo_id)
memo_confirm_text = str(msg.memo_id)
elif msg.memo_type in (StellarMemoType.HASH, StellarMemoType.RETURN):
elif memo_type in (StellarMemoType.HASH, StellarMemoType.RETURN):
# Hash/Return: 32 byte hash
if msg.memo_hash is None:
raise DataError("Stellar: Missing memo hash")
@ -116,4 +96,27 @@ async def _memo(ctx: Context, w: Writer, msg: StellarSignTx) -> None:
memo_confirm_text = hexlify(msg.memo_hash).decode()
else:
raise ProcessError("Stellar invalid memo type")
await layout.require_confirm_memo(ctx, msg.memo_type, memo_confirm_text)
await layout.require_confirm_memo(ctx, memo_type, memo_confirm_text)
# ---------------------------------
# OPERATION
# ---------------------------------
writers.write_uint32(w, num_operations)
for _ in range(num_operations):
op = await ctx.call_any(StellarTxOpRequest(), *consts.op_codes.keys())
await process_operation(ctx, w, op) # type: ignore [Argument of type "MessageType" cannot be assigned to parameter "op" of type "StellarMessageType" in function "process_operation"]
# ---------------------------------
# FINAL
# ---------------------------------
# 4 null bytes representing a (currently unused) empty union
writers.write_uint32(w, 0)
# final confirm
await layout.require_confirm_final(ctx, msg.fee, num_operations)
# sign
digest = sha256(w).digest()
signature = ed25519.sign(node.private_key(), digest)
# Add the public key for verification that the right account was used for signing
return StellarSignedTx(public_key=pubkey, signature=signature)

@ -1,16 +1,11 @@
from typing import TYPE_CHECKING
from apps.common.writers import (
write_bytes_fixed,
write_bytes_unchecked,
write_uint32_be,
write_uint64_be,
)
import apps.common.writers as writers
from .helpers import public_key_from_address
write_uint32 = write_uint32_be
write_uint64 = write_uint64_be
# Reexporting to other modules
write_bytes_fixed = writers.write_bytes_fixed
write_uint32 = writers.write_uint32_be
write_uint64 = writers.write_uint64_be
if TYPE_CHECKING:
from typing import AnyStr
@ -20,26 +15,24 @@ if TYPE_CHECKING:
def write_string(w: Writer, s: AnyStr) -> None:
"""Write XDR string padded to a multiple of 4 bytes."""
if isinstance(s, str):
buf = s.encode()
else:
buf = s
# NOTE: 2 bytes smaller than if-else
buf = s.encode() if isinstance(s, str) else s
write_uint32(w, len(buf))
write_bytes_unchecked(w, buf)
writers.write_bytes_unchecked(w, buf)
# if len isn't a multiple of 4, add padding bytes
remainder = len(buf) % 4
if remainder:
write_bytes_unchecked(w, bytes([0] * (4 - remainder)))
writers.write_bytes_unchecked(w, bytes([0] * (4 - remainder)))
def write_bool(w: Writer, val: bool) -> None:
if val:
write_uint32(w, 1)
else:
write_uint32(w, 0)
# NOTE: 10 bytes smaller than if-else
write_uint32(w, 1 if val else 0)
def write_pubkey(w: Writer, address: str) -> None:
from .helpers import public_key_from_address
# first 4 bytes of an address are the type, there's only one type (0)
write_uint32(w, 0)
write_bytes_fixed(w, public_key_from_address(address), 32)
writers.write_bytes_fixed(w, public_key_from_address(address), 32)

Loading…
Cancel
Save